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
|
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
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.trade_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.trade_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.trade_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')
|
