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
|
from serenitas.utils.db2 import dbconn
from lz4.block import decompress
import datetime
from construct import Enum, CString, Struct, Int8ul, Int32ul, Int32sl, Int64ul, Int64sl, Float64l, this
from pyisda.curve import YieldCurve as CYieldCurve, SpreadCurve as CSpreadCurve
TRatePt = Struct("fDate" / Int64sl,
"fRate" / Float64l)
TCurve = Struct("fNumItems" / Int32sl,
"fArray" / TRatePt[this.fNumItems],
"fBaseDate" / Int64sl,
"fBasis" / Float64l,
"fDayCountConv" / Int64sl)
YieldCurve = Struct("curve" / TCurve,
"n_dates" / Int64ul,
"dates" / Int64sl[this.n_dates])
CurveName = Struct("Seniority" / Enum(Int32ul, Senior=0, Subordinated=1),
"DocClause" / Enum(Int32ul, XR14=0, MR14=1, MM14=2, CR14=3),
"ticker" / CString("utf8"))
SpreadCurve = Struct("curve" / TCurve,
"recovery" / Float64l[this.curve.fNumItems],
"defaulted" / Int64sl,
"name" / CurveName)
TCurve_v1 = Struct("fNumItems" / Int32sl,
"fDayCountConv" / Int32sl,
"fBaseDate" / Int64sl,
"fBasis" / Float64l,
"fArray" / TRatePt[this.fNumItems])
YieldCurve_v1 = Struct("curve" / TCurve_v1)
CurveName_v1 = Struct("Seniority" / Enum(Int8ul, Senior=0, Subordinated=1),
"DocClause" / Enum(Int8ul, XR14=0, MR14=1, MM14=2, CR14=3),
"ticker" / CString("utf8"))
SpreadCurve_v1 = Struct("curve" / TCurve_v1,
"recovery" / Float64l[this.curve.fNumItems],
"defaulted" / Int64sl,
"name" / CurveName_v1)
serenitasdb = dbconn("serenitasdb")
with serenitasdb.cursor(binary=True) as c:
c.execute('SELECT curve FROM rate_curves WHERE effective_date=%s AND curve_type=%s',
(datetime.date(2023, 1, 9), 532))
(state,) = c.fetchone()
curve = decompress(state, 500)
yc = YieldCurve.parse(curve)
yc_prime = CYieldCurve.from_bytes(state)
curve_v1 = yc_prime.as_bytes(False)
yc_v1 = YieldCurve_v1.parse(curve_v1)
c.execute('SELECT curve FROM cds_curves WHERE date=%s AND redcode=%s',
(datetime.date(2023, 1, 9), 'EKFHAH'))
(state,) = c.fetchone()
curve = decompress(state, 600)
sc = SpreadCurve.parse(curve)
breakpoint()
sc_prime = CSpreadCurve.from_bytes(state, compressed=True)
curve_v1 = sc_prime.as_bytes(False)
sc_v1 = SpreadCurve_v1.parse(curve_v1)
new_yc = TCurve_v1.build({"fNumItems":yc.curve.fNumItems, "fDayCountConv":yc.curve.fDayCountConv, "fBaseDate":yc.curve.fBaseDate,
"fBasis":yc.curve.fBasis, "fArray":yc.curve.fArray})
|
