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
|
from analytics.curve_trades import on_the_run
from analytics.index_data import get_index_quotes, index_returns
from db import dbengine
import pandas as pd
import math
import datetime
dawndb = dbengine('dawndb')
serenitasdb = dbengine('serenitasdb')
def hist_var(portf, index_type='IG', quantile=.05, years=5):
df = index_returns(index=index_type, years=years,
tenor=['3yr', '5yr', '7yr', '10yr'])
df = (df.reset_index(['index'], drop=True).
reorder_levels(['date', 'series', 'tenor']))
returns = df.spread_return.dropna().reset_index('series')
returns['dist_on_the_run'] = (returns.
groupby('date')['series'].
transform(lambda x: x.max() - x))
del returns['series']
returns = returns.set_index('dist_on_the_run', append=True).unstack('tenor')
returns.columns = returns.columns.droplevel(0)
portf.reset_pv()
otr = on_the_run(index_type)
spreads = pd.DataFrame({'spread': portf.spread,
'tenor': [ind.tenor for ind in portf.indices],
'dist_on_the_run': [otr - ind.series for ind in portf.indices]})
spreads = spreads.set_index(['dist_on_the_run', 'tenor'])
r = []
for k, g in returns.groupby(level='date', as_index=False):
shocks = g.reset_index('date', drop=True).stack('tenor')
shocks.name = 'shocks'
portf.spread = spreads.spread * (1 + spreads.join(shocks).shocks)
r.append((k, portf.pnl))
pnl = pd.DataFrame.from_records(r, columns=['date', 'pnl'], index=['date'])
return pnl.quantile(quantile) * math.sqrt(12)
|
