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
|
from analytics.curve_trades import on_the_run
from analytics.index_data import get_index_quotes, index_returns
from utils.db import serenitas_engine, dawn_engine
import pandas as pd
import math
import datetime
def hist_var(portf, index_type="IG", quantile=0.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(20)
|
