from .index_data import get_index_quotes, get_singlenames_curves from .db import _engine from .utils import tenor_t from functools import partial from pandas.util import hash_pandas_object 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 from scipy.optimize import brentq from pandas.tseries.offsets import Day, BDay from pyisda.cdsone import upfront_charge 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], *, value_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']) if self.index_desc.empty: raise ValueError(f"Index {index_type} {series} doesn't exist") 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] self.index_desc = self.index_desc.sort_values('maturity') self.tenors = {t: m.date() for t, m in self.index_desc.maturity.items()} maturities = self.index_desc.maturity.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, value_date) self.currency = "EUR" if index_type in ["XO", "EU"] else "USD" self.yc = get_curve(value_date, self.currency) self.step_in_date = value_date + Day() self.cash_settle_date = value_date + 3 * BDay() self.tweaks = [] super().__init__(self.issue_date, maturities, curves, value_date=value_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).sum() elif isinstance(v, list): return hash(tuple(v)) elif type(v) is np.ndarray: return hash(v.tobytes()) else: return hash(v) ignore = set(['_Z', '_w', '_skew']) for k, v in vars(self).items(): if k not in ignore: print(k, aux(v)) return hash((CreditIndex.__hash__(self), hash(frozenset([(k, aux(v)) for k, v in vars(self).items() \ if k not in ignore])))) def _query_version(self, i): for lastdate, *data in self._version: if lastdate >= self.value_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 value_date = property(CreditIndex.value_date.__get__) @value_date.setter def value_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.cash_settle_date = d + 3 * BDay() CreditIndex.value_date.__set__(self, d) @property def recovery_rates(self): return np.array([c.recovery_rates[0] for c in self.curves]) 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.cash_settle_date, m, self.yc, 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.cash_settle_date, maturity, self.yc, 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.cash_settle_date, m, self.yc)) return pd.Series(r, index=self.index_desc.tenor, name='duration') else: return super().duration(self.step_in_date, self.cash_settle_date, maturity, self.yc) def theta(self, maturity=None, coupon=None, theta_date=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 theta_date is provided, computes the theta to that specific date instead of one-year theta""" if hasattr(self, "index_quotes"): index_quotes = self._get_quotes() else: index_quotes = {} if maturity is None: r = [] for m in self.maturities: coupon = self.index_desc.coupon[m] index_quote = index_quotes.get(m, np.nan) r.append(super().theta(self.step_in_date, self.cash_settle_date, m, self.yc, coupon, index_quote, theta_date)) 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.cash_settle_date, maturity, self.yc, coupon, np.nan, theta_date) 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 in self.maturities: if m not in quotes: self.tweaks.append(np.nan) continue else: index_quote = quotes[m] 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(f"couldn't calibrate for date: {self.value_date} and maturity: {m}") self.tweaks.append(np.NaN) continue self.tweaks.append(eps) self.tweak_portfolio(eps, m) def _snacpv(self, spread, coupon, recov): return upfront_charge(self.value_date, self.cash_settle_date, self.start_date, self.step_in_date, self.start_date, self.maturity, coupon, self.yc, spread, recov) class MarkitBasketIndex(BasketIndex): def __init__(self, index_type: str, series: int, tenors: List[str], *, value_date: pd.Timestamp=pd.Timestamp.today().normalize() - BDay()): super().__init__(index_type, series, tenors, value_date=value_date) self.index_quotes = (get_index_quotes(index_type, series, tenors, years=None, remove_holidays=False)[['close_price', 'id']]. groupby(level=['date', 'tenor'], as_index=True). nth(0)) self.index_quotes.close_price = 1 - self.index_quotes.close_price / 100 def _get_quotes(self): quotes = self.index_quotes.loc[self.value_date, "close_price"] return {self.tenors[t]: q for t, q in quotes.items()} 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)