from .index_data import get_index_quotes, get_singlenames_curves from .db import _engine from .utils import tenor_t from dateutil.relativedelta import relativedelta from functools import partial from pyisda.credit_index import CreditIndex from typing import List from yieldcurve import get_curve import numpy as np import pandas as pd from math import exp import datetime from scipy.optimize import brentq from pandas.tseries.offsets import Day, BDay def make_index(t, d, args): instance = t.__new__(t) CreditIndex.__init__(instance, *args) instance.__dict__.update(d) return instance class BasketIndex(CreditIndex): index_type: str series: int recovery: float step_in_date: pd.Timestamp value_date: pd.Timestamp tweaks: List[float] def __init__(self, index_type: str, series: int, tenors: List[str], *, trade_date: pd.Timestamp=pd.Timestamp.today().normalize() - BDay()): self.index_type = index_type self.series = series if index_type == 'IG' or index_type == 'EU': self.recovery = 0.4 else: self.recovery = 0.3 self.index_desc = pd.read_sql_query("SELECT tenor, maturity, coupon * 1e-4 AS coupon, " \ "issue_date "\ "FROM index_maturity " \ "WHERE index=%s AND series=%s", _engine, index_col='tenor', params=(index_type, series), parse_dates=['maturity', 'issue_date']) r = _engine.execute("SELECT lastdate, indexfactor/100 AS factor, cumulativeloss/100, version " \ "FROM index_version " \ "WHERE index = %s AND series = %s" \ "ORDER BY lastdate", (index_type, series)) self._version = tuple(tuple(t) for t in r) self.issue_date = self.index_desc.issue_date[0] self.index_desc = self.index_desc.loc[tenors] maturities = self.index_desc.maturity.sort_values().dt.to_pydatetime() self.index_desc = self.index_desc.reset_index().set_index('maturity') self.index_desc.tenor = self.index_desc.tenor.astype(tenor_t) curves = get_singlenames_curves(index_type, series, trade_date) self.currency = "EUR" if index_type in ["XO", "EU"] else "USD" self.yc = get_curve(trade_date, self.currency) self.step_in_date = trade_date + Day() self.value_date = trade_date + 3 * BDay() self.tweaks = [] super().__init__(self.issue_date, maturities, curves, trade_date=trade_date) def __reduce__(self): _, args = CreditIndex.__reduce__(self) d = vars(self) return partial(make_index, self.__class__), (d, args) def __hash__(self): def aux(v): if isinstance(v, pd.DataFrame): return hash_pandas_object(v) elif isinstance(v, list): return hash(tuple(v)) else: return hash(v) hash(CreditIndex.__hash__(self), hash(frozenset([(k, aux(v)) for k, v in dirs(self)]))) def _query_version(self, i): for lastdate, *data in self._version: if lastdate >= self.trade_date: return data[i] @property def factor(self): return self._query_version(0) @property def cumloss(self): return self._query_version(1) @property def version(self): return self._query_version(2) def _get_quotes(self): pass trade_date = property(CreditIndex.trade_date.__get__) @trade_date.setter def trade_date(self, d: pd.Timestamp): self.curves = get_singlenames_curves(self.index_type, self.series, d) self.yc = get_curve(d, self.currency) self.step_in_date = d + Day() self.value_date = d + 3 * BDay() CreditIndex.trade_date.__set__(self, d) def pv(self, maturity=None, epsilon=0., coupon=None): if maturity is None: r = [] for m in self.maturities: coupon = self.index_desc.coupon[m] r.append(super().pv(self.step_in_date, self.value_date, m, self.yc, self.recovery, coupon, epsilon)) return pd.Series(r, index=self.index_desc.tenor, name='pv') else: if coupon is None: try: coupon = self.index_desc.coupon[maturity] except KeyError: raise ValueError("Non standard maturity: coupon must be provided") return super().pv(self.step_in_date, self.value_date, maturity, self.yc, self.recovery, coupon, epsilon) def coupon_leg(self, maturity=None): return self.index_desc.coupon.values * self.duration() def protection_leg(self, maturity=None): return self.pv() + self.coupon_leg() def spread(self, maturity=None, coupon=None): if maturity is None: return (self.index_desc.coupon.values + self.pv() / self.duration()) * 1e4 else: return (coupon + self.pv(maturity, coupon=coupon) / self.duration(maturity)) * 1e4 def duration(self, maturity=None): if maturity is None: r = [] for m in self.maturities: r.append(super().duration(self.step_in_date, self.value_date, m, self.yc)) return pd.Series(r, index=self.index_desc.tenor, name='duration') else: return super().duration(self.step_in_date, self.value_date, maturity, self.yc) def theta(self, maturity=None, coupon=None): """ index thetas if maturity is None, returns a series of theta for all tenors. Otherwise computes the theta for that specific maturity (which needs not be an existing tenor)""" if hasattr(self, "index_quotes"): index_quotes = self.index_quotes.loc[self.trade_date] else: index_quotes = None if maturity is None: r = [] for m in self.maturities: coupon = self.index_desc.coupon[m] index_quote = np.nan if index_quotes is None else index_quotes[m] r.append(super().theta(self.step_in_date, self.value_date, m, self.yc, self.recovery, coupon, index_quote)) return pd.Series(r, index=self.index_desc.tenor, name='theta') else: if coupon is None: try: coupon = self.index_desc.coupon[maturity] except KeyError: raise ValueError("Non standard maturity: coupon must be provided") return super().theta(self.step_in_date, self.value_date, maturity, self.yc, self.recovery, coupon, np.nan) def coupon(self, maturity=None): if maturity is None: return self.index_desc.set_index('tenor').coupon else: return self.index_desc.coupon[maturity] def tweak(self, *args): """ tweak the singlename curves to match index quotes""" quotes = self._get_quotes(*args) self.tweaks = [] for m, index_quote in quotes.items(): lo, hi = -0.3, 0.3 hi_tilde = exp(hi) -1 while hi_tilde < 5: # map range to (-1, +inf) lo_tilde = exp(lo) -1 hi_tilde = exp(hi) -1 try: eps = brentq(lambda epsilon: self.pv(m, epsilon) - index_quote, lo_tilde, hi_tilde) except ValueError: lo *= 1.1 hi *= 1.1 else: break else: print("couldn't calibrate for date: {} and maturity: {}". format(self.trade_date.date(), m.date())) self.tweaks.append(np.NaN) continue self.tweaks.append(eps) self.tweak_portfolio(eps, m) class MarkitBasketIndex(BasketIndex): def __init__(self, index_type: str, series: int, tenors: List[str], *, trade_date: pd.Timestamp=pd.Timestamp.today().normalize() - BDay()): super().__init__(index_type, series, tenors, trade_date=trade_date) self.index_quotes = (get_index_quotes(index_type, series, tenors, years=None)['closeprice']. unstack(). groupby(level='date', as_index=False).nth(0). reset_index(['index', 'series', 'version'], drop=True)) self.index_quotes.columns = (self.index_desc.reset_index(). set_index('tenor'). loc[self.index_quotes.columns, "maturity"]) self.index_quotes = 1 - self.index_quotes / 100 def _get_quotes(self): return self.index_quotes.loc[self.trade_date] if __name__ == "__main__": ig28 = BasketIndex("IG", 28, ["3yr", "5yr", "7yr", "10yr"]) from quantlib.time.api import Schedule, Rule, Date, Period, WeekendsOnly from quantlib.settings import Settings settings = Settings() cds_schedule = Schedule.from_rule(settings.evaluation_date, Date.from_datetime(ig28.maturities[-1]), Period('3M'), WeekendsOnly(), date_generation_rule=Rule.CDS2015) sp = ig28.survival_matrix(cds_schedule.to_npdates().view('int') + 134774)