from dataclasses import dataclass, field, fields from typing import ClassVar from decimal import Decimal from typing import Literal import datetime from enum import Enum from psycopg2.extensions import register_adapter, AsIs from serenitas.analytics.dates import next_business_day, previous_twentieth from serenitas.analytics.index import CreditIndex from serenitas.utils.db import dbconn from process_queue import rename_keys Fund = Literal["SERCGMAST", "BRINKER", "BOWDST"] Portfolio = Literal[ "OPTIONS", "IR", "MORTGAGES", "CURVE", "TRANCHE", "CLO", "HEDGE_MAC" ] # deprecated IG, HY, STRUCTURED class BusDayConvention(str, Enum): modified_following = "Modified Following" following = "Following" modified_preceding = "Modified Preceding" second_day_after = "Second-Day-After" end_of_month = "End-of-Month" DayCount = Literal["ACT/360", "ACT/ACT", "30/360", "ACT/365"] IsdaDoc = Literal["ISDA2014", "ISDA2003Cred"] class Frequency(Enum): Quarterly = 4 Monthly = 12 Ccy = Literal["USD", "CAD", "EUR", "YEN"] SwapType = Literal[ "CD_INDEX", "CD_INDEX_TRANCHE", "CD_BASKET_TRANCHE", "ABS_CDS", "BESPOKE" ] ClearingFacility = Literal["ICE-CREDIT", "NOT CLEARED"] CdsStrat = Literal[ "HEDGE_CSO", "HEDGE_CLO", "HEDGE_MAC", "HEDGE_MBS", "SER_IGSNR", "SER_IGMEZ", "SER_IGEQY", "SER_IGINX", "SER_HYSNR", "SER_HYMEZ", "SER_HYEQY", "SER_HYINX", "SER_HYCURVE", "SER_IGCURVE", "SER_ITRXCURVE", "XCURVE", "MBSCDS", "IGOPTDEL", "HYOPTDEL", "HYEQY", "HYMEZ", "HYSNR", "HYINX", "IGEQY", "IGMEZ", "IGSNR", "IGINX", "XOEQY", "XOMEZ", "XOINX", "EUEQY", "EUMEZ", "EUSNR", "EUINX", "BSPK", "*", ] BondStrat = Literal[ "M_STR_MAV", "M_STR_MEZZ", "CSO_TRANCH", "M_CLO_BB20", "M_CLO_AAA", "M_CLO_BBB", "M_MTG_IO", "M_MTG_THRU", "M_MTG_GOOD", "M_MTG_B4PR", "M_MTG_RW", "M_MTG_FP", "M_MTG_LMG", "M_MTG_SD", "M_MTG_PR", "M_MTG_CRT_SD", "CRT_LD", "CRT_LD_JNR", "CRT_SD", "IGNORE", "MTG_REPO", ] AssetClass = Literal["CSO", "Subprime", "CLO", "CRT"] @dataclass class Counterparty: name: str register_adapter(Frequency, lambda f: AsIs(f.value)) class Deal: _conn: ClassVar = dbconn("dawndb", application_name="autobooker") _table_name: None _sql_fields: ClassVar[list[str]] _sql_insert: ClassVar[str] _sql_select: ClassVar[str] _insert_queue: ClassVar[list] = [] def __init_subclass__(cls, table_name: str): super().__init_subclass__() cls._table_name = table_name cls._sql_fields = list(cls.__annotations__) _sql_insert_fields = list( c for c in cls.__annotations__ if c not in ("id", "dealid") ) insert_place_holders = ",".join(["%s"] * len(_sql_insert_fields)) insert_columns = ",".join(c for c in _sql_insert_fields) select_columns = ",".join(c for c in cls._sql_fields) cls._sql_insert = f"INSERT INTO {cls._table_name}({insert_columns}) VALUES({insert_place_holders})" cls._sql_select = f"SELECT {select_columns} FROM {cls._table_name} WHERE id=%s" def stage(self): self._insert_queue.append([getattr(self, f) for f in self._sql_fields]) @classmethod def commit(cls): with cls._conn.cursor() as c: c.executemany(cls._sql_insert, cls._insert_queue) cls._conn.commit() cls._insert_queue.clear() @classmethod def from_tradeid(cls, trade_id: int): with cls._conn.cursor() as c: c.execute(cls._sql_select, (trade_id,)) r = c.fetchone() return cls(*r) @dataclass class CDSDeal(Deal, table_name="cds"): id: field(default=None) dealid: field(default=None) initial_margin_percentage: field(default=None) fund: Fund account_code: str cp_code: str security_id: str security_desc: str maturity: datetime.date currency: Ccy protection: Literal["Buy", "Sell"] notional: float fixed_rate: float upfront: float traded_level: Decimal effective_date: datetime.date = field(default=None) portfolio: Portfolio = field(default=None) folder: CdsStrat = field(default=None) payment_rolldate: BusDayConvention = BusDayConvention.following day_count: DayCount = "ACT/360" frequency: Frequency = Frequency.Quarterly trade_date: datetime.date = field(default_factory=datetime.date.today()) upfront_settle_date: datetime.date = field( default_factory=lambda: next_business_day(datetime.date.today()) ) swap_type: SwapType = "CD_INDEX" clearing_facility: ClearingFacility = "ICE-CREDIT" isda_definition: IsdaDoc = "ISDA2014" def __post_init__(self): self.effective_date = previous_twentieth(self.trade_date) def credit_index(self): index = CreditIndex( redcode=self.security_id, maturity=self.maturity, notional=self.notional, value_date=self.trade_date, ) index.direction = self.protection def to_markit(self): obj = self.__dict__ rename_keys( obj, { "dealid": "Swap ID", "cp_code": "Broker Id", "trade_date": "Trade Date", "effective_date": "Effective Date", "maturity": "Maturity Date", "notional": "1st Leg Notional", "fixed_rate": "1st Leg Rate", "upfront": "Initial Payment", "security_id": "RED", "orig_attach": "Attachment Point", "orig_detach": "Exhaustion Point", "currency": "Currency Code", "upfront_settle_date": "First Payment Date", "cp_code": "Broker Id", "fund": "Account Abbreviation", }, ) if obj["Initial Payment"] >= 0: obj["Transaction Code"] = "Receive" else: obj["Initial Payment"] = abs(round(obj["Initial Payment"], 2)) obj["Transaction Code"] = "Pay" obj["Trade ID"] = obj["Swap ID"] obj["Product Type"] = "TRN" obj["Transaction Type"] = "NEW" obj["Protection"] = "Buy" if obj["protection"] == "Buyer" else "Sell" obj["Entity Matrix"] = "Publisher" obj["Definitions Type"] = "ISDA2014Credit" obj["Independent Amount (%)"] = obj["initial_margin_percentage"] if "ITRX" in obj["security_desc"]: obj["Include Contractual Supplement"] = "Y" obj["Contractual Supplement"] = "StandardiTraxxEuropeTranche" return obj @dataclass class BondDeal(Deal, table_name="bonds"): buysell: bool description: str faceamount: float price: float cp_code: str cusip: str = None isin: str = None identifier: str = None trade_date: datetime.date = field(default_factory=datetime.date.today()) settle_date: datetime.date = field( default_factory=lambda: next_business_day(datetime.date.today()) ) folder: BondStrat = field(default=None) portfolio: Portfolio = field(default=None) asset_class: AssetClass = field(default=None)