diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/risk/tranches.py | 64 |
1 files changed, 64 insertions, 0 deletions
diff --git a/python/risk/tranches.py b/python/risk/tranches.py index 4647ce0c..2040b494 100644 --- a/python/risk/tranches.py +++ b/python/risk/tranches.py @@ -1,8 +1,11 @@ import numpy as np from pyisda.date import cds_accrued from serenitas.analytics.api import Portfolio, DualCorrTranche +from serenitas.analytics.index import BasketIndex +from serenitas.analytics.index_data import hist_skews, on_the_run from serenitas.analytics.dates import prev_business_day from serenitas.analytics.utils import get_fx +from serenitas.analytics.yieldcurve import get_curve, hist_curves import logging logger = logging.getLogger(__name__) @@ -243,3 +246,64 @@ def insert_tranche_risk(portf, conn): ) conn.commit() + + +def skew_diff(prev_skew, curr_skew) -> dict[str, int]: + r = {} + for index, skews in curr_skew.items(): + prev_skews = prev_skew[index] + for serie, (otr, skew) in skews.items(): + if serie in prev_skews: + dS = skew - prev_skews[serie][1] + r[index, otr] = dS + return r + + +def VaR(conn, portf: Portfolio, quantile=0.05, years: int = 5): + """let's do IR VaR for now""" + value_date = portf.value_date + curves = ( + hist_curves("USD", delta_in_years=years), + hist_curves("EUR", delta_in_years=years), + ) + skew_hist = hist_skews(conn, end_date=value_date, lookback=years) + yc_hist = [] + for (d1, c1), (d2, c2) in zip(*curves): + if d1 != d2: + raise ValueError() + else: + yc_hist.append((d1, c1, c2)) + + orig_usd, orig_eur = get_curve(value_date, "USD"), get_curve(value_date, "EUR") + l = [] + d = [] + otr = {k: on_the_run(k, value_date) for k in ("EU", "XO", "IG", "HY")} + skew_prev = yc_hist[0][0] + for prev, curr in zip(yc_hist, yc_hist[1:]): + d_curr, curr_usd, curr_eur = curr + d_prev, prev_usd, prev_eur = prev + if not skew_hist[d_curr]: + dS = None + skew_prev = d_prev + else: + dS = skew_diff(skew_hist[skew_prev], skew_hist[d_curr]) + skew_prev = d_curr + print(d_curr) + usd_shock = curr_usd - prev_usd + eur_shock = curr_eur - prev_eur + for k, v in BasketIndex._cache.items(): + if k[0] in ("HY", "IG"): + v.yc = orig_usd + usd_shock + elif k[0] in ("XO", "EU"): + v.yc = orig_eur + eur_shock + for t in portf.trades: + if dS is not None: + for i in range(3): + v = otr[t.index_type] - t.series - i + if (t.index_type, v) in dS: + t.rho = (t._skew + dS[t.index_type, v])(t.moneyness) + break + t.cs.df = t._index.yc.discount_factor(t.cs.payment_dates) + l.append(portf.pv) + d.append(d_curr) + return l, d |