diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/analytics/scenarios.py | 12 | ||||
| -rw-r--r-- | python/tests/test_scenarios.py | 42 |
2 files changed, 48 insertions, 6 deletions
diff --git a/python/analytics/scenarios.py b/python/analytics/scenarios.py index 2169df46..334a27a1 100644 --- a/python/analytics/scenarios.py +++ b/python/analytics/scenarios.py @@ -36,7 +36,7 @@ def run_swaption_scenarios(swaption, date_range, spread_shock, vol_shock, 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_shock', 'vol_shock'] + params) + df = pd.DataFrame.from_records(r, columns=['date', 'spread', 'vol_shock'] + params) return df.set_index('date') @@ -50,7 +50,7 @@ def run_index_scenarios(index, date_range, spread_shock, params=['pnl']): 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_shock'] + params) + df = pd.DataFrame.from_records(r, columns=['date', 'spread'] + params) return df.set_index('date') def _aux(portf, curr_vols, params, vs): @@ -79,19 +79,19 @@ def run_portfolio_scenarios(portf, date_range, spread_shock, vol_shock, if nproc > 0 run with nproc processes. """ portf = deepcopy(portf) - spreads = portf.index.spread * (1 + spread_shock) + spreads = np.hstack([index.spread * (1 + spread_shock) for index in portf.indices]) r = [] with MaybePool(nproc) as pool: pmap = pool.map if pool else map for date in date_range: - portf.index.trade_date = date.date() + portf.trade_date = date.date() t = [swaption.T for swaption in portf.swaptions] for s in spreads: - portf.index.ref = s + portf.ref = 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_shock', 'vol_shock'] + params) + df = pd.DataFrame.from_records(chain(*r), columns=['date', 'spread', 'vol_shock'] + params) return df.set_index('date') diff --git a/python/tests/test_scenarios.py b/python/tests/test_scenarios.py new file mode 100644 index 00000000..b7f5f9f0 --- /dev/null +++ b/python/tests/test_scenarios.py @@ -0,0 +1,42 @@ +import unittest +import datetime +import numpy as np +import pandas as pd + +from analytics import Index, BlackSwaption, Portfolio, VolatilitySurface +from pandas.tseries.offsets import BDay +from analytics.scenarios import run_portfolio_scenarios, run_swaption_scenarios, run_index_scenarios + +class TestSenarios(unittest.TestCase): + option_delta = Index.from_tradeid(874) + option1 = BlackSwaption.from_tradeid(7, option_delta) + option2 = BlackSwaption.from_tradeid(8, option_delta) + portf = Portfolio([option1, option2, option_delta]) + date_range = pd.bdate_range(option_delta.trade_date, pd.Timestamp('2017-05-17') - BDay(), freq = '5B') + + def test_portfolio(self): + """ check that run_portfolio_scenarios match the sum of the individual pieces""" + vol_shock = np.arange(-0.15, 0.3, 0.01) + spread_shock = np.arange(-0.2, 0.3, 0.01) + vs = VolatilitySurface("IG", 28, trade_date=self.option_delta.trade_date) + vol_surface = vs[vs.list(model="black", source="BAML")[-1]] + df = run_portfolio_scenarios(self.portf, self.date_range, + spread_shock, vol_shock, vol_surface) + df = df.set_index(['spread', 'vol_shock'], append=True) + + df1 = run_swaption_scenarios(self.option1, self.date_range, + spread_shock, vol_shock, vol_surface, ["pnl"]) + df2 = run_swaption_scenarios(self.option2, self.date_range, + spread_shock, vol_shock, vol_surface, ["pnl"]) + df_index = run_index_scenarios(self.option_delta, self.date_range, spread_shock) + df1 = df1.set_index(['spread', 'vol_shock'], append=True) + df2 = df2.set_index(['spread', 'vol_shock'], append=True) + df_index = df_index.set_index(['spread'], append=True) + df_swaptions = df1 + df2 + df_swaptions = df_swaptions.reset_index(level='vol_shock') + df_orig = df_index.add(df_swaptions, fill_value=0) + df_orig = df_orig.set_index('vol_shock', append=True) + self.assertFalse(np.any((df-df_orig).values)) + +if __name__=="__main__": + unittest.main() |