path: root/python/position.py

from serenitas.analytics.bbg_helpers import init_bbg_session, retrieve_data, BBG_IP
from itertools import product
import datetime
import numpy as np
import pandas as pd
from pandas import bdate_range
import re
import os
import logging


def get_list(
    engine,
    workdate: datetime.date = None,
    asset_class=None,
    include_unsettled: bool = True,
    fund="SERCGMAST",
):
    if workdate:
        positions = pd.read_sql_query(
            "SELECT identifier, bbg_type FROM list_positions(%s, %s, %s, %s)",
            engine,
            params=(workdate, asset_class, include_unsettled, fund),
        )
    else:
        positions = pd.read_sql_table("securities", engine)
        positions.identifier = positions.figi
    positions["bbg_id"] = positions.identifier + " " + positions.bbg_type
    positions.set_index("bbg_id", inplace=True)
    return positions


def get_list_range(engine, begin, end, asset_class=None):
    begin = pd.Timestamp(begin).date()
    end = pd.Timestamp(end).date()
    positions = pd.read_sql_query(
        "select identifier, bbg_type, strategy from list_positions_range(%s, %s, %s)",
        engine,
        params=(begin, end, asset_class),
    )
    positions.loc[
        positions.identifier.str.len() <= 11, "cusip"
    ] = positions.identifier.str.slice(stop=9)
    positions.loc[positions.identifier.str.len() == 12, "isin"] = positions.identifier
    positions["bbg_id"] = (
        positions.cusip.where(positions.cusip.notnull(), positions["isin"])
        + " "
        + positions.bbg_type
    )
    positions.set_index("bbg_id", inplace=True)
    return positions


def backpopulate_marks(begin_str="2015-01-15", end_str="2015-07-15"):
    pattern = re.compile(r"\d{4}-\d{2}-\d{2}")
    list_of_daily_folder = (
        fullpath
        for (fullpath, _, _) in os.walk("/home/serenitas/Daily")
        if pattern.match(os.path.basename(fullpath))
    )
    list_of_bdays = bdate_range(start=begin_str, end=end_str)
    for path in list_of_daily_folder:
        date = pd.to_datetime(os.path.basename(path))
        if date in list_of_bdays:
            marks_file = [f for f in os.listdir(path) if f.startswith("securitiesNpv")]
            if marks_file:
                marks_file.sort(
                    key=lambda x: x[13:], reverse=True
                )  # sort by lexicographic order which is what we want since we use ISO dates
                marks = pd.read_csv(os.path.join(path, marks_file[0]))
                positions = get_list(pd.to_datetime(date))
                positions = positions.merge(
                    marks, left_on="identifier", right_on="IDENTIFIER"
                )
                positions.drop(["IDENTIFIER", "last_settle_date"], axis=1, inplace=True)
                positions["date"] = date
                positions.rename(columns={"Price": "price"}, inplace=True)
                positions = positions.drop_duplicates()
                positions.to_sql("position", engine, if_exists="append", index=False)


def update_securities(engine, session, workdate):
    field = {"Corp": "PREV_CPN_DT", "Mtge": "START_ACC_DT"}
    securities = get_list(engine)
    securities = securities[securities.paid_down.isnull()]
    data = retrieve_data(
        session,
        securities.index.tolist(),
        ["PREV_CPN_DT", "START_ACC_DT", "CUR_CPN", "CPN_ASOF_DT", "MTG_FACE_AMT"],
    )
    data = pd.DataFrame.from_dict(data, orient="index")
    data = data[
        data.CPN_ASOF_DT.isnull() | (data.CPN_ASOF_DT <= pd.Timestamp(workdate))
    ]

    m = securities.merge(data, left_index=True, right_index=True)
    conn = engine.raw_connection()
    with conn.cursor() as c:
        for r in m.to_dict("records"):
            accrued_field = field[r["bbg_type"]]
            if r[accrued_field].date() < workdate:
                c.execute(
                    f"UPDATE securities SET start_accrued_date=%({accrued_field})s "
                    ",coupon=%(CUR_CPN)s, face_amount=%(MTG_FACE_AMT)s "
                    "WHERE identifier=%(identifier)s",
                    r,
                )
    conn.commit()
    conn.close()


