from analytics import ATMstrike from joblib import delayed, Parallel import pandas as pd from copy import deepcopy import numpy as np from contextlib import contextmanager from itertools import chain from functools import partial from multiprocessing import Pool from .index_data import _get_singlenames_curves from .curve_trades import curve_shape from scipy.interpolate import RectBivariateSpline def run_swaption_scenarios(swaption, date_range, spread_shock, vol_shock, vol_surface, params=["pv"], vol_time_roll=True): """computes the pv of a swaption for a range of scenarios Parameters ---------- swaption : Swaption date_range : `pandas.Datetime.Index` spread_shock : `np.array` vol_shock : `np.array` vol_surface params : list of strings list of attributes to call on the swaption object. """ swaption = deepcopy(swaption) spreads = swaption.ref * (1 + spread_shock) T = swaption.T r = [] for date in date_range: swaption.index.value_date = date.date() if vol_time_roll: T = swaption.T for s in spreads: swaption.ref = s curr_vol = max(0, float(vol_surface(T, swaption.moneyness))) for vs in vol_shock: swaption.sigma = curr_vol * (1 + vs) r.append([date, s, vs] + [getattr(swaption, p) for p in params]) df = pd.DataFrame.from_records(r, columns=['date', 'spread', 'vol_shock'] + params) return df.set_index('date') def run_index_scenarios(index, date_range, spread_shock, params=['pnl']): index = deepcopy(index) spreads = index.spread * (1 + spread_shock) r = [] for date in date_range: index.value_date = date.date() for s in spreads: index.spread = s r.append([date, s] + [getattr(index, p) for p in params]) df = pd.DataFrame.from_records(r, columns=['date', 'spread'] + params) return df.set_index('date') def _aux(portf, curr_vols, params, vs): for swaption, curr_vol in zip(portf.swaptions, curr_vols): swaption.sigma = curr_vol * (1 + vs) return [vs] + [getattr(portf, p) for p in params] @contextmanager def MaybePool(nproc): yield Pool(nproc) if nproc > 0 else None def run_portfolio_scenarios(portf, date_range, spread_shock, vol_shock, vol_surface, params=["pnl"], nproc=-1, vol_time_roll=True): """computes the pnl of a portfolio for a range of scenarios Parameters ---------- swaption : Swaption date_range : `pandas.Datetime.Index` spread_shock : `np.array` vol_shock : `np.array` vol_surface : VolSurface params : list of strings list of attributes to call on the Portfolio object. nproc : int if nproc > 0 run with nproc processes. """ portf = deepcopy(portf) spreads = np.hstack([index.spread * (1 + spread_shock) for index in portf.indices]) t = [swaption.T for swaption in portf.swaptions] r = [] with MaybePool(nproc) as pool: pmap = pool.map if pool else map for date in date_range: portf.value_date = date.date() if vol_time_roll: t = [swaption.T for swaption in portf.swaptions] for s in spreads: portf.spread = s mon = [swaption.moneyness for swaption in portf.swaptions] curr_vols = np.maximum(vol_surface.ev(t, mon), 0) temp = pmap(partial(_aux, portf, curr_vols, params), vol_shock) r.append([[date, s] + rec for rec in temp]) df = pd.DataFrame.from_records(chain(*r), columns=['date', 'spread', 'vol_shock'] + params) return df.set_index('date') def run_tranche_scenarios(tranche, spread_range, date_range, corr_map=False): """computes the pnl of a tranche for a range of spread scenarios Parameters ---------- tranche : TrancheBasket spread_range : `np.array`, spread range to run (different from swaption) corr_map: static correlation or mapped correlation """ if np.isnan(tranche.rho[2]): tranche.build_skew() temp_tranche = deepcopy(tranche) _get_singlenames_curves.cache_clear() orig_tranche_pvs = tranche.tranche_pvs().bond_price results = [] index_pv = np.empty_like(spread_range) tranche_pv = np.empty((len(spread_range), tranche.K.size - 1)) tranche_delta = np.empty((len(spread_range), tranche.K.size - 1)) for d in date_range: temp_tranche.value_date = d.date() for i, spread in enumerate(spread_range): temp_tranche.tweak(spread) if corr_map: temp_tranche.rho = tranche.map_skew(temp_tranche, 'TLP') index_pv[i] = temp_tranche._snacpv(spread * 1e-4, temp_tranche.coupon(temp_tranche.maturity), temp_tranche.recovery) tranche_pv[i] = temp_tranche.tranche_pvs().bond_price tranche_delta[i] = temp_tranche.tranche_deltas()['delta'] carry = temp_tranche.tranche_quotes.running * \ (d.date() - tranche.value_date).days / 360 df = pd.concat({'pv': pd.DataFrame(tranche_pv, index=spread_range, columns=tranche._row_names), 'delta': pd.DataFrame(tranche_delta, index=spread_range, columns=tranche._row_names), 'carry': pd.DataFrame( np.tile(carry, (len(spread_range), 1)), index=spread_range, columns=tranche._row_names)}, axis=1) df = df.join( pd.concat({'pnl': df['pv'].sub(orig_tranche_pvs), 'index_price_snac_pv': pd.Series(index_pv, index=spread_range, name='pv')}, axis=1)) results.append(df) results = pd.concat(results, keys=date_range) results.index.names = ['date', 'spread_range'] return results def run_tranche_scenarios_rolldown(tranche, spread_range, date_range, corr_map=False): """computes the pnl of a tranche for a range of spread scenarios curve roll down from the back, and valuations interpolated in the dates in between Parameters ---------- tranche : TrancheBasket spread_range : `np.array`, spread range to run (different from swaption) corr_map: static correlation or mapped correlation """ if np.isnan(tranche.rho[2]): tranche.build_skew() temp_tranche = deepcopy(tranche) orig_tranche_pvs = tranche.tranche_pvs().bond_price #create blanks index_pv, tranche_pv, tranche_delta = [], [], [] tranche_pv_f, tranche_delta_f = [], [] #do less scenarios, takes less time since the convexity is not as strong as swaptions days = np.diff((tranche.cs.index - date_range[0]).days.values) num_shortened = np.sum(tranche.cs.index < date_range[-1]) shorten_by = np.arange(0, max(1, num_shortened)+1, 1) days = np.append(0, np.cumsum(np.flip(days,0))[:len(shorten_by)-1]) smaller_spread_range = np.linspace(spread_range[0], spread_range[-1], 10) for i, spread in enumerate(smaller_spread_range): for shortened in shorten_by: if shortened > 0: temp_tranche.cs = tranche.cs.iloc[:-shortened] else: temp_tranche.cs = tranche.cs temp_tranche.tweak(spread) if corr_map: temp_tranche.rho = tranche.map_skew(temp_tranche, 'TLP') index_pv.append(temp_tranche.index_pv().bond_price) tranche_pv.append(temp_tranche.tranche_pvs().bond_price) tranche_delta.append(temp_tranche.tranche_deltas()['delta']) index_pv = np.reshape(index_pv, (smaller_spread_range.shape[0], days.shape[0])).transpose() tranche_pv = np.array(tranche_pv).transpose() tranche_delta = np.array(tranche_delta).transpose() index_pv_f = RectBivariateSpline(days, smaller_spread_range, index_pv, kx=1, ky=1) for pv, delta in zip(tranche_pv, tranche_delta): pv = np.reshape(pv, (smaller_spread_range.shape[0], days.shape[0])).transpose() delta = np.reshape(delta, (smaller_spread_range.shape[0], days.shape[0])).transpose() tranche_pv_f.append(RectBivariateSpline(days, smaller_spread_range, pv, kx=1, ky=1)) tranche_delta_f.append(RectBivariateSpline(days, smaller_spread_range, delta, kx=1, ky=1)) #Reset the blanks date_range_days = (date_range - date_range[0]).days.values tranche_pv = np.empty((tranche.K.size - 1, len(date_range_days), len(spread_range))) tranche_delta = np.empty((tranche.K.size - 1, len(date_range_days), len(spread_range))) index_pv = index_pv_f(date_range_days, spread_range) for i in range(len(tranche_pv_f)): tranche_pv[i] = tranche_pv_f[i](date_range_days, spread_range) tranche_delta[i] = tranche_delta_f[i](date_range_days, spread_range) index_pv = index_pv.reshape(1,len(date_range_days) * len(spread_range)).T tranche_pv = tranche_pv.reshape(len(tranche._row_names),len(date_range_days) * len(spread_range)).T tranche_delta = tranche_delta.reshape(len(tranche._row_names),len(date_range_days) * len(spread_range)).T days_diff = np.tile(((date_range - date_range[0]).days/360).values, len(tranche._row_names)) carry = pd.DataFrame(days_diff.reshape(len(tranche._row_names),len(date_range)).T, index=date_range, columns=pd.MultiIndex.from_product([['carry'], tranche._row_names])) carry.index.name = 'date' df = pd.concat({'index_pv': pd.DataFrame(index_pv, index=pd.MultiIndex.from_product([date_range, spread_range]), columns=['index_pv']), 'pv': pd.DataFrame(tranche_pv, index=pd.MultiIndex.from_product([date_range, spread_range]), columns=tranche._row_names), 'delta': pd.DataFrame(tranche_delta, index=pd.MultiIndex.from_product([date_range, spread_range]), columns=tranche._row_names)}, axis=1) df.index.names = ['date', 'spread_range'] df = df.join(carry) df = df.join(pd.concat({'pnl': df['pv'].sub(orig_tranche_pvs)}, axis=1)) return df def run_curve_scenarios(portf, spread_range, date_range, curve_per): """computes the pnl of a portfolio of indices for a range of spread/curve scenarios Parameters ---------- portf : Portfolio spread_range : `np.array` date_range : `pandas.Datetime.Index` """ portf.reset_pv() portf = deepcopy(portf) index = portf.indices[0].index_type r = [] for p in curve_per: new_curve = curve_shape(date_range[0], index, p, 100) for date in date_range: portf.value_date = date.date() for s in spread_range: for ind in portf.indices: ind.spread = new_curve((pd.to_datetime(ind.end_date) - date).days/365) * s/100 r.append([[date, s, p] + [portf.pnl]]) df = pd.DataFrame.from_records(chain(*r), columns=['date', 'spread', 'curve_per', 'pnl']) return df.set_index('date')