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, Actual365Fixed 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 .utils import previous_twentieth from yieldcurve import YC, ql_to_jp, roll_yc, rate_helpers serenitasdb = dbconn('serenitasdb') def g(index, spread : float, exercise_date : datetime.date, use_rolled_curve = True): """ computes the strike clean price using the expected forward yield curve """ if use_rolled_curve: rolled_curve = roll_yc(index._yc, exercise_date) else: rolled_curve = index._yc step_in_date = exercise_date + datetime.timedelta(days=1) exercise_date_settle = (pd.Timestamp(exercise_date) + 3* BDay()).date() sc = SpreadCurve(exercise_date, rolled_curve, index.start_date, step_in_date, exercise_date_settle, [index.end_date], array.array('d', [spread * 1e-4]), index.recovery) a = index._fee_leg.pv(exercise_date, step_in_date, exercise_date_settle, rolled_curve, sc, True) return (spread - index.fixed_rate) * a *1e-4 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._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 @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 def forward_annuity(self, exercise_date): step_in_date = exercise_date + datetime.timedelta(days=1) value_date = (pd.Timestamp(exercise_date) + 3* BDay()).date() a = self._fee_leg.pv(self.trade_date, step_in_date, self.trade_date, self._yc, self._sc, False) Delta = self._fee_leg.accrued(step_in_date) df = self._yc.discount_factor(value_date) q = self._sc.survival_probability(exercise_date) return a - Delta * df * q 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.trade_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._sc.survival_probability(exercise_date) clean_forward_annuity = a - Delta * df * q forward_price = self.notional * clean_forward_annuity * (self._spread - self.fixed_rate*1e-4) fep = self.notional * (1 - self.recovery) * (1 - q) return forward_price / 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 = previous_twentieth(d) 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)