1 files changed, 160 insertions, 0 deletions
diff --git a/python/trade_dataclasses.py b/python/trade_dataclasses.py
new file mode 100644
index 00000000..e64ab63a
--- /dev/null
+++ b/python/trade_dataclasses.py
@@ -0,0 +1,160 @@
+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
+
+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",
+    "*",
+]
+
+
+@dataclass
+class Counterparty:
+    name: str
+
+
+register_adapter(Frequency, lambda f: AsIs(f.value))
+
+
+class Deal:
+    _conn: ClassVar = dbconn("dawndb")
+    _sql_fields: ClassVar[list[str]] = []
+    _sql_insert: ClassVar[str] = ""
+    _sql_select: ClassVar[str] = ""
+    _insert_queue: ClassVar[list] = []
+
+    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._insert_str, cls._insert_queue)
+        cls._conn.commit()
+
+    @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):
+    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
+
+
+CDSDeal._sql_fields = [f.name for f in fields(CDSDeal)]
+place_holders = ",".join(["%s"] * len(fields(CDSDeal)))
+columns = ",".join(c for c in CDSDeal._sql_fields)
+CDSDeal._sql_insert = f"INSERT INTO cds({columns}) VALUES({place_holders})"
+CDSDeal._sql_select = f"SELECT {columns} FROM cds WHERE id=%s"