from serenitas.ops.trade_dataclasses import Deal from dataclasses import dataclass, field import datetime from typing import Literal from serenitas.utils.db2 import dbconn firmness = Literal["FIRM", "INDICATIVE"] asset_class = Literal["CD"] @dataclass class QuoteDetails(Deal, table_name="markit_quote_details", deal_type=None): quotetime: datetime quote_source: str id: int sender: str = None @dataclass class Quote(Deal, table_name="markit_quotes", deal_type=None): quoteid: int assetclass: asset_class redcode: str ticker: str maturity: datetime.date tenor: int runningcoupon: int bidconventionalspread: float bidupfront: float bidsize: float askconventionalspread: float askupfront: float asksize: float firmness: firmness msg_id: str quotedate: datetime quotesource: str @classmethod def from_markit_line(cls, d): additional_attributes = { "maturity": datetime.date( int(d["maturityyear"]), int(d["maturitymonth"]), int(d["maturityday"]) ), "msg_id": d["message"]["id"], "quotedate": d["receiveddatetime"], "quotesource": d["sourceshortname"], "tenor": f"{d['tenor']}Y", } d.update(additional_attributes) return cls.from_dict(**d) @property def message(self): return QuoteDetails.from_tradeid(self.msg_id) Quote.init_dbconn(dbconn("serenitasdb")) Quote._sql_insert = Quote._sql_insert.replace( "RETURNING *", "ON CONFLICT DO NOTHING RETURNING *" )