1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
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
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
"""
#create empty lists
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))
tranche.build_skew()
temp_tranche = deepcopy(tranche)
_get_singlenames_curves.cache_clear()
orig_tranche_pvs = tranche.tranche_pvs().bond_price
results = []
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
|
