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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
from libc.stdlib cimport malloc, free
from curve cimport JpmcdsBuildIRZeroCurve, JpmcdsZeroPrice
from date cimport (JpmcdsStringToDateInterval, pydate_to_TDate, dcc,
JpmcdsDateIntervalToFreq, JpmcdsDateFwdThenAdjust, TDate_to_pydate,
JpmcdsDateFromBusDaysOffset)
from cdsone cimport JpmcdsStringToStubMethod, TStubMethod
cdef int SUCCESS = 0
cpdef public enum BadDay:
FOLLOW = <long>'F'
PREVIOUS = <long>'P'
NONE = <long>'N'
MODIFIED = <long>'M'
cdef class Curve:
def __dealloc__(self):
if self._thisptr is not NULL:
JpmcdsFreeTCurve(self._thisptr)
def inspect(self):
return {'base_date': TDate_to_pydate(self._thiptr.fBaseDate)}
cdef class ZeroCurve(Curve):
def __init__(self, date, str types,
list periods, double[:] rates,
str mm_dcc, str fixed_swap_period, str float_swap_period,
str fixed_swap_dcc, str float_swap_dcc, BadDay bad_day_conv):
""" Initialize a zero coupon curve
instruments need to be sorted by tenor
"""
cdef:
double fixed_freq
double float_freq
TDateInterval ivl
char* routine = 'zerocurve'
TDate value_date = pydate_to_TDate(date)
self._dates = <TDate*>malloc(len(periods) * sizeof(TDate))
self._ninstr = len(periods)
cdef TDate settle_date
if JpmcdsDateFromBusDaysOffset(value_date, 2, "None", &settle_date)!= SUCCESS:
raise ValueError
cdef TDateInterval tmp
for i, p in enumerate(periods):
period_bytes = p.encode('utf-8')
if JpmcdsStringToDateInterval(period_bytes, routine, &tmp) != SUCCESS:
raise ValueError
if JpmcdsDateFwdThenAdjust(settle_date, &tmp, NONE,
"None", &self._dates[i]) != SUCCESS:
raise ValueError('Invalid interval')
fixed_bytes = fixed_swap_period.encode('utf-8')
float_bytes = float_swap_period.encode('utf-8')
types_bytes = types.encode('utf-8')
if JpmcdsStringToDateInterval(fixed_bytes, routine, &ivl) != SUCCESS:
raise ValueError
if JpmcdsDateIntervalToFreq(&ivl, &fixed_freq) != SUCCESS:
raise ValueError
if JpmcdsStringToDateInterval(float_bytes, routine, &ivl) != SUCCESS:
raise ValueError
if JpmcdsDateIntervalToFreq(&ivl, &float_freq) != SUCCESS:
raise ValueError
self._thisptr = JpmcdsBuildIRZeroCurve(
value_date, types_bytes, self._dates,
&rates[0], len(periods), dcc(mm_dcc), <long> fixed_freq,
<long> float_freq, dcc(fixed_swap_dcc), dcc(float_swap_dcc),
bad_day_conv, b"None"
)
def __dealloc__(self):
if self._dates is not NULL:
free(self._dates)
def discount_factor(self, date):
if self._thisptr is NULL:
raise ValueError('curve is empty')
cdef TDate discount_date = pydate_to_TDate(date)
return JpmcdsZeroPrice(self._thisptr, discount_date)
def list_dates(self):
cdef size_t i
return [TDate_to_pydate(self._dates[i]) for i in range(self._ninstr)]
cdef class SpreadCurve(Curve):
def __init__(self, today, ZeroCurve zc, start_date, step_in_date,
cash_settle_date, list end_dates, double[:] coupon_rates,
double recovery_rate, int pay_accrued_on_default):
cdef TDate today_c = pydate_to_TDate(today)
cdef TDate step_in_date_c = pydate_to_TDate(step_in_date)
cdef TDate cash_settle_date_c = pydate_to_TDate(cash_settle_date)
cdef TDate start_date_c = pydate_to_TDate(start_date)
cdef TDate* end_dates_c = <TDate*>malloc(len(end_dates) * sizeof(TDate))
self._thisptr = NULL
cdef size_t i
for i, d in enumerate(end_dates):
end_dates_c[i] = pydate_to_TDate(d)
cdef TStubMethod stub_type
if JpmcdsStringToStubMethod(b"f/s", &stub_type) != 0:
raise ValueError("can't convert stub")
self._thisptr = JpmcdsCleanSpreadCurve(today_c,
zc._thisptr,
start_date_c,
step_in_date_c,
cash_settle_date_c,
len(end_dates),
end_dates_c,
&coupon_rates[0],
NULL,
recovery_rate,
pay_accrued_on_default,
NULL,
dcc('ACT/360'),
&stub_type,
<long>'M',
b'NONE')
|