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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
|
from libc.stdlib cimport malloc, free
from curve cimport (JpmcdsBuildIRZeroCurve, JpmcdsZeroPrice, JpmcdsMakeTCurve,
Basis, CONTINUOUS, JpmcdsCleanSpreadCurve)
from date cimport (JpmcdsStringToDateInterval, pydate_to_TDate, dcc,
JpmcdsDateIntervalToFreq, JpmcdsDateFwdThenAdjust, TDate_to_pydate,
JpmcdsDateFromBusDaysOffset)
from date import dcc_tostring
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):
""" method to inspect the content of the C struct
Returns
-------
dict
contains `base_date`, `basis`, `day_count_counvention` and `data`
"""
return {'base_date': TDate_to_pydate(self._thisptr.fBaseDate),
'basis': self._thisptr.fBasis,
'day_count_convention': dcc_tostring(self._thisptr.fDayCountConv),
'data': fArray_to_list(self._thisptr.fArray, self._thisptr.fNumItems)}
cdef fArray_to_list(TRatePt* fArray, int fNumItems):
cdef size_t i
cdef list l = []
for i in range(fNumItems):
l.append((TDate_to_pydate(fArray[i].fDate), fArray[i].fRate))
return l
cdef class YieldCurve(Curve):
""" Initialize a yield curve from a list of zero coupon rates
Parameters
----------
types : str
string containing only the letters 'M' (for Money Market ) and
'S' (for swaps) to describe the type of quotes
periods : list of str
Describe the maturity of each instrument (Example: ['3M', '2Y'])
rates: array.array
Array of double containing the quotes
mm_dcc : str
Day count convention for the money market instrument.
fixed_swap_period : str
Period of the fixed leg of the swap.
float_swap_period : str
Period of the floating leg of the swap.
fixed_swap_dcc : str
Day count convention for the fixed leg of the swap.
float_swap_dcc : str
Day count convention for the floating leg of the swap.
bad_day_conv : int
Business day convention.
.. warning:: Instruments need to be sorted by tenor!
"""
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):
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
cdef long period_adjust
for i, p in enumerate(periods):
period_bytes = p.encode('utf-8')
if JpmcdsStringToDateInterval(period_bytes, routine, &tmp) != SUCCESS:
raise ValueError
if types[i] == 'M':
period_adjust = MODIFIED
else:
priod_adjust = NONE
if JpmcdsDateFwdThenAdjust(settle_date, &tmp, period_adjust,
"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)
@classmethod
def from_discount_factors(cls, base_date, list dates, double[:] dfs, str day_count_conv):
""" build a yield curve from a list of discount factors """
cdef TDate base_date_c = pydate_to_TDate(base_date)
cdef YieldCurve yc = cls.__new__(cls)
yc._dates = <TDate*>malloc(sizeof(TDate) * len(dates))
cdef size_t i
cdef double* rates = <double*>malloc(sizeof(double) * len(dfs))
for i, d in enumerate(dates):
yc._dates[i] = pydate_to_TDate(d)
JpmcdsDiscountToRateYearFrac(dfs[i], <double>(yc._dates[i]-base_date_c)/365.,
<double>1, &rates[i]);
yc._thisptr = JpmcdsMakeTCurve(base_date_c, yc._dates, rates, len(dfs),
<double>1, dcc(day_count_conv))
return yc
def discount_factor(self, date):
""" computes the discount factor at a given date.
Parameters
----------
date : :class:`datetime.date`
Returns
-------
float
"""
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):
""" returns the list of instrument dates
"""
cdef size_t i
return [TDate_to_pydate(self._dates[i]) for i in range(self._ninstr)]
cdef class SpreadCurve(Curve):
"""
Initialize a SpreadCurve from a list of spreads and maturity.
Parameters
----------
today : :class:`datetime.date`
yc : :class:`~pyisda.curve.YieldCurve`
start_date : :class:`datetime.date`
step_in_date : :class:`datetime.date`
cash_settle_date: :class:`datetime.date`
end_dates : list of :class:`datetime.date`
coupon_rates : :class:`array.array` of double
recovery_rate : float
pay_accrued_on_default : bool, optional
"""
def __init__(self, today, YieldCurve yc, start_date, step_in_date,
cash_settle_date, list end_dates, double[:] coupon_rates,
double recovery_rate, int pay_accrued_on_default = True):
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
if cash_settle_date < yc.inspect()['base_date']:
raise ValueError("cash_settle_date: {0} is anterior to yc's base_date: {1}".
format(cash_settle_date, yc.inspect()['base_date']))
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,
yc._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')
if self._thisptr == NULL:
raise ValueError("something went wrong")
@classmethod
def from_flat_hazard(cls, base_date, double rate, Basis basis = CONTINUOUS,
str day_count_conv = 'Actual/365F'):
"""
Alternative constructor for flat hazard rate Curve.
Parameters
----------
base_date : datetime.date
Starting date of the curve
rate : float
Flat hazard rate.
basis : int, optional
Default to :data:`CONTINUOUS`
day_count_cont : str, optional
Default to 'Actual/365F'
"""
cdef TDate base_date_c = pydate_to_TDate(base_date)
cdef SpreadCurve sc = cls.__new__(cls)
cdef TDate max_date = 200000 # can go higher but this should be more than enough
cdef TDate* dates = <TDate*>malloc(sizeof(TDate))
cdef double* rates = <double*>malloc(sizeof(double))
dates[0] = max_date
rates[0] = rate
sc._thisptr = JpmcdsMakeTCurve(base_date_c, dates, rates, 1,
<double>basis, dcc(day_count_conv))
return sc
|
