summaryrefslogtreecommitdiffstats
path: root/curve.pyx
blob: 88b1098d048f9a25054c4369b9a050507cb08ff5 (plain)
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
from cpython cimport datetime
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)

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,
                 TBoolean pay_accrued_on_default):

        cdef today_c = pydate_to_TDate(today)
        cdef step_in_date_c = pydate_to_TDate(step_in_date)
        cdef cash_settle_date_c = pydate_to_TDate(cash_settle_date)
        cdef start_date_c = pydate_to_TDate(start_date)
        cdef TDate* end_dates_c = <TDate*>malloc(len(end_dates) * sizeof(TDate))

        cdef size_t i

        for i, d in enumerate(end_dates):
            end_dates_c[i] = pydate_to_TDate(end_dates)
        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,
                                               len(end_dates),
                                               end_dates_c,
                                               &coupon_rates[0],
                                               NULL,
                                               recovery_rate,
                                               pay_accrued_on_default,
                                               NULL,
                                               dcc('ACT/360'),
                                               &stub_type,
                                               <long>'M',
                                               NULL)