import analytics import array import datetime import pandas as pd from .credit_default_swap import CreditDefaultSwap from . import serenitas_engine, dawn_engine, DataError try: from bbg_helpers import BBG_IP, retrieve_data, init_bbg_session except ModuleNotFoundError: pass from pandas.tseries.offsets import BDay from pyisda.curve import SpreadCurve from pyisda.date import previous_twentieth from termcolor import colored from .utils import build_table def g(index, spread, exercise_date, pv=None): """computes the strike clean price using the expected forward yield curve. """ step_in_date = exercise_date + datetime.timedelta(days=1) exercise_date_settle = pd.Timestamp(exercise_date) + 3 * BDay() if spread is None and index._sc is not None: sc = index._sc prot = index._default_leg.pv( exercise_date, step_in_date, exercise_date_settle, index._yc, index._sc, index.recovery, ) else: rates = array.array("d", [spread * 1e-4]) upfront = 0.0 if pv is None else pv sc = SpreadCurve( exercise_date, index._yc, index.start_date, step_in_date, exercise_date_settle, [index.end_date], rates, array.array("d", [upfront]), array.array("d", [index.recovery]), ) a = index._fee_leg.pv( exercise_date, step_in_date, exercise_date_settle, index._yc, sc, True ) if pv is not None: return 1e4 * pv / a + spread else: if spread is None: return prot - a * index.fixed_rate * 1e-4 else: return (spread - index.fixed_rate) * a * 1e-4 class CreditIndex(CreditDefaultSwap): __slots__ = ( "_indic", "_version", "_cumloss", "index_type", "series", "tenor", "_quote_is_price", ) def __init__( self, index_type=None, series=None, tenor=None, value_date=datetime.date.today(), notional=10_000_000, redcode=None, maturity=None, ): if all([redcode, maturity]): r = serenitas_engine.execute( "SELECT index, series, tenor FROM index_desc " "WHERE redindexcode=%s AND maturity = %s", (redcode, maturity), ) index_type, series, tenor = next(r) if all([index_type, series, tenor]): sql_str = ( "SELECT indexfactor, lastdate, maturity, coupon, " "issue_date, version, cumulativeloss " "FROM index_desc WHERE index=%s AND series=%s AND tenor = %s " "ORDER BY lastdate ASC" ) params = (index_type.upper(), series, tenor) else: raise ValueError("Not enough information to load the index.") try: df = pd.read_sql_query( sql_str, serenitas_engine, parse_dates=["lastdate", "issue_date"], params=params, ) maturity = df.maturity[0] coupon = df.coupon[0] if tenor is None: tenor = df.tenor[0] index_type = index_type.upper() if index_type else df.loc[0, "index"] series = series if series else df.series.iat[0] df.loc[df.lastdate.isnull(), "lastdate"] = maturity except DataError as e: print(e) return None else: recovery = 0.3 if index_type == "HY" else 0.4 super().__init__( previous_twentieth(value_date), maturity, recovery, coupon, notional, df.issue_date[0], ) self._quote_is_price = index_type == "HY" self._indic = tuple( (ld, factor / 100, cumloss, version) for ld, factor, cumloss, version in ( df[ ["lastdate", "indexfactor", "cumulativeloss", "version"] ].itertuples(index=False) ) ) self.index_type = index_type self.series = series self.tenor = tenor tenor = tenor.upper() if tenor.endswith("R"): tenor = tenor[:-1] if index_type in ("IG", "HY"): self.name = "CDX {} CDSI S{} {}".format(index_type, series, tenor) elif index_type == "EU": self.name = f"ITRX EUR CDSI S{series} {tenor}" elif index_type == "XO": self.name = f"ITRX XOVER CDSI S{series} {tenor}" if index_type in ("IG", "HY"): self.currency = "USD" else: self.currency = "EUR" self.value_date = value_date @classmethod def from_tradeid(cls, trade_id): r = dawn_engine.execute( """ SELECT index, series, tenor, trade_date, notional, security_desc, protection, upfront FROM cds LEFT JOIN index_desc ON security_id = redindexcode AND cds.maturity = index_desc.maturity WHERE id=%s""", (trade_id,), ) rec = r.fetchone() if rec is None: raise ValueError(f"No index trade for id: {trade_id}") instance = cls(rec.index, rec.series, rec.tenor, rec.trade_date, rec.notional) instance.name = rec.security_desc instance.direction = rec.protection instance.value_date = rec.trade_date instance.pv = rec.upfront instance.reset_pv() return instance @property def hy_equiv(self): try: ontr = analytics._ontr except AttributeError: return float("nan") risk = self.notional * self.risky_annuity / ontr.risky_annuity if self.index_type != "HY": risk *= analytics._beta[self.index_type] return risk @property def ref(self): if self._quote_is_price: return self.price else: return self.spread @ref.setter def ref(self, val): if self._quote_is_price: self.price = val else: self.spread = val def mark(self, **kwargs): if "ref" in kwargs: self.ref = kwargs["ref"] return if self.value_date == datetime.date.today(): with init_bbg_session(BBG_IP) as session: security = self.name + " Corp" field = "PX_LAST" ref_data = retrieve_data(session, [security], field) self.ref = ref_data[security][field] else: run = serenitas_engine.execute( """SELECT * FROM index_quotes WHERE index=%s AND series=%s AND tenor=%s AND date=%s""", (self.index_type, self.series, self.tenor, self.value_date), ) rec = run.fetchone() self.spread = rec.closespread value_date = property(CreditDefaultSwap.value_date.__get__) @value_date.setter def value_date(self, d): CreditDefaultSwap.value_date.__set__(self, d) for lastdate, factor, cumloss, version in self._indic: if lastdate >= self.value_date: self._factor = factor self._version = version self._cumloss = cumloss break else: self._factor = 1.0 self._version = 1 @property def factor(self): return self._factor @property def version(self): return self._version @property def cumloss(self): return self._cumloss def __repr__(self): if not self.spread: raise ValueError("Market spread is missing!") if self.days_accrued > 1: accrued_str = "Accrued ({} Days)".format(self.days_accrued) else: accrued_str = "Accrued ({} Day)".format(self.days_accrued) s = [ "{:<20}\tNotional {:>5.2f}MM {}\tFactor {:>28.5f}".format( "Buy Protection" if self.notional > 0.0 else "Sell Protection", abs(self.notional) / 1_000_000, self.currency, self._factor, ), "{:<20}\t{:>15}".format("CDS Index", colored(self.name, attrs=["bold"])), "", ] rows = [ ["Trd Sprd (bp)", self.spread, "Coupon (bp)", self.fixed_rate], ["1st Accr Start", self.issue_date, "Payment Freq", "Quarterly"], ["Maturity Date", self.end_date, "Rec Rate", self.recovery], ["Bus Day Adj", "Following", "DayCount", "ACT/360"], ] format_strings = [ [None, "{:.2f}", None, "{:.0f}"], [None, "{:%m/%d/%y}", None, None], [None, "{:%m/%d/%y}", None, None], [None, None, None, None], ] s += build_table(rows, format_strings, "{:<20}{:>19}\t\t{:<20}{:>15}") s += ["", colored("Calculator", attrs=["bold"])] rows = [ ["Valuation Date", self.value_date], ["Cash Settled On", self._cash_settle_date], ] format_strings = [[None, "{:%m/%d/%y}"], [None, "{:%m/%d/%y}"]] s += build_table(rows, format_strings, "{:<20}\t{:>15}") s += [""] rows = [ ["Price", self.price, "Spread DV01", self.DV01], ["Principal", self.clean_pv, "IR DV01", self.IRDV01], [accrued_str, self.accrued, "Rec Risk (1%)", self.rec_risk], ["Cash Amount", self.pv, "Def Exposure", self.jump_to_default], ] format_strings = [ [None, "{:.8f}", None, "{:,.2f}"], [None, "{:,.0f}", None, "{:,.2f}"], [None, "{:,.0f}", None, "{:,.2f}"], [None, "{:,.0f}", None, "{:,.0f}"], ] s += build_table(rows, format_strings, "{:<20}{:>19}\t\t{:<20}{:>15}") return "\n".join(s) class ForwardIndex: __slots__ = ( "index", "forward_date", "exercise_date_settle", "df", "_forward_annuity", "_forward_pv", "_forward_spread", "__weakref__", ) def __init__(self, index, forward_date, observer=True): self.index = index if isinstance(forward_date, pd.Timestamp): self.forward_date = forward_date.date() else: self.forward_date = forward_date self.exercise_date_settle = pd.Timestamp(forward_date) + 3 * BDay() self._update() if observer: self.index.observe(self) @classmethod def from_name( cls, index_type, series, tenor, forward_date, value_date=datetime.date.today(), notional=10e6, ): index = CreditIndex(index_type, series, tenor, value_date, notional) return cls(index, forward_date) @property def forward_annuity(self): return self._forward_annuity @property def forward_pv(self): return self._forward_pv @property def forward_spread(self): return self._forward_spread * 1e4 @property def ref(self): return self.index.ref @ref.setter def ref(self, val): self.index.ref = val def __hash__(self): return hash(tuple(getattr(self, k) for k in ForwardIndex.__slots__[:-1])) def _update(self, *args): self.df = self.index._yc.discount_factor(self.exercise_date_settle) if self.index.value_date > self.forward_date: raise ValueError( f"Option expired: value_date {self.index.value_date}" f" is greater than forward_date: {self.forward_date}" ) if self.index._sc is not None: step_in_date = self.forward_date + datetime.timedelta(days=1) a = self.index._fee_leg.pv( self.index.value_date, step_in_date, self.index.value_date, self.index._yc, self.index._sc, False, ) Delta = self.index._fee_leg.accrued(step_in_date) q = self.index._sc.survival_probability(self.forward_date) self._forward_annuity = a - Delta * self.df * q self._forward_pv = ( self._forward_annuity * (self.index.spread - self.index.fixed_rate) * 1e-4 ) fep = (1 - self.index.recovery) * (1 - q) self._forward_pv = self._forward_pv / self.df + fep self._forward_spread = ( self.index._spread + fep * self.df / self._forward_annuity ) else: self._forward_annuity, self._forward_pv, self._forward_spread = ( None, None, None, )