def init_fx(session, engine, startdate):
    currencies = ["EURUSD", "CADUSD"]
    securities = [c + " Curncy" for c in currencies]
    data = retrieve_data(session, securities, ["PX_LAST"], start_date=startdate)
    data = data["EURUSD Curncy"].merge(
        data["CADUSD Curncy"], left_on="date", right_on="date"
    )
    data.rename(columns={"PX_LAST_x": "eurusd", "PX_LAST_y": "cadusd"}, inplace=True)
    data.to_sql("fx", engine, if_exists="append")


def update_fx(conn, session, currencies):
    securities = [c + " Curncy" for c in currencies]
    data = retrieve_data(session, securities, ["FIXED_CLOSING_PRICE_NY", "PX_CLOSE_DT"])
    colnames = ["date"]
    values = []
    for k, v in data.items():
        currency_pair = k.split(" ")[0].lower()
        colnames.append(currency_pair)
        values.append(v["FIXED_CLOSING_PRICE_NY"])
    values = [v["PX_CLOSE_DT"]] + values
    sqlstr = "INSERT INTO fx({0}) VALUES({1}) ON CONFLICT DO NOTHING".format(
        ",".join(colnames), ",".join(["%s"] * len(values))
    )

    with conn.cursor() as c:
        c.execute(sqlstr, values)
    conn.commit()


def init_swap_rates(
    conn,
    session,
    tenors=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 30],
    start_date=datetime.date(1998, 10, 7),
):
    securities = [f"USISDA{t:02} Index" for t in tenors]
    data = retrieve_data(
        session, securities, ["PX_LAST"], start_date=datetime.date(1998, 10, 7)
    )
    for t in tenors:
        ticker = f"USISDA{t:02} Index"
        sql_str = (
            f'INSERT INTO USD_swap_fixings(fixing_date, "{t}y") '
            + "VALUES(%s, %s) ON CONFLICT (fixing_date)"
            + f' DO UPDATE SET "{t}y" = excluded."{t}y"'
        )

        with conn.cursor() as c:
            c.executemany(sql_str, [(d, r) for d, r in data[ticker]["PX_LAST"].items()])
        conn.commit()


def init_swaption_vol(
    session,
    tenors=["A", "C", "F", "I"] + list(range(1, 11)) + [15, 20, 25, 30],
    source="BBIR",
    vol_type="N",
    start_date=datetime.date(1990, 1, 1),
):
    tickers = []
    for t1 in tenors:
        for t2 in tenors[4:]:
            tickers.append(f"USS{vol_type}{t1:0>2}{t2} {source} Curncy")
    data = retrieve_data(session, tickers, ["PX_LAST"], start_date=start_date)
    return data


def split_tenor_expiry(ticker, vol_type="N"):
    m = re.match("USS" + vol_type + r"(.{{2}})([^\s]*) ([^\s]*) Curncy", ticker)
    expiry, tenor, _ = m.groups()
    if expiry[0] == "0":
        expiry = expiry[1:]
    if not expiry.isalpha():
        expiry = int(expiry)
    tenor = int(tenor)
    return expiry, tenor


def insert_swaption_vol(data, conn, source, vol_type="N"):
    tenors = ["A", "C", "F", "I"] + list(range(1, 11)) + [15, 20, 25, 30]

    df = pd.concat(data, axis=1)
    df.columns = df.columns.get_level_values(0)
    df.columns = pd.MultiIndex.from_tuples(
        [split_tenor_expiry(c, vol_type) for c in df.columns]
    )

    table_name = "swaption_normal_vol" if vol_type == "N" else "swaption_lognormal_vol"
    for t in tenors[-14:]:
        sql_str = (
            f'INSERT INTO {table_name}(date, "{t}y", source) '
            + "VALUES(%s, %s, %s) ON CONFLICT (date, source)"
            + f' DO UPDATE SET "{t}y" = excluded."{t}y", source = excluded.source'
        )
        with conn.cursor() as c:
            df_temp = df.xs(t, axis=1, level=1).reindex(tenors, axis=1)
            for k, v in df_temp.iterrows():
                if np.all(np.isnan(v.values)):
                    continue
                c.execute(sql_str, (k, v.tolist(), source))
            conn.commit()


def update_swaption_vol(
    conn,
    session,
    expiries=["A", "C", "F", "I"] + list(range(1, 11)) + [15, 20, 25, 30],
    tenors=[1, 2, 3, 4, 5, 7, 10, 12, 15, 20, 25, 30],
    start_from=datetime.date.today(),
    *,
    sources=("BBIR", "CMPN", "ICPL"),
    vol_type="N",
):
    """
    Parameters
    ----------
    vol_type : one of 'N' or 'V' (normal or log-normal)
    """
    db_vol_type = "Normal" if vol_type == "N" else "LogNormal"
    mappings = {"A": "1M", "C": "3M", "F": "6M", "I": "9M"}
    for source in sources:
        tickers = {
            f"USS{vol_type}{e:0>2}{t} {source} Curncy": (e, t)
            for e, t in product(expiries, tenors)
        }
        data = retrieve_data(session, tickers, ["PX_LAST"], start_date=start_from)
        data = pd.concat(data, names=["ticker", "date"])
        for date, df in data.groupby(level="date"):
            with conn.cursor() as c:
                for ticker, val in df.reset_index("date", drop=True).itertuples():
                    e, t = tickers[ticker]
                    tenor = f"{t}Y"
                    expiry = mappings.get(e, f"{e}Y")
                    c.execute(
                        "INSERT INTO swaption_vol VALUES (%s, %s, %s, %s, %s, %s)"
                        " ON CONFLICT (date, expiry, tenor, vol_type, source) "
                        "DO UPDATE SET vol=EXCLUDED.vol",
                        (date, expiry, tenor, db_vol_type, source, val),
                    )
        conn.commit()


def update_swap_rates(
    conn, session, tenors=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 30]
):
    securities = [f"USISDA{t:02} Index" for t in tenors]
    data = retrieve_data(session, securities, ["PX_LAST", "LAST_UPDATE_DT"])
    for t in tenors:
        ticker = f"USISDA{t:02} Index"
        sql_str = (
            f'INSERT INTO USD_swap_fixings(fixing_date, "{t}y") '
            + "VALUES(%(LAST_UPDATE_DT)s, %(PX_LAST)s) ON CONFLICT (fixing_date)"
            + f' DO UPDATE SET "{t}y" = %(PX_LAST)s'
        )
        with conn.cursor() as c:
            c.execute(sql_str, data[ticker])
        conn.commit()


def update_cash_rates(conn, session, start_date=None):
    securities = {
        "FEDL01 Index": "FED_FUND",
        "US0001M Index": "1M_LIBOR",
        "US0003M Index": "3M_LIBOR",
    }
    if start_date is None:
        data = retrieve_data(
            session, list(securities.keys()), ["PX_LAST", "LAST_UPDATE_DT"]
        )
    else:
        data = retrieve_data(
            session, list(securities.keys()), ["PX_LAST"], start_date=start_date.date()
        )
    sql_str = "INSERT INTO rates VALUES(%s, %s, %s) ON CONFLICT DO NOTHING"
    with conn.cursor() as c:
        if start_date is None:
            for k, v in data.items():
                c.execute(sql_str, (v["LAST_UPDATE_DT"], securities[k], v["PX_LAST"]))
        else:
            for k, v in data.items():
                for d, r in v["PX_LAST"].items():
                    c.execute(sql_str, (d, securities[k], r))
    conn.commit()


def populate_cashflow_history(engine, session, workdate=None, funds=("SERCGMAST",)):
    securities = {}
    for fund in funds:
        secs = get_list(engine, workdate, fund=fund)
        for sec in secs.itertuples():
            if sec.Index not in securities:
                securities[sec.Index] = sec.identifier
    data = retrieve_data(
        session,
        securities,
        ["HIST_CASH_FLOW", "MTG_HIST_CPN", "FLT_CPN_HIST", "HIST_INTEREST_DISTRIBUTED"],
    )
    fixed_coupons = {"XS0306416982 Mtge": 7.62, "91927RAD1 Mtge": 6.77}
    conn = engine.raw_connection()
    for k, v in data.items():
        if "HIST_CASH_FLOW" in v:
            to_insert = v["HIST_CASH_FLOW"].merge(
                v["MTG_HIST_CPN"],
                how="left",
                left_on="Payment Date",
                right_on="Payment Date",
            )
            to_insert.rename(
                columns={
                    "Coupon_y": "coupon",
                    "Interest": "interest",
                    "Payment Date": "date",
                    "Principal Balance": "principal_bal",
                    "Principal Paid": "principal",
                },
                inplace=True,
            )
            to_insert.drop(["Period Number", "Coupon_x"], axis=1, inplace=True)
        elif "FLT_CPN_HIST" in v:
            to_insert = v["FLT_CPN_HIST"]
            to_insert.rename(
                columns={"Coupon Rate": "coupon", "Accrual Start Date": "date"},
                inplace=True,
            )
            to_insert.coupon = to_insert.coupon.shift(1)
        elif "HIST_INTEREST_DISTRIBUTED" in v:
            to_insert = v["HIST_INTEREST_DISTRIBUTED"]
            to_insert.rename(
                columns={"Interest": "interest", "Historical Date": "date"},
                inplace=True,
            )
            if k in fixed_coupons:
                to_insert["coupon"] = fixed_coupons[k]
            else:  # damn you XS0299146992 !
                continue
        else:
            logging.error(f"No cashflows for security {securities[k]}")
            continue

        to_insert["identifier"] = securities[k]
        with conn.cursor() as c:
            c.execute(
                "DELETE FROM cashflow_history WHERE identifier=%s", (securities[k],)
            )
        conn.commit()
        to_insert.to_sql("cashflow_history", engine, if_exists="append", index=False)
    with conn.cursor() as c:
        c.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY factors_history")
    conn.commit()
    conn.close()


if __name__ == "__main__":
    from serenitas.utils.db import serenitas_pool, dawn_engine
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "workdate",
        nargs="?",
        type=datetime.date.fromisoformat,
        default=datetime.date.today(),
    )
    args = parser.parse_args()

    dawn_conn = dawn_engine.raw_connection()
    serenitas_conn = serenitas_pool.getconn()
    with init_bbg_session(BBG_IP) as session:
        update_securities(dawn_engine, session, args.workdate)
        populate_cashflow_history(
            dawn_engine, session, args.workdate, ("SERCGMAST", "BRINKER", "BOWDST")
        )
        update_fx(dawn_conn, session, ["EURUSD", "CADUSD"])
        update_swap_rates(serenitas_conn, session)
        update_cash_rates(serenitas_conn, session)
        for vol_type in ["N", "V"]:
            update_swaption_vol(serenitas_conn, session, vol_type=vol_type)
    serenitas_pool.putconn(serenitas_conn)
    # with init_bbg_session(BBG_IP) as session:
    #     init_fx(session, engine, pd.datetime(2013, 1, 1))
    # with init_bbg_session(BBG_IP) as session:
    #     init_swap_rates(serenitas_conn, session, start_date=pd.datetime(2012, 2, 2))
    # for source in ['BBIR', 'ICPL', 'CMPN']:
    #     for vol_type in ["N", "V"]:
    #         with init_bbg_session(BBG_IP) as session:
    #             data = init_swaption_vol(session, source=source,
    #                                      vol_type=vol_type,
    #                                      start_date=datetime.date(2001, 1, 1))
    #         insert_swaption_vol(data, serenitas_conn, source,
    #                             vol_type=vol_type)