Coverage for tests/test_marketdata.py: 80%
93 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-05 14:27 +0000
1import unittest
2import numpy as np
3import datetime as dt
5import rivapy
6from rivapy.marketdata import VolatilityGridParametrization, VolatilityParametrizationSABR, VolatilitySurface
7from rivapy.marketdata import (
8 DiscountCurve,
9 EquityForwardCurve,
10 enums,
11 DiscountCurveComposition,
12 DiscountCurveParametrized,
13 ConstantRate,
14 LinearRate,
15)
16from rivapy.tools import SimpleSchedule
17from rivapy import enums
18from rivapy import _pyvacon_available
21class VolatilityGridParamTest(unittest.TestCase):
22 def test_exceptions(self):
23 expiries = np.linspace(1.0 / 365.0, 4.0, 10)
24 strikes = np.linspace(0.4, 1.6, 100)
25 # test for exception if number of strikes does not match number of cols
26 vols = np.empty((expiries.shape[0], 1))
27 try:
28 vol_grid_param = VolatilityGridParametrization(expiries, strikes, vols)
29 self.assertFalse(False)
30 except:
31 self.assertFalse(True)
33 def test_calc_implied_vol(self):
34 """Simple tests for class VolatilityGridParametrization"""
35 if not _pyvacon_available:
36 self.assertEqual(1, 1)
37 return
38 expiries = np.linspace(1.0 / 365.0, 4.0, 10)
39 strikes = np.linspace(0.4, 1.6, 100)
40 vols = 0.3 * np.ones((expiries.shape[0], strikes.shape[0]))
41 vol_grid_param = VolatilityGridParametrization(expiries, strikes, vols)
42 self.assertAlmostEqual(0.3, vol_grid_param.calc_implied_vol(1.0, 1.3), delta=1e-7)
43 # now add to volatility surface
44 refdate = dt.datetime(2021, 1, 1)
45 dummy = DiscountCurve("", refdate=refdate, dates=[refdate, refdate + dt.timedelta(days=10 * 365)], df=[1.0, 1.0])
46 fwd = EquityForwardCurve(100.0, funding_curve=dummy, borrow_curve=dummy, div_table=None)
47 vol_surface = VolatilitySurface("", refdate, fwd, enums.DayCounterType.Act365Fixed, vol_grid_param)
48 if not _pyvacon_available:
49 self.assertAlmostEqual(1, 1)
50 return
51 vol = vol_surface.calc_implied_vol(refdate + dt.timedelta(days=365), 100.0)
52 self.assertAlmostEqual(vol, vol_grid_param.calc_implied_vol(1.0, 1.3), delta=1e-7)
55class VolatilitySABRParamTest(unittest.TestCase):
57 def test_calc_implied_vol(self):
58 # 1. Parametrization
59 expiries = [1.0 / 12.0, 1.0, 2.0, 3.0]
60 sabr_params = np.array(
61 [
62 [0.1, 0.1, 0.9, -0.8],
63 [0.3, 0.1, 0.1, 0.1],
64 [
65 0.5,
66 0.3,
67 0.9,
68 -0.75,
69 ],
70 [
71 0.5,
72 0.3,
73 0.9,
74 -0.85,
75 ],
76 ]
77 )
78 sabr_param = VolatilityParametrizationSABR(expiries, sabr_params)
79 self.assertAlmostEqual(0.30118, sabr_param.calc_implied_vol(ttm=1.0, strike=1.0), delta=1e-7)
81 # 2. Vol Surface
82 obj_id = "Test Surface"
83 refdate = dt.datetime(2021, 1, 1)
84 dc = DiscountCurve("", refdate=refdate, dates=[refdate, refdate + dt.timedelta(days=10 * 365)], df=[1.0, 1.0])
85 fc = EquityForwardCurve(100.0, funding_curve=dc, borrow_curve=dc, div_table=None)
86 vol_surf = VolatilitySurface(obj_id, refdate, fc, enums.DayCounterType.Act365Fixed, sabr_param)
87 if not _pyvacon_available:
88 self.assertAlmostEqual(1, 1)
89 return
90 vol = vol_surf.calc_implied_vol(refdate + dt.timedelta(days=365), 100.0, refdate)
91 self.assertAlmostEqual(vol, sabr_param.calc_implied_vol(ttm=1.0, strike=1.0), delta=1e-7)
94class VolatilitySurfaceTest(unittest.TestCase):
96 def test_calc_implied_vol_single_expiry(self):
97 self.assertAlmostEqual(0, 0)
98 # Fix to make the the commented code below working
99 return
100 """Test if calc_implied_vol works for single expiry
101 """
102 expiries = np.array([1])
103 strikes = np.linspace(0.4, 1.6, 100)
104 vols = 0.3 * np.ones((expiries.shape[0], strikes.shape[0]))
105 vol_grid_param = VolatilityGridParametrization(expiries, strikes, vols)
106 self.assertAlmostEqual(0.3, vol_grid_param.calc_implied_vol(1.0, 1.3), delta=1e-7)
107 refdate = dt.datetime(2021, 1, 1)
108 dummy = DiscountCurve("", refdate=refdate, dates=[refdate, refdate + dt.timedelta(days=10 * 365)], df=[1.0, 1.0])
109 fwd = EquityForwardCurve(100.0, funding_curve=dummy, borrow_curve=dummy, div_table=None)
110 vol_surface = VolatilitySurface("", refdate, fwd, enums.DayCounterType.Act365Fixed, vol_grid_param)
111 vol = vol_surface.calc_implied_vol(refdate + dt.timedelta(days=365), 100.0)
112 self.assertAlmostEqual(vol, vol_grid_param.calc_implied_vol(1.0, 1.3), delta=1e-7)
115# class PowerPriceForwardCurveTest(unittest.TestCase):
116# def test_simple_schedule(self):
117# """Test value with SimpleSchedule"""
118# simple_schedule = SimpleSchedule(dt.datetime(2022, 12, 1), dt.datetime(2023, 11, 1, 4, 0, 0), freq="1H")
119# values = np.ones((len(simple_schedule.get_schedule()),)).cumsum()
120# hpfc = PowerPriceForwardCurve(dt.datetime(2022, 12, 1), dt.datetime(2022, 12, 1), dt.datetime(2023, 11, 1, 4, 0, 0), freq="1H", values=values)
121# simple_schedule = SimpleSchedule(dt.datetime(2022, 12, 1), dt.datetime(2022, 12, 1, 4, 0, 0), freq="1H")
122# values = hpfc.value(dt.datetime(2022, 1, 1), simple_schedule)
123# self.assertEqual(values.shape[0], 4)
124# self.assertEqual(values[0], 1)
125# self.assertEqual(values[-1], 4)
126# # same with hours excluded
127# simple_schedule = SimpleSchedule(dt.datetime(2022, 12, 1), dt.datetime(2022, 12, 1, 4, 0, 0), freq="1H", hours=[2])
128# values = hpfc.value(dt.datetime(2022, 1, 1), simple_schedule)
129# self.assertEqual(values.shape[0], 1)
130# self.assertEqual(values[0], 3)
132# def test_exceptions(self):
133# """Test consistency checks in forward curve."""
134# simple_schedule = SimpleSchedule(dt.datetime(2022, 12, 1), dt.datetime(2023, 11, 1, 4, 0, 0), freq="1H")
135# values = np.ones((len(simple_schedule.get_schedule()),)).cumsum()
136# hpfc = PowerPriceForwardCurve(dt.datetime(2022, 12, 1), dt.datetime(2022, 12, 1), dt.datetime(2023, 11, 1, 4, 0, 0), freq="1H", values=values)
137# # schedule starts before first date of forward curve
138# simple_schedule = SimpleSchedule(dt.datetime(2022, 11, 30), dt.datetime(2022, 12, 1, 4, 0, 0), freq="1H")
139# self.assertRaises(Exception, lambda: hpfc.value(dt.datetime(2022, 1, 1), simple_schedule))
140# # schedule ends after last date of forward curve
141# simple_schedule = SimpleSchedule(dt.datetime(2022, 11, 30), dt.datetime(2023, 12, 1, 4, 0, 0), freq="1H")
142# self.assertRaises(Exception, lambda: hpfc.value(dt.datetime(2022, 1, 1), simple_schedule))
145class DiscountCurveCompositionTest(unittest.TestCase):
146 def test_curve_addition(self):
147 """Simple test adding two curves testing"""
148 ref_date = dt.datetime(2023, 1, 1)
149 c1 = DiscountCurveParametrized("C1", ref_date, ConstantRate(0.01))
150 c2 = DiscountCurveParametrized("C2", ref_date, ConstantRate(0.025))
151 # add two constant curves
152 c = c1 + c2
153 d = ref_date + dt.timedelta(days=10 * 365)
154 self.assertAlmostEqual(c1.value_rate(ref_date, d) + c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
155 self.assertAlmostEqual(c1.value(ref_date, d) * c2.value(ref_date, d), c.value(ref_date, d), places=6)
156 # add one constant and one linear curve
157 c2 = DiscountCurveParametrized("C2", ref_date, LinearRate(0.01, 0.05, max_maturity=10.0))
158 c = c1 + c2
159 self.assertAlmostEqual(c1.value_rate(ref_date, d) + c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
160 self.assertAlmostEqual(c1.value(ref_date, d) * c2.value(ref_date, d), c.value(ref_date, d), places=6)
161 # add a curve and a float
162 c = 0.01 + c2
163 self.assertAlmostEqual(c1.value_rate(ref_date, d) + c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
164 self.assertAlmostEqual(c1.value(ref_date, d) * c2.value(ref_date, d), c.value(ref_date, d), places=6)
166 def test_curve_multiplication(self):
167 """Simple test multiplying two curves"""
168 ref_date = dt.datetime(2023, 1, 1)
169 c1 = DiscountCurveParametrized("C1", ref_date, ConstantRate(0.01))
170 c2 = DiscountCurveParametrized("C2", ref_date, ConstantRate(0.025))
171 # multiply two constant curves
172 c = c1 * c2
173 d = ref_date + dt.timedelta(days=10 * 365)
174 self.assertAlmostEqual(c1.value_rate(ref_date, d) * c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
175 df = np.exp(-c1.value_rate(ref_date, d) * c2.value_rate(ref_date, d) * c1._dc.yf(ref_date, d))
176 self.assertAlmostEqual(df, c.value(ref_date, d), places=6)
178 # multiply one constant and one linear curve
179 c2 = DiscountCurveParametrized("C2", ref_date, LinearRate(0.01, 0.05, max_maturity=10.0))
180 c = c1 * c2
181 self.assertAlmostEqual(c1.value_rate(ref_date, d) * c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
182 df = np.exp(-c1.value_rate(ref_date, d) * c2.value_rate(ref_date, d) * c1._dc.yf(ref_date, d))
183 self.assertAlmostEqual(df, c.value(ref_date, d), places=6)
184 # multiply a curve and a float
185 c = 0.01 * c2
186 self.assertAlmostEqual(0.01 * c2.value_rate(ref_date, d), c.value_rate(ref_date, d), places=6)
187 df = np.exp(-0.01 * c2.value_rate(ref_date, d) * c1._dc.yf(ref_date, d))
188 self.assertAlmostEqual(df, c.value(ref_date, d), places=6)
191if __name__ == "__main__":
192 unittest.main()