1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
from serenitas.ops.trade_dataclasses import Deal
from dataclasses import dataclass
import datetime
from typing import Literal
from serenitas.utils.db2 import dbconn
firmness = Literal["FIRM", "INDICATIVE"]
asset_class = Literal["CD"]
def maturity_dt(d):
try:
return datetime.date(
int(d["maturityyear"]), int(d["maturitymonth"]), int(d["maturityday"])
)
except (
ValueError,
KeyError,
): # Sometimes maturity isn't included but we still have tenor
return
@dataclass
class Quote(Deal, table_name="markit_singlename_quotes", deal_type=None):
quoteid: int
assetclass: asset_class
msg_id: str
quotesource: str
confidence: int
redcode: str = None
ticker: str = None
maturity: datetime.date = None
tenor: int = None
runningcoupon: int = None
bidconventionalspread: float = None
bidupfront: float = None
bidsize: float = None
askconventionalspread: float = None
askupfront: float = None
asksize: float = None
firmness: firmness = None
quotedate: datetime.datetime = None
@classmethod
def from_markit_line(cls, d):
additional_attributes = {
"maturity": maturity_dt(d),
"msg_id": d["message"]["id"],
"quotedate": datetime.datetime.fromtimestamp(
d["receiveddatetime"] / 1000
).replace(tzinfo=datetime.timezone.utc),
"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 (quoteid) DO NOTHING RETURNING *"
)
|
