import datetime import pandas as pd from serenitas.analytics.dates import next_business_day from . import DAILY_DIR, SftpClient from .common import compare_notionals, compare_notionals_rates, STRATEGY_CASH_MAPPING def download_files(*args, **kwargs): sftp = SftpClient.from_creds("gs") sftp.download_files("outgoing", DAILY_DIR / "BowdSt" / "GS_fcm_reports") def get_filename(d: datetime.date, name: str): try: fname = next( (DAILY_DIR / "BowdSt" / "GS_fcm_reports").glob( f"GS-*-{name}-*-{d:%Y%m%d}-*.csv" ) ) except StopIteration: raise FileNotFoundError(f"GS fcm file {name} not found for date {d}") else: return fname def get_file_date(d: datetime.date): # GS weekend timestamp is now same date instead of prev_date + 1 if d == datetime.date(2023, 6, 2): return next_business_day(d) else: return d + datetime.timedelta(days=1) def collateral(d: datetime.date, positions, *, engine, fund, positions_irs, **kwargs): file_date = get_file_date(d) df = pd.read_csv( get_filename(file_date, "Open_Trades_Redcode"), usecols=[ "Notional", "Direction", "NPV (local)", "Maturity Date", "COB Date", "Red Code", "Instrument", "CCP Trade ID", "Pay Notional", "Pay Leg Rate/Index", "Receive Leg Rate/Index", ], parse_dates=["COB Date", "Maturity Date"], index_col=["Red Code", "Maturity Date"], thousands=",", ) df_rates = df[df["Instrument"] == "IRS"].drop(["Instrument", "Notional"], axis=1) df_rates = df_rates.reset_index() df_rates.loc[df_rates["Direction"] == "Receive", "Pay Notional"] *= -1 df_rates["fixed_rate"] = df_rates.apply( lambda row: float(row["Receive Leg Rate/Index"]) if row["Direction"] == "Receive" else float(row["Pay Leg Rate/Index"]), axis=1, result_type="reduce", ) _rate_index_mapping = {"1Y-USD-SOFR-COMPOUND": "SOFRRATE"} df_rates["float_index"] = df_rates.apply( lambda row: _rate_index_mapping[row["Receive Leg Rate/Index"]] if row["Direction"] == "Pay" else _rate_index_mapping[row["Pay Leg Rate/Index"]], axis=1, result_type="reduce", ) # Going to fix roll day at IMM for now until we have the column updated df_rates["roll_day"] = "IMM" df_rates = df_rates.rename(columns={"Pay Notional": "NOTIONAL"}) df_rates = df_rates.groupby( by=["Maturity Date", "fixed_rate", "roll_day", "float_index"] )[["NOTIONAL", "NPV (local)"]].sum() df_rates.index.names = ["maturity_date", "fixed_rate", "roll_day", "float_index"] df = df[df["Instrument"] == "CDS"].drop( ["Instrument", "CCP Trade ID", "Pay Notional"], axis=1 ) df.Notional = df.Notional.where(df.Direction == "Buy", -df.Notional) df.index.names = ["security_id", "maturity"] cob_date = df["COB Date"][0] df = df.groupby(level=["security_id", "maturity"])[ ["Notional", "NPV (local)"] ].sum() df = df.rename(columns={"Notional": "NOTIONAL"}) compare_notionals(df, positions, "GS", fund) df_rates = df_rates.rename(columns={"Pay Notional": "NOTIONAL"}) compare_notionals_rates(df_rates, positions_irs, "GS") positions_irs["Amount"] = df_rates.reindex(positions_irs.index)["NPV (local)"] df_rates = ( positions_irs.groupby(["folder", "currency"]) .agg({"Amount": "sum"}) .reset_index(["folder", "currency"]) ) df_rates = df_rates.rename(columns={"folder": "Strategy", "currency": "Currency"}) df["dirty_upfront"] = df["NPV (local)"] / df.NOTIONAL positions["dirty_upfront"] = df.reindex(positions.index)["dirty_upfront"] positions["Amount"] = positions["notional"] * positions["dirty_upfront"] positions.folder = positions.folder.replace(STRATEGY_CASH_MAPPING) def aux(row): if row.folder == "XCURVE": return "SER_IGCVECSH" if row.currency == "USD" else "SER_ITRXCVCSH" else: return row.folder positions.folder = positions.apply(aux, axis=1) df = ( positions.groupby(["folder", "currency"]) .agg({"Amount": "sum"}) .reset_index(["folder", "currency"]) ) df = df.rename(columns={"folder": "Strategy", "currency": "Currency"}) df = pd.concat([df, df_rates]) df["date"] = cob_date df_margin = pd.read_csv( get_filename(file_date, "Account_Balances_and_Margin_extended"), parse_dates=["COB Date"], thousands=",", ) col_mapping = { "Opening Balance (local)": "beginning_balance", "Ending Balance (local)": "ending_balance", "PAI (local)": "pai", "Account Value + IM Collateral (local)": "account_value_market", "Initial Margin Requirement (local)": "current_im", "Excess/Deficit (local)": "current_excess_deficit", "Currency": "currency", "GS Account Number": "account", "COB Date": "date", "CCP Fees (local)": "clearing_fees", "Commissions (local)": "transaction_fees", } df_margin = df_margin.rename(columns=col_mapping) df_margin["current_im"] *= -1.0 df_margin["account"] = df_margin["account"].str.replace(" ", "") cols = col_mapping.values() df_margin_futures = pd.read_csv( get_filename(d + datetime.timedelta(days=1), "Account_Balances_By_Currency"), parse_dates=["COB Date"], thousands=",", ) col_mapping = { "Beginning Account Balance (Local)": "beginning_balance", "Ending Account Balance (Local)": "ending_balance", "Initial Margin Requirement (Local)": "current_im", "GS Account Number": "account", "COB Date": "date", "Excess/Deficit (Local)": "current_excess_deficit", "Currency": "currency", "Gross Realized P&L (Local)": "realized_pnl", "Commission & Fees (Local)": "transaction_fees", } df_margin_futures = df_margin_futures.rename(columns=col_mapping) df_margin_futures["account_value_market"] = ( df_margin_futures["current_im"] + df_margin_futures["Total Equity (Local)"] ) df_margin_futures["pai"] = 0.0 df_margin = df_margin.groupby(["date", "account", "currency"], as_index=False).sum() df_margin = pd.concat([df_margin, df_margin_futures]) place_holders = ",".join(["%s"] * len(cols)) engine.execute( f"INSERT INTO fcm_moneyline({','.join(cols)}) VALUES({place_holders})" "ON CONFLICT (date, account, currency) " "DO NOTHING", list(df_margin[cols].itertuples(index=False)), ) return df.set_index("Strategy")