aboutsummaryrefslogtreecommitdiffstats
path: root/python/tests/test_scenarios.py
blob: 7be1bd415999c465f6d32ac3b09797699f0233e6 (plain)
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
import unittest
import numpy as np
import pandas as pd

from serenitas.analytics.api import (
    CreditIndex,
    BlackSwaption,
    Portfolio,
    BlackSwaptionVolSurface,
)
from pandas.tseries.offsets import BDay
from serenitas.analytics.scenarios import (
    run_portfolio_scenarios,
    run_swaption_scenarios,
    run_index_scenarios,
)


class TestSenarios(unittest.TestCase):
    option_delta = CreditIndex.from_tradeid(874)
    option1 = BlackSwaption.from_tradeid(7, option_delta)
    option2 = BlackSwaption.from_tradeid(8, option_delta)
    portf = Portfolio(
        [option1, option2, option_delta], trade_ids=["opt1", "opt2", "delta"]
    )
    date_range = pd.bdate_range(
        option_delta.value_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 = BlackSwaptionVolSurface("IG", 28, value_date=self.option_delta.value_date)
        vol_surface = vs[vs.list(source="BAML")[-1]]
        df = run_portfolio_scenarios(
            self.portf,
            self.date_range,
            spread_shock=spread_shock,
            vol_shock=vol_shock,
            vol_surface=vol_surface,
        )
        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)
        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.sum(axis=1).values - df_orig["pnl"].values)))


if __name__ == "__main__":
    unittest.main()