diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/analytics/tranche_basket.py | 118 |
1 files changed, 110 insertions, 8 deletions
diff --git a/python/analytics/tranche_basket.py b/python/analytics/tranche_basket.py index bd6a5b34..4b3e90f0 100644 --- a/python/analytics/tranche_basket.py +++ b/python/analytics/tranche_basket.py @@ -1,12 +1,10 @@ from .basket_index import BasketIndex -from .db import _engine from .tranche_functions import ( credit_schedule, adjust_attachments, GHquad, BCloss_recov_dist, BCloss_recov_trunc, tranche_cl, tranche_pl) -from .index_data import get_singlenames_curves, get_tranche_quotes +from .index_data import get_tranche_quotes from collections import namedtuple from copy import deepcopy -from pandas.tseries.offsets import BDay from pyisda.date import cds_accrued from scipy.optimize import brentq from scipy.interpolate import CubicSpline, PchipInterpolator @@ -15,6 +13,110 @@ import datetime import pandas as pd import numpy as np + +class DualCorrTranche(BasketIndex): + def __init__(self, index_type: str, series: int, tenor: str, *, + attach: float, detach: float, corr_attach: float, + corr_detach: float, tranche_running: float, + notional: float=10_000_000, + value_date: pd.Timestamp=pd.Timestamp.today().normalize()): + super().__init__(index_type, series, [tenor], value_date=value_date) + self.tenor = tenor + self.K_orig = np.array([attach, detach]) / 100 + self.attach, self.detach = attach, detach + self.K = adjust_attachments(self.K_orig, self.cumloss, self.factor) + self._Ngh = 250 + self._Ngrid = 201 + self._Z, self._w = GHquad(self._Ngh) + self.rho = np.array([corr_attach, corr_detach]) + self.notional = notional + self.tranche_running = tranche_running + self._direction = -1. if notional > 0 else 1. + self.start_date, self.cs = credit_schedule(value_date, self.tenor[:-1], + 1., self.yc) + self.default_prob, _ = super().survival_matrix(self.cs.index.values.astype('M8[D]'). + view('int') + 134774) + self._accrued = cds_accrued(value_date, self.tranche_running * 1e-4) + + value_date = property(BasketIndex.value_date.__get__) + + @value_date.setter + def value_date(self, d: pd.Timestamp): + BasketIndex.value_date.__set__(self, d) + self.start_date, self.cs = credit_schedule(d, self.tenor[:-1], + 1., self.yc) + self.default_prob, _ = super().survival_matrix(self.cs.index.values.astype('M8[D]'). + view('int') + 134774) + self._accrued = cds_accrued(d, self.tranche_running * 1e-4) + + def tranche_legs(self, K, rho): + if K == 0.: + return 0., 0. + elif K == 1.: + return self.index_pv()[:-1] + elif np.isnan(rho): + raise ValueError("rho needs to be a real number between 0. and 1.") + else: + L, R = BCloss_recov_dist(self.default_prob.values, + self.weights, + self.recovery_rates, + rho, + self._Z, self._w, self._Ngrid) + Legs = namedtuple('TrancheLegs', 'coupon_leg, protection_leg') + return Legs(tranche_cl(L, R, self.cs, 0., K), tranche_pl(L, self.cs, 0., K)) + + @property + def direction(self): + if self._direction == -1.: + return "Buyer" + else: + return "Seller" + + @direction.setter + def direction(self, d): + if d == "Buyer": + self._direction = -1. + elif d == "Seller": + self._direction = 1. + else: + raise ValueError("Direction needs to be either 'Buyer' or 'Seller'") + + @property + def pv(self): + """ computes coupon leg, protection leg and bond price. + + coupon leg is *dirty*. + bond price is *clean*.""" + cl = np.zeros(2) + pl = np.zeros(2) + + i = 0 + for rho, k in zip(self.rho, self.K): + cl[i], pl[i] = self.tranche_legs(k, rho) + i += 1 + dK = np.diff(self.K) + pl = np.diff(pl) / dK + cl = np.diff(cl) / dK * self.tranche_running * 1e-4 + bp = 1 + pl + cl - self._accrued + Pvs = namedtuple('TranchePvs', 'coupon_leg, protection_leg, bond_price') + return Pvs(cl, pl, bp) + + def reset_pv(self): + self._original_pv = self.pv.bond_price + self._trade_date = self._value_date + + def pnl(self): + if self._original_pv is None: + raise ValueError("original pv not set") + else: + # TODO: handle factor change + days_accrued = (self.value_date - self._trade_date).days / 360 + return self.notional * self._direction * (self.pv.bond_price - self._original_pv + + self.tranche_running * days_accrued) + + def shock(self, params=['pnl'], *, corr_shock, kwargs): + pass + class TrancheBasket(BasketIndex): def __init__(self, index_type: str, series: int, tenor: str, *, value_date: pd.Timestamp=pd.Timestamp.today().normalize()): @@ -214,10 +316,10 @@ class TrancheBasket(BasketIndex): if rho is None: rho = expit(self._skew(logit(K))) L, _ = BCloss_recov_dist(self.default_prob.values[:,-(1+shortened),np.newaxis], - self.weights, - self.recovery_rates, - rho, - self._Z, self._w, self._Ngrid) + self.weights, + self.recovery_rates, + rho, + self._Z, self._w, self._Ngrid) p = np.cumsum(L) support = np.linspace(0, 1, self._Ngrid) probfun = PchipInterpolator(support, p) @@ -331,7 +433,7 @@ class TrancheBasket(BasketIndex): else: newrho = expit(index1._skew(logit(x))) assert newrho >= 0 and newrho <= 1, "Something went wrong" - return self.expected_loss_trunc(x, rho=newrho) /el1 - \ + return self.expected_loss_trunc(x, rho=newrho) / el1 - \ index2.expected_loss_trunc(K2, newrho, shortened) / el2 def aux2(x, index1, index2, K2, shortened): |