aboutsummaryrefslogtreecommitdiffstats
path: root/python/analytics
diff options
context:
space:
mode:
Diffstat (limited to 'python/analytics')
-rw-r--r--python/analytics/tranche_basket.py65
1 files changed, 59 insertions, 6 deletions
diff --git a/python/analytics/tranche_basket.py b/python/analytics/tranche_basket.py
index 53514824..83b69508 100644
--- a/python/analytics/tranche_basket.py
+++ b/python/analytics/tranche_basket.py
@@ -2,12 +2,14 @@ from .basket_index import BasketIndex
from .db import _engine
from .tranche_functions import (
credit_schedule, adjust_attachments, cds_accrued, GHquad, BCloss_recov_dist,
- tranche_cl, tranche_pl)
+ BCloss_recov_trunc, tranche_cl, tranche_pl)
from .index_data import get_singlenames_curves, get_tranche_quotes
from copy import deepcopy
from pyisda.cdsone import upfront_charge
from pandas.tseries.offsets import BDay
from scipy.optimize import brentq
+from scipy.interpolate import CubicSpline
+from scipy.special import logit, expit
import pandas as pd
import numpy as np
@@ -117,20 +119,58 @@ class TrancheBasket(BasketIndex):
bp = 1 + pl + cl
return cl, pl, bp
- def index_pv(self, discounted=True):
- DP = self.default_prob.values
+ def index_pv(self, discounted=True, shortened=0):
+ if shortened > 0:
+ DP = self.default_prob.values[:,-shortened]
+ df = self.cs.df.values[:-shortened]
+ coupons = self.cs.coupons.values[:-shortened]
+ else:
+ DP = self.default_prob.values
+ df = self.cs.df.values
+ coupons = self.cs.coupons
ELvec = self.weights * (1 - self.recovery_rates) @ DP
size = 1 - self.weights @ DP
sizeadj = 0.5 * (np.hstack((1., size[:-1])) + size)
if not discounted:
pl = - ELvec[-1]
- cl = self.cs.coupons.values() @ sizeadj
+ cl = coupons @ sizeadj
else:
- pl = - np.diff(np.hstack((0., ELvec))) @ self.cs.df.values
- cl = self.cs.coupons.values @ (sizeadj * self.cs.df.values)
+ pl = - np.diff(np.hstack((0., ELvec))) @ df
+ cl = coupons @ (sizeadj * df)
bp = 1 + cl * self.coupon(self.maturity) + pl
return cl, pl, bp
+ def expected_loss(self, discounted=True, shortened=0):
+ if shortened > 0:
+ DP = self.default_prob.values[:,:-shortened]
+ df = self.cs.df.values[:-shortened]
+ else:
+ DP = self.default_prob.values
+ df = self.cs.df.values
+
+ ELvec = self.weights * (1 - self.recovery_rates) @ DP
+ if not discounted:
+ return ELvec[-1]
+ else:
+ return np.diff(np.hstack((0., ELvec))) @ df
+
+ def expected_loss_trunc(self, K, rho=None):
+ if rho is None:
+ rho = expit(self._skew(logit(K)))
+ L, _ = BCloss_recov_dist(self.default_prob.values,
+ self.weights,
+ self.recovery_rates,
+ rho,
+ self._Z, self._w, self._Ngrid)
+ Ltrunc, _ = BCloss_recov_trunc(self.default_prob.values,
+ self.weights,
+ self.recovery_rates,
+ rho,
+ K,
+ self._Z, self._w, self._Ngrid)
+ return - tranche_pl(L, self.cs, 0., K), Ltrunc
+
+
@property
def recovery_rates(self):
return np.array([c.recovery_rates[0] for c in self.curves])
@@ -191,3 +231,16 @@ class TrancheBasket(BasketIndex):
else:
print(res.flag)
break
+ self._skew = CubicSpline(logit(self.K[1:-1]),
+ logit(self.rho[1:-1]), bc_type='natural')
+
+ def map_skew(self, index2, method="ATM"):
+ if method not in ["ATM", "TLP", "PM"]:
+ raise ValueError("method needs to be one of 'ATM', 'TLP' or 'PM'")
+
+ if method in ["ATM", "TLP"]:
+ el1 = self.expected_loss()
+ el2 = index2.expected_loss()
+
+ if method == "ATM":
+ K1eq = el1 / el2 * index2.K[1:-1]