diff options
Diffstat (limited to 'python/analytics')
| -rw-r--r-- | python/analytics/basket_index.py | 4 | ||||
| -rw-r--r-- | python/analytics/tranche_basket.py | 61 |
2 files changed, 52 insertions, 13 deletions
diff --git a/python/analytics/basket_index.py b/python/analytics/basket_index.py index 0347a224..7e70866f 100644 --- a/python/analytics/basket_index.py +++ b/python/analytics/basket_index.py @@ -133,7 +133,9 @@ class MarkitBasketIndex(BasketIndex): unstack(). groupby(level='date', as_index=False).nth(0). reset_index(['index', 'series', 'version'], drop=True)) - self.index_quotes.columns = self.index_desc.loc[self.index_quotes.columns, "maturity"] + 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): diff --git a/python/analytics/tranche_basket.py b/python/analytics/tranche_basket.py index ad96cb59..9d1dbbda 100644 --- a/python/analytics/tranche_basket.py +++ b/python/analytics/tranche_basket.py @@ -1,17 +1,17 @@ from .basket_index import BasketIndex from .db import _engine -from .tranche_functions import credit_schedule, adjust_attachments, cds_accrued +from .tranche_functions import ( + credit_schedule, adjust_attachments, cds_accrued, GHquad) from index_data import get_singlenames_curves, get_tranche_quotes from pyisda.cdsone import upfront_charge from pandas.tseries.offsets import BDay - +from scipy.optimize import minimize_scalar import pandas as pd import numpy as np class TrancheBasket(BasketIndex): def __init__(self, index_type: str, series: int, tenor: str, *, trade_date: pd.Timestamp=pd.Timestamp.today().normalize()): - print(trade_date) super().__init__(index_type, series, [tenor], trade_date=trade_date) self.tranche_quotes = get_tranche_quotes(index_type, series, tenor, trade_date.date()) index_desc = self.index_desc.reset_index('maturity').set_index('tenor') @@ -58,8 +58,8 @@ class TrancheBasket(BasketIndex): self._Z, self._w = GHquad(self._Ngh) def _get_quotes(self): - refprice = self.tranche_quotes[0]['indexrefprice'] - refspread = self.tranche_quotes[0]['indexrefspread'] + refprice = self.tranche_quotes.indexrefprice.iat[0] + refspread = self.tranche_quotes.indexrefspread.iat[0] if refprice is not None: return {self.maturity: 1 - refprice / 100} if refspread is not None: @@ -77,9 +77,9 @@ class TrancheBasket(BasketIndex): return 1 - super().survival_matrix(self.cs.index.values.astype('M8[D]').view('int') + 134774) def tranche_legs(self, K, rho, complement=False): - if ((K == 0 and not complement) or (K == 1 and complement)): - return {'cl': 0., 'pl': 0.} - elif ((K == 1 and not complement) or (K == 0 and complement)): + if ((K == 0. and not complement) or (K == 1. and complement)): + return 0., 0. + elif ((K == 1. and not complement) or (K == 0. and complement)): return BCindexpv(index) else: L, R = BCloss_recov_dist(self.default_prob, @@ -88,8 +88,45 @@ class TrancheBasket(BasketIndex): rho, self._Z, self._w, self._Ngrid) if complement: - return {'cl': tranche_cl(L, R, self.cs, K, 1), - 'pl': tranche_pl(L, self.cs, K, 1)} + return tranche_cl(L, R, self.cs, K, 1.), tranche_pl(L, self.cs, K, 1.) else: - return {'cl': tranche_cl(L, R, self.cs, 0, K), - 'pl': tranche_pl(L, self.cs, 0, K)} + return tranche_cl(L, R, self.cs, 0., K), tranche_pl(L, self.cs, 0., K) + + def index_pv(self): + ELvec = self.weights @ self.default_prob + def build_skew(self, skew_type="bottomup"): + assert(skew_type == "bottomup" or skew_type == "topdown") + rhovec = np.full(self.K.size, np.nan) + dK = np.diff(self.K) + + def aux(rho, obj, K, quote, spread, complement): + cl, pl = obj.tranche_legs(K, rho, complement) + return abs(pl + cl * spread + quote) + + if skew_type == "bottomup": + for j in range(len(dK) - 1): + cl, pl = self.tranche_legs(self.K[j], rhovec[j]) + q = self.tranche_quotes.quotes.iat[j] * dK[j] - \ + pl - cl * self.tranche_quotes.running.iat[j] + res = minimize_scalar(aux, + bounds=(0, 1), + args=(self, self.K[j+1], q, self.tranche_quotes.running.iat[j], False) ) + if res.success: + rhovec[j+1] = res.x + else: + print(res.message) + break + elif skew_type == "topdown": + for j in range(len(dK) - 1, 0, -1): + cl, pl = self.tranche_legs(self.K[j+1], rhovec[j+1]) + q = self.tranche_quotes.quotes.iat[j] * dK[j] - \ + pl - cl * self.tranche_quotes.running.iat[j] + res = minimize_scalar(aux, + bounds=(0, 1), + args=(self, self.K[j], q, self.tranche_quotes.running.iat[j], False) ) + if res.success: + rhovec[j+1] = res.x + else: + print(res.message) + break + return rhovec |