Coverage for tests / test_ir_swap.py: 98%
473 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1import unittest
2import unittest.mock
3from datetime import datetime, timedelta
4from dateutil.relativedelta import relativedelta
5import numpy as np
6import math
9from rivapy.marketdata.curves import DiscountCurve
10from rivapy.instruments.ir_swap_specification import (
11 IrSwapLegSpecification,
12 IrFixedLegSpecification,
13 IrFloatLegSpecification,
14 IrOISLegSpecification,
15 InterestRateSwapSpecification,
16)
17from rivapy.instruments.components import ConstNotionalStructure, ResettingNotionalStructure
18from rivapy.tools.enums import DayCounterType, IrLegType, Currency, RollConvention, SecuritizationLevel, Rating, InterpolationType, ExtrapolationType
19from rivapy.pricing.interest_rate_swap_pricing import get_projected_notionals, InterestRateSwapPricer
20from rivapy.instruments.components import CashFlow
21from rivapy.tools.datetools import DayCounter
23# of class InterestRateSwapPricer as Pricer
24# #non-static methods are:
25# price
26# #static methods are
27# _populate_cashflow_fixed
28# _populate_cashflow_float
29# _populate_cashflow_ois
30# price_leg_pricing_data # not fully implemented - test later
31# price_leg
32# compute_swap_rate
33# compute_swap_spread # not yett implemented
34# compute_basis_spread # not yet implemented
37class TestIrSwapLegSpecification(unittest.TestCase):
38 def setUp(self):
39 """General setup for tests with some default values e.g., dates and notional"""
40 self.start_dates = [datetime(2024, 1, 1)]
41 self.end_dates = [datetime(2025, 1, 1)]
42 self.pay_dates = [datetime(2025, 1, 1)]
43 self.notional = 1000.0
45 def test_init_and_properties(self):
46 """Test for the initialization and properties of the IrSwapLegSpecification class"""
47 leg = IrSwapLegSpecification(
48 obj_id="leg1",
49 notional=self.notional,
50 start_dates=self.start_dates,
51 end_dates=self.end_dates,
52 pay_dates=self.pay_dates,
53 currency="EUR",
54 day_count_convention=DayCounterType.ThirtyU360,
55 )
56 self.assertEqual(leg.obj_id, "leg1")
57 self.assertEqual(leg.currency, "EUR")
58 self.assertEqual(leg.start_dates, self.start_dates)
59 self.assertEqual(leg.end_dates, self.end_dates)
60 self.assertEqual(leg.pay_dates, self.pay_dates)
61 self.assertIsInstance(leg.notional_structure, ConstNotionalStructure)
63 def test_notional_structure_setter(self):
64 """Test for use of NotionalStructure class used in IrSwapLegSpecification.
65 For more spepcific tests of NotionalStructure, see its own test class.
67 #TODO add uses cases of different notional structures
68 """
69 ns = ConstNotionalStructure(5000.0)
70 leg = IrSwapLegSpecification(
71 obj_id="leg2",
72 notional=ns,
73 start_dates=self.start_dates,
74 end_dates=self.end_dates,
75 pay_dates=self.pay_dates,
76 currency="USD",
77 )
78 self.assertIs(leg.notional_structure, ns)
81class TestIrFixedLegSpecification(unittest.TestCase):
82 """Similar to TestIrSwapLegSpecification but for fixed leg specific
84 Args:
85 unittest (_type_): _description_
86 """
88 def setUp(self):
89 self.start_dates = [datetime(2024, 1, 1)]
90 self.end_dates = [datetime(2025, 1, 1)]
91 self.pay_dates = [datetime(2025, 1, 1)]
92 self.notional = 1000.0
94 def test_fixed_leg(self):
95 leg = IrFixedLegSpecification(
96 fixed_rate=0.01,
97 obj_id="fixed_leg",
98 notional=self.notional,
99 start_dates=self.start_dates,
100 end_dates=self.end_dates,
101 pay_dates=self.pay_dates,
102 currency="EUR",
103 )
104 self.assertEqual(leg.leg_type, IrLegType.FIXED)
105 self.assertAlmostEqual(leg.fixed_rate, 0.01)
106 self.assertEqual(leg.udl_id, "")
109class TestIrFloatLegSpecification(unittest.TestCase):
110 """Similar to TestIrSwapLegSpecification but for float leg specific
112 Args:
113 unittest (_type_): _description_
114 """
116 def setUp(self):
117 self.start_dates = [datetime(2024, 1, 1)]
118 self.end_dates = [datetime(2025, 1, 1)]
119 self.pay_dates = [datetime(2025, 1, 1)]
120 self.reset_dates = [datetime(2024, 1, 1)]
121 self.rate_start_dates = [datetime(2024, 1, 1)]
122 self.rate_end_dates = [datetime(2025, 1, 1)]
123 self.notional = 1000.0
125 def test_float_leg(self):
126 leg = IrFloatLegSpecification(
127 obj_id="float_leg",
128 notional=self.notional,
129 reset_dates=self.reset_dates,
130 start_dates=self.start_dates,
131 end_dates=self.end_dates,
132 rate_start_dates=self.rate_start_dates,
133 rate_end_dates=self.rate_end_dates,
134 pay_dates=self.pay_dates,
135 currency="USD",
136 udl_id="SOFR",
137 fixing_id="SOFR_FIX",
138 spread=0.002,
139 )
140 self.assertEqual(leg.leg_type, IrLegType.FLOAT)
141 self.assertEqual(leg.udl_id, "SOFR")
142 self.assertEqual(leg.fixing_id, "SOFR_FIX")
143 self.assertAlmostEqual(leg.spread, 0.002)
144 self.assertEqual(leg.reset_dates, self.reset_dates)
147class TestInterestRateSwapSpecification(unittest.TestCase):
148 """Full swap with both leg tests
150 Args:
151 unittest (_type_): _description_
152 """
154 def setUp(self):
155 self.start_dates = [datetime(2024, 1, 1)]
156 self.end_dates = [datetime(2025, 1, 1)]
157 self.pay_dates = [datetime(2025, 1, 1)]
158 self.notional = 1000.0
159 self.fixed_leg = IrFixedLegSpecification(
160 fixed_rate=0.01,
161 obj_id="fixed_leg",
162 notional=self.notional,
163 start_dates=self.start_dates,
164 end_dates=self.end_dates,
165 pay_dates=self.pay_dates,
166 currency="EUR",
167 )
168 self.float_leg = IrFloatLegSpecification(
169 obj_id="float_leg",
170 notional=self.notional,
171 reset_dates=self.start_dates,
172 start_dates=self.start_dates,
173 end_dates=self.end_dates,
174 rate_start_dates=self.start_dates,
175 rate_end_dates=self.end_dates,
176 pay_dates=self.pay_dates,
177 currency="USD",
178 udl_id="SOFR",
179 fixing_id="SOFR_FIX",
180 spread=0.002,
181 )
183 def test_swap_specification(self):
184 issue_date = datetime(2024, 1, 1)
185 maturity_date = datetime(2025, 1, 1)
186 spec = InterestRateSwapSpecification(
187 obj_id="swap1",
188 notional=self.notional,
189 issue_date=issue_date,
190 maturity_date=maturity_date,
191 pay_leg=self.fixed_leg,
192 receive_leg=self.float_leg,
193 currency="EUR",
194 day_count_convention=DayCounterType.ThirtyU360,
195 business_day_convention=RollConvention.FOLLOWING,
196 issuer="TestIssuer",
197 securitization_level=SecuritizationLevel.NONE,
198 rating=Rating.NONE,
199 )
200 self.assertEqual(spec.obj_id, "swap1")
201 self.assertEqual(spec.issue_date, issue_date)
202 self.assertEqual(spec.maturity_date, maturity_date)
203 self.assertEqual(spec.pay_leg, self.fixed_leg)
204 self.assertEqual(spec.receive_leg, self.float_leg)
205 self.assertEqual(spec.currency, "EUR")
206 self.assertEqual(spec.issuer, "TestIssuer")
207 self.assertEqual(spec.securitization_level, SecuritizationLevel.to_string(SecuritizationLevel.NONE))
208 self.assertEqual(spec.rating, Rating.to_string(Rating.NONE))
209 self.assertIsInstance(spec.notional_structure, ConstNotionalStructure)
211 def test_get_fixed_and_float_leg(self):
212 spec = InterestRateSwapSpecification(
213 obj_id="swap2",
214 notional=self.notional,
215 issue_date=datetime(2024, 1, 1),
216 maturity_date=datetime(2025, 1, 1),
217 pay_leg=self.fixed_leg,
218 receive_leg=self.float_leg,
219 )
220 self.assertIs(spec.get_fixed_leg(), self.fixed_leg)
221 self.assertIs(spec.get_float_leg(), self.float_leg)
223 def test_get_fixed_leg_error(self):
224 # Both legs fixed should raise
225 fixed_leg2 = IrFixedLegSpecification(
226 fixed_rate=0.01,
227 obj_id="fixed_leg2",
228 notional=self.notional,
229 start_dates=self.start_dates,
230 end_dates=self.end_dates,
231 pay_dates=self.pay_dates,
232 currency="EUR",
233 )
234 spec = InterestRateSwapSpecification(
235 obj_id="swap3",
236 notional=self.notional,
237 issue_date=datetime(2024, 1, 1),
238 maturity_date=datetime(2025, 1, 1),
239 pay_leg=self.fixed_leg,
240 receive_leg=fixed_leg2,
241 )
242 with self.assertRaises(ValueError):
243 spec.get_fixed_leg()
245 def test_get_float_leg_error(self):
246 # Both legs fixed should raise
247 fixed_leg2 = IrFixedLegSpecification(
248 fixed_rate=0.01,
249 obj_id="fixed_leg2",
250 notional=self.notional,
251 start_dates=self.start_dates,
252 end_dates=self.end_dates,
253 pay_dates=self.pay_dates,
254 currency="EUR",
255 )
256 spec = InterestRateSwapSpecification(
257 obj_id="swap4",
258 notional=self.notional,
259 issue_date=datetime(2024, 1, 1),
260 maturity_date=datetime(2025, 1, 1),
261 pay_leg=self.fixed_leg,
262 receive_leg=fixed_leg2,
263 )
264 with self.assertRaises(ValueError):
265 spec.get_float_leg()
268#######################################################
269# Tests for Pricing
272class TestIRSwapSpecificationPricing(unittest.TestCase):
273 """Test suite for pricing functionality of interest rate swaps.
274 Note OIS is modelled after a plain IRS except with its own pricing function
275 for the OIS/float leg.
277 Args:
278 unittest (_type_): _description_
279 """
281 def setUp(self):
282 """Setup of default values used throughout pricing"""
283 # by hand calculations to compare to
284 # #discount curve
285 # ttm = [0.5, 1.0, 1.5]
286 # rates = [0.1, 0.105, 0.11]
287 # N = 100 # Notional
288 # m = 2 # compounding frequency i.e. every half year
290 # ref_date = datetime(2017, 1, 1)
292 # days_to_maturity = [180, 360, 540]
293 # dates = [ref_date + timedelta(days=d) for d in days_to_maturity]
294 # df = [math.exp(-r * t) for r, t in zip(rates, ttm)]
296 # Discount curve - we use these discount factors to get the present values of both the fixed and floating leg as well as
297 object_id = "TEST_DC"
298 refdatedc = datetime(2017, 1, 1)
299 days_to_maturity = [180, 360, 540]
300 dates_dc = [refdatedc + timedelta(days=d) for d in days_to_maturity]
301 # discount factors from constant rate
302 rates = [0.10, 0.105, 0.11]
303 df = [math.exp(-r * d / 360) for r, d in zip(rates, days_to_maturity)]
304 dc = DiscountCurve(
305 id=object_id, refdate=refdatedc, dates=dates_dc, df=df, interpolation=InterpolationType.LINEAR, extrapolation=ExtrapolationType.LINEAR
306 )
308 dcc = "Act360"
309 ccy = "EUR"
310 # Create the vectors defining the statdates, enddates, paydates and reset dates
311 refdate = datetime(2017, 1, 1)
312 days_to_maturity = [0, 180, 360, 540]
313 dates = [refdate + timedelta(d) for d in days_to_maturity]
315 startdates = dates[:-1]
316 enddates = dates[1:]
317 paydates = enddates
318 resetdates = startdates
320 fixed_leg = IrFixedLegSpecification(
321 fixed_rate=0.08,
322 obj_id="dummy_fixed_leg",
323 notional=100.0,
324 start_dates=startdates,
325 end_dates=enddates,
326 pay_dates=paydates,
327 currency=ccy,
328 day_count_convention=dcc,
329 )
331 spread = 0.00
332 self.notional_amount = 100
333 ns = ConstNotionalStructure(self.notional_amount)
335 float_leg = IrFloatLegSpecification(
336 obj_id="dummy_float_leg",
337 notional=ns,
338 reset_dates=resetdates,
339 start_dates=startdates,
340 end_dates=enddates,
341 rate_start_dates=startdates,
342 rate_end_dates=enddates,
343 pay_dates=paydates,
344 currency=ccy,
345 udl_id="test_udl_id",
346 fixing_id="test_fixing_id",
347 day_count_convention=dcc,
348 spread=spread,
349 )
351 # in my other example, was an OIS with 6M tenor, and 6M maturity. with only 1 "interval"
352 # here we will have more if using these startdates and enddates
353 res = IrOISLegSpecification.ois_scheduler_2D(startdates, enddates)
354 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
355 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
356 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
357 ois_pay_dates = res[3]
359 ois_leg = IrOISLegSpecification(
360 obj_id="dummy_ois_leg",
361 notional=ns,
362 rate_reset_dates=daily_rate_reset_dates, # reflects the overnight nature of OIS
363 start_dates=startdates, # same as before
364 end_dates=enddates, # same as before
365 rate_start_dates=daily_rate_start_dates, # reflects the overnight nature of OIS
366 rate_end_dates=daily_rate_end_dates, # reflects the overnight nature of OIS
367 pay_dates=ois_pay_dates,
368 currency=ccy,
369 udl_id="test_udl_id",
370 fixing_id="test_fixing_id",
371 day_count_convention=dcc,
372 rate_day_count_convention=dcc,
373 spread=spread,
374 )
376 maturity_date = refdate + timedelta(600)
377 # ir_swap = InterestRateSwapSpecification('TEST_SWAP', 'DBK', 'COLLATERALIZED', 'EUR', paydates[-1], fixedleg, floatleg)
378 ir_swap = InterestRateSwapSpecification(
379 obj_id="dummy_swap_6m",
380 notional=ns,
381 issue_date=refdate,
382 maturity_date=maturity_date,
383 pay_leg=fixed_leg,
384 receive_leg=float_leg,
385 currency=ccy,
386 day_count_convention=dcc,
387 issuer="dummy_issuer",
388 securitization_level="COLLATERALIZED",
389 )
391 oi_swap = InterestRateSwapSpecification(
392 obj_id="dummy_ois_6M",
393 notional=ns,
394 issue_date=refdate,
395 maturity_date=maturity_date,
396 pay_leg=fixed_leg,
397 receive_leg=ois_leg,
398 currency=ccy,
399 day_count_convention=dcc,
400 issuer="dummy_issuer",
401 securitization_level="COLLATERALIZED",
402 )
404 self.refdate = refdate
405 self.start_dates = startdates
406 self.end_dates = enddates
407 self.reset_dates = resetdates
408 self.pay_dates = paydates
409 self.maturity_date = maturity_date
411 self.day_count_convention = dcc
412 self.ccy = ccy
413 self.dc = dc
414 self.dc_rates = rates
415 self.dc_dates = dates_dc
416 self.dc_df = df
418 self.ns = ns
420 self.fixed_leg = fixed_leg
421 self.float_leg = float_leg
422 self.ois_leg = ois_leg
423 self.ir_swap = ir_swap
424 self.oi_swap = oi_swap
426 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter.get")
427 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter")
428 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.get_projected_notionals")
429 def test_populate_cashflow_fixed(self, mock_get_notionals, mock_daycounter, mock_daycounter_get):
430 """
431 Test to output correct present value calculation of FIXED leg of an IR swap.
433 We provide a fixed IR leg specification
434 We contrtol the output of finding forward rates from a discount curve through mock so that we are
435 independent of this helper function.
436 We control the yearfraction also for ease of calculation and verification.
439 """
440 notional_amount = self.notional_amount
441 fixed_rate = 0.08
442 # Arrange
443 mock_get_notionals.return_value = [notional_amount]
445 mock_instance = unittest.mock.Mock()
446 mock_instance.yf.return_value = 1.0
447 mock_daycounter_get.return_value = mock_instance
449 mock_daycounter.return_value.yf.return_value = 1.0 # just assume year fraction will be 1 for ease of calc
451 # Mock discount curve
452 discount_curve = unittest.mock.Mock()
453 discount_curve.daycounter = "Act360"
454 discount_curve.value.return_value = 0.95 # talk the discount factor to be 0.95 for control
456 print("DEBUG: DayCounter mock:", mock_daycounter)
457 print("DEBUG: DayCounter instance:", mock_daycounter.return_value)
458 print("DEBUG: DayCounter.yf mock:", mock_daycounter.return_value.yf)
459 print("DEBUG: DayCounter.yf.return_value:", mock_daycounter.return_value.yf.return_value)
460 # Act
461 result = InterestRateSwapPricer._populate_cashflows_fix(
462 self.refdate,
463 fixed_leg_spec=self.fixed_leg,
464 discount_curve=discount_curve,
465 fx_forward_curve=unittest.mock.Mock(), # since they are not required
466 fixing_map=unittest.mock.Mock(), # since they are not required for calculation
467 set_rate=False,
468 desired_rate=0.05, # will not be accessed if set_rate = False
469 )
471 # Assert
472 self.assertEqual(len(result), 1)
474 # Check the interest cashflow
475 interest_cf = [cf for cf in result if getattr(cf, "interest_cashflow", False)][0]
476 print("DEBUG: entry.interest_yf =", interest_cf.interest_yf, type(interest_cf.interest_yf))
478 # DEBUG PRINTS
479 print("DEBUG: interest_yf =", interest_cf.interest_yf, type(interest_cf.interest_yf))
480 print("DEBUG: interest_amount =", interest_cf.interest_amount, type(interest_cf.interest_amount))
481 print("DEBUG: present_value =", interest_cf.present_value, type(interest_cf.present_value))
482 print("DEBUG: discount_factor =", interest_cf.discount_factor, type(interest_cf.discount_factor))
484 self.assertAlmostEqual(interest_cf.interest_amount, notional_amount * fixed_rate * 1.0) # 8
485 self.assertAlmostEqual(interest_cf.present_value, 8 * 0.95)
487 # Not yet implemented in populate_fixed fully
488 # # Check notional outflow
489 # outflow_cf = [cf for cf in result if getattr(cf, "notional_cashflow", False) and cf.pay_amount < 0][0]
490 # self.assertEqual(outflow_cf.pay_amount, -notional_amount)
492 # # Check notional inflow
493 # inflow_cf = [cf for cf in result if getattr(cf, "notional_cashflow", False) and cf.pay_amount > 0][0]
494 # self.assertEqual(inflow_cf.pay_amount, notional_amount)
496 # Verify mocks
497 mock_get_notionals.assert_called_once()
498 mock_daycounter.return_value.yf.assert_called_once()
500 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.get_projected_notionals")
501 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter")
502 def test_cashflows_fix_notional_start_end(self, mock_daycounter_class, mock_get_notionals):
503 """
504 Test _populate_cashflows_fix generates entries for both start and end notional cashflows.
505 """
507 # --- Arrange ---
508 val_date = datetime(2025, 1, 1)
510 # Mock fixed leg spec
511 fixed_leg_spec = unittest.mock.Mock()
512 fixed_leg_spec.fixed_rate = 0.05
513 fixed_leg_spec.start_dates = [val_date, val_date + timedelta(days=180)]
514 fixed_leg_spec.end_dates = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
515 fixed_leg_spec.pay_dates = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
517 # Mock Notional structure with start/end dates
518 notional_structure = unittest.mock.Mock()
519 notional_structure.get_pay_date_start.side_effect = [val_date, val_date + timedelta(days=180)]
520 notional_structure.get_pay_date_end.side_effect = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
521 fixed_leg_spec.get_NotionalStructure.return_value = notional_structure
523 # Mock projected notionals
524 mock_get_notionals.return_value = [100.0, 200.0]
526 # Mock DiscountCurve
527 discount_curve = unittest.mock.Mock()
528 discount_curve.value.side_effect = lambda val, pay: 0.95 # always 0.95
530 # Mock fx_forward_curve
531 fx_forward_curve = unittest.mock.Mock()
533 # Mock DayCounter
534 mock_daycounter_instance = unittest.mock.Mock()
535 mock_daycounter_instance.yf.side_effect = lambda start, end: (end - start).days / 360.0
536 mock_daycounter_class.return_value = mock_daycounter_instance
538 # Mock FixingMap (not used)
539 fixing_map = unittest.mock.Mock()
541 # --- Act ---
542 from rivapy.pricing.interest_rate_swap_pricing import InterestRateSwapPricer
544 entries = InterestRateSwapPricer._populate_cashflows_fix(
545 val_date=val_date,
546 fixed_leg_spec=fixed_leg_spec,
547 discount_curve=discount_curve,
548 fx_forward_curve=fx_forward_curve,
549 fixing_map=fixing_map,
550 set_rate=False,
551 )
553 # --- Assert ---
554 # There should be 2 start + 2 end notional cashflows + 2 interest cashflows = 6
555 self.assertEqual(len(entries), 6)
557 # Check first notional start/outflow
558 start_cf = entries[0]
559 self.assertTrue(getattr(start_cf, "notional_cashflow", False))
560 self.assertEqual(start_cf.pay_amount, -100.0)
561 self.assertEqual(start_cf.discount_factor, 0.95)
562 self.assertAlmostEqual(start_cf.present_value, -95.0)
564 # Check first interest cashflow
565 interest_cf = entries[1]
566 self.assertTrue(getattr(interest_cf, "interest_cashflow", False))
567 self.assertEqual(interest_cf.notional, 100.0)
568 expected_interest = 100.0 * 0.05 * ((180) / 360) # rate * yf
569 self.assertAlmostEqual(interest_cf.interest_amount, expected_interest)
570 self.assertAlmostEqual(interest_cf.present_value, expected_interest * 0.95)
572 # Check first notional end/inflow
573 end_cf = entries[2]
574 self.assertTrue(getattr(end_cf, "notional_cashflow", False))
575 self.assertEqual(end_cf.pay_amount, 100.0)
576 self.assertEqual(end_cf.discount_factor, 0.95)
577 self.assertAlmostEqual(end_cf.present_value, 95.0)
579 # Optionally, check second cashflow sequence similarly
580 start_cf2, interest_cf2, end_cf2 = entries[3], entries[4], entries[5]
581 self.assertEqual(start_cf2.pay_amount, -200.0)
582 self.assertEqual(end_cf2.pay_amount, 200.0)
583 self.assertAlmostEqual(interest_cf2.interest_amount, 200.0 * 0.05 * (180 / 360))
585 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter.get")
586 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter")
587 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.get_projected_notionals")
588 def test_populate_cashflow_float(self, mock_get_notionals, mock_daycounter, mock_daycounter_get):
589 """
590 Test correct present value calculation of a FLOAT leg of an IR swap.
591 """
593 # --- ARRANGE ---
594 notional_amount = self.notional_amount
595 leg_spread = 0.00 # 1%
596 fwd_floating_rate = 0.05 # 5% forward rate
598 # Mock notionals
599 mock_get_notionals.return_value = [notional_amount]
601 # Mock daycounter for year fractions
602 mock_instance = unittest.mock.Mock()
603 mock_instance.yf.return_value = 1.0
604 mock_daycounter_get.return_value = mock_instance
605 mock_daycounter.return_value.yf.return_value = 1.0
607 # Mock discount curve
608 discount_curve = unittest.mock.Mock()
609 discount_curve.daycounter = "Act360"
610 discount_curve.value.return_value = 0.95
612 # Mock forward curve
613 forward_curve = unittest.mock.Mock()
614 fwd_factor = 1.0 / (1.0 + fwd_floating_rate)
615 forward_curve.value_fwd.return_value = fwd_factor # given fwd_floating_rate = (1/fwd_factor -1 )/yf
616 # Mock fx curve (not used)
617 fx_forward_curve = unittest.mock.Mock()
619 # Mock fixing map (not used)
620 fixing_map = unittest.mock.Mock()
622 # Create a minimal float leg spec #simpler than what we predefined in setup
623 float_leg = unittest.mock.Mock()
624 float_leg.start_dates = [datetime(2025, 1, 1)]
625 float_leg.end_dates = [datetime(2026, 1, 1)]
626 float_leg.pay_dates = [datetime(2026, 1, 1)]
627 float_leg.reset_dates = [datetime(2025, 1, 1)]
628 float_leg.rate_start_dates = [datetime(2025, 1, 1)]
629 float_leg.rate_end_dates = [datetime(2026, 1, 1)]
630 float_leg.udl_id = "LIBOR3M"
631 float_leg.spread = leg_spread
632 float_leg.rate_day_count_convention = "Act360"
633 float_leg.get_NotionalStructure.return_value.get_pay_date_start.return_value = datetime(2025, 1, 1)
634 float_leg.get_NotionalStructure.return_value.get_pay_date_end.return_value = datetime(2026, 1, 1)
636 val_date = datetime(2025, 1, 1)
638 # --- ACT ---
639 result = InterestRateSwapPricer._populate_cashflows_float(
640 val_date=val_date,
641 float_leg_spec=float_leg,
642 discount_curve=discount_curve,
643 forward_curve=forward_curve,
644 fx_forward_curve=fx_forward_curve,
645 fixing_map=fixing_map,
646 fixing_grace_period=0.0,
647 set_spread=False,
648 spread=None,
649 )
651 # --- ASSERT ---
652 # There should be 3 cashflows: initial notional outflow, interest cashflow, end notional inflow
653 self.assertEqual(len(result), 3)
655 # Check interest cashflow
656 interest_cf = [cf for cf in result if getattr(cf, "interest_cashflow", False)][0]
658 print(" - - - DEBUG - - - ")
659 for attr, value in vars(interest_cf).items():
660 print(attr, value)
662 # expected_rate = leg_spread + (1.0 / fwd_rate - 1.0) / 1.0 # as per your formula
663 expected_interest = notional_amount * fwd_floating_rate * 1.0 # yf = 1
664 expected_pv = expected_interest * 0.95 / fwd_factor # discount factor
666 self.assertAlmostEqual(interest_cf.rate, fwd_floating_rate)
667 self.assertAlmostEqual(interest_cf.interest_amount, expected_interest)
668 self.assertAlmostEqual(interest_cf.present_value, expected_pv)
670 # Check mocks called
671 mock_get_notionals.assert_called_once()
672 mock_daycounter.return_value.yf.assert_called()
673 forward_curve.value_fwd.assert_called()
674 discount_curve.value.assert_called()
676 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.get_projected_notionals")
677 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.DayCounter")
678 def test_cashflows_float_notional_start_end(self, mock_daycounter_class, mock_get_notionals):
679 """
680 Test _populate_cashflows_float generates entries for both start and end notional cashflows,
681 including interest cashflows with mocked forward rates.
682 """
684 # --- Arrange ---
685 val_date = datetime(2025, 1, 1)
687 # Mock floating leg spec
688 float_leg_spec = unittest.mock.Mock()
689 float_leg_spec.udl_id = "LIBOR3M"
690 float_leg_spec.spread = 0.01
691 float_leg_spec.start_dates = [val_date, val_date + timedelta(days=180)]
692 float_leg_spec.end_dates = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
693 float_leg_spec.pay_dates = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
694 float_leg_spec.rate_start_dates = [val_date, val_date + timedelta(days=180)]
695 float_leg_spec.rate_end_dates = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
696 float_leg_spec.reset_dates = [val_date - timedelta(days=1), val_date + timedelta(days=180)]
697 float_leg_spec.rate_day_count_convention = "ACT/360"
699 # Mock Notional structure with start/end dates
700 notional_structure = unittest.mock.Mock()
701 notional_structure.get_pay_date_start.side_effect = [val_date, val_date + timedelta(days=180)]
702 notional_structure.get_pay_date_end.side_effect = [val_date + timedelta(days=180), val_date + timedelta(days=360)]
703 float_leg_spec.get_NotionalStructure.return_value = notional_structure
705 # Mock projected notionals
706 mock_get_notionals.return_value = [100.0, 200.0]
708 # Mock DiscountCurve
709 discount_curve = unittest.mock.Mock()
710 discount_curve.value.side_effect = lambda val, pay: 0.95 # always 0.95
712 # Mock ForwardCurve
713 forward_curve = unittest.mock.Mock()
714 forward_curve.value_fwd.side_effect = lambda val, start, end: 0.02 # constant forward rate
716 # Mock fx_forward_curve (not used)
717 fx_forward_curve = unittest.mock.Mock()
719 # Mock DayCounter
720 mock_daycounter_instance = unittest.mock.Mock()
721 mock_daycounter_instance.yf.side_effect = lambda start, end: (end - start).days / 360.0
722 mock_daycounter_class.return_value = mock_daycounter_instance
724 # Mock FixingMap (return None to force using forward rate logic)
725 fixing_map = unittest.mock.Mock()
726 fixing_map.get_fixing.return_value = None
728 # --- Act ---
729 from rivapy.pricing.interest_rate_swap_pricing import InterestRateSwapPricer
731 entries = InterestRateSwapPricer._populate_cashflows_float(
732 val_date=val_date,
733 float_leg_spec=float_leg_spec,
734 discount_curve=discount_curve,
735 forward_curve=forward_curve,
736 fx_forward_curve=fx_forward_curve,
737 fixing_map=fixing_map,
738 fixing_grace_period=30,
739 set_spread=False,
740 spread=0.0,
741 )
743 # --- Assert ---
744 # There should be 2 start + 2 end notional cashflows + 2 interest cashflows = 6
745 self.assertEqual(len(entries), 6)
747 # --- First cashflow sequence ---
748 start_cf = entries[0]
749 interest_cf = entries[1]
750 end_cf = entries[2]
752 # Notional start
753 self.assertTrue(getattr(start_cf, "notional_cashflow", False))
754 self.assertEqual(start_cf.pay_amount, -100.0)
755 self.assertEqual(start_cf.discount_factor, 0.95)
756 self.assertAlmostEqual(start_cf.present_value, -95.0)
758 # Interest cashflow
759 self.assertTrue(getattr(interest_cf, "interest_cashflow", False))
760 self.assertEqual(interest_cf.notional, 100.0)
761 expected_interest = 100.0 * (1.0 / 0.02 - 1.0) / ((180) / 360) * ((180) / 360) + 0.01 * 100.0 * (180 / 360)
762 # Simplified calculation here, in practice it should match the formula in the code
763 self.assertAlmostEqual(interest_cf.interest_amount, interest_cf.notional * interest_cf.rate * interest_cf.interest_yf)
764 self.assertAlmostEqual(interest_cf.present_value, interest_cf.pay_amount * 0.95)
766 # Notional end
767 self.assertTrue(getattr(end_cf, "notional_cashflow", False))
768 self.assertEqual(end_cf.pay_amount, 100.0)
769 self.assertEqual(end_cf.discount_factor, 0.95)
770 self.assertAlmostEqual(end_cf.present_value, 95.0)
772 # --- Second cashflow sequence ---
773 start_cf2, interest_cf2, end_cf2 = entries[3], entries[4], entries[5]
775 self.assertEqual(start_cf2.pay_amount, -200.0)
776 self.assertTrue(getattr(interest_cf2, "interest_cashflow", False))
777 self.assertEqual(end_cf2.pay_amount, 200.0)
779 def populate_cashflow_ois(self):
780 # TODO
781 pass
783 @unittest.mock.patch.object(InterestRateSwapPricer, "price_leg")
784 def price(self, mock_price_leg):
785 """Simple test of swap leg aggregation ensuring it makes the expected calls to the
786 internal function price_leg properly and aggregates properly
787 essentially: test_price_subtracts_pay_leg_from_receive_leg
788 """
789 # Arrange
790 swap = self.ir_swap
792 # Arrange: create pricer with dummy data
793 pricer = InterestRateSwapPricer(
794 val_date=datetime(2025, 1, 1),
795 spec=swap,
796 discount_curve_pay_leg=unittest.mock.Mock(name="discount_pay"),
797 discount_curve_receive_leg=unittest.mock.Mock(name="discount_receive"),
798 fixing_curve_pay_leg=unittest.mock.Mock(name="fixing_pay"),
799 fixing_curve_receive_leg=unittest.mock.Mock(name="fixing_receive"),
800 fx_fwd_curve_pay_leg=unittest.mock.Mock(name="fx_pay"),
801 fx_fwd_curve_receive_leg=unittest.mock.Mock(name="fx_receive"),
802 pricing_request=unittest.mock.Mock(name="InterestRateSwapPricingRequest"),
803 pricing_param={"fixing_grace_period": 1},
804 )
806 # Stub price_leg results
807 mock_price_leg.side_effect = [150.0, 40.0] # first call receive leg, second pay leg
809 # Act
810 result = pricer.price()
812 # Assert
813 self.assertEqual(result, 110.0) # 150 - 40
814 self.assertEqual(mock_price_leg.call_count, 2)
816 def price_leg_pricing_data(self): # using the pricing data container structure
817 pass
819 @unittest.mock.patch.object(InterestRateSwapPricer, "_populate_cashflows_fix")
820 def test_price_fixed_leg(self, mock_populate_fix):
821 # Arrange
822 leg_spec = self.fixed_leg
823 cf1 = CashFlow()
824 cf1.present_value = 10
825 cf2 = CashFlow()
826 cf2.present_value = 20
827 mock_populate_fix.return_value = [cf1, cf2]
829 # Act
830 result = InterestRateSwapPricer.price_leg(
831 self.refdate,
832 discount_curve=self.dc,
833 forward_curve=self.dc,
834 fxForward_curve=self.dc,
835 spec=leg_spec,
836 # fixing_map=self.fixing_map,
837 pricing_params={"set_rate": False, "desired_rate": 0.05},
838 )
840 # Assert
841 self.assertEqual(result, 30.0)
842 mock_populate_fix.assert_called_once()
844 @unittest.mock.patch.object(InterestRateSwapPricer, "_populate_cashflows_float")
845 def test_price_float_leg(self, mock_populate_float):
846 # Arrange
847 leg_spec = self.float_leg
848 cf1 = CashFlow()
849 cf1.present_value = -5
850 cf2 = CashFlow()
851 cf2.present_value = 20
852 mock_populate_float.return_value = [cf1, cf2]
854 # Act
855 result = InterestRateSwapPricer.price_leg(
856 self.refdate,
857 discount_curve=self.dc,
858 forward_curve=self.dc,
859 fxForward_curve=self.dc,
860 spec=leg_spec,
861 # fixing_map=self.fixing_map,
862 pricing_params={"set_rate": False, "desired_rate": 0.05},
863 )
865 # Assert
866 self.assertEqual(result, 15.0)
867 mock_populate_float.assert_called_once()
869 @unittest.mock.patch.object(InterestRateSwapPricer, "_populate_cashflows_ois")
870 def test_price_ois_leg(self, mock_populate_ois):
871 # Arrange
872 leg_spec = self.ois_leg
873 cf1 = CashFlow()
874 cf1.present_value = 10
875 cf2 = CashFlow()
876 cf2.present_value = 40
877 mock_populate_ois.return_value = [cf1, cf2]
879 # Act
880 result = InterestRateSwapPricer.price_leg(
881 self.refdate,
882 discount_curve=self.dc,
883 forward_curve=self.dc,
884 fxForward_curve=self.dc,
885 spec=leg_spec,
886 # fixing_map=self.fixing_map,
887 pricing_params={"set_rate": False, "desired_rate": 0.05},
888 )
890 # Assert
891 self.assertEqual(result, 50.0)
892 mock_populate_ois.assert_called_once()
894 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.InterestRateSwapPricer.price_leg")
895 def test_compute_swap_rate(self, mock_price_leg):
896 """
897 Test fair swap rate calculation:
898 swap_rate = PV_float / Annuity_fixed
900 The function essentially calls price_leg on a fixed and floating leg, where for the
901 fixed leg, the desired rate is set = 1 in order to calculate the Annuity instead
902 as used for the computation of a fair swap rate st. PV_Fixed = PV_float
904 """
906 # Arrange: control PVs from mocked price_leg
907 mock_price_leg.side_effect = [200.0, 800.0] # float PV, fixed annuity
909 ref_date = datetime(2024, 1, 1)
910 discount_curve = unittest.mock.Mock()
911 fixing_curve = unittest.mock.Mock()
912 float_leg = unittest.mock.Mock()
913 fixed_leg = unittest.mock.Mock()
914 fixing_map = unittest.mock.Mock()
916 # Act
917 swap_rate = InterestRateSwapPricer.compute_swap_rate(
918 ref_date=ref_date,
919 discount_curve=discount_curve,
920 fixing_curve=fixing_curve,
921 float_leg=float_leg,
922 fixed_leg=fixed_leg,
923 fixing_map=fixing_map,
924 pricing_params={
925 "fixing_grace_period": 0.0
926 }, # then ensures that desired_rate = 1 in price_leg for fixed since it will go with default value
927 )
929 # Assert
930 expected_rate = 200.0 / 800.0 # 0.25
931 self.assertAlmostEqual(swap_rate, expected_rate)
933 # Verify calls
934 self.assertEqual(mock_price_leg.call_count, 2)
935 # First call -> float leg
936 args_float = mock_price_leg.call_args_list[0][0]
937 # Second call -> fixed leg
938 args_fixed = mock_price_leg.call_args_list[1][0]
940 self.assertIs(args_float[2], fixing_curve) # forward curve used for float leg
941 self.assertIs(args_fixed[2], fixing_curve) # forward curve also passed for fixed leg annuity
943 @unittest.mock.patch("rivapy.pricing.interest_rate_swap_pricing.InterestRateSwapPricer.price_leg")
944 def test_compute_basis_spread(self, mock_price_leg):
945 """
946 Test fair basis spread calculation:
947 basis_spread = (PV_receive_leg - PV_pay_leg) / PV01_fixed_leg
949 The function calls price_leg three times:
950 1. receive_leg PV (using receiveLegFixingCurve)
951 2. pay_leg PV (using payLegFixingCurve)
952 3. fixed_leg PV01 (using payLegFixingCurve, desired_rate=1)
953 """
955 # --- Arrange ---
956 mock_price_leg.side_effect = [300.0, 250.0, 20.0] # receive_leg PV, pay_leg PV, fixed_leg PV01
958 ref_date = datetime(2024, 1, 1)
960 # create mock curves and legs
961 discount_curve = unittest.mock.Mock(name="discount_curve")
962 pay_fixing_curve = unittest.mock.Mock(name="pay_fixing_curve")
963 receive_fixing_curve = unittest.mock.Mock(name="receive_fixing_curve")
964 pay_leg = unittest.mock.Mock(name="pay_leg")
965 receive_leg = unittest.mock.Mock(name="receive_leg")
966 spread_leg = unittest.mock.Mock(name="spread_leg")
967 fixing_map = unittest.mock.Mock(name="fixing_map")
969 pricing_params = {"fixing_grace_period": 0.0}
971 # --- Act ---
972 result = InterestRateSwapPricer.compute_basis_spread(
973 ref_date=ref_date,
974 discount_curve=discount_curve,
975 payLegFixingCurve=pay_fixing_curve,
976 receiveLegFixingCurve=receive_fixing_curve,
977 pay_leg=pay_leg,
978 receive_leg=receive_leg,
979 spread_leg=spread_leg,
980 fixing_map=fixing_map,
981 pricing_params=pricing_params,
982 )
984 # --- Assert ---
985 # expected = (receive_leg_PV - pay_leg_PV) / fixed_leg_PV01 = (300 - 250) / 20 = 2.5
986 expected_spread = (300.0 - 250.0) / 20.0
987 self.assertAlmostEqual(result, expected_spread)
989 # Check calls
990 self.assertEqual(mock_price_leg.call_count, 3)
992 # Check that the right arguments were passed for each call
993 args_receive = mock_price_leg.call_args_list[0][0]
994 args_pay = mock_price_leg.call_args_list[1][0]
995 args_fixed = mock_price_leg.call_args_list[2][0]
997 # receive leg should use receiveLegFixingCurve
998 self.assertIs(args_receive[2], receive_fixing_curve)
999 self.assertIs(args_receive[4], receive_leg)
1001 # pay leg should use payLegFixingCurve
1002 self.assertIs(args_pay[2], pay_fixing_curve)
1003 self.assertIs(args_pay[4], pay_leg)
1005 # fixed leg should also use payLegFixingCurve
1006 self.assertIs(args_fixed[2], pay_fixing_curve)
1007 self.assertIs(args_fixed[4], spread_leg)
1010class TestInterestRateSwapPricerInit(unittest.TestCase):
1011 """Unit tests for InterestRateSwapPricer initialization logic."""
1013 def setUp(self):
1014 self.refdate = datetime(2025, 1, 1)
1015 self.maturity = self.refdate + timedelta(days=365) # FIX: must be later than issue_date
1016 self.ccy = "EUR"
1018 # --- Discount curves ---
1019 self.dc = DiscountCurve(
1020 id="TEST_DC",
1021 refdate=self.refdate,
1022 dates=[self.refdate],
1023 df=[1.0],
1024 interpolation=InterpolationType.LINEAR,
1025 extrapolation=ExtrapolationType.LINEAR,
1026 )
1028 # --- Legs ---
1029 ns = ConstNotionalStructure(100)
1030 self.fixed_leg = IrFixedLegSpecification(
1031 fixed_rate=0.05,
1032 obj_id="fixed_leg",
1033 notional=100.0,
1034 start_dates=[self.refdate],
1035 end_dates=[self.maturity],
1036 pay_dates=[self.maturity],
1037 currency=self.ccy,
1038 day_count_convention="Act360",
1039 )
1041 self.float_leg = IrFloatLegSpecification(
1042 obj_id="float_leg",
1043 notional=ns,
1044 reset_dates=[self.refdate],
1045 start_dates=[self.refdate],
1046 end_dates=[self.maturity],
1047 rate_start_dates=[self.refdate],
1048 rate_end_dates=[self.maturity],
1049 pay_dates=[self.maturity],
1050 currency=self.ccy,
1051 udl_id="test_udl",
1052 fixing_id="test_fixing",
1053 day_count_convention="Act360",
1054 spread=0.0,
1055 )
1057 self.spec = InterestRateSwapSpecification(
1058 obj_id="test_swap",
1059 notional=ns,
1060 issue_date=self.refdate,
1061 maturity_date=self.maturity,
1062 pay_leg=self.fixed_leg,
1063 receive_leg=self.float_leg,
1064 currency=self.ccy,
1065 day_count_convention="Act360",
1066 issuer="issuer",
1067 securitization_level="COLLATERALIZED",
1068 )
1070 self.pricing_request = unittest.mock.Mock(name="InterestRateSwapPricingRequest")
1072 def test_init_success_minimal(self):
1073 """Should initialize successfully with minimal valid inputs."""
1074 pricer = InterestRateSwapPricer(
1075 val_date=self.refdate,
1076 spec=self.spec,
1077 discount_curve_pay_leg=self.dc,
1078 discount_curve_receive_leg=self.dc,
1079 fixing_curve_pay_leg=self.dc,
1080 fixing_curve_receive_leg=self.dc,
1081 fx_fwd_curve_pay_leg=self.dc,
1082 fx_fwd_curve_receive_leg=self.dc,
1083 pricing_request=self.pricing_request,
1084 )
1086 self.assertEqual(pricer._val_date, self.refdate)
1087 self.assertIs(pricer._spec, self.spec)
1088 self.assertIs(pricer._pay_leg, self.spec.pay_leg)
1089 self.assertIs(pricer._receive_leg, self.spec.receive_leg)
1090 self.assertEqual(pricer._fx_pay_leg, 1.0)
1091 self.assertEqual(pricer._fx_receive_leg, 1.0)
1092 self.assertIsInstance(pricer._pricing_param, dict)
1093 self.assertEqual(pricer._pricing_param, {})
1095 def test_init_success_with_optional_args(self):
1096 """Should properly assign optional args like pricing_param and fixing_map."""
1097 mock_fixing_map = unittest.mock.Mock(name="FixingTable")
1098 pricing_param = {"fixing_grace_period": 2}
1100 pricer = InterestRateSwapPricer(
1101 val_date=self.refdate,
1102 spec=self.spec,
1103 discount_curve_pay_leg=self.dc,
1104 discount_curve_receive_leg=self.dc,
1105 fixing_curve_pay_leg=self.dc,
1106 fixing_curve_receive_leg=self.dc,
1107 fx_fwd_curve_pay_leg=self.dc,
1108 fx_fwd_curve_receive_leg=self.dc,
1109 pricing_request=self.pricing_request,
1110 pricing_param=pricing_param,
1111 fixing_map=mock_fixing_map,
1112 fx_pay_leg=1.2,
1113 fx_receive_leg=0.8,
1114 )
1116 self.assertEqual(pricer._pricing_param, pricing_param)
1117 self.assertIs(pricer._fixing_map, mock_fixing_map)
1118 self.assertEqual(pricer._fx_pay_leg, 1.2)
1119 self.assertEqual(pricer._fx_receive_leg, 0.8)
1121 def test_init_raises_if_missing_required_args(self):
1122 """Should raise TypeError if required arguments are missing."""
1123 with self.assertRaises(TypeError):
1124 InterestRateSwapPricer(
1125 val_date=self.refdate,
1126 spec=self.spec,
1127 discount_curve_pay_leg=self.dc,
1128 discount_curve_receive_leg=self.dc,
1129 fixing_curve_pay_leg=self.dc,
1130 fixing_curve_receive_leg=self.dc,
1131 fx_fwd_curve_pay_leg=self.dc,
1132 # missing fx_fwd_curve_receive_leg
1133 pricing_request=self.pricing_request,
1134 )
1136 def test_init_all_curves_assigned_correctly(self):
1137 """Ensure all curve attributes are assigned to the correct legs."""
1138 pricer = InterestRateSwapPricer(
1139 val_date=self.refdate,
1140 spec=self.spec,
1141 discount_curve_pay_leg=self.dc,
1142 discount_curve_receive_leg=self.dc,
1143 fixing_curve_pay_leg=self.dc,
1144 fixing_curve_receive_leg=self.dc,
1145 fx_fwd_curve_pay_leg=self.dc,
1146 fx_fwd_curve_receive_leg=self.dc,
1147 pricing_request=self.pricing_request,
1148 )
1150 self.assertIs(pricer._discount_curve_pay_leg, self.dc)
1151 self.assertIs(pricer._discount_curve_receive_leg, self.dc)
1152 self.assertIs(pricer._fixing_curve_pay_leg, self.dc)
1153 self.assertIs(pricer._fixing_curve_receive_leg, self.dc)
1154 self.assertIs(pricer._fx_fwd_curve_pay_leg, self.dc)
1155 self.assertIs(pricer._fx_fwd_curve_receive_leg, self.dc)
1157 def test_init_pricing_param_is_new_dict(self):
1158 """Ensure pricing_param defaults to a new dict to avoid mutable default issues."""
1159 pricer1 = InterestRateSwapPricer(
1160 val_date=self.refdate,
1161 spec=self.spec,
1162 discount_curve_pay_leg=self.dc,
1163 discount_curve_receive_leg=self.dc,
1164 fixing_curve_pay_leg=self.dc,
1165 fixing_curve_receive_leg=self.dc,
1166 fx_fwd_curve_pay_leg=self.dc,
1167 fx_fwd_curve_receive_leg=self.dc,
1168 pricing_request=self.pricing_request,
1169 )
1170 pricer2 = InterestRateSwapPricer(
1171 val_date=self.refdate,
1172 spec=self.spec,
1173 discount_curve_pay_leg=self.dc,
1174 discount_curve_receive_leg=self.dc,
1175 fixing_curve_pay_leg=self.dc,
1176 fixing_curve_receive_leg=self.dc,
1177 fx_fwd_curve_pay_leg=self.dc,
1178 fx_fwd_curve_receive_leg=self.dc,
1179 pricing_request=self.pricing_request,
1180 )
1181 self.assertIsNot(pricer1._pricing_param, pricer2._pricing_param)
1184class TestGetProjectedNotionals(unittest.TestCase):
1186 def setUp(self):
1187 self.val_date = datetime(2025, 1, 1)
1188 self.val_notional = 1000
1190 def test_const_notional_structure(self):
1191 ns = ConstNotionalStructure(1000)
1192 result = get_projected_notionals(self.val_date, ns, 0, 3, fx_forward_curve=None)
1193 self.assertEqual(result, [1000, 1000, 1000])
1195 def test_resetting_notional_structure_with_fx(self):
1196 amounts = [100, 200, 300]
1197 fixing_dates = [
1198 datetime(2025, 2, 1),
1199 datetime(2025, 3, 1),
1200 datetime(2025, 4, 1),
1201 ]
1202 ns = ResettingNotionalStructure(
1203 ref_currency="USD",
1204 fx_fixing_id="EUR/USD",
1205 notionals=amounts,
1206 pay_date_start=[datetime(2025, 1, 1)] * 3,
1207 pay_date_end=[datetime(2025, 6, 1)] * 3,
1208 fixing_dates=fixing_dates,
1209 )
1210 # Mock FX forward curve: return 2.0 regardless of input
1211 fx_curve = unittest.mock.Mock()
1212 fx_curve.value.return_value = 2.0
1214 result = get_projected_notionals(self.val_date, ns, 0, 3, fx_curve)
1215 self.assertEqual(result, [200.0, 400.0, 600.0])
1216 fx_curve.value.assert_called() # ensure it was used
1218 def test_resetting_notional_structure_without_fx_raises(self):
1219 fixing_dates = [
1220 datetime(2025, 2, 1),
1221 datetime(2025, 3, 1),
1222 datetime(2025, 4, 1),
1223 ]
1224 ns = ResettingNotionalStructure(
1225 ref_currency="USD",
1226 fx_fixing_id="EUR/USD",
1227 notionals=[100, 200, 300],
1228 pay_date_start=[datetime(2025, 1, 1)] * 3,
1229 pay_date_end=[datetime(2025, 6, 1)] * 3,
1230 fixing_dates=fixing_dates,
1231 )
1232 with self.assertRaises(ValueError):
1233 get_projected_notionals(self.val_date, ns, 0, 1, fx_forward_curve=None)
1236# TODO once implemented: computeSwapSpread,
1239if __name__ == "__main__":
1240 unittest.main()