path: root/python/analytics/index.py

import array
import datetime
import math
import pandas as pd

from pyisda.legs import ContingentLeg, FeeLeg
from quantlib.settings import Settings
from quantlib.time.api import (
    Date, Schedule, WeekendsOnly, CDS, Following,
    Unadjusted, Period, pydate_from_qldate )
from termcolor import colored
from pandas.tseries.offsets import BDay
from dates import prev_immdate
from db import dbconn
from psycopg2 import DataError
from pyisda.curve import SpreadCurve
from yieldcurve import YC, ql_to_jp, roll_yc, rate_helpers
from quantlib.time.api import Actual365Fixed

serenitasdb  = dbconn('serenitasdb')

class Index():
    """ minimal class to represent a credit index """
    def __init__(self, start_date, end_date, recovery, fixed_rate,
                 notional = 10e6):
        """
        start_date : :class:`datetime.date`
            index start_date (Could be issue date, or last imm date)
        end_date : :class:`datetime.date`
            index last date
        recovery :
            recovery rate (between 0 and 1)
        fixed_rate :
            fixed coupon (in bps)
        """
        self.fixed_rate = fixed_rate
        self.notional = notional
        self._sched = Schedule(Date.from_datetime(start_date),
                               Date.from_datetime(end_date),
                               Period("3M"),
                               WeekendsOnly(),
                               Following,
                               Unadjusted,
                               CDS)
        self._start_date = start_date
        self._end_date = end_date
        self.recovery = recovery

        self._fee_leg = FeeLeg(self._start_date, end_date, True, 1, 1)
        self._default_leg = ContingentLeg(self._start_date, end_date, 1)
        self._trade_date = None
        self._yc = None
        self._risky_annuity = None
        self._spread = None
        self.name = None

    @property
    def start_date(self):
        return self._start_date

    @property
    def end_date(self):
        return self._end_date

    @start_date.setter
    def start_date(self, d):
        self._fee_leg = FeeLeg(d, self.end_date, True, 1, 1)
        self._default_leg = ContingentLeg(d, self.end_date, 1)
        self._start_date = d
        self._sched = Schedule(Date.from_datetime(d),
                               Date.from_datetime(self.end_date),
                               Period("3M"),
                               WeekendsOnly(),
                               Following,
                               Unadjusted,
                               CDS)

    @end_date.setter
    def end_date(self, d):
        self._fee_leg = FeeLeg(self.start_date, d, True, 1, 1)
        self._default_leg = ContingentLeg(self.start_date, d, 1)
        self._end_date = d
        self._sched = Schedule(self.start_date,
                               d,
                               Period("3M"),
                               WeekendsOnly(),
                               Following,
                               Unadjusted,
                               CDS)

    def survival_probability(self, d):
        if d > self.trade_date:
            return math.exp( - self.flat_hazard * (d - self.trade_date).days/365)
        else:
            return 1

    def forward_pv(self, exercise_date):
        """This is default adjusted forward price at time exercise_date"""
        step_in_date = exercise_date + datetime.timedelta(days=1)
        a = self._fee_leg.pv(self.trade_date, step_in_date, self._value_date,
                             self._yc, self._sc, False)
        Delta = self._fee_leg.accrued(step_in_date)
        value_date = (pd.Timestamp(exercise_date) + 3* BDay()).date()
        df = self._yc.discount_factor(value_date)
        q = self.survival_probability(exercise_date)
        clean_forward_annuity = a - Delta * df * q
        dl_pv = self._default_leg.pv(
            self.trade_date, step_in_date, self._value_date,
            self._yc, self._sc, self.recovery)
        forward_price = self.notional * (dl_pv - clean_forward_annuity * self.fixed_rate*1e-4)
        fep = self.notional * (1 - self.recovery) * (1 - q)
        return forward_price * self._yc.discount_factor(self._value_date) / df + fep

    @property
    def spread(self):
        return self._spread * 1e4

    def _update(self):
        self._sc = SpreadCurve(self.trade_date, self._yc, self.start_date,
                               self._step_in_date, self._value_date,
                               [self.end_date], array.array('d', [self._spread]),
                               self.recovery)
        self._risky_annuity = self._fee_leg.pv(self.trade_date, self._step_in_date,
                                               self._value_date, self._yc,
                                               self._sc, False)
        self._dl_pv = self._default_leg.pv(
            self.trade_date, self._step_in_date, self._value_date,
            self._yc, self._sc, self.recovery)
        self._pv = self._dl_pv - self._risky_annuity * self.fixed_rate * 1e-4
        self._clean_pv = self._pv + self._accrued * self.fixed_rate * 1e-4
        self._price = 100 * (1 - self._clean_pv)

    @spread.setter
    def spread(self, s: float):
        """ s: spread in bps """
        self._spread = s * 1e-4
        self._update()

    @property
    def flat_hazard(self):
        sc_data = self._sc.inspect()['data']
        ## conversion to continuous compounding
        return math.log(1 + sc_data[0][1])

    @property
    def pv(self):
        return self.notional * self._pv

    @property
    def accrued(self):
        return - self.notional * self._accrued * self.fixed_rate * 1e-4

    @property
    def days_accrued(self):
        return int(self._accrued * 360)

    @property
    def clean_pv(self):
        return self.notional * self._clean_pv

    @property
    def price(self):
        return self._price

    @price.setter
    def price(self, val):
        pass

    @property
    def DV01(self):
        old_pv = self.pv
        self.spread += 1
        dv01 = self.pv - old_pv
        self.spread -= 1
        return dv01

    @property
    def IRDV01(self):
        old_pv = self.pv
        old_yc = self._yc
        for rh in self._helpers:
            rh.quote += 1e-4
        self._yc = ql_to_jp(self._ql_yc)
        self._update() ## to force recomputation
        new_pv = self.pv
        for r in self._helpers:
            r.quote -= 1e-4
        self._yc = old_yc
        self._update()
        return new_pv - old_pv

    @property
    def rec_risk(self):
        old_pv = self.pv
        old_recovery = self.recovery
        self.recovery = old_recovery - 0.01
        self._update()
        pv_minus = self.pv
        self.recovery = old_recovery + 0.01
        self._update()
        pv_plus = self.pv
        self.recovery = old_recovery
        self._update()
        return (pv_plus - pv_minus)/2

    @property
    def jump_to_default(self):
        return self.notional * (1 - self.recovery) - self.clean_pv

    @property
    def risky_annuity(self):
        return self._risky_annuity - self._accrued

    @property
    def trade_date(self):
        if self._trade_date is None:
            raise AttributeError('Please set trade_date first')
        else:
            return self._trade_date

    @trade_date.setter
    def trade_date(self, d):
        settings = Settings()
        settings.evaluation_date = Date.from_datetime(d)
        self.start_date = pydate_from_qldate(
            self._sched.previous_date(settings.evaluation_date))
        self._helpers = rate_helpers(self.currency)
        self._ql_yc = YC(self._helpers)
        self._yc = ql_to_jp(self._ql_yc)
        self._trade_date = d
        self._step_in_date = self.trade_date + datetime.timedelta(days=1)
        self._accrued = self._fee_leg.accrued(self._step_in_date)
        self._value_date = (pd.Timestamp(self._trade_date) + 3* BDay()).date()
        if self._spread is not None:
            self._update()

    @classmethod
    def from_name(cls, index, series, tenor, trade_date = datetime.date.today(),
                  notional = 10e6):
        try:
            with serenitasdb.cursor() as c:
                c.execute("SELECT maturity, coupon FROM index_maturity " \
                          "WHERE index=%s AND series=%s AND tenor = %s",
                          (index.upper(), series, tenor))
                maturity, coupon = next(c)
        except DataError as e:
            raise
        else:
            recovery = 0.4 if index.lower() == "ig" else 0.3
            instance = cls(trade_date, maturity, recovery, coupon)
            instance.name = "MARKIT CDX.NA.{}.{} {:%m/%y} ".format(
                index.upper(),
                series,
                maturity)
            if index.upper() in ["IG", "HY"]:
                instance.currency = "USD"
            else:
                instance.currency = "EUR"
            instance.notional = notional
            instance.trade_date = trade_date
            return instance

    def __repr__(self):
        if self.days_accrued > 1:
            accrued_str = "Accrued ({} Days)".format(self.days_accrued)
        else:
            accrued_str = "Accrued ({} Day)".format(self.days_accrued)
        s = ["{:<20}\t{:>15}".format("CDS Index", colored(self.name, attrs = ['bold'])),
             "",
             "{:<20}\t{:>15}".format("Trade Date", ('{:%m/%d/%y}'.
                                                    format(self.trade_date))),
             "{:<20}\t{:>15.2f}\t\t{:<20}\t{:>10,.2f}".format("Trd Sprd (bp)",
                                                              self.spread,
                                                              "Coupon (bp)",
                                                              self.fixed_rate),
             "{:<20}\t{:>15.2f}\t\t{:<20}\t{:>10}".format("1st Accr Start",
                                                          self.spread,
                                                          "Payment Freq",
                                                          "Quarterly"),
             "{:<20}\t{:>15}\t\t{:<20}\t{:>10.2f}".format("Maturity Date",
                                                          ('{:%m/%d/%y}'.
                                                           format(self.end_date)),
                                                          "Rec Rate",
                                                          self.recovery),
             "{:<20}\t{:>15}\t\t{:<20}\t{:>10}".format("Bus Day Adj",
                                                       "Following",
                                                       "Day Count",
                                                       "ACT/360"),
             "",
             colored("Calculator", attrs = ['bold']),
             "{:<20}\t{:>15}".format("Valuation Date", ('{:%m/%d/%y}'.
                                                        format(self.trade_date))),
             "{:<20}\t{:>15}".format("Cash Settled On", ('{:%m/%d/%y}'.
                                                         format(self._value_date))),
             "",
             "{:<20}\t{:>15.8f}\t\t{:<20}\t{:>10,.2f}".format("Price",
                                                              self.price,
                                                              "Spread DV01",
                                                              self.DV01),
             "{:<20}\t{:>15,.0f}\t\t{:<20}\t{:>10,.2f}".format("Principal",
                                                               self.clean_pv,
                                                               "IR DV01",
                                                               self.IRDV01),
             "{:<20}\t{:>15,.0f}\t\t{:<20}\t{:>10,.2f}".format(accrued_str,
                                                               self.accrued,
                                                               "Rec Risk (1%)",
                                                               self.rec_risk),
             "{:<20}\t{:>15,.0f}\t\t{:<20}\t{:>10,.0f}".format("Cash Amount",
                                                               self.pv,
                                                               "Def Exposure",
                                                               self.jump_to_default)
        ]
        return "\n".join(s)