path: root/pyisda/curve.pyx

from libc.stdlib cimport malloc, free
from libc.string cimport memcpy
from libc.math cimport log1p, expm1
from date cimport (JpmcdsStringToDateInterval, pydate_to_TDate, dcc,
                   JpmcdsDateIntervalToFreq, JpmcdsDateFwdThenAdjust, TDate_to_pydate,
                   JpmcdsDateFromBusDaysOffset, JpmcdsStringToDayCountConv)
from date import dcc_tostring
from cdsone cimport JpmcdsStringToStubMethod, TStubMethod
cimport cython
cimport numpy as np
np.import_array()

cdef extern from "numpy/arrayobject.h":
    void PyArray_ENABLEFLAGS(np.ndarray arr, int flags)

cdef int SUCCESS = 0

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

cdef class Curve(object):
    def __dealloc__(self):
        if self._thisptr is not NULL:
            JpmcdsFreeTCurve(self._thisptr)

    def __getstate__(self):
        cdef int num_items = self._thisptr.fNumItems
        return (num_items,
                <bytes>(<char*>self._thisptr.fArray)[:sizeof(TRatePt)*num_items],
                self._thisptr.fBaseDate,
                self._thisptr.fBasis,
                self._thisptr.fDayCountConv)

    def __setstate__(self, state):
        num_items, rates, base_date, basis, dcc = state
        self._thisptr = <TCurve*>malloc(sizeof(TCurve))
        self._thisptr.fNumItems = num_items
        self._thisptr.fArray = <TRatePt*>malloc(sizeof(TRatePt) * num_items)
        memcpy(self._thisptr.fArray, <char*> rates, sizeof(TRatePt) * num_items)
        self._thisptr.fBaseDate = base_date
        self._thisptr.fBasis = basis
        self._thisptr.fDayCountConv = dcc

    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': self.base_date,
                'basis': self._thisptr.fBasis,
                'day_count_convention': dcc_tostring(self._thisptr.fDayCountConv),
                'data': fArray_to_list(self._thisptr.fArray, self._thisptr.fNumItems)}

    @property
    def forward_hazard_rates(self):
        cdef double t1, h1, t2, h2
        cdef np.npy_intp shape = self._thisptr.fNumItems
        t1 = 0
        h1 = 0
        cdef double* data = <double*>malloc(self._thisptr.fNumItems * sizeof(double))
        for i in range(self._thisptr.fNumItems):
            h2 = log1p(self._thisptr.fArray[i].fRate)
            t2 = (self._thisptr.fArray[i].fDate - self._thisptr.fBaseDate)/365.
            data[i] = (h2 * t2 - h1 * t1) / (t2 - t1)
            h1 = h2
            t1 = t2
        cdef np.ndarray[np.float64_t] out = \
            np.PyArray_SimpleNewFromData(1, &shape, np.NPY_DOUBLE, data)
        PyArray_ENABLEFLAGS(out, np.NPY_OWNDATA)
        return out

    @property
    def base_date(self):
        return TDate_to_pydate(self._thisptr.fBaseDate)

    def __forward_zero_price(self, d2, d1=None):
        """ computes the forward zero price at a given date.

        Parameters
        ----------
        date : :class:`datetime.date`

        Returns
        -------
        float
        """
        if self._thisptr is NULL:
            raise ValueError('curve is empty')
        cdef TDate start_date
        if d1 is None:
            start_date = self._thisptr.fBaseDate
        else:
            start_date = pydate_to_TDate(d1)
        return JpmcdsForwardZeroPrice(self._thisptr,
                                      start_date,
                                      pydate_to_TDate(d2))

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):
        ## __dealloc__ of superclass is called by cython so no need to call here
        if self._dates is not NULL:
            free(self._dates)

    def __getstate__(self):
        cdef Py_ssize_t size = sizeof(TRatePt) * self._ninstr
        cdef bytes dates = (<char*>self._dates)[:size]
        return super().__getstate__() + (self._ninstr, dates)

    def __setstate__(self, state):
        super().__setstate__(state[:5])
        self._ninstr = <int>state[5]
        cdef Py_ssize_t size = sizeof(TRatePt) * self._ninstr
        self._dates = <TDate*>malloc(size)
        memcpy(self._dates, <char*> state[6], size)

    @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))
        yc._ninstr = len(dates)
        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, dfs.shape[0],
                                       <double>1, dcc(day_count_conv))
        return yc

    discount_factor = Curve.__forward_zero_price

    @property
    def 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)]

    def expected_forward_curve(self, forward_date):
        """ returns the expected forward curve """
        cdef TDate forward_date_c = pydate_to_TDate(forward_date)
        cdef YieldCurve yc = YieldCurve.__new__(YieldCurve)
        cdef size_t i = 0
        while self._dates[i] < forward_date_c:
            i += 1
        yc._ninstr = self._ninstr - i
        yc._dates = <TDate*>malloc(sizeof(TDate) * (self._ninstr-i))
        cdef double* rates = <double*>malloc(sizeof(double) * yc._ninstr)
        cdef size_t k
        cdef double df
        for k in range(yc._ninstr):
            yc._dates[k] = self._dates[i]
            df = JpmcdsForwardZeroPrice(self._thisptr, forward_date_c, self._dates[i])
            JpmcdsDiscountToRateYearFrac(
                df, <double>(self._dates[i] - forward_date_c)/365.,
                <double>1, &rates[k])
            i += 1
        yc._thisptr = JpmcdsMakeTCurve(forward_date_c, yc._dates, rates, yc._ninstr,
                                       <double>1, self._thisptr.fDayCountConv)
        return yc

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
        Default to True

    """
    @cython.boundscheck(False)
    @cython.wraparound(False)
    def __init__(self, today, YieldCurve yc, start_date, step_in_date,
                 cash_settle_date, list end_dates,
                 double[:] coupon_rates, double[:] upfront_rates,
                 double[:] recovery_rates, bint 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 int n_dates = len(end_dates)
        cdef TDate* end_dates_c = <TDate*>malloc(n_dates * sizeof(TDate))
        cdef TCurve* curve = NULL
        cdef size_t i
        if cash_settle_date_c < yc._thisptr.fBaseDate:
            raise ValueError("cash_settle_date: {0} is anterior to yc's base_date: {1}".
                             format(cash_settle_date, yc.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")
        cdef long dc
        with nogil:
            JpmcdsStringToDayCountConv('ACT/360', &dc)
            curve = JpmcdsCleanSpreadCurve(today_c,
                                           yc._thisptr,
                                           start_date_c,
                                           step_in_date_c,
                                           cash_settle_date_c,
                                           n_dates,
                                           end_dates_c,
                                           &coupon_rates[0],
                                           &upfront_rates[0],
                                           NULL,
                                           &recovery_rates[0],
                                           pay_accrued_on_default,
                                           NULL,
                                           dc,
                                           &stub_type,
                                           <long>'M',
                                           b'NONE')
            free(end_dates_c)
        if curve == NULL:
            raise ValueError("Didn't init the survival curve properly")
        else:
            self._thisptr = curve

    survival_probability = Curve.__forward_zero_price

    @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

        sc._thisptr = JpmcdsMakeTCurve(base_date_c, &max_date, &rate, 1,
                                       <double>basis, dcc(day_count_conv))
        return sc