path: root/python/external_deriv_marks.py

import datetime
import pandas as pd
import re
from serenitas.utils.env import DAILY_DIR
from collateral.baml_isda import load_excel
from collateral.citi import load_pdf, get_col
from collateral.jpm import load_positions
from serenitas.analytics.dates import next_business_day, prev_business_day
from serenitas.analytics.utils import get_fx

# local_nav is the nav in the trade's own currency
COLUMNS = ["trade_date", "buy/sell", "notional", "local_nav", "base_nav", "ia"]


def gs_navs(date: datetime.date = None, fund: str = "Serenitas"):
    d = {}
    date_str = date.strftime("%d_%b_%Y") if date else ""
    for fname in (DAILY_DIR / fund / "GS_reports").glob(
        f"Trade_Detail*{date_str}*.xls"
    ):
        try:
            df = pd.read_excel(fname, skiprows=9, skipfooter=77, index_col="Trade Id")
        except ValueError:
            continue
        df = df.dropna(subset=["GS Entity"])
        df["Trade Date"] = pd.to_datetime(df["Trade Date"])
        name = fname.name.replace("9972734", "")
        if m := re.match(r"[^\d]*(\d{2}_.{3}_\d{4})", name):
            (date_string,) = m.groups()
            date = datetime.datetime.strptime(date_string, "%d_%b_%Y").date()
        df["fx"] = df["Not1Ccy"].apply(lambda s: get_fx(date, s))
        df["local_navs"] = df["NPV (USD)"] / df["fx"]
        df = df[
            [
                "Trade Date",
                "Buy/Sell",
                "Notional (USD)",
                "local_navs",
                "NPV (USD)",
                "Initial Margin Required",
            ]
        ]
        df.columns = COLUMNS
        d[date] = df
    if d:
        df = pd.concat(d)
        # nav is from Goldman's point of view
        df[["local_nav", "base_nav"]] *= -1.0
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def ms_navs(date: datetime.date = None, fund: str = "Serenitas"):
    d = {}
    date_str = date.strftime("%Y%m%d") if date else "*"
    for fname in (DAILY_DIR / fund / "MS_reports").glob(f"Trade_Detail_{date_str}.xls"):
        df = pd.read_excel(fname, index_col="trade_id")
        df.trade_date = pd.to_datetime(df.trade_date)
        df = df[
            [
                "trade_date",
                "pay_rec",
                "notional_in_trade_ccy",
                "exposure_in_rpt_ccy",
                "exposure_in_rpt_ccy",
                "upfront_in_rpt_ccy",
            ]
        ]
        df.columns = COLUMNS
        df.ia *= -1.0
        if m := re.match(r"[^\d]*(\d{8})", fname.name):
            (date_string,) = m.groups()
            date = datetime.datetime.strptime(date_string, "%Y%m%d")
        d[date] = df
    if d:
        df = pd.concat(d)
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def citi_navs(date: datetime.date = None, **kwargs):
    date = next_business_day(date)
    dfs = []
    glob_str = f"{date:%Y%m%d}*" if date else "*"
    for fname in (DAILY_DIR / "CITI_reports").glob(f"262966_Portfolio_{glob_str}.xlsx"):
        df = pd.read_excel(
            fname, skiprows=6, skipfooter=2, parse_dates=["Trade Date", "Value Date"]
        )
        df = df.dropna(subset=["Operations File"]).set_index(
            ["Value Date", "Operations File"]
        )
        df = df[
            [
                "Trade Date",
                "Party Position",
                "Notional",
                "Market Value",
                "Market Value",
                "BasicAmt",
            ]
        ]
        df.columns = COLUMNS
        dfs.append(df)
    # there can be multiple files per day, we take the latest one
    df = (
        pd.concat(dfs)
        .sort_index()
        .groupby(level=["Value Date", "Operations File"])
        .last()
    )
    # nav is from Citi's point of view
    df[["local_nav", "base_nav"]] *= -1.0
    return df


def baml_navs(date: datetime.date = None, fund: str = "Serenitas"):
    dfs = []
    glob_str = f"{next_business_day(date):%m%d%Y}" if date else "*"
    match fund:
        case "Serenitas":
            tag = "TSLP"
        case "BowdSt":
            tag = "TLLC"
        case "Selene":
            tag = "INC"
        case "Brinker":
            tag = "BOGUS"
    for fname in (DAILY_DIR / fund / "BoA_reports").glob(
        f"301__LMCG_INVESTMEN{tag}_CSA_{glob_str}_*.xls"
    ):
        df = load_excel(fname)
        df = df.set_index(["Market Value Date", "Trade ID"])
        df = df[
            [
                "Trade Date",
                "Buy/Sell",
                "Notional 1",
                "local_nav",
                "base_nav",
                "ia",
            ]
        ]
        df.columns = COLUMNS
        dfs.append(df)
    if dfs:
        df = pd.concat(dfs)
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def bnp_navs(date: datetime.date = None, fund: str = "Serenitas"):
    d = {}
    date_str = date.strftime("%Y%m%d") if date else ""
    for fname in (DAILY_DIR / fund / "BNP_reports").glob(f"Exposure*{date_str}.XLS"):
        try:
            df = pd.read_excel(fname, skiprows=7)
        except ValueError:
            continue
        df["Trade Ref"] = df["Trade Ref"].str.replace("(FOC-|MBO-)", "", regex=True)
        df = df.set_index("Trade Ref")
        df["Trade Date"] = pd.to_datetime(df["Trade Date"], dayfirst=True)
        df["Exposure Amount"] = df["Exposure Amount"].where(
            df["Notional 1 Ccy"] == "EUR", df["Exposure Amount (Agmt Ccy)"]
        )
        df = df[
            [
                "Trade Date",
                "Buy/Sell",
                "Notional 1",
                "Exposure Amount",
                "Exposure Amount (Agmt Ccy)",
                "Lock Up (Agmt Ccy)",
            ]
        ]
        df.columns = COLUMNS
        d[datetime.datetime.strptime(fname.stem[-8:], "%Y%m%d").date()] = df
    if d:
        df = pd.concat(d)
        # nav is from BNP's point of view
        df[["local_nav", "base_nav"]] *= -1.0
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def cs_navs_old(date: datetime.date = None, fund: str = "Serenitas"):
    d = {}
    DATA_DIR = DAILY_DIR / fund / "CS_reports"
    glob_str = f"{date:%b%d%Y}" if date else "*"
    g = DATA_DIR.glob(f"DERV048829_{glob_str}.xlsx")
    for fname in g:
        try:
            df = pd.read_excel(fname, skiprows=9, skipfooter=50, thousands=",")
        except ValueError:
            continue
        df["Mid Price"] = df["Mid Price"].apply(
            lambda s: -float(s[1:-1].replace(",", ""))
            if s.startswith("(") and s.endswith(")")
            else float(s)
        )
        df["Order No"] = df["Order No"].astype("str")
        df["Trade Date"] = pd.to_datetime(df["Trade Date"])
        df = df.set_index("Order No")
        df = df[["Trade Date", "Buy/Sell", "Notional", "Mid Price", "Mid Price"]]
        df.columns = COLUMNS[:-1]
        # TODO: fix this
        df_ia = get_ia(date, fund)
        df = df.join(df_ia)
        d[datetime.datetime.strptime(fname.stem.split("_")[1], "%b%d%Y").date()] = df
    if d:
        df = pd.concat(d)
        # nav is from CS's point of view
        df[["local_nav", "base_nav"]] *= -1.0
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def cs_navs(date: datetime.date = None, fund: str = "Serenitas"):
    if date:
        date = next_business_day(date)
        glob_str = f"{date:%m%d%Y}"
    else:
        glob_str = "*"
    d = {}
    DATA_DIR = DAILY_DIR / fund / "CS_reports"
    full_name = {
        "Serenitas": "SerenitasCGMF",
        "BowdSt": "BostonBPStLLC",
        "Brinker": "",
        "Selene": "",
    }
    g = DATA_DIR.glob(f"CollateralCptyStatement161{full_name[fund]}RVM_{glob_str}.xls")
    for fname in g:
        try:
            df = pd.read_excel(fname, header=5, skipfooter=29)
        except ValueError:
            continue
        if df.empty:
            raise ValueError(f"CS: empty position statement for {date}")
        df.columns = [c.replace("\n", " ").strip() for c in df.columns]
        df = df[1:]
        df = df.set_index("Structure ID")
        df = df[
            [
                "Trade Date",
                "Buy/Sell",
                "Notional1",
                "PV (USD)",
                "PV (USD)",
                "Initial Margin (USD)",
            ]
        ]
        df.columns = COLUMNS
        date = datetime.datetime.strptime(fname.stem.split("_")[1], "%m%d%Y").date()
        d[prev_business_day(date)] = df
    if d:
        df = pd.concat(d)
        # nav is from CS's point of view
        df[["local_nav", "base_nav"]] *= -1.0
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def jpm_navs(date: datetime.date = None, fund: str = "BowdSt"):
    account = {"Serenitas": 923550, "BowdSt": 909271}
    DATA_DIR = DAILY_DIR / fund / "JPM_reports"
    if date:
        glob_str = f"{date:%y%m%d}"
    else:
        glob_str = "*"
    g = DATA_DIR.glob(f"CSCFTCSTMT-*-{glob_str}-{account.get(fund)}_2.pdf")
    d = {}
    for fname in g:
        pages = load_pdf(fname, pages=True)
        df = load_positions(pages[4])
        date = datetime.datetime.strptime(fname.stem.split("-")[2], "%y%m%d").date()
        df["fx"] = df["Pay CCY"].apply(lambda s: get_fx(date, s))
        df["local_navs"] = df["MTM Amount"] / df["fx"]
        df = df[
            [
                "Deal ID",
                "Trade Date",
                "Long/ Short",
                "Pay Notional",
                "local_navs",
                "MTM Amount",
                "IM Amount",
            ]
        ]
        df = df.set_index("Deal ID")
        df["IM Amount"] *= -1.0
        df.columns = COLUMNS
        d[date] = df
    if d:
        df = pd.concat(d)
    else:
        df = pd.DataFrame(columns=COLUMNS)
    return df


def get_ia(date: datetime.date = None, fund: str = "Serenitas"):
    date = next_business_day(date)
    glob_str = f"{date:%m%d%Y}"
    for fname in (DAILY_DIR / fund / "CS_reports").glob(
        f"CollateralCptyStatement161SerenitasCGMFRVM_{glob_str}.pdf"
    ):
        l = load_pdf(fname)
        top, bottom = get_box_dimension(l)
        trade_ids = get_col(l, top, bottom, 20, 70)
        ia = get_col(l, top, bottom, 850, 1000)
        df = pd.DataFrame({"trade_ids": trade_ids, "ia": ia})
        df.ia = pd.to_numeric(df.ia.str.strip().str.replace(",", ""))
        return df.set_index("trade_ids")


def get_box_dimension(l):
    for e in l:
        if e.text == "**CD Swaption":
            top = int(e["top"])
        if e.text == "**CD Swaption Total:":
            bottom = int(e["top"])
    return (top + 1, bottom - 1)


# def bnp_navs_old(date: datetime.date = None):
#     d = {}
#     date_str = date.strftime("%d%b%Y") if date else ""
#     for fname in (DAILY_DIR / "BNP_reports").glob(f"SERENITAS*0_*{date_str}.csv"):
#         try:
#             df = pd.read_csv(fname)
#         except ValueError:
#             continue
#         df = df.set_index("Contract")
#         df["COB Date"] = pd.to_datetime(df["COB Date"])
#         df = df[["COB Date", "B/S", "Notional", "Reval PV"]]
#         df.columns = ["trade_date", "buy/sell", "notional", "nav"]
#         d[datetime.datetime.strptime(fname.name.split("_")[3], "%d%b%Y").date()] = df
#     df = pd.concat(d)
#     return df


if __name__ == "__main__":
    import argparse
    import logging
    from serenitas.utils.pool import dawn_pool

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "date",
        type=datetime.datetime.fromisoformat,
        nargs="?",
        default=datetime.date.today(),
        help="this is today's date, we load marks as of previous day cob",
    )
    parser.add_argument(
        "-a", "--all", action="store_true", default=False, help="download everything"
    )
    parser.add_argument(
        "-d", "--debug", action="store_true", default=False, help="more verbose logging"
    )
    args = parser.parse_args()
    date = None if args.all else prev_business_day(args.date)
    logging.basicConfig()
    logger = logging.getLogger("external_marks")
    logger.setLevel(logging.DEBUG if args.debug else logging.INFO)
    for cp in ("MS", "CITI", "GS", "BAML", "BNP", "CS", "JPM"):
        for fund in ("Serenitas", "Brinker", "BowdSt", "Selene"):
            logger.info(f"{cp} at {fund}")
            try:
                df = globals()[f"{cp.lower()}_navs"](date, fund=fund)
            except ValueError as e:
                logger.error(e)
                continue
            if df.empty and cp == "CS":
                df = globals()[f"{cp.lower()}_navs_old"](date, fund=fund)
            logger.debug(df)
            with dawn_pool.connection() as conn:
                with conn.cursor() as c:
                    for k, v in df[["local_nav", "base_nav", "ia"]].iterrows():
                        c.execute(
                            "INSERT INTO external_marks_deriv "
                            "VALUES(%s, %s, %s, %s, %s, %s) "
                            "ON CONFLICT (identifier, date) "
                            "DO UPDATE SET local_nav=excluded.local_nav, "
                            "base_nav=excluded.base_nav, ia=excluded.ia",
                            (
                                *k,
                                float(v.local_nav),
                                float(v.base_nav),
                                cp,
                                float(v.ia),
                            ),
                        )