summaryrefslogtreecommitdiffstats
path: root/curve.pyx
blob: a5eb2d20f00de5d91835854cade4106d69af7680 (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
from cpython cimport datetime
from libc.stdlib cimport malloc, free
from pyisda.zerocurve cimport JpmcdsBuildIRZeroCurve, JpmcdsZeroPrice
from pyisda.yearfrac cimport dcc
from pyisda.date cimport (JpmcdsStringToDateInterval, pydate_to_TDate,
                          JpmcdsDateIntervalToFreq, JpmcdsDateFwdThenAdjust, TDate_to_pydate,
                          JpmcdsDateFromBusDaysOffset)

cdef int SUCCESS = 0

cpdef public enum BadDay:
    FOLLOW = <long>'F'
    PREVIOUS = <long>'P'
    NONE = <long>'N'
    MODIFIED = <long>'M'

cdef class ZeroCurve:

    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._thisptr is not NULL:
            JpmcdsFreeTCurve(self._thisptr)
        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)]