Coverage for tests / test_bootstrap.py: 96%
1019 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1# 2025.09.09 Bootstrapping without pyvacon
2import unittest
3import sys
5from tests.setup_logging import setup_logging_for_tests
7# Configure logging once per test module
8setup_logging_for_tests("tests/rivapy_test.log")
9import logging
11logger = logging.getLogger("rivapy.tests.test_bootstrap")
13import math
14import pandas as pd
15from datetime import date, datetime, timedelta
16from dateutil.relativedelta import relativedelta
17import numpy as np
19from rivapy.marketdata.bootstrapping import (
20 bootstrap_curve,
21 error_fn,
22 find_bracket,
23 get_quote,
24)
25from rivapy.marketdata.curves import DiscountCurve
26from rivapy.instruments.deposit_specifications import DepositSpecification
27from rivapy.instruments.fra_specifications import ForwardRateAgreementSpecification
28from rivapy.instruments.ir_swap_specification import (
29 InterestRateSwapSpecification,
30 IrFixedLegSpecification,
31 IrFloatLegSpecification,
32 IrOISLegSpecification,
33 InterestRateBasisSwapSpecification,
34)
35from rivapy.tools.enums import DayCounterType, InterpolationType, ExtrapolationType, Instrument
36from rivapy.instruments.components import ConstNotionalStructure
37from rivapy.tools.datetools import DayCounter, Period, Schedule, calc_end_day, calc_start_day
40# for specification from file tests
41import rivapy.instruments.specification_from_csv as sfc
42from rivapy.tools.holidays_compat import HolidayBase as _HolidayBase, EuropeanCentralBank as _ECB
45# Helper functions
46def tolerance_from_quote(q: float) -> float:
47 """
48 Determine an appropriate delta for assertAlmostEqual
49 based on the number of decimals in the quote.
50 """
51 s = format(q, "f").rstrip("0").rstrip(".")
52 decimals = len(s.split(".")[1]) if "." in s else 0
53 delta = 0.5 * 10 ** (-decimals)
54 return delta
57def deep_equal_OLD(obj1, obj2):
58 if type(obj1) != type(obj2):
59 return False
60 if hasattr(obj1, "__dict__") and hasattr(obj2, "__dict__"):
61 return all(deep_equal(v, obj2.__dict__[k]) for k, v in obj1.__dict__.items())
62 if isinstance(obj1, (list, tuple)):
63 return all(deep_equal(x, y) for x, y in zip(obj1, obj2))
64 return obj1 == obj2
67def deep_equal(obj1, obj2, path="root"):
68 """Recursively compare two objects and print where they differ."""
69 if type(obj1) != type(obj2):
70 print(f"Type mismatch at {path}: {type(obj1)} != {type(obj2)}")
71 return False
73 # handle objects with __dict__ (custom classes)
74 if hasattr(obj1, "__dict__") and hasattr(obj2, "__dict__"):
75 all_equal = True
76 keys1, keys2 = set(obj1.__dict__.keys()), set(obj2.__dict__.keys())
78 for key in keys1 | keys2:
79 if key not in obj1.__dict__:
80 print(f"Missing key {path}.{key} in obj1")
81 all_equal = False
82 continue
83 if key not in obj2.__dict__:
84 print(f"Missing key {path}.{key} in obj2")
85 all_equal = False
86 continue
88 if not deep_equal(obj1.__dict__[key], obj2.__dict__[key], f"{path}.{key}"):
89 all_equal = False
90 return all_equal
92 # handle lists and tuples
93 if isinstance(obj1, (list, tuple)):
94 all_equal = True
95 for i, (x, y) in enumerate(zip(obj1, obj2)):
96 if not deep_equal(x, y, f"{path}[{i}]"):
97 all_equal = False
98 if len(obj1) != len(obj2):
99 print(f"Length mismatch at {path}: {len(obj1)} != {len(obj2)}")
100 all_equal = False
101 return all_equal
103 # base case: primitive comparison
104 if obj1 != obj2:
105 print(f"Value mismatch at {path}: {obj1} != {obj2}")
106 return False
108 return True
111def update_fra_tenors(df):
112 """
113 Overwrite UnderlyingTenor for FRA instruments based on Maturity like '3MX6M'.
114 """
116 def fra_tenor_delta(maturity_str):
117 if pd.isna(maturity_str):
118 return maturity_str
119 maturity_str = maturity_str.upper().strip()
120 if "X" in maturity_str:
121 try:
122 start, end = maturity_str.split("X")
124 # Convert start and end to months
125 def to_months(s):
126 if s.endswith("D"):
127 return float(s[:-1]) / 30
128 elif s.endswith("W"):
129 return float(s[:-1]) * 7 / 30
130 elif s.endswith("M"):
131 return float(s[:-1])
132 elif s.endswith("Y"):
133 return float(s[:-1]) * 12
134 else:
135 return 0.0
137 start_m = to_months(start)
138 end_m = to_months(end)
139 delta_m = end_m - start_m
141 # Return as string with 'M'
142 return f"{int(delta_m)}M" if delta_m.is_integer() else f"{delta_m:.2f}M"
143 except Exception:
144 return maturity_str
145 else:
146 return maturity_str
148 # Only apply to FRA instruments
149 mask = df["Instrument"] == "FRA" # adjust column name if necessary
150 df.loc[mask, "UnderlyingTenor"] = df.loc[mask, "Maturity"].apply(fra_tenor_delta)
152 return df
155# Minimal instrument specification classes for testing
156class DummyDepositSpec(DepositSpecification):
157 def __init__(self, maturity_date=None, issue_date=None, ref_date=None):
158 """Setting up base deposit specification for tests.
159 For now, as O/N deposit with 1 day accrual.
160 """
161 ##########################################
162 # setting up deposit
163 # calculation date
164 if ref_date is None:
165 ref_date = datetime(2019, 8, 31)
167 # start date of the accrual period with spot lag equal to 2 days
168 if issue_date is None:
169 issue_date = ref_date + timedelta(days=2)
171 # end date of the accrual period is 1 day after startdate
172 if maturity_date is None:
173 maturity_date = issue_date + timedelta(days=1)
175 super().__init__(
176 obj_id="dummy_deposit",
177 issuer="dummy_issuer",
178 currency="EUR",
179 issue_date=issue_date,
180 maturity_date=maturity_date,
181 notional=1.0,
182 rate=0.01,
183 day_count_convention="Act360",
184 )
187class TestBootstrapCurveFunctions(unittest.TestCase):
189 def test_input_length_assertion(self):
190 """Test that providing different lengths of instruments and quotes raises an AssertionError."""
191 with self.assertRaises(AssertionError):
192 bootstrap_curve(
193 ref_date=date(2024, 1, 1),
194 curve_id="curve1",
195 day_count_convention=DayCounterType.ThirtyU360,
196 instruments=[DummyDepositSpec(maturity_date=datetime(2025, 1, 1))],
197 quotes=[],
198 )
200 def test_duplicate_end_dates(self):
201 """Test that duplicate end dates in instruments raises an error.
202 This is a design choice for the moment to keep bootstrapping logic simple.
203 """
205 inst1 = DummyDepositSpec(date(2025, 1, 1))
206 inst2 = DummyDepositSpec(date(2025, 1, 1))
207 with self.assertRaises(Exception) as cm:
208 bootstrap_curve(
209 ref_date=datetime(2024, 1, 1),
210 curve_id="curve1",
211 day_count_convention=DayCounterType.ThirtyU360,
212 instruments=[inst1, inst2],
213 quotes=[0.01, 0.02],
214 )
215 self.assertIn("Duplicate expiry date", str(cm.exception))
217 def test_successful_bootstrap(self):
218 """Test if bootstrap_curve runs without error and returns a DiscountCurve
219 Does not check for correctness of the curve.
220 """
221 # here this end_date is being saved into the maturity date..., and the end date is calculated internally in deposit spec
222 # inst = DummyDepositSpec(ref_date=datetime(2024, 1, 1), start_date=datetime(2024, 1, 1), end_date=datetime(2025, 1, 1))
223 # here we input an end date that would coincide with the maturity date by desgien
224 inst = DummyDepositSpec(ref_date=datetime(2024, 1, 1), issue_date=datetime(2024, 1, 1), maturity_date=datetime(2024, 1, 3))
225 result = bootstrap_curve(
226 ref_date=datetime(2024, 1, 1),
227 curve_id="curve1",
228 day_count_convention=DayCounterType.ThirtyU360,
229 instruments=[inst],
230 quotes=[0.01],
231 )
233 self.assertIsInstance(result, DiscountCurve)
234 self.assertEqual(result.get_dates()[0], datetime(2024, 1, 1))
235 self.assertEqual(result.get_dates()[1], datetime(2024, 1, 3))
238class TestErrorFn(unittest.TestCase):
239 """Tests on the error function used for the brentq solver used for iteratively solving for discount factors
240 regardless of instrument type.
242 Args:
243 unittest (_type_): _description_
244 """
246 def test_error_fn_returns_float(self):
247 """Test function output type check"""
248 inst = DummyDepositSpec(datetime(2024, 1, 3))
249 dfs = [1.0, 0.99]
250 yc_dates = [datetime(2024, 1, 1), datetime(2024, 1, 3)]
251 curves = {
252 "discount_curve": DiscountCurve(
253 "test", datetime(2024, 1, 1), yc_dates, dfs, InterpolationType.LINEAR, ExtrapolationType.LINEAR, DayCounterType.ThirtyU360
254 )
255 }
256 result = error_fn(
257 df_val=0.98,
258 index=1,
259 dfs=dfs.copy(),
260 yc_dates=yc_dates,
261 instrument_spec=inst,
262 ref_date=datetime(2024, 1, 1),
263 ref_quote=0.01,
264 curves=curves,
265 interpolation_type=InterpolationType.LINEAR,
266 extrapolation_type=ExtrapolationType.LINEAR,
267 )
268 self.assertIsInstance(result, float)
271class TestFindBracket(unittest.TestCase):
272 """Test for the helper function used in initial testing for bretnq solver.
273 Helper function was made for the case that inital guesses
274 were far off from potential root values to find better initial brackets.
276 Args:
277 unittest (_type_): _description_
278 """
280 def test_find_bracket_success(self):
281 """Caveat is that the inital guess should not produce a boundary value that is also the intended root."""
283 def fake_error_fn(x, *args):
284 return x - 1
286 ARGS = ()
287 lower, upper = find_bracket(fake_error_fn, 1.5, ARGS)
288 self.assertLess(lower, 1)
289 self.assertGreater(upper, 1)
291 def test_find_bracket_failure(self):
292 def fake_error_fn(x, *args):
293 return 1
295 ARGS = ()
296 with self.assertRaises(RuntimeError):
297 find_bracket(fake_error_fn, 2, ARGS)
300class TestGetQuote(unittest.TestCase):
301 """Testing get_quote for different instrument types
303 Args:
304 unittest (_type_): _description_
305 """
307 def test_get_quote_deposit(self):
308 refdate = datetime(2024, 1, 1)
309 inst = DummyDepositSpec(ref_date=refdate, issue_date=refdate, maturity_date=datetime(2024, 1, 3))
310 days_to_maturity = [1, 180, 365, 720, 3 * 365, 4 * 365, 10 * 365]
311 dates = [refdate + timedelta(days=d) for d in days_to_maturity]
312 flat_rate = 0.025
313 df = [math.exp(-d / 365.0 * flat_rate) for d in days_to_maturity]
314 curve = DiscountCurve(
315 id="dummy discount curve",
316 refdate=refdate,
317 dates=dates,
318 df=df,
319 interpolation=InterpolationType.LINEAR,
320 extrapolation=ExtrapolationType.LINEAR,
321 )
322 result = get_quote(
323 ref_date=refdate,
324 instrument_spec=inst,
325 curve_dict={"discount_curve": curve},
326 )
327 # self.assertEqual(result, 0.01)
328 self.assertAlmostEqual(result, 0.024508664452316253, delta=1e-8)
330 def test_get_quote_fra(self):
332 ref_date = datetime(2023, 1, 28)
333 start_date = datetime(2023, 7, 28) # 6mx3m
334 end_date = datetime(2023, 10, 28)
335 inst = ForwardRateAgreementSpecification(
336 obj_id="dummy_id",
337 trade_date=ref_date,
338 # maturity_date=mat_date,
339 notional=1000.0,
340 rate=0.04,
341 start_date=start_date,
342 end_date=end_date,
343 udlID="dummy_underlying_index",
344 rate_start_date=start_date,
345 rate_end_date=end_date,
346 day_count_convention="Act360",
347 rate_day_count_convention="Act360",
348 currency="EUR",
349 spot_days=1,
350 payment_days=1,
351 issuer="dummy_issuer",
352 securitization_level="NONE",
353 )
355 # fwd curve
356 object_id = "TEST_fwd"
357 fwd_rate = 0.05
358 days_to_maturity = [1, 180, 365, 720, 3 * 365, 4 * 365, 10 * 365]
359 dates = [ref_date + timedelta(days=d) for d in days_to_maturity]
360 fwd_df = [math.exp(-d / 365.0 * fwd_rate) for d in days_to_maturity]
361 fwd_dc = DiscountCurve(
362 id=object_id,
363 refdate=ref_date,
364 dates=dates,
365 df=fwd_df,
366 interpolation=InterpolationType.LINEAR,
367 extrapolation=ExtrapolationType.LINEAR,
368 daycounter=DayCounterType.ACT360,
369 )
371 # note that since we are getting to the FRA pricing through the 'get_quoute' function which
372 # is in the framework of curve bootstrapping a single curve, where the discounting,
373 # and forecast of forward is done with the same curve and built into the usage of FRAs in
374 # context of single curve bootstrapping here
375 # proper unit tests of the pricing or fair rate computation of FRAs is to be done in the FRAs section.
376 result = get_quote(
377 ref_date=ref_date,
378 instrument_spec=inst,
379 curve_dict={"discount_curve": fwd_dc},
380 )
381 self.assertAlmostEqual(result, 0.049315806932066504, delta=1e-8)
383 def test_get_quote_irs(self):
385 # 1Y maturity with quartlery payment
386 ref_date = datetime(2019, 8, 31)
387 start_dates = [ref_date + relativedelta(months=3 * i) for i in range(4)]
389 # reset dates are equal to start dates if spot lag is 0.
390 reset_dates = start_dates
392 # the end dates of the accral periods
393 end_dates = [x + relativedelta(months=3) for x in start_dates]
394 pay_dates = end_dates
395 ns = ConstNotionalStructure(100.0)
396 spread = 0.00
398 # float leg spec
399 float_leg = IrFloatLegSpecification(
400 obj_id="dummy_float_leg",
401 notional=ns,
402 reset_dates=reset_dates,
403 start_dates=start_dates,
404 end_dates=end_dates,
405 rate_start_dates=start_dates,
406 rate_end_dates=end_dates,
407 pay_dates=pay_dates,
408 currency="EUR",
409 udl_id="test_udl_id",
410 fixing_id="test_fixing_id",
411 day_count_convention="Act365Fixed",
412 spread=spread,
413 )
415 # # definition of the fixed leg
416 fixed_leg = IrFixedLegSpecification(
417 fixed_rate=0.01,
418 obj_id="dummy_fixed_leg",
419 notional=100.0,
420 start_dates=start_dates,
421 end_dates=end_dates,
422 pay_dates=pay_dates,
423 currency="EUR",
424 day_count_convention="Act365Fixed",
425 )
427 # # definition of the IR swap
428 ir_swap = InterestRateSwapSpecification(
429 obj_id="3M_SWAP",
430 notional=ns,
431 issue_date=ref_date,
432 maturity_date=pay_dates[-1],
433 pay_leg=fixed_leg,
434 receive_leg=float_leg,
435 currency="EUR",
436 day_count_convention="Act365Fixed",
437 issuer="dummy_issuer",
438 securitization_level="COLLATERALIZED",
439 )
441 curve = DiscountCurve(
442 "test",
443 ref_date,
444 [ref_date, ref_date + relativedelta(years=1)],
445 [1.0, 0.99],
446 InterpolationType.LINEAR,
447 ExtrapolationType.LINEAR,
448 DayCounterType.ThirtyU360,
449 )
450 result = get_quote(
451 ref_date=ref_date,
452 instrument_spec=ir_swap,
453 curve_dict={"discount_curve": curve, "fixing_curve": curve},
454 ) # in the context of bootstrapping a single curve, we make the assumption that discount curve and fixing curve are the same
455 # in case of irswap- compute swap rate implies calculating the annuity =>
456 print(result)
457 self.assertAlmostEqual(result, 0.010090861477238481, delta=1e-8)
460class TestBootstrapCurveInstruments(unittest.TestCase):
461 """Test for various combination of instruments supplied to the bootstrapper
463 Args:
464 unittest (_type_): _description_
465 """
467 def setUp(self):
468 self.ref_date = datetime(2019, 8, 31)
469 self.curve_id = "EUR_DISC"
470 self.day_count = DayCounterType.Act365Fixed
471 self.interp = InterpolationType.LINEAR
472 self.extrap = ExtrapolationType.LINEAR
474 def test_deposit_bootstrap(self):
475 # Minimal deposit: 6M, 2% rate
476 logger.debug("Creating 1 deposit instrument")
477 issue_date = self.ref_date + timedelta(days=2) # spot lag of 2 days
478 maturity_date = issue_date + timedelta(days=1) # 1 day after startdate
479 deposit = DepositSpecification(
480 obj_id="dep1",
481 notional=100,
482 issue_date=issue_date,
483 maturity_date=maturity_date,
484 currency="EUR",
485 day_count_convention=self.day_count,
486 rate=0.01, # rate different from "market quote" to ensure that rate here is NOT used in deposit bootstrapping
487 )
488 logger.debug("Running single curve bootstrapping")
489 curve = bootstrap_curve(
490 ref_date=self.ref_date,
491 curve_id=self.curve_id,
492 day_count_convention=self.day_count,
493 instruments=[deposit],
494 quotes=[0.025],
495 interpolation_type=self.interp,
496 extrapolation_type=self.extrap,
497 )
498 # print(curve.get_dates())
499 logger.debug("Checking assertions")
500 self.assertIsInstance(curve, DiscountCurve)
501 self.assertEqual(curve.get_dates()[0], self.ref_date)
502 self.assertEqual(curve.get_dates()[1], maturity_date)
504 # the discount curve needs to be able to get the same market quote for the instrument
505 model_quote = get_quote(self.ref_date, deposit, {"discount_curve": curve})
506 self.assertAlmostEqual(model_quote, 0.025, delta=1e-8) # not good enough, what is going on? not enough data points?
507 logger.debug("TestBootstrapCurveInstruments.test_deposit_bootstrap completed")
509 def test_fra_bootstrap(self):
510 # Minimal FRA: 6Mx3M, 2.5% rate
511 # start_date = self.ref_date + relativedelta(months=6)
512 # end_date = self.ref_date + relativedelta(months=9)
513 ref_date = datetime(2023, 1, 28)
514 issue_date = datetime(2023, 7, 28)
515 maturity_date = datetime(2023, 10, 28)
516 fra = ForwardRateAgreementSpecification(
517 obj_id="dummy_id",
518 trade_date=ref_date,
519 # maturity_date=mat_date,
520 notional=1000.0,
521 rate=0.025,
522 start_date=issue_date,
523 end_date=maturity_date,
524 udlID="dummy_underlying_index",
525 rate_start_date=issue_date,
526 rate_end_date=maturity_date,
527 day_count_convention=self.day_count,
528 rate_day_count_convention=self.day_count,
529 currency="EUR",
530 spot_days=1,
531 payment_days=1,
532 issuer="dummy_issuer",
533 securitization_level="NONE",
534 )
536 # Need a deposit for the first period to anchor the curve # i.e. in front of the FRA
537 deposit = DepositSpecification(
538 obj_id="dep1",
539 notional=1000.0,
540 issue_date=ref_date,
541 maturity_date=issue_date,
542 currency="EUR",
543 day_count_convention=self.day_count,
544 rate=0.02,
545 )
547 curve = bootstrap_curve(
548 ref_date=self.ref_date,
549 curve_id=self.curve_id,
550 day_count_convention=self.day_count,
551 instruments=[deposit, fra],
552 quotes=[0.02, 0.025],
553 interpolation_type=self.interp,
554 extrapolation_type=self.extrap,
555 )
557 self.assertIsInstance(curve, DiscountCurve)
558 self.assertEqual(curve.get_dates()[0], self.ref_date)
559 self.assertEqual(curve.get_dates()[1], issue_date)
560 self.assertEqual(curve.get_dates()[2], maturity_date)
562 def test_irs_bootstrap(self):
563 # test that at least, the bootstrap completes
564 # Minimal IRS: 1Y, fixed 3%, float 6M reset
565 start_date = self.ref_date
566 end_date = self.ref_date + timedelta(days=365)
567 pay_dates = [end_date]
568 ns = ConstNotionalStructure(100.0)
569 fixed_leg = IrFixedLegSpecification(
570 fixed_rate=0.03,
571 obj_id="fixed_leg",
572 notional=ns,
573 start_dates=[start_date],
574 end_dates=[end_date],
575 pay_dates=pay_dates,
576 currency="EUR",
577 day_count_convention=self.day_count,
578 )
579 float_leg = IrFloatLegSpecification(
580 obj_id="float_leg",
581 notional=ns,
582 reset_dates=[start_date],
583 start_dates=[start_date],
584 end_dates=[end_date],
585 rate_start_dates=[start_date],
586 rate_end_dates=[end_date],
587 pay_dates=pay_dates,
588 currency="EUR",
589 udl_id="EURIBOR6M",
590 fixing_id="EURIBOR6M",
591 day_count_convention=self.day_count,
592 rate_day_count_convention=self.day_count,
593 spread=0.0,
594 )
595 irs = InterestRateSwapSpecification(
596 obj_id="swap1",
597 notional=ns,
598 issue_date=start_date,
599 maturity_date=end_date,
600 pay_leg=fixed_leg,
601 receive_leg=float_leg,
602 currency="EUR",
603 day_count_convention=self.day_count,
604 )
605 # Need a deposit to anchor the curve
606 deposit = DepositSpecification(
607 obj_id="dep1",
608 notional=1_000_000,
609 issue_date=self.ref_date,
610 maturity_date=self.ref_date + timedelta(days=182),
611 currency="EUR",
612 day_count_convention=self.day_count,
613 rate=0.02,
614 )
615 curve = bootstrap_curve(
616 ref_date=self.ref_date,
617 curve_id=self.curve_id,
618 day_count_convention=self.day_count,
619 instruments=[deposit, irs],
620 quotes=[0.02, 0.03],
621 interpolation_type=self.interp,
622 extrapolation_type=self.extrap,
623 )
624 self.assertIsInstance(curve, DiscountCurve)
625 self.assertIn(end_date, curve.get_dates())
627 # test that it reproduces the model quotes
628 model_quotes = []
629 # pricing_params = {"fixing_grace_period": 0.0, "set_rate": True, "desired_rate": 1.0}
630 curves_dict = {"discount_curve": curve, "fixing_curve": curve}
632 for i in range(len([deposit, irs])):
634 model_quote = get_quote(self.ref_date, [deposit, irs][i], curves_dict)
635 # print(model_quote)
636 model_quotes.append(model_quote)
638 self.assertAlmostEqual(model_quotes[0], 0.02, delta=1e-8)
639 self.assertAlmostEqual(model_quotes[1], 0.03, delta=1e-8)
641 def test_deposit_fra_irs_combination(self):
642 # Deposit, FRA, IRS in sequence
643 # d1 = self.ref_date + timedelta(days=182)
644 # d2 = self.ref_date + timedelta(days=365)
645 # d3 = self.ref_date + timedelta(days=730)
647 ref_date = datetime(2023, 1, 28)
648 # start_date = datetime(2023, 7, 28)
649 # end_date = datetime(2023, 10, 28)
651 d1 = datetime(2023, 7, 28)
652 d2 = datetime(2023, 10, 28)
653 d3 = datetime(2024, 10, 28)
655 deposit = DepositSpecification(
656 obj_id="dep1",
657 notional=1_000_000,
658 issue_date=ref_date,
659 maturity_date=d1,
660 currency="EUR",
661 day_count_convention=self.day_count,
662 rate=0.02,
663 )
664 fra = fra = ForwardRateAgreementSpecification(
665 obj_id="dummy_id",
666 trade_date=ref_date,
667 # maturity_date=mat_date,
668 notional=1_000_000.0,
669 rate=0.025,
670 start_date=d1,
671 end_date=d2,
672 udlID="dummy_underlying_index",
673 rate_start_date=d1,
674 rate_end_date=d2,
675 day_count_convention=self.day_count,
676 rate_day_count_convention=self.day_count,
677 currency="EUR",
678 spot_days=1,
679 payment_days=1,
680 issuer="dummy_issuer",
681 securitization_level="NONE",
682 )
683 # IRS 1Y from d2 to d3
684 ns = ConstNotionalStructure(1_000_000.0)
685 pay_dates = [d3]
686 fixed_leg = IrFixedLegSpecification(
687 fixed_rate=0.03,
688 obj_id="fixed_leg",
689 notional=ns,
690 start_dates=[d2],
691 end_dates=[d3],
692 pay_dates=pay_dates,
693 currency="EUR",
694 day_count_convention=self.day_count,
695 )
696 float_leg = IrFloatLegSpecification(
697 obj_id="float_leg",
698 notional=ns,
699 reset_dates=[d2],
700 start_dates=[d2],
701 end_dates=[d3],
702 rate_start_dates=[d2],
703 rate_end_dates=[d3],
704 pay_dates=pay_dates,
705 currency="EUR",
706 udl_id="EURIBOR6M",
707 fixing_id="EURIBOR6M",
708 day_count_convention=self.day_count,
709 rate_day_count_convention=self.day_count,
710 spread=0.0,
711 )
712 irs = InterestRateSwapSpecification(
713 obj_id="swap1",
714 notional=ns,
715 issue_date=d2,
716 maturity_date=d3,
717 pay_leg=fixed_leg,
718 receive_leg=float_leg,
719 currency="EUR",
720 day_count_convention=self.day_count,
721 )
723 instruments = [deposit, fra, irs]
724 quotes = [0.02, 0.025, 0.03]
725 curve = bootstrap_curve(
726 ref_date=ref_date,
727 curve_id=self.curve_id,
728 day_count_convention=self.day_count,
729 instruments=instruments,
730 quotes=quotes,
731 interpolation_type=self.interp,
732 extrapolation_type=self.extrap,
733 )
734 self.assertIsInstance(curve, DiscountCurve)
735 self.assertIn(d1, curve.get_dates())
736 self.assertIn(d2, curve.get_dates())
737 self.assertIn(d3, curve.get_dates())
739 # test that it reproduces the model quotes
740 model_quotes = []
741 # pricing_params = {"fixing_grace_period": 0.0, "set_rate": True, "desired_rate": 1.0}
742 curves_dict = {"discount_curve": curve, "fixing_curve": curve}
744 for i in range(len(instruments)):
746 model_quote = get_quote(ref_date, instruments[i], curves_dict)
747 # print(model_quote)
748 model_quotes.append(model_quote)
750 self.assertAlmostEqual(model_quotes[0], quotes[0], delta=1e-8)
751 self.assertAlmostEqual(model_quotes[1], quotes[1], delta=1e-8)
752 self.assertAlmostEqual(model_quotes[2], quotes[2], delta=1e-8)
754 def test_fra_irs_combination_forward_curve(self):
755 """Here we test the case where a discount curve is GIVEN, and that we want the FRA to also contribute to the fwd curve"""
756 # Deposit, FRA, IRS in sequence
757 # d1 = self.ref_date + timedelta(days=182)
758 # d2 = self.ref_date + timedelta(days=365)
759 # d3 = self.ref_date + timedelta(days=730)
761 ref_date = datetime(2023, 1, 28)
762 # start_date = datetime(2023, 7, 28)
763 # end_date = datetime(2023, 10, 28)
765 d1 = datetime(2023, 7, 28)
766 d2 = datetime(2023, 10, 28)
767 d3 = datetime(2024, 10, 28)
769 fra = fra = ForwardRateAgreementSpecification(
770 obj_id="dummy_id",
771 trade_date=ref_date,
772 # maturity_date=mat_date,
773 notional=1_000_000.0,
774 rate=0.025,
775 start_date=d1,
776 end_date=d2,
777 udlID="dummy_underlying_index",
778 rate_start_date=d1,
779 rate_end_date=d2,
780 day_count_convention=self.day_count,
781 rate_day_count_convention=self.day_count,
782 currency="EUR",
783 spot_days=1,
784 payment_days=1,
785 issuer="dummy_issuer",
786 securitization_level="NONE",
787 )
788 # IRS 1Y from d2 to d3
789 ns = ConstNotionalStructure(1_000_000.0)
790 pay_dates = [d3]
791 fixed_leg = IrFixedLegSpecification(
792 fixed_rate=0.03,
793 obj_id="fixed_leg",
794 notional=ns,
795 start_dates=[d2],
796 end_dates=[d3],
797 pay_dates=pay_dates,
798 currency="EUR",
799 day_count_convention=self.day_count,
800 )
801 float_leg = IrFloatLegSpecification(
802 obj_id="float_leg",
803 notional=ns,
804 reset_dates=[d2],
805 start_dates=[d2],
806 end_dates=[d3],
807 rate_start_dates=[d2],
808 rate_end_dates=[d3],
809 pay_dates=pay_dates,
810 currency="EUR",
811 udl_id="EURIBOR6M",
812 fixing_id="EURIBOR6M",
813 day_count_convention=self.day_count,
814 rate_day_count_convention=self.day_count,
815 spread=0.0,
816 )
817 irs = InterestRateSwapSpecification(
818 obj_id="swap1",
819 notional=ns,
820 issue_date=d2,
821 maturity_date=d3,
822 pay_leg=fixed_leg,
823 receive_leg=float_leg,
824 currency="EUR",
825 day_count_convention=self.day_count,
826 )
828 instruments = [fra, irs]
829 quotes = [0.025, 0.03]
831 days_to_maturity = [180, 360, 540]
832 dates = [ref_date + timedelta(days=d) for d in days_to_maturity]
833 # discount factors from constant rate
834 rates = [0.10, 0.105, 0.11]
835 df = [math.exp(-r * d / 360) for r, d in zip(rates, days_to_maturity)]
837 dc = DiscountCurve("bootstrappedDC", ref_date, dates, df, self.interp, self.extrap, self.day_count)
838 curve = bootstrap_curve(
839 ref_date=ref_date,
840 curve_id=self.curve_id,
841 day_count_convention=self.day_count,
842 instruments=instruments,
843 quotes=quotes,
844 interpolation_type=self.interp,
845 extrapolation_type=self.extrap,
846 curves={"discount_curve": dc},
847 )
848 self.assertIsInstance(curve, DiscountCurve)
849 # self.assertIn(d1, curve.get_dates()) # was the end date for the deposits that we removed for this test
850 self.assertIn(d2, curve.get_dates())
851 self.assertIn(d3, curve.get_dates())
853 # test that it reproduces the model quotes
854 model_quotes = []
855 # pricing_params = {"fixing_grace_period": 0.0, "set_rate": True, "desired_rate": 1.0}
856 curves_dict = {"discount_curve": dc, "fixing_curve": curve}
858 for i in range(len(instruments)):
860 model_quote = get_quote(ref_date, instruments[i], curves_dict, {"flag_multi_curve": True})
861 # print(model_quote)
862 model_quotes.append(model_quote)
864 self.assertAlmostEqual(model_quotes[0], quotes[0], delta=1e-8)
865 self.assertAlmostEqual(model_quotes[1], quotes[1], delta=1e-8)
866 # self.assertAlmostEqual(model_quotes[2], quotes[2], delta=1e-8)
868 def test_many_instruments(self):
869 """Case where lots of instruments are given"""
870 # calculation date
871 ref_date = datetime(2019, 8, 31) # refdate = dt.datetime(2017, 8, 31)
872 # start date of the accrual period with spot lag equal to 2 days
873 start_date = ref_date + timedelta(days=2)
875 end_date_deposits = [
876 start_date + timedelta(days=1),
877 start_date + timedelta(days=7),
878 start_date + timedelta(days=30),
879 start_date + timedelta(days=60),
880 start_date + timedelta(days=90),
881 start_date + timedelta(days=181), # 180 - error due to BCC roll oveer mismatch between start and end date... look into! #TODO
882 start_date + timedelta(days=270),
883 ]
885 # Provide a list of market quotes from which to derive the discount curve
886 quotes_deposits = [0.025, 0.028, 0.0283, 0.029, 0.0305, 0.0315, 0.0348]
888 # List where the multiple deposit specfications are stored
889 multiple_deposits = []
891 for i in range(len(quotes_deposits)):
892 print(i)
893 temp_deposit = DepositSpecification(
894 obj_id="DEPOSIT_" + str(i + 1),
895 issuer="dummy_issuer",
896 currency="EUR",
897 issue_date=start_date,
898 maturity_date=end_date_deposits[i],
899 notional=100.0,
900 rate=quotes_deposits[i],
901 day_count_convention="Act365Fixed",
902 )
903 multiple_deposits.append(temp_deposit)
905 # setting up swaps
906 start_dates = [ref_date + relativedelta(months=3 * i) for i in range(4)]
907 # reset dates are equal to start dates if spot lag is 0.
908 reset_dates = start_dates
909 # the end dates of the accral periods
910 end_dates = [x + relativedelta(months=3) for x in start_dates]
911 # the actual payment dates of the cashflows may differ from the end of the accrual period (e.g. OIS).
912 # in the standard case these two sets of dates coincide
913 pay_dates = end_dates
915 ns = ConstNotionalStructure(100.0)
916 spread = 0.00
917 # # definition of the floating leg
918 float_leg = IrFloatLegSpecification(
919 obj_id="dummy_float_leg",
920 notional=ns,
921 reset_dates=reset_dates,
922 start_dates=start_dates,
923 end_dates=end_dates,
924 rate_start_dates=start_dates,
925 rate_end_dates=end_dates,
926 pay_dates=pay_dates,
927 currency="EUR",
928 udl_id="test_udl_id",
929 fixing_id="test_fixing_id",
930 day_count_convention="Act365Fixed",
931 spread=spread,
932 )
934 # # definition of the fixed leg
935 # Note that a fixed rate is given for the specification as it is required.
936 # However, for the creation of the bootrstrapped curve, the market quotes are used as the target swap par rate
937 fixed_leg = IrFixedLegSpecification(
938 fixed_rate=0.01,
939 obj_id="dummy_fixed_leg",
940 notional=100.0,
941 start_dates=start_dates,
942 end_dates=end_dates,
943 pay_dates=pay_dates,
944 currency="EUR",
945 day_count_convention="Act365Fixed",
946 )
947 # # definition of the IR swap
948 ir_swap = InterestRateSwapSpecification(
949 obj_id="3M_SWAP",
950 notional=ns,
951 issue_date=ref_date,
952 maturity_date=pay_dates[-1],
953 pay_leg=fixed_leg,
954 receive_leg=float_leg,
955 currency="EUR",
956 day_count_convention="Act365Fixed",
957 issuer="dummy_issuer",
958 securitization_level="COLLATERALIZED",
959 )
961 # 2Y maturity 3M swap
962 start_dates2 = [ref_date + relativedelta(months=3 * i) for i in range(4 * 2)]
963 reset_dates2 = start_dates2
964 end_dates2 = [x + relativedelta(months=3) for x in start_dates2]
965 pay_dates2 = end_dates2
967 ns = ConstNotionalStructure(100.0)
968 spread = 0.00
970 # # definition of the floating leg
971 float_leg2 = IrFloatLegSpecification(
972 obj_id="dummy_float_leg2",
973 notional=ns,
974 reset_dates=reset_dates2,
975 start_dates=start_dates2,
976 end_dates=end_dates2,
977 rate_start_dates=start_dates2,
978 rate_end_dates=end_dates2,
979 pay_dates=pay_dates2,
980 currency="EUR",
981 udl_id="test_udl_id",
982 fixing_id="test_fixing_id",
983 day_count_convention="Act365Fixed",
984 spread=spread,
985 )
987 # # definition of the fixed leg
988 fixed_leg2 = IrFixedLegSpecification(
989 fixed_rate=0.01,
990 obj_id="dummy_fixed_leg2",
991 notional=100.0,
992 start_dates=start_dates2,
993 end_dates=end_dates2,
994 pay_dates=pay_dates2,
995 currency="EUR",
996 day_count_convention="Act365Fixed",
997 )
999 # # definition of the IR swap
1000 ir_swap2 = InterestRateSwapSpecification(
1001 obj_id="3M_SWAP2",
1002 notional=ns,
1003 issue_date=ref_date,
1004 maturity_date=pay_dates2[-1],
1005 pay_leg=fixed_leg2,
1006 receive_leg=float_leg2,
1007 currency="EUR",
1008 day_count_convention="Act365Fixed",
1009 issuer="dummy_issuer",
1010 securitization_level="COLLATERALIZED",
1011 )
1013 # 3Y maturity 3M swap
1015 start_dates3 = [ref_date + relativedelta(months=3 * i) for i in range(4 * 3)]
1016 reset_dates3 = start_dates3
1017 end_dates3 = [x + relativedelta(months=3) for x in start_dates3]
1018 pay_dates3 = end_dates3
1019 ns = ConstNotionalStructure(100.0)
1020 spread = 0.00
1022 # # definition of the floating leg
1023 float_leg3 = IrFloatLegSpecification(
1024 obj_id="dummy_float_leg3",
1025 notional=ns,
1026 reset_dates=reset_dates3,
1027 start_dates=start_dates3,
1028 end_dates=end_dates3,
1029 rate_start_dates=start_dates3,
1030 rate_end_dates=end_dates3,
1031 pay_dates=pay_dates3,
1032 currency="EUR",
1033 udl_id="test_udl_id",
1034 fixing_id="test_fixing_id",
1035 day_count_convention="Act365Fixed",
1036 spread=spread,
1037 )
1039 # # definition of the fixed leg
1040 fixed_leg3 = IrFixedLegSpecification(
1041 fixed_rate=0.01,
1042 obj_id="dummy_fixed_leg3",
1043 notional=100.0,
1044 start_dates=start_dates3,
1045 end_dates=end_dates3,
1046 pay_dates=pay_dates3,
1047 currency="EUR",
1048 day_count_convention="Act365Fixed",
1049 )
1051 # # definition of the IR swap
1052 ir_swap3 = InterestRateSwapSpecification(
1053 obj_id="3M_SWAP3",
1054 notional=ns,
1055 issue_date=ref_date,
1056 maturity_date=pay_dates3[-1],
1057 pay_leg=fixed_leg3,
1058 receive_leg=float_leg3,
1059 currency="EUR",
1060 day_count_convention="Act365Fixed",
1061 issuer="dummy_issuer",
1062 securitization_level="COLLATERALIZED",
1063 )
1065 # organising the swaps and their taarget market quotes
1066 multiple_swaps = [ir_swap, ir_swap2, ir_swap3]
1067 quotes_swaps = [0.05, 0.06, 0.07]
1069 instruments_both = multiple_deposits + multiple_swaps
1070 quotes_both = quotes_deposits + quotes_swaps
1072 boot_curve = bootstrap_curve(
1073 ref_date,
1074 "bootstrapped_DC",
1075 DayCounterType.Act365Fixed,
1076 instruments_both,
1077 quotes_both,
1078 interpolation_type=InterpolationType.LINEAR_LOG,
1079 extrapolation_type=ExtrapolationType.LINEAR_LOG,
1080 )
1082 model_quotes = []
1083 # pricing_params = {"fixing_grace_period": 0.0, "set_rate": True, "desired_rate": 1.0}
1084 curves_dict = {"discount_curve": boot_curve, "fixing_curve": boot_curve}
1086 for i in range(len(instruments_both)):
1087 model_quote = get_quote(ref_date, instruments_both[i], curves_dict)
1088 # print(model_quote)
1089 model_quotes.append(model_quote)
1091 for i in range(len(quotes_both)):
1093 self.assertAlmostEqual(model_quotes[i], quotes_both[i], delta=1e-8)
1095 def test_ois_multicurve_bootstrap(self):
1096 """Test for multicurve boot strapping.
1097 Create a discount curve based on 3M tenor, and create a discount curve from OIS instruments
1098 Create a forward curve based off this input curve and other instruments
1099 """
1100 ref_date = self.ref_date
1102 # setting up OIS
1103 ######################################################################
1104 # 1M Maturity, 1 M tenor
1106 start_dates = [ref_date + relativedelta(months=1 * i) for i in range(1)]
1108 # the end dates of the accrual periods
1109 end_dates = [x + relativedelta(months=1) for x in start_dates]
1111 # the actual payment dates of the cashflows may differ from the end of the accrual period (e.g. OIS).
1112 # in the standard case these two sets of dates coincide
1113 # pay_dates = end_dates
1115 print(end_dates)
1116 ns = ConstNotionalStructure(100.0)
1117 spread = 0.00
1119 # the difference here is that rate_date arrays are excpected to be 2 dimensional, i.e. keep track of the daily resetting per accrual period
1120 res = IrOISLegSpecification.ois_scheduler_2D(start_dates, end_dates)
1122 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
1123 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
1124 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
1125 pay_dates = res[3]
1127 float_leg = IrOISLegSpecification(
1128 obj_id="dummy_float_leg",
1129 notional=ns,
1130 rate_reset_dates=daily_rate_reset_dates,
1131 start_dates=start_dates,
1132 end_dates=end_dates,
1133 rate_start_dates=daily_rate_start_dates,
1134 rate_end_dates=daily_rate_end_dates,
1135 pay_dates=pay_dates,
1136 currency="EUR",
1137 udl_id="test_udl_id",
1138 fixing_id="test_fixing_id",
1139 day_count_convention="Act365Fixed",
1140 rate_day_count_convention="Act365Fixed",
1141 spread=spread,
1142 )
1144 # # definition of the fixed leg
1145 # Note that a fixed rate is given for the specification as it is required.
1146 # However, for the creation of the bootrstrapped curve, the market quotes are used as the target swap par rate
1147 fixed_leg = IrFixedLegSpecification(
1148 fixed_rate=0.01,
1149 obj_id="dummy_fixed_leg",
1150 notional=100.0,
1151 start_dates=start_dates,
1152 end_dates=end_dates,
1153 pay_dates=pay_dates,
1154 currency="EUR",
1155 day_count_convention="Act365Fixed",
1156 )
1158 # # definition of the IR swap
1159 ois_swap_1M = InterestRateSwapSpecification(
1160 obj_id="1M_SWAP",
1161 notional=ns,
1162 issue_date=ref_date,
1163 maturity_date=pay_dates[-1],
1164 pay_leg=fixed_leg,
1165 receive_leg=float_leg,
1166 currency="EUR",
1167 day_count_convention="Act365Fixed",
1168 issuer="dummy_issuer",
1169 securitization_level="COLLATERALIZED",
1170 )
1172 # 3M maturity 3M underlying tenor swap, i.e. the floating leg is reset every 3M
1173 # since it is OIS
1175 # start dates of the accrual periods corresponding to the tenor of the underlying index (3 months). The spot lag is set to 0.
1176 start_dates = [ref_date + relativedelta(months=3 * i) for i in range(1)]
1177 end_dates = [x + relativedelta(months=3) for x in start_dates]
1178 ns = ConstNotionalStructure(100.0)
1179 spread = 0.00
1180 res = IrOISLegSpecification.ois_scheduler_2D(start_dates, end_dates)
1182 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
1183 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
1184 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
1185 pay_dates = res[3]
1187 float_leg = IrOISLegSpecification(
1188 obj_id="dummy_float_leg",
1189 notional=ns,
1190 rate_reset_dates=daily_rate_reset_dates,
1191 start_dates=start_dates,
1192 end_dates=end_dates,
1193 rate_start_dates=daily_rate_start_dates,
1194 rate_end_dates=daily_rate_end_dates,
1195 pay_dates=pay_dates,
1196 currency="EUR",
1197 udl_id="test_udl_id",
1198 fixing_id="test_fixing_id",
1199 day_count_convention="Act365Fixed",
1200 rate_day_count_convention="Act365Fixed",
1201 spread=spread,
1202 )
1204 # # definition of the fixed leg
1205 # Note that a fixed rate is given for the specification as it is required.
1206 # However, for the creation of the bootrstrapped curve, the market quotes are used as the target swap par rate
1207 fixed_leg = IrFixedLegSpecification(
1208 fixed_rate=0.01,
1209 obj_id="dummy_fixed_leg",
1210 notional=100.0,
1211 start_dates=start_dates,
1212 end_dates=end_dates,
1213 pay_dates=pay_dates,
1214 currency="EUR",
1215 day_count_convention="Act365Fixed",
1216 )
1218 # # definition of the IR swap
1219 ois_swap_3M = InterestRateSwapSpecification(
1220 obj_id="3M_SWAP",
1221 notional=ns,
1222 issue_date=ref_date,
1223 maturity_date=pay_dates[-1],
1224 pay_leg=fixed_leg,
1225 receive_leg=float_leg,
1226 currency="EUR",
1227 day_count_convention="Act365Fixed",
1228 issuer="dummy_issuer",
1229 securitization_level="COLLATERALIZED",
1230 )
1232 # 6M Maturity, 6M tenor
1233 start_dates = [ref_date + relativedelta(months=6 * i) for i in range(1)]
1234 end_dates = [x + relativedelta(months=6) for x in start_dates]
1235 ns = ConstNotionalStructure(100.0)
1236 spread = 0.00
1237 res = IrOISLegSpecification.ois_scheduler_2D(start_dates, end_dates)
1239 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
1240 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
1241 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
1242 pay_dates = res[3]
1244 float_leg = IrOISLegSpecification(
1245 obj_id="dummy_float_leg",
1246 notional=ns,
1247 rate_reset_dates=daily_rate_reset_dates,
1248 start_dates=start_dates,
1249 end_dates=end_dates,
1250 rate_start_dates=daily_rate_start_dates,
1251 rate_end_dates=daily_rate_end_dates,
1252 pay_dates=pay_dates,
1253 currency="EUR",
1254 udl_id="test_udl_id",
1255 fixing_id="test_fixing_id",
1256 day_count_convention="Act365Fixed",
1257 rate_day_count_convention="Act365Fixed",
1258 spread=spread,
1259 )
1261 # # definition of the fixed leg
1263 fixed_leg = IrFixedLegSpecification(
1264 fixed_rate=0.01,
1265 obj_id="dummy_fixed_leg",
1266 notional=100.0,
1267 start_dates=start_dates,
1268 end_dates=end_dates,
1269 pay_dates=pay_dates,
1270 currency="EUR",
1271 day_count_convention="Act365Fixed",
1272 )
1274 # # definition of the IR swap
1275 ois_swap_6M = InterestRateSwapSpecification(
1276 obj_id="6M_SWAP",
1277 notional=ns,
1278 issue_date=ref_date,
1279 maturity_date=pay_dates[-1],
1280 pay_leg=fixed_leg,
1281 receive_leg=float_leg,
1282 currency="EUR",
1283 day_count_convention="Act365Fixed",
1284 issuer="dummy_issuer",
1285 securitization_level="COLLATERALIZED",
1286 )
1288 # 9M Maturity, 9M tenor
1289 start_dates = [ref_date + relativedelta(months=9 * i) for i in range(1)]
1290 end_dates = [x + relativedelta(months=9) for x in start_dates]
1291 ns = ConstNotionalStructure(100.0)
1292 spread = 0.00
1293 res = IrOISLegSpecification.ois_scheduler_2D(start_dates, end_dates)
1295 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
1296 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
1297 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
1298 pay_dates = res[3]
1300 float_leg = IrOISLegSpecification(
1301 obj_id="dummy_float_leg",
1302 notional=ns,
1303 rate_reset_dates=daily_rate_reset_dates,
1304 start_dates=start_dates,
1305 end_dates=end_dates,
1306 rate_start_dates=daily_rate_start_dates,
1307 rate_end_dates=daily_rate_end_dates,
1308 pay_dates=pay_dates,
1309 currency="EUR",
1310 udl_id="test_udl_id",
1311 fixing_id="test_fixing_id",
1312 day_count_convention="Act365Fixed",
1313 rate_day_count_convention="Act365Fixed",
1314 spread=spread,
1315 )
1317 # # definition of the fixed leg
1318 fixed_leg = IrFixedLegSpecification(
1319 fixed_rate=0.01,
1320 obj_id="dummy_fixed_leg",
1321 notional=100.0,
1322 start_dates=start_dates,
1323 end_dates=end_dates,
1324 pay_dates=pay_dates,
1325 currency="EUR",
1326 day_count_convention="Act365Fixed",
1327 )
1329 # # definition of the IR swap
1330 ois_swap_9M = InterestRateSwapSpecification(
1331 obj_id="39_SWAP",
1332 notional=ns,
1333 issue_date=ref_date,
1334 maturity_date=pay_dates[-1],
1335 pay_leg=fixed_leg,
1336 receive_leg=float_leg,
1337 currency="EUR",
1338 day_count_convention="Act365Fixed",
1339 issuer="dummy_issuer",
1340 securitization_level="COLLATERALIZED",
1341 )
1343 #####################################################
1344 # combine the multiple instruments to be given to the bootstrapper
1345 instruments_ois = [ois_swap_1M, ois_swap_3M, ois_swap_6M, ois_swap_9M]
1346 quotes_ois = [-0.00358, -0.00358, -0.00358, -0.00357] # taken from the .csv as example
1348 # bootsrap OIS discount curve
1349 boot_curve_ois = bootstrap_curve(
1350 ref_date,
1351 "bootstrapped_ois_DC",
1352 DayCounterType.Act365Fixed,
1353 instruments_ois,
1354 quotes_ois,
1355 interpolation_type=InterpolationType.LINEAR_LOG,
1356 extrapolation_type=ExtrapolationType.LINEAR_LOG,
1357 )
1359 # setting up 3M tenor instruments
1360 ##########################################
1361 # 1Y IRS
1362 # 1Y maturity 3M swap, i.e. the floating leg is reset every 3M
1363 # start dates of the accrual periods corresponding to the tenor of the underlying index (3 months). The spot lag is set to 0.
1364 start_dates = [ref_date + relativedelta(months=3 * i) for i in range(4)]
1365 reset_dates = start_dates
1366 end_dates = [x + relativedelta(months=3) for x in start_dates]
1367 pay_dates = end_dates
1369 print(end_dates)
1370 ns = ConstNotionalStructure(100.0)
1371 spread = 0.00
1373 # # definition of the floating leg
1374 float_leg_1Y = IrFloatLegSpecification(
1375 obj_id="dummy_float_leg",
1376 notional=ns,
1377 reset_dates=reset_dates,
1378 start_dates=start_dates,
1379 end_dates=end_dates,
1380 rate_start_dates=start_dates,
1381 rate_end_dates=end_dates,
1382 pay_dates=pay_dates,
1383 currency="EUR",
1384 udl_id="test_udl_id",
1385 fixing_id="test_fixing_id",
1386 day_count_convention="Act365Fixed",
1387 spread=spread,
1388 )
1390 # # definition of the fixed leg
1391 fixed_leg_1Y = IrFixedLegSpecification(
1392 fixed_rate=0.01,
1393 obj_id="dummy_fixed_leg",
1394 notional=100.0,
1395 start_dates=start_dates,
1396 end_dates=end_dates,
1397 pay_dates=pay_dates,
1398 currency="EUR",
1399 day_count_convention="Act365Fixed",
1400 )
1402 # # definition of the IR swap
1403 irs_1Y = InterestRateSwapSpecification(
1404 obj_id="3M_SWAP_1Y",
1405 notional=ns,
1406 issue_date=ref_date,
1407 maturity_date=pay_dates[-1],
1408 pay_leg=fixed_leg_1Y,
1409 receive_leg=float_leg_1Y,
1410 currency="EUR",
1411 day_count_convention="Act365Fixed",
1412 issuer="dummy_issuer",
1413 securitization_level="COLLATERALIZED",
1414 )
1416 ##########################################
1417 # 2 Yr IRS
1418 # 2Y maturity 3M swap, i.e. the floating leg is reset every 3M
1420 start_dates = [ref_date + relativedelta(months=3 * i) for i in range(4 * 2)]
1421 reset_dates = start_dates
1422 end_dates = [x + relativedelta(months=3) for x in start_dates]
1423 pay_dates = end_dates
1424 ns = ConstNotionalStructure(100.0)
1425 spread = 0.00
1427 # # definition of the floating leg
1428 float_leg_2Y = IrFloatLegSpecification(
1429 obj_id="dummy_float_leg",
1430 notional=ns,
1431 reset_dates=reset_dates,
1432 start_dates=start_dates,
1433 end_dates=end_dates,
1434 rate_start_dates=start_dates,
1435 rate_end_dates=end_dates,
1436 pay_dates=pay_dates,
1437 currency="EUR",
1438 udl_id="test_udl_id",
1439 fixing_id="test_fixing_id",
1440 day_count_convention="Act365Fixed",
1441 spread=spread,
1442 )
1444 # # definition of the fixed leg
1445 fixed_leg_2Y = IrFixedLegSpecification(
1446 fixed_rate=0.01,
1447 obj_id="dummy_fixed_leg",
1448 notional=100.0,
1449 start_dates=start_dates,
1450 end_dates=end_dates,
1451 pay_dates=pay_dates,
1452 currency="EUR",
1453 day_count_convention="Act365Fixed",
1454 )
1456 # # definition of the IR swap
1457 irs_2Y = InterestRateSwapSpecification(
1458 obj_id="3M_SWAP_2Y",
1459 notional=ns,
1460 issue_date=ref_date,
1461 maturity_date=pay_dates[-1],
1462 pay_leg=fixed_leg_2Y,
1463 receive_leg=float_leg_2Y,
1464 currency="EUR",
1465 day_count_convention="Act365Fixed",
1466 issuer="dummy_issuer",
1467 securitization_level="COLLATERALIZED",
1468 )
1470 instruments_3M = [irs_1Y, irs_2Y]
1471 quotes_3M = [-0.003204, -0.002615]
1473 # bootstrap forward curve
1474 euribor3MCurve = bootstrap_curve(
1475 ref_date,
1476 "euribor3M_DC",
1477 DayCounterType.Act365Fixed,
1478 instruments_3M,
1479 quotes_3M,
1480 curves={"discount_curve": boot_curve_ois},
1481 interpolation_type=InterpolationType.LINEAR_LOG,
1482 extrapolation_type=ExtrapolationType.LINEAR_LOG,
1483 )
1485 # assertions
1487 model_quotes = []
1488 # pricing_params = {"fixing_grace_period": 0.0, "set_rate": True, "desired_rate": 1.0}
1489 curves_dict = {"discount_curve": euribor3MCurve, "fixing_curve": euribor3MCurve}
1491 for i in range(len(instruments_3M)):
1492 model_quote = get_quote(ref_date, instruments_3M[i], curves_dict)
1493 # print(model_quote)
1494 model_quotes.append(model_quote)
1496 for i in range(len(quotes_3M)):
1497 self.assertAlmostEqual(model_quotes[i], quotes_3M[i], places=4)
1499 def test_multicurve_with_deposit_raises(self):
1500 # Deposit in multicurve should raise
1501 d1 = self.ref_date + timedelta(days=182)
1502 deposit = DepositSpecification(
1503 obj_id="dep1",
1504 notional=1_000_000,
1505 issue_date=self.ref_date,
1506 maturity_date=d1,
1507 currency="EUR",
1508 day_count_convention=self.day_count,
1509 rate=0.02,
1510 )
1511 discount_curve = bootstrap_curve(
1512 ref_date=self.ref_date,
1513 curve_id="DISC",
1514 day_count_convention=self.day_count,
1515 instruments=[deposit],
1516 quotes=[0.02],
1517 interpolation_type=self.interp,
1518 extrapolation_type=self.extrap,
1519 )
1520 curves = {"discount_curve": discount_curve}
1521 with self.assertRaises(Exception) as cm:
1522 bootstrap_curve(
1523 ref_date=self.ref_date,
1524 curve_id="FWD",
1525 day_count_convention=self.day_count,
1526 instruments=[deposit],
1527 quotes=[0.02],
1528 curves=curves,
1529 interpolation_type=self.interp,
1530 extrapolation_type=self.extrap,
1531 )
1532 self.assertIn("Deposits cannot be used in multi-curve bootstrapping", str(cm.exception))
1535class TestAutomaticInstrumentCreation(unittest.TestCase):
1536 """Helper functions used to create instrument speficiations from dataframes.
1538 Args:
1539 unittest (_type_): _description_
1540 """
1542 def setUp(self):
1543 """Setup input file"""
1544 # set directory and file name for Input Quotes
1545 dirName = "./sample_data" # "./"
1546 fileName = "/inputQuotes_includeFRAs.csv" # "/inputQuotes.csv"
1547 # fileName = "/multi_dates_tbs.csv" # "/inputQuotes.csv"
1549 df = pd.read_csv(dirName + fileName, sep=";", decimal=",")
1550 column_names = list(df.columns)
1552 self.quotes_df = df
1553 self.column_names = column_names
1555 def test_create_deposits_from_df(self):
1556 """ """
1557 df = self.quotes_df.copy()
1558 df_deposits = df[df["Instrument"] == "DEPOSIT"]
1560 example_dep = df_deposits.iloc[0]
1561 input_data = example_dep.copy()
1563 # these inputs must be given by user
1564 refDate = datetime(2019, 3, 1)
1565 holidays = _ECB()
1567 # the following is read for every instrument type
1568 instr = input_data["Instrument"]
1569 fixDayCount = input_data["DayCountFixed"]
1570 floatDayCount = input_data["DayCountFloat"]
1571 basisDayCount = input_data["DayCountBasis"]
1572 maturity = input_data["Maturity"]
1573 tenor = input_data["UnderlyingTenor"]
1574 underlyingPayFreq = input_data["UnderlyingPaymentFrequency"]
1575 basisTenor = input_data["BasisTenor"]
1576 basisPayFreq = input_data["BasisPaymentFrequency"]
1577 fixPayFreq = input_data["PaymentFrequencyFixed"]
1578 rollConvFloat = input_data["RollConventionFloat"]
1579 rollConvFix = input_data["RollConventionFixed"]
1580 rollConvBasis = input_data["RollConventionBasis"]
1581 spotLag = input_data["SpotLag"] # expect form "1D", i.e 1 day
1582 parRate = float(input_data["Quote"])
1583 currency = input_data["Currency"]
1584 label = instr + "_" + maturity
1586 #######################
1587 # so from the file, we know the TERM for sure, the MATURITY for sure, and we give the REFERENCE DATE
1589 # our deposit spepcificaiton can be created using the refdate, spotlag, and MATURITY to calculate the term, start, end_date
1591 dep_spec = DepositSpecification(
1592 obj_id=label,
1593 issue_date=refDate,
1594 currency=currency,
1595 # notional: float = 100.0, # we let notional default to 100
1596 rate=parRate,
1597 term=maturity,
1598 day_count_convention=floatDayCount,
1599 business_day_convention=rollConvFloat,
1600 # roll_convention: _Union[RollRule, str] = RollRule.EOM, # leave as default
1601 spot_days=int(spotLag[:-1]), # make assumption it is always given in DAYS convert -> int
1602 calendar=holidays,
1603 issuer="dummy_issuer",
1604 securitization_level="NONE",
1605 )
1607 dep_spec2 = sfc.make_deposit_spec(input_data, refDate, holidays)
1609 self.assertIsInstance(dep_spec, DepositSpecification)
1610 self.assertIsInstance(dep_spec2, DepositSpecification)
1611 # self.assertEqual(dep_spec.__dict__, dep_spec2.__dict__)
1613 def test_create_IRS_from_df(self):
1614 """ """
1615 df = self.quotes_df.copy()
1616 df_irs = df[df["Instrument"] == "IRS"]
1618 example_irs = df_irs.iloc[0]
1619 input_data = example_irs.copy()
1621 # these inputs must be given by user
1622 refDate = datetime(2019, 3, 1)
1623 holidays = _ECB()
1625 # the following is read for every instrument type
1626 instr = input_data["Instrument"]
1627 fixDayCount = input_data["DayCountFixed"]
1628 floatDayCount = input_data["DayCountFloat"]
1629 basisDayCount = input_data["DayCountBasis"]
1630 maturity = input_data["Maturity"]
1631 underlyingIndex = input_data["UnderlyingIndex"]
1632 tenor = input_data["UnderlyingTenor"]
1633 underlyingPayFreq = input_data["UnderlyingPaymentFrequency"]
1634 basisTenor = input_data["BasisTenor"]
1635 basisPayFreq = input_data["BasisPaymentFrequency"]
1636 fixPayFreq = input_data["PaymentFrequencyFixed"]
1637 rollConvFloat = input_data["RollConventionFloat"]
1638 rollConvFix = input_data["RollConventionFixed"]
1639 rollConvBasis = input_data["RollConventionBasis"]
1640 spotLag = input_data["SpotLag"]
1641 parRate = float(input_data["Quote"])
1642 currency = input_data["Currency"]
1643 label = instr + "_" + maturity
1645 spot_date = calc_end_day(start_day=refDate, term=spotLag, business_day_convention=rollConvFix, calendar=holidays)
1646 expiry = calc_end_day(spot_date, maturity, rollConvFix, holidays)
1648 # start_day = calc_start_day(ref)
1649 # end_day = calc_end_day()
1650 # generate_dates
1651 fix_schedule = Schedule(
1652 start_day=spot_date, end_day=expiry, time_period=fixPayFreq, business_day_convention=rollConvFix, calendar=holidays, ref_date=refDate
1653 ).generate_dates(False)
1655 # fix_schedule = get_schedule(self.refDate, self.maturity, pay_freq, roll_conv, self.holidays, spot_days)
1656 fix_start_dates = fix_schedule[:-1]
1657 fix_end_dates = fix_schedule[1:]
1658 fix_pay_dates = fix_end_dates
1660 flt_schedule = Schedule(
1661 start_day=spot_date,
1662 end_day=expiry,
1663 time_period=underlyingPayFreq,
1664 business_day_convention=rollConvFloat,
1665 calendar=holidays,
1666 ref_date=refDate,
1667 ).generate_dates(False)
1669 # flt_schedule = get_schedule(self.refDate, self.maturity, pay_freq, roll_conv, self.holidays, spot_days)
1670 flt_start_dates = flt_schedule[:-1]
1671 flt_end_dates = flt_schedule[1:]
1672 flt_pay_dates = flt_end_dates
1674 flt_reset_schedule = Schedule(
1675 start_day=spot_date, end_day=expiry, time_period=tenor, business_day_convention=rollConvFloat, calendar=holidays, ref_date=refDate
1676 ).generate_dates(False)
1678 # flt_reset_schedule = get_schedule(self.refDate, self.maturity, reset_freq, roll_conv, self.holidays, spot_days)
1679 flt_reset_dates = flt_reset_schedule[:-1]
1681 print(flt_reset_dates)
1683 # start_dates3 = [ref_date + relativedelta(months=3*i) for i in range(4*3)]
1684 # reset_dates3 = start_dates3
1685 # end_dates3 = [x + relativedelta(months=3) for x in start_dates3]
1686 # pay_dates3 = end_dates3
1687 ns = ConstNotionalStructure(1.0)
1688 spread = 0.00
1690 # # definition of the floating leg
1691 float_leg = IrFloatLegSpecification(
1692 obj_id=label + "_float_leg",
1693 notional=ns,
1694 reset_dates=flt_reset_dates,
1695 start_dates=flt_start_dates,
1696 end_dates=flt_end_dates,
1697 rate_start_dates=flt_start_dates,
1698 rate_end_dates=flt_end_dates,
1699 pay_dates=flt_pay_dates,
1700 currency=currency,
1701 udl_id=underlyingIndex,
1702 fixing_id="test_fixing_id",
1703 day_count_convention=floatDayCount,
1704 spread=spread,
1705 )
1707 # # definition of the fixed leg
1708 fixed_leg = IrFixedLegSpecification(
1709 fixed_rate=parRate,
1710 obj_id=label + "_fixed_leg3",
1711 notional=1.0,
1712 start_dates=fix_start_dates,
1713 end_dates=fix_end_dates,
1714 pay_dates=fix_pay_dates,
1715 currency=currency,
1716 day_count_convention=fixDayCount,
1717 )
1719 # get expiry of swap (cannot be before last paydate of legs)
1720 # spot_date = get_end_date(self.refDate, self.spotLag)
1721 # expiry = get_end_date(spot_date, self.maturity)
1722 # # definition of the IR swap
1723 ir_swap = InterestRateSwapSpecification(
1724 obj_id=label,
1725 notional=ns,
1726 issue_date=refDate,
1727 maturity_date=expiry,
1728 pay_leg=fixed_leg,
1729 receive_leg=float_leg,
1730 currency=currency,
1731 day_count_convention=floatDayCount,
1732 issuer="dummy_issuer",
1733 securitization_level="COLLATERALIZED",
1734 )
1736 ir_swap2 = sfc.make_irswap_spec(input_data, refDate, holidays)
1737 self.maxDiff = None
1738 self.assertIsInstance(ir_swap, InterestRateSwapSpecification)
1739 self.assertIsInstance(ir_swap2, InterestRateSwapSpecification)
1740 self.assertTrue(deep_equal(ir_swap, ir_swap2))
1742 def test_create_OIS_from_df(self):
1743 """ """
1744 df = self.quotes_df.copy()
1745 df_ois = df[df["Instrument"] == "OIS"]
1747 example_ois = df_ois.iloc[0]
1748 input_data = example_ois.copy()
1750 # these inputs must be given by user
1751 refDate = datetime(2019, 3, 1)
1752 holidays = _ECB()
1754 # the following is read for every instrument type
1755 instr = input_data["Instrument"]
1756 fixDayCount = input_data["DayCountFixed"]
1757 floatDayCount = input_data["DayCountFloat"]
1758 basisDayCount = input_data["DayCountBasis"]
1759 maturity = input_data["Maturity"]
1760 underlyingIndex = input_data["UnderlyingIndex"]
1761 tenor = input_data["UnderlyingTenor"]
1762 underlyingPayFreq = input_data["UnderlyingPaymentFrequency"]
1763 basisTenor = input_data["BasisTenor"]
1764 basisPayFreq = input_data["BasisPaymentFrequency"]
1765 fixPayFreq = input_data["PaymentFrequencyFixed"]
1766 rollConvFloat = input_data["RollConventionFloat"]
1767 rollConvFix = input_data["RollConventionFixed"]
1768 rollConvBasis = input_data["RollConventionBasis"]
1769 spotLag = input_data["SpotLag"]
1770 parRate = float(input_data["Quote"])
1771 currency = input_data["Currency"]
1772 label = instr + "_" + maturity
1774 # get swap leg schedule # assume same for both fix and float legs?
1775 # we use the helper function with spotlag in place of maturity to effctively shift the date
1776 spot_date = calc_end_day(start_day=refDate, term=spotLag, business_day_convention=rollConvFix, calendar=holidays)
1777 expiry = calc_end_day(spot_date, maturity, rollConvFix, holidays)
1778 expiry_unadjusted = calc_end_day(start_day=spot_date, term=maturity, calendar=holidays)
1780 # start_day = calc_start_day(ref)
1781 # end_day = calc_end_day()
1782 # generate_dates
1783 fix_schedule = Schedule(
1784 start_day=spot_date,
1785 end_day=expiry_unadjusted, # expiry,
1786 time_period=fixPayFreq,
1787 business_day_convention=rollConvFix,
1788 calendar=holidays,
1789 ref_date=refDate,
1790 ).generate_dates(False)
1792 # fix_schedule = get_schedule(self.refDate, self.maturity, pay_freq, roll_conv, self.holidays, spot_days)
1793 fix_start_dates = fix_schedule[:-1]
1794 fix_end_dates = fix_schedule[1:]
1795 fix_pay_dates = fix_end_dates
1797 flt_schedule = Schedule(
1798 start_day=spot_date,
1799 end_day=expiry_unadjusted, # expiry,
1800 time_period=underlyingPayFreq,
1801 business_day_convention=rollConvFloat,
1802 calendar=holidays,
1803 ref_date=refDate,
1804 ).generate_dates(False)
1806 flt_start_dates = flt_schedule[:-1]
1807 flt_end_dates = flt_schedule[1:]
1808 flt_pay_dates = flt_end_dates
1810 flt_reset_schedule = Schedule(
1811 start_day=spot_date, end_day=expiry, time_period=tenor, business_day_convention=rollConvFloat, calendar=holidays, ref_date=refDate
1812 ).generate_dates(False)
1814 flt_reset_dates = flt_reset_schedule[:-1]
1816 res = IrOISLegSpecification.ois_scheduler_2D(flt_start_dates, flt_end_dates)
1818 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
1819 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
1820 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
1821 daily_rate_pay_dates = res[3]
1823 # print(flt_reset_dates)
1824 # print(daily_rate_reset_dates)
1826 ns = ConstNotionalStructure(1.0)
1827 spread = 0.00
1829 # # definition of the floating leg
1831 ois_leg = IrOISLegSpecification(
1832 obj_id=label + "_float_leg",
1833 notional=ns,
1834 rate_reset_dates=daily_rate_reset_dates,
1835 start_dates=flt_start_dates,
1836 end_dates=flt_end_dates,
1837 rate_start_dates=daily_rate_start_dates,
1838 rate_end_dates=daily_rate_end_dates,
1839 pay_dates=daily_rate_pay_dates,
1840 currency=currency,
1841 udl_id=underlyingIndex,
1842 fixing_id="test_fixing_id",
1843 day_count_convention=floatDayCount,
1844 rate_day_count_convention=floatDayCount,
1845 spread=spread,
1846 )
1848 # # definition of the fixed leg
1849 fixed_leg = IrFixedLegSpecification(
1850 fixed_rate=parRate,
1851 obj_id=label + "_fixed_leg3",
1852 notional=1.0,
1853 start_dates=fix_start_dates,
1854 end_dates=fix_end_dates,
1855 pay_dates=fix_pay_dates,
1856 currency=currency,
1857 day_count_convention=fixDayCount,
1858 )
1860 # # definition of the IR swap
1861 oi_swap = InterestRateSwapSpecification(
1862 obj_id=label,
1863 notional=ns,
1864 issue_date=refDate,
1865 maturity_date=expiry,
1866 pay_leg=fixed_leg,
1867 receive_leg=ois_leg,
1868 currency=currency,
1869 day_count_convention=floatDayCount,
1870 issuer="dummy_issuer",
1871 securitization_level="COLLATERALIZED",
1872 )
1874 oi_swap2 = sfc.make_ois_spec(input_data, refDate, holidays)
1875 self.maxDiff = None
1876 self.assertIsInstance(oi_swap, InterestRateSwapSpecification)
1877 self.assertIsInstance(oi_swap2, InterestRateSwapSpecification)
1878 self.assertTrue(deep_equal(oi_swap, oi_swap2))
1880 def test_create_TBS_from_df(self):
1881 """
1882 Create a tenor basis swap (TBS) specification:
1883 - Pay short floating leg
1884 - Receive long floating leg
1885 - Pay fixed spread leg (represents market quote)
1886 """
1887 df = self.quotes_df.copy()
1888 df_irs = df[df["Instrument"] == "TBS"]
1890 example_irs = df_irs.iloc[0]
1891 row = example_irs.copy()
1893 # these inputs must be given by user
1894 refDate = datetime(2019, 3, 1)
1895 holidays = _ECB()
1897 # --- Extract general fields ---
1898 instr = row["Instrument"]
1899 currency = row["Currency"]
1900 maturity = row["Maturity"]
1901 spot_lag = row["SpotLag"]
1902 roll_conv = row["RollConventionFloat"]
1903 fixDayCount = row["DayCountFloat"]
1904 floatDayCount = row["DayCountFloat"]
1905 basisDayCount = row["DayCountBasis"]
1906 rollConvFix = row["RollConventionFixed"]
1907 rollConvBasis = row["RollConventionBasis"]
1908 # --- Long (receive) leg info ---
1909 long_index = row["UnderlyingIndex"]
1910 long_tenor = row["UnderlyingTenor"]
1911 long_freq = row["UnderlyingPaymentFrequency"]
1913 # --- Short (pay) leg info ---
1914 short_index = row["UnderlyingIndex"]
1915 short_tenor = row["UnderlyingTenorShort"]
1916 short_freq = row["UnderlyingPaymentFrequencyShort"]
1918 # --- Spread (basis quote) ---
1919 spread_rate = float(row["Quote"]) / 10000.0 # e.g. 8.5 bps -> 0.00085
1921 # --- Spot and maturity dates ---
1922 spot_date = calc_end_day(refDate, spot_lag, roll_conv, holidays)
1923 expiry = calc_end_day(spot_date, maturity, roll_conv, holidays)
1924 label = f"{instr}_{maturity}"
1926 ns = ConstNotionalStructure(1.0)
1928 # --------------------------------------------
1929 # PAY FLOATING LEG (short tenor, pays basis)
1930 short_schedule = Schedule(
1931 start_day=spot_date,
1932 end_day=expiry,
1933 time_period=short_freq,
1934 business_day_convention=roll_conv,
1935 calendar=holidays,
1936 ref_date=refDate,
1937 ).generate_dates(False)
1939 short_start = short_schedule[:-1]
1940 short_end = short_schedule[1:]
1941 short_pay = short_end
1942 short_reset = Schedule(
1943 start_day=spot_date,
1944 end_day=expiry,
1945 time_period=short_tenor,
1946 business_day_convention=roll_conv,
1947 calendar=holidays,
1948 ref_date=refDate,
1949 ).generate_dates(False)[:-1]
1951 pay_leg = IrFloatLegSpecification(
1952 obj_id=label + "_pay_leg",
1953 notional=ns,
1954 reset_dates=short_reset,
1955 start_dates=short_start,
1956 end_dates=short_end,
1957 rate_start_dates=short_start,
1958 rate_end_dates=short_end,
1959 pay_dates=short_pay,
1960 currency=currency,
1961 udl_id=short_index,
1962 fixing_id="test_fixing_id",
1963 day_count_convention=floatDayCount,
1964 spread=0.0, # this is the quoted basis
1965 )
1967 # --------------------------------------------
1968 # RECEIVE FLOATING LEG (long tenor)
1970 long_schedule = Schedule(
1971 start_day=spot_date,
1972 end_day=expiry,
1973 time_period=long_freq,
1974 business_day_convention=roll_conv,
1975 calendar=holidays,
1976 ref_date=refDate,
1977 ).generate_dates(False)
1979 long_start = long_schedule[:-1]
1980 long_end = long_schedule[1:]
1981 long_pay = long_end
1982 long_reset = Schedule(
1983 start_day=spot_date,
1984 end_day=expiry,
1985 time_period=long_tenor,
1986 business_day_convention=roll_conv,
1987 calendar=holidays,
1988 ref_date=refDate,
1989 ).generate_dates(False)[:-1]
1991 receive_leg = IrFloatLegSpecification(
1992 obj_id=label + "_receive_leg",
1993 notional=ns,
1994 reset_dates=long_reset,
1995 start_dates=long_start,
1996 end_dates=long_end,
1997 rate_start_dates=long_start,
1998 rate_end_dates=long_end,
1999 pay_dates=long_pay,
2000 currency=currency,
2001 udl_id=long_index,
2002 fixing_id="test_fixing_id",
2003 day_count_convention=floatDayCount,
2004 spread=0.0,
2005 )
2007 # --------------------------------------------
2008 # FIXED SPREAD LEG
2009 # The spread leg represents the fixed +x bps cashflows applied to the pay leg
2010 spread_schedule = Schedule(
2011 start_day=spot_date,
2012 end_day=expiry,
2013 time_period=short_freq, # same freq as short leg
2014 business_day_convention=rollConvFix,
2015 calendar=holidays,
2016 ref_date=refDate,
2017 ).generate_dates(False)
2019 spread_start = spread_schedule[:-1]
2020 spread_end = spread_schedule[1:]
2021 spread_pay = spread_end
2023 spread_leg = IrFixedLegSpecification(
2024 fixed_rate=spread_rate,
2025 obj_id=label + "_spread_leg",
2026 notional=1.0,
2027 start_dates=spread_start,
2028 end_dates=spread_end,
2029 pay_dates=spread_pay,
2030 currency=currency,
2031 day_count_convention=fixDayCount,
2032 )
2034 # --------------------------------------------
2035 # Combine into full TBS object
2036 basis_swap = InterestRateBasisSwapSpecification(
2037 obj_id=label,
2038 notional=ns,
2039 issue_date=refDate,
2040 maturity_date=expiry,
2041 pay_leg=pay_leg,
2042 receive_leg=receive_leg,
2043 spread_leg=spread_leg,
2044 currency=currency,
2045 day_count_convention=floatDayCount,
2046 issuer="dummy_issuer",
2047 securitization_level="COLLATERALIZED",
2048 )
2050 basis_swap2 = sfc.make_basis_swap_spec(row, refDate, holidays)
2052 self.maxDiff = None
2053 self.assertIsInstance(basis_swap, InterestRateBasisSwapSpecification)
2054 self.assertIsInstance(basis_swap2, InterestRateBasisSwapSpecification)
2055 self.assertTrue(deep_equal(basis_swap, basis_swap2))
2057 def test_create_FRA_from_df(self):
2058 """Assuming a 3Mx6M Forward rate agreement instrument"""
2059 df = self.quotes_df.copy()
2060 df_fra = df[df["Instrument"] == "FRA"]
2062 example_fra = df_fra.iloc[0]
2063 input_data = example_fra.copy()
2064 self.assertEqual(input_data["Maturity"], "3Mx6M")
2065 self.assertEqual(input_data["SpotLag"], "2D")
2066 # these inputs must be given by user
2067 refDate = datetime(2019, 4, 1)
2068 holidays = _ECB()
2069 # spot_days = 2
2070 start_date = datetime(2019, 4 + 3, 3)
2071 end_date = datetime(2019, 4 + 3 + 3, 3)
2072 label = input_data["Instrument"] + "_" + input_data["Maturity"]
2074 fra_spec = ForwardRateAgreementSpecification(
2075 obj_id=label,
2076 trade_date=refDate,
2077 notional=1,
2078 rate=float(input_data["Quote"]),
2079 start_date=start_date,
2080 end_date=end_date,
2081 udlID=input_data["UnderlyingIndex"],
2082 rate_start_date=start_date,
2083 rate_end_date=end_date,
2084 # maturity_date=,
2085 day_count_convention=input_data["DayCountFixed"],
2086 business_day_convention=input_data["RollConventionFixed"],
2087 rate_day_count_convention=input_data["DayCountFloat"],
2088 rate_business_day_convention=input_data["RollConventionFloat"],
2089 calendar=holidays,
2090 currency=input_data["Currency"],
2091 # payment_days: int = 0,
2092 spot_days=int(input_data["SpotLag"][:-1]),
2093 # start_period: int = None,
2094 # end_period: int = None,
2095 # ir_index: str = None,
2096 # issuer: str = None,
2097 )
2099 fra_spec2 = sfc.make_fra_spec(input_data, refDate, holidays)
2100 self.assertIsInstance(fra_spec, ForwardRateAgreementSpecification)
2101 self.assertIsInstance(fra_spec2, ForwardRateAgreementSpecification)
2102 self.assertEqual(fra_spec.start_date, fra_spec2.start_date)
2103 self.assertEqual(fra_spec.end_date, fra_spec2.end_date)
2104 # self.assertEqual(fra_spec.__dict__, fra_spec2.__dict__)
2106 def test_bootstrap_deposits_from_df(self):
2107 """Test of the bootstrap function using deposit specifications
2108 parsed from a datafram of expected format
2110 Assumption is that the deposits are already ordered by maturity...
2112 """
2113 # these inputs must be given by user
2114 refDate = datetime(2019, 3, 1)
2115 holidays = _ECB()
2117 df = self.quotes_df.copy()
2118 df_ins = df[df["Instrument"] == "DEPOSIT"]
2120 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
2121 ins_quotes = df_ins["Quote"].tolist()
2123 print("--------------DEBUG PARSING")
2124 print(ins_quotes[0])
2125 print(df_ins["DayCountFixed"][0])
2126 print(df_ins)
2127 print("--------------Starting bootstrapper")
2128 curve = bootstrap_curve(
2129 ref_date=refDate,
2130 curve_id="dc_deposits",
2131 day_count_convention=df_ins["DayCountFixed"][0], # taken the first entry and assume is valid for all other deposits
2132 instruments=ins_spec,
2133 quotes=ins_quotes,
2134 interpolation_type=InterpolationType.LINEAR,
2135 extrapolation_type=ExtrapolationType.LINEAR,
2136 )
2137 # print(curve.get_dates())
2138 self.assertIsInstance(curve, DiscountCurve)
2139 self.assertEqual(curve.get_dates()[0], refDate)
2141 # the discount curve needs to be able to get the same market quote for the instrument
2142 for i in range(len(ins_spec)):
2143 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve})
2144 self.assertAlmostEqual(model_quote, ins_quotes[i], delta=1e-6)
2145 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2146 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2148 # self.assertEqual(1, 1)
2150 def test_bootstrap_FRAs_from_df(self):
2151 """Test of the bootstrap function using deposit specifications
2152 parsed from a datafram of expected format
2154 Assumption is that the deposits are already ordered by maturity...
2156 """
2157 # these inputs must be given by user
2158 refDate = datetime(2019, 3, 1)
2159 holidays = _ECB()
2161 df = self.quotes_df.copy()
2162 df_ins = df[df["Instrument"] == "FRA"]
2164 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
2165 ins_quotes = df_ins["Quote"].tolist()
2167 print("--------------DEBUG PARSING")
2168 print(ins_quotes[0])
2169 print(df_ins["DayCountFixed"].tolist()[0])
2170 print(df_ins)
2171 print("--------------Starting bootstrapper")
2172 curve = bootstrap_curve(
2173 ref_date=refDate,
2174 curve_id="dc_deposits",
2175 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2176 instruments=ins_spec,
2177 quotes=ins_quotes,
2178 interpolation_type=InterpolationType.LINEAR,
2179 extrapolation_type=ExtrapolationType.LINEAR,
2180 )
2181 # print(curve.get_dates())
2182 self.assertIsInstance(curve, DiscountCurve)
2183 self.assertEqual(curve.get_dates()[0], refDate)
2185 # the discount curve needs to be able to get the same market quote for the instrument
2186 for i in range(len(ins_spec)):
2187 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve})
2188 self.assertAlmostEqual(model_quote, ins_quotes[i], delta=1e-6)
2189 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2190 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2192 # self.assertEqual(1, 1)
2194 def test_bootstrap_ois_from_df(self):
2195 """Test of the bootstrap function using ois specifications
2196 parsed from a datafram of expected format
2198 Assumption is that the ois are already ordered by maturity...
2200 """
2201 logger.debug(f"--------------------------------------------------------")
2202 logger.debug("test_bootstrap_ois_from_df start")
2203 # these inputs must be given by user
2204 refDate = datetime(2019, 3, 1)
2205 holidays = _ECB()
2207 df = self.quotes_df.copy()
2208 df_ins = df[df["Instrument"] == "OIS"]
2210 logger.debug("CSV loaded")
2212 min_i = 0
2213 max_i = 18 # 19-25 problematic?
2214 min_i2 = 26
2215 max_i2 = len(df_ins)
2216 ins_spec = sfc.load_specifications_from_pd(df_ins.iloc[np.r_[min_i:max_i, min_i2:max_i2]], refDate, holidays)
2217 # ins_quotes = df_ins["Quote"].tolist()[min_i:max_i]
2218 ins_quotes = df_ins["Quote"].tolist()[min_i:max_i] + df_ins["Quote"].tolist()[min_i2:max_i2]
2220 logger.debug("instrument specifications created")
2222 # print("--------------DEBUG PARSING")
2223 # print(ins_quotes[0])
2224 # print(df_ins["DayCountFixed"].tolist()[0])
2225 # print(df_ins.iloc[min_i:max_i].copy())
2226 # print(len(ins_spec), len(ins_quotes))
2227 # print(len(df_ins["Quote"].tolist()))
2228 # print(len(df_ins.iloc[np.r_[min_i:max_i, min_i2:max_i2]]))
2229 # for i in range(len(ins_quotes)):
2230 # print(i, ins_quotes[i])
2232 print("--------------Starting bootstrapper")
2233 logger.debug(f"starting bootstrapper of {len(ins_quotes)} instruments")
2234 curve = bootstrap_curve(
2235 ref_date=refDate,
2236 curve_id="OIS_estr",
2237 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2238 instruments=ins_spec,
2239 quotes=ins_quotes,
2240 interpolation_type=InterpolationType.LINEAR_LOG,
2241 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2242 )
2243 # print(curve.get_dates())
2244 self.assertIsInstance(curve, DiscountCurve)
2245 self.assertEqual(curve.get_dates()[0], refDate)
2246 logger.debug("bootstrapped curve dates matched")
2247 # the discount curve needs to be able to get the same market quote for the instrument
2248 for i in range(len(ins_spec)):
2249 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve, "fixing_curve": curve})
2250 self.assertAlmostEqual(model_quote, ins_quotes[i], delta=1e-6)
2251 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2252 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2254 logger.debug(f"asserted market quote matched -done")
2255 logger.debug(f"--------------------------------------------------------")
2256 # self.assertEqual(1, 1)
2258 def test_multicurve_bootstrap_ois_3M(self):
2259 """Test of the bootstrap function in the context of multicurve bootstrapping
2260 using ois and irs specifications parsed from a datafram of expected format
2262 """
2264 # these inputs must be given by user
2265 refDate = datetime(2019, 3, 1)
2266 holidays = _ECB()
2268 ###############################
2269 # PREPARE discount curve
2270 df = self.quotes_df.copy()
2271 df_ins = df[df["Instrument"] == "OIS"]
2273 min_i = 0
2274 max_i = 17 # up to 3 years ...
2276 ins_spec = sfc.load_specifications_from_pd(df_ins.iloc[min_i:max_i], refDate, holidays)
2277 ins_quotes = df_ins["Quote"].tolist()[min_i:max_i]
2279 print("--------------DEBUG PARSING")
2280 print(ins_quotes[0])
2281 print(df_ins["DayCountFixed"].tolist()[0])
2282 print(df_ins.iloc[min_i:max_i].copy())
2283 print("--------------Starting bootstrapper")
2284 curve_ois = bootstrap_curve(
2285 ref_date=refDate,
2286 curve_id="OIS_estr",
2287 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2288 instruments=ins_spec,
2289 quotes=ins_quotes,
2290 interpolation_type=InterpolationType.LINEAR,
2291 extrapolation_type=ExtrapolationType.LINEAR,
2292 )
2293 # print(curve.get_dates())
2294 self.assertIsInstance(curve_ois, DiscountCurve)
2295 self.assertEqual(curve_ois.get_dates()[0], refDate)
2297 # the discount curve needs to be able to get the same market quote for the instrument
2298 for i in range(len(ins_spec)):
2299 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve_ois, "fixing_curve": curve_ois})
2300 self.assertAlmostEqual(model_quote, ins_quotes[i], delta=1e-6)
2301 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2302 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2304 # self.assertEqual(1, 1)
2306 ##################################################
2307 # select for 3M instruments
2308 min_i = 0
2309 max_i = -1
2311 # df_ins_3M = df[(df["UnderlyingIndex"] == "EURIBOR") & (df["UnderlyingTenor"] == "3M")]
2312 df_ins_3M = df[(df["UnderlyingIndex"] == "EURIBOR") & (df["UnderlyingTenor"] == "3M") & (df["Instrument"] == "IRS")]
2313 ins_spec_3M = sfc.load_specifications_from_pd(df_ins_3M.iloc[min_i:max_i], refDate, holidays)
2314 ins_quotes_3M = df_ins_3M["Quote"].tolist()[min_i:max_i]
2316 # bootstrap forward curve
2317 euribor3MCurve = bootstrap_curve(
2318 refDate,
2319 "euribor3M_DC",
2320 DayCounterType.Act365Fixed,
2321 ins_spec_3M,
2322 ins_quotes_3M,
2323 curves={"discount_curve": curve_ois},
2324 interpolation_type=InterpolationType.LINEAR_LOG,
2325 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2326 )
2328 for i in range(len(ins_spec_3M)):
2329 model_quote = get_quote(refDate, ins_spec_3M[i], {"discount_curve": curve_ois, "fixing_curve": euribor3MCurve})
2330 self.assertAlmostEqual(model_quote, ins_quotes_3M[i], delta=1e-6)
2331 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2332 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2335class TestBSBootstrap(unittest.TestCase):
2336 """ """
2338 def setUp(self):
2339 """Set up input file"""
2340 # set directory and file name for Input Quotes
2341 dirName = "./sample_data" # "./"
2342 fileName = "/multi_dates_tbs.csv" # "/inputQuotes.csv"
2344 df = pd.read_csv(dirName + fileName, sep=";", decimal=",")
2345 column_names = list(df.columns)
2347 self.quotes_df = df
2348 self.column_names = column_names
2350 def test_tbs_3m_6m(self):
2351 """Using 2025 09 24 as a control date, to ensure the proper bootstrapping from Frontmark data example.
2352 Goes through the complete process of bootstrapping 3 times
2353 1. produce OIS curve for discounting
2354 2. produce fwd curve e.g. 3M euribor from IRS instruments
2355 3. produce 6M euribor from TBS instruments and 3m Euribor"""
2357 logger.debug(f"--------------------------------------------------------")
2358 logger.debug("start")
2359 # these inputs must be given by user
2360 mon = "09"
2361 day = "24"
2362 year = "2025"
2363 date_str = f"{day}.{mon}.{year}"
2364 refDate = datetime(int(year), int(mon), int(day))
2365 holidays = _ECB()
2367 df = self.quotes_df.copy()
2369 A = df[df["Date"] == date_str]
2371 logger.debug(f"Creating OIS discount Curve")
2372 # df_ins = df[df["Instrument"] == "OIS"]
2373 # dc_df_temp = df[(df["Date"] == selected_date) & (df["Currency"] == selected_currency) & (df["UnderlyingIndex"] == "ESTR") & (df["Instrument"] == "OIS") ]
2374 df_ins = df[(df["Date"] == date_str) & (df["Currency"] == "EUR") & (df["UnderlyingIndex"] == "EONIA") & (df["Instrument"] == "OIS")]
2376 logger.debug("CSV loaded")
2378 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
2379 ins_quotes = df_ins["Quote"].tolist()
2380 logger.debug("instrument specifications created")
2382 print("--------------Starting bootstrapper")
2383 logger.debug(f" OIS starting bootstrapper of {len(ins_quotes)} instruments")
2384 curve = bootstrap_curve(
2385 ref_date=refDate,
2386 curve_id="OIS_estr",
2387 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2388 instruments=ins_spec,
2389 quotes=ins_quotes,
2390 interpolation_type=InterpolationType.LINEAR_LOG,
2391 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2392 # interpolation_type=InterpolationType.HAGAN_DF,
2393 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2394 )
2396 # OUtput discoutn curve dates and values for test:
2397 print(" OIS: curve valueus (date, DF)")
2398 dates_ois = curve.get_dates()
2399 df_ois = curve.get_df()
2400 for i in range(len(dates_ois)):
2401 print(f"{dates_ois[i]} , {df_ois[i]}")
2403 # --------------------------------- IR for 3M
2404 logger.debug(f"Loading instruments for reference date and, currency, and EURIBOR for SHORT LEG, e.g. 3M in this case")
2405 df_ins_3m = df[
2406 (df["Date"] == date_str)
2407 & (df["Currency"] == "EUR")
2408 & (df["UnderlyingIndex"] == "EURIBOR")
2409 & (df["Instrument"] == "IRS")
2410 & (df["UnderlyingTenor"] == "3M")
2411 ]
2413 ins_spec_3m = sfc.load_specifications_from_pd(df_ins_3m, refDate, holidays)
2414 ins_quotes_3m = df_ins_3m["Quote"].tolist()
2415 logger.debug("instrument specifications created")
2417 curves = {"discount_curve": curve}
2418 print("--------------Starting bootstrapper")
2419 logger.debug(f" IRS starting bootstrapper of {len(ins_quotes_3m)} instruments")
2420 curve_3m = bootstrap_curve(
2421 ref_date=refDate,
2422 curve_id="euribor3m",
2423 day_count_convention=df_ins_3m["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2424 instruments=ins_spec_3m,
2425 quotes=ins_quotes_3m,
2426 curves=curves,
2427 interpolation_type=InterpolationType.LINEAR_LOG,
2428 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2429 # interpolation_type=InterpolationType.HAGAN_DF,
2430 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2431 )
2433 # OUtput discoutn curve dates and values for test:
2434 print(" 3m euribor: curve valueus (date, DF)")
2435 dates_3m = curve_3m.get_dates()
2436 df_3m = curve_3m.get_df()
2437 for i in range(len(dates_3m)):
2438 print(f"{dates_3m[i]} , {df_3m[i]}")
2440 # --------------------------------- TBS
2441 logger.debug(f"Loading TBS instruments for reference date and, currency, and EURIBOR, 6M long tenor")
2442 df_ins_tbs = df[
2443 (df["Date"] == date_str)
2444 & (df["Currency"] == "EUR")
2445 & (df["UnderlyingIndex"] == "EURIBOR")
2446 & (df["Instrument"] == "TBS")
2447 & (df["UnderlyingTenor"] == "6M")
2448 & (df["UnderlyingTenorShort"] == "3M")
2449 ]
2451 if df_ins_tbs.empty:
2452 raise ValueError(f"No TBS instruments found for date {date_str} with EUR/EURIBOR.")
2454 print("--------- TBS instruments used ...")
2455 for _, item in df_ins_tbs.iterrows():
2456 print(item["Date"], item["Instrument"], item["Maturity"], item["Quote"], item["UnderlyingTenorShort"], item["UnderlyingTenor"])
2458 ins_spec_tbs = sfc.load_specifications_from_pd(df_ins_tbs, refDate, holidays)
2459 # ins_quotes_tbs = df_ins_tbs["Quote"].tolist()
2460 ins_quotes_tbs = (df_ins_tbs["Quote"] / 10000.0).tolist()
2461 logger.debug("instrument specifications created")
2463 curves["basis_curve"] = curve_3m
2465 print("--------------Starting bootstrapper")
2466 logger.debug(f" TBS starting bootstrapper of {len(df_ins_tbs)} instruments")
2467 curve_6m = bootstrap_curve(
2468 ref_date=refDate,
2469 curve_id="euribor_6m",
2470 day_count_convention=df_ins_tbs["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2471 instruments=ins_spec_tbs,
2472 quotes=ins_quotes_tbs,
2473 curves=curves,
2474 interpolation_type=InterpolationType.LINEAR_LOG,
2475 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2476 # interpolation_type=InterpolationType.HAGAN_DF,
2477 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2478 )
2479 print(" 6m euribor: curve valueus (date, DF)")
2480 dates_6m = curve_6m.get_dates()
2481 df_6m = curve_6m.get_df()
2482 for i in range(len(dates_6m)):
2483 print(f"{dates_6m[i]} , {df_6m[i]}")
2484 # print(curve.get_dates())
2485 self.assertIsInstance(curve_6m, DiscountCurve)
2486 self.assertEqual(curve_6m.get_dates()[0], refDate)
2487 logger.debug("bootstrapped curve dates matched")
2488 # the discount curve needs to be able to get the same market quote for the instrument
2489 for i in range(len(ins_spec_tbs)):
2490 model_quote = get_quote(refDate, ins_spec_tbs[i], {"discount_curve": curve, "fixing_curve": curve_6m, "basis_curve": curve_3m})
2491 # compare to given basis points
2492 self.assertAlmostEqual(model_quote, ins_quotes_tbs[i], delta=1e-5) # since the quotes are only to 5 decimals
2493 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2494 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2495 # print(i, model_quote, curve.get_df()[i])
2497 logger.debug(f"asserted market quote matched -done")
2498 logger.debug(f"--------------------------------------------------------")
2499 # self.assertEqual(1, 1)
2501 def test_initial_curve_tbs_extend(self):
2502 """Using 2025 07 23 as a control date, to ensure the proper bootstrapping from Frontmark data example.
2503 Goes through the complete process of bootstrapping 34 times
2504 1. produce OIS curve for discounting
2505 2. produce fwd curve e.g. 3M euribor from IRS instruments
2506 3. produce 6M euribor from IRS instruments, where we assume there are not many maturities
2507 4. we EXTEND the 6M curve with TBS instruments given also the 3M euribor as basis curve"""
2509 logger.debug(f"--------------------------------------------------------")
2510 logger.debug("start")
2511 # these inputs must be given by user
2512 mon = "07"
2513 day = "23"
2514 year = "2025"
2515 date_str = f"{day}.{mon}.{year}"
2516 refDate = datetime(int(year), int(mon), int(day))
2517 holidays = _ECB()
2519 df = self.quotes_df.copy()
2520 df = update_fra_tenors(df) # correct the FRA underlying tenors column
2522 A = df[df["Date"] == date_str]
2524 logger.debug(f"Creating OIS discount Curve")
2525 # df_ins = df[df["Instrument"] == "OIS"]
2526 # dc_df_temp = df[(df["Date"] == selected_date) & (df["Currency"] == selected_currency) & (df["UnderlyingIndex"] == "ESTR") & (df["Instrument"] == "OIS") ]
2527 df_ins = df[(df["Date"] == date_str) & (df["Currency"] == "EUR") & (df["UnderlyingIndex"] == "EONIA") & (df["Instrument"] == "OIS")]
2529 logger.debug("CSV loaded")
2531 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
2532 ins_quotes = df_ins["Quote"].tolist()
2533 logger.debug("instrument specifications created")
2535 print("--------------Starting bootstrapper")
2536 logger.debug(f" OIS starting bootstrapper of {len(ins_quotes)} instruments")
2537 curve = bootstrap_curve(
2538 ref_date=refDate,
2539 curve_id="OIS_estr",
2540 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2541 instruments=ins_spec,
2542 quotes=ins_quotes,
2543 interpolation_type=InterpolationType.LINEAR_LOG,
2544 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2545 # interpolation_type=InterpolationType.HAGAN_DF,
2546 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2547 )
2549 # OUtput discoutn curve dates and values for test:
2550 print(" OIS: curve valueus (date, DF)")
2551 dates_ois = curve.get_dates()
2552 df_ois = curve.get_df()
2553 for i in range(len(dates_ois)):
2554 print(f"{dates_ois[i]} , {df_ois[i]}")
2556 # --------------------------------- IR for 3M
2557 logger.debug(f"Loading instruments for reference date and, currency, and EURIBOR for SHORT LEG, e.g. 3M in this case")
2558 df_ins_3m = df[
2559 (df["Date"] == date_str)
2560 & (df["Currency"] == "EUR")
2561 & (df["UnderlyingIndex"] == "EURIBOR")
2562 & (df["Instrument"] == "IRS")
2563 & (df["UnderlyingTenor"] == "3M")
2564 ]
2566 ins_spec_3m = sfc.load_specifications_from_pd(df_ins_3m, refDate, holidays)
2567 ins_quotes_3m = df_ins_3m["Quote"].tolist()
2568 logger.debug("instrument specifications created")
2570 curves = {"discount_curve": curve}
2571 print("--------------Starting bootstrapper")
2572 logger.debug(f" IRS starting bootstrapper of {len(ins_quotes_3m)} instruments")
2573 curve_3m = bootstrap_curve(
2574 ref_date=refDate,
2575 curve_id="euribor3m",
2576 day_count_convention=df_ins_3m["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2577 instruments=ins_spec_3m,
2578 quotes=ins_quotes_3m,
2579 curves=curves,
2580 interpolation_type=InterpolationType.LINEAR_LOG,
2581 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2582 # interpolation_type=InterpolationType.HAGAN_DF,
2583 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2584 )
2586 # OUtput discoutn curve dates and values for test:
2587 print(" 3m euribor: curve valueus (date, DF)")
2588 dates_3m = curve_3m.get_dates()
2589 df_3m = curve_3m.get_df()
2590 for i in range(len(dates_3m)):
2591 print(f"{dates_3m[i]} , {df_3m[i]}")
2593 # --------------------------------- FRA for 6M
2594 logger.debug(f"Loading instruments for reference date and, currency, and EURIBOR for SHORT LEG, e.g. 6M in this case")
2595 df_ins_6m = df[
2596 (df["Date"] == date_str)
2597 & (df["Currency"] == "EUR")
2598 & (df["UnderlyingIndex"] == "EURIBOR")
2599 & (df["Instrument"] == "FRA")
2600 & (df["UnderlyingTenor"] == "6M") # the test data has wrong FRA underlying tenor, but we know is 6M
2601 ]
2603 ins_spec_6m = sfc.load_specifications_from_pd(df_ins_6m, refDate, holidays)
2604 ins_quotes_6m = df_ins_6m["Quote"].tolist()
2605 logger.debug("instrument specifications created")
2607 curves = {"discount_curve": curve}
2608 print("--------------Starting bootstrapper")
2609 logger.debug(f" FRA starting bootstrapper of {len(ins_quotes_6m)} instruments")
2610 curve_6m = bootstrap_curve(
2611 ref_date=refDate,
2612 curve_id="euribor6m",
2613 day_count_convention=df_ins_6m["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2614 instruments=ins_spec_6m,
2615 quotes=ins_quotes_6m,
2616 curves=curves,
2617 interpolation_type=InterpolationType.LINEAR_LOG,
2618 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2619 # interpolation_type=InterpolationType.HAGAN_DF,
2620 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2621 )
2623 # OUtput discoutn curve dates and values for test:
2624 print(" 6m euribor: curve valueus (date, DF)")
2625 dates_6m = curve_6m.get_dates()
2626 df_6m = curve_6m.get_df()
2627 for i in range(len(dates_6m)):
2628 print(f"{dates_6m[i]} , {df_6m[i]}")
2630 # --------------------------------- TBS
2631 logger.debug(f"Loading TBS instruments for reference date and, currency, and EURIBOR, 6M long tenor")
2632 df_ins_tbs = df[
2633 (df["Date"] == date_str)
2634 & (df["Currency"] == "EUR")
2635 & (df["UnderlyingIndex"] == "EURIBOR")
2636 & (df["Instrument"] == "TBS")
2637 & (df["UnderlyingTenor"] == "6M")
2638 & (df["UnderlyingTenorShort"] == "3M")
2639 ]
2641 if df_ins_tbs.empty:
2642 raise ValueError(f"No TBS instruments found for date {date_str} with EUR/EURIBOR.")
2644 print("--------- TBS instruments used ...")
2645 for _, item in df_ins_tbs.iterrows():
2646 print(item["Date"], item["Instrument"], item["Maturity"], item["Quote"], item["UnderlyingTenorShort"], item["UnderlyingTenor"])
2648 ins_spec_tbs = sfc.load_specifications_from_pd(df_ins_tbs, refDate, holidays)
2649 # ins_quotes_tbs = df_ins_tbs["Quote"].tolist()
2650 ins_quotes_tbs = (df_ins_tbs["Quote"] / 10000.0).tolist()
2651 logger.debug("instrument specifications created")
2653 curves["basis_curve"] = curve_3m
2654 curves["initial_curve"] = curve_6m
2656 print("--------------Starting bootstrapper")
2657 logger.debug(f" TBS starting bootstrapper of {len(df_ins_tbs)} instruments")
2658 curve_6m_ext = bootstrap_curve(
2659 ref_date=refDate,
2660 curve_id="euribor_6m_extended",
2661 day_count_convention=df_ins_tbs["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2662 instruments=ins_spec_tbs,
2663 quotes=ins_quotes_tbs,
2664 curves=curves,
2665 interpolation_type=InterpolationType.LINEAR_LOG,
2666 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2667 # interpolation_type=InterpolationType.HAGAN_DF,
2668 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2669 )
2671 # print(curve.get_dates())
2672 self.assertIsInstance(curve_6m_ext, DiscountCurve)
2673 self.assertEqual(curve_6m_ext.get_dates()[0], refDate)
2674 logger.debug("bootstrapped curve dates matched")
2675 # the discount curve needs to be able to get the same market quote for the instrument
2676 for i in range(len(ins_spec_tbs)):
2677 model_quote = get_quote(refDate, ins_spec_tbs[i], {"discount_curve": curve, "fixing_curve": curve_6m_ext, "basis_curve": curve_3m})
2678 # compare to given basis points
2679 self.assertAlmostEqual(model_quote, ins_quotes_tbs[i], delta=1e-5) # since the quotes are only to 5 decimals
2680 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2681 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2682 # print(i, model_quote, curve.get_df()[i])
2684 logger.debug(f"asserted market quote matched -done")
2685 logger.debug(f"--------------------------------------------------------")
2687 def test_tbs_premade(self):
2688 """To speed up the test, and test only the production of the TBS curve given a discount curve and basis curve (e.g. 3M euribor)"""
2690 logger.debug(f"--------------------------------------------------------")
2691 logger.debug("start")
2692 # these inputs must be given by user
2693 mon = "09"
2694 day = "24"
2695 year = "2025"
2696 date_str = f"{day}.{mon}.{year}"
2697 refDate = datetime(int(year), int(mon), int(day))
2698 holidays = _ECB()
2700 df = self.quotes_df.copy()
2702 A = df[df["Date"] == date_str]
2704 logger.debug(f"Creating OIS discount Curve")
2705 # df_ins = df[df["Instrument"] == "OIS"]
2706 # dc_df_temp = df[(df["Date"] == selected_date) & (df["Currency"] == selected_currency) & (df["UnderlyingIndex"] == "ESTR") & (df["Instrument"] == "OIS") ]
2707 df_ins = df[(df["Date"] == date_str) & (df["Currency"] == "EUR") & (df["UnderlyingIndex"] == "EONIA") & (df["Instrument"] == "OIS")]
2709 logger.debug("CSV loaded")
2711 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
2712 ins_quotes = df_ins["Quote"].tolist()
2713 logger.debug("instrument specifications created")
2715 print("--------------Starting bootstrapper")
2716 logger.debug(f" OIS starting bootstrapper of {len(ins_quotes)} instruments")
2717 # curve = bootstrap_curve(
2718 # ref_date=refDate,
2719 # curve_id="OIS_estr",
2720 # day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2721 # instruments=ins_spec,
2722 # quotes=ins_quotes,
2723 # interpolation_type=InterpolationType.LINEAR_LOG,
2724 # extrapolation_type=ExtrapolationType.LINEAR_LOG,
2725 # # interpolation_type=InterpolationType.HAGAN_DF,
2726 # # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2727 # )
2728 ois_dates = [
2729 datetime(2025, 9, 24, 0, 0),
2730 datetime(2025, 10, 3, 0, 0),
2731 datetime(2025, 10, 10, 0, 0),
2732 datetime(2025, 10, 27, 0, 0),
2733 datetime(2025, 11, 26, 0, 0),
2734 datetime(2025, 12, 29, 0, 0),
2735 datetime(2026, 1, 26, 0, 0),
2736 datetime(2026, 2, 26, 0, 0),
2737 datetime(2026, 3, 26, 0, 0),
2738 datetime(2026, 4, 27, 0, 0),
2739 datetime(2026, 5, 26, 0, 0),
2740 datetime(2026, 6, 26, 0, 0),
2741 datetime(2026, 7, 27, 0, 0),
2742 datetime(2026, 8, 26, 0, 0),
2743 datetime(2026, 9, 28, 0, 0),
2744 datetime(2027, 3, 30, 0, 0),
2745 datetime(2027, 9, 27, 0, 0),
2746 datetime(2028, 9, 26, 0, 0),
2747 datetime(2029, 9, 26, 0, 0),
2748 datetime(2030, 9, 26, 0, 0),
2749 datetime(2031, 9, 26, 0, 0),
2750 datetime(2032, 9, 27, 0, 0),
2751 datetime(2033, 9, 26, 0, 0),
2752 datetime(2034, 9, 26, 0, 0),
2753 datetime(2035, 9, 26, 0, 0),
2754 datetime(2040, 9, 26, 0, 0),
2755 datetime(2045, 9, 26, 0, 0),
2756 datetime(2055, 9, 27, 0, 0),
2757 ]
2759 ois_dfs = [
2760 1.0,
2761 0.9995183313174537,
2762 0.9991111934028121,
2763 0.9981592783704503,
2764 0.9964906189442938,
2765 0.9946699855622928,
2766 0.9931280849220783,
2767 0.9914487731890309,
2768 0.9899089952356281,
2769 0.9882165869484604,
2770 0.9866937212054259,
2771 0.9850722027277637,
2772 0.983433328030071,
2773 0.9818472981388028,
2774 0.9801584057930596,
2775 0.9706595090105641,
2776 0.9604274957650127,
2777 0.9388051694832871,
2778 0.9160391166426118,
2779 0.8922971398755308,
2780 0.8681004825478764,
2781 0.8432558412799663,
2782 0.8181088759777948,
2783 0.7927910510425162,
2784 0.7677179931517396,
2785 0.6488447684675088,
2786 0.5517820851696722,
2787 0.41133831341116706,
2788 ]
2789 # ACT365FIXED, LINEAR, NONE - EXPECT ERROR TO BE THROWN for EXTRAPOLATIOn
2790 curve = DiscountCurve(
2791 "OIS_estr",
2792 refDate,
2793 ois_dates,
2794 ois_dfs,
2795 InterpolationType.LINEAR_LOG,
2796 ExtrapolationType.LINEAR_LOG,
2797 DayCounterType.ACT360,
2798 )
2800 # OUtput discoutn curve dates and values for test:
2801 print(" OIS: curve valueus (date, DF)")
2802 dates_ois = curve.get_dates()
2803 df_ois = curve.get_df()
2804 for i in range(len(dates_ois)):
2805 print(f"{dates_ois[i]} , {df_ois[i]}")
2807 # --------------------------------- IR for 3M
2808 logger.debug(f"Loading instruments for reference date and, currency, and EURIBOR for SHORT LEG, e.g. 3M in this case")
2809 df_ins_3m = df[
2810 (df["Date"] == date_str)
2811 & (df["Currency"] == "EUR")
2812 & (df["UnderlyingIndex"] == "EURIBOR")
2813 & (df["Instrument"] == "IRS")
2814 & (df["UnderlyingTenor"] == "3M")
2815 ]
2817 ins_spec_3m = sfc.load_specifications_from_pd(df_ins_3m, refDate, holidays)
2818 ins_quotes_3m = df_ins_3m["Quote"].tolist()
2819 logger.debug("instrument specifications created")
2821 curves = {"discount_curve": curve}
2822 print("--------------Starting bootstrapper")
2823 logger.debug(f" IRS starting bootstrapper of {len(ins_quotes_3m)} instruments")
2824 # curve_3m = bootstrap_curve(
2825 # ref_date=refDate,
2826 # curve_id="euribor3m",
2827 # day_count_convention=df_ins_3m["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2828 # instruments=ins_spec_3m,
2829 # quotes=ins_quotes_3m,
2830 # curves=curves,
2831 # interpolation_type=InterpolationType.LINEAR_LOG,
2832 # extrapolation_type=ExtrapolationType.LINEAR_LOG,
2833 # # interpolation_type=InterpolationType.HAGAN_DF,
2834 # # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2835 # )
2837 eur3m_dates = [
2838 datetime(2025, 9, 24, 0, 0),
2839 datetime(2026, 9, 28, 0, 0),
2840 datetime(2027, 9, 27, 0, 0),
2841 datetime(2028, 9, 26, 0, 0),
2842 datetime(2029, 9, 26, 0, 0),
2843 datetime(2030, 9, 26, 0, 0),
2844 datetime(2031, 9, 26, 0, 0),
2845 datetime(2032, 9, 27, 0, 0),
2846 datetime(2033, 9, 26, 0, 0),
2847 datetime(2034, 9, 26, 0, 0),
2848 datetime(2035, 9, 26, 0, 0),
2849 datetime(2036, 9, 26, 0, 0),
2850 datetime(2037, 9, 28, 0, 0),
2851 datetime(2040, 9, 26, 0, 0),
2852 datetime(2045, 9, 26, 0, 0),
2853 datetime(2050, 9, 26, 0, 0),
2854 datetime(2055, 9, 27, 0, 0),
2855 datetime(2065, 9, 28, 0, 0),
2856 datetime(2075, 9, 26, 0, 0),
2857 datetime(2085, 9, 26, 0, 0),
2858 ]
2860 eur3m_dfs = [
2861 1.0,
2862 0.9799030353171693,
2863 0.9597990484748371,
2864 0.9380048482094476,
2865 0.9149505395343006,
2866 0.8912209769075741,
2867 0.8670805932120672,
2868 0.8423413056820628,
2869 0.8173570936292627,
2870 0.7921402251628151,
2871 0.7670503220512309,
2872 0.7422680023945799,
2873 0.7177013315988795,
2874 0.6486808461523088,
2875 0.5525449999385698,
2876 0.4763668626646936,
2877 0.4129322567797108,
2878 0.31451512666881004,
2879 0.24795753632011802,
2880 0.20081662961055713,
2881 ]
2882 # ACT365FIXED, LINEAR, NONE - EXPECT ERROR TO BE THROWN for EXTRAPOLATIOn
2883 curve_3m = DiscountCurve(
2884 "euribor3m",
2885 refDate,
2886 eur3m_dates,
2887 eur3m_dfs,
2888 InterpolationType.LINEAR_LOG,
2889 ExtrapolationType.LINEAR_LOG,
2890 DayCounterType.ACT360,
2891 )
2893 # OUtput discoutn curve dates and values for test:
2894 print(" 3m euribor: curve valueus (date, DF)")
2895 dates_3m = curve_3m.get_dates()
2896 df_3m = curve_3m.get_df()
2897 for i in range(len(dates_3m)):
2898 print(f"{dates_3m[i]} , {df_3m[i]}")
2900 # --------------------------------- TBS
2901 logger.debug(f"Loading TBS instruments for reference date and, currency, and EURIBOR, 6M long tenor")
2902 df_ins_tbs = df[
2903 (df["Date"] == date_str)
2904 & (df["Currency"] == "EUR")
2905 & (df["UnderlyingIndex"] == "EURIBOR")
2906 & (df["Instrument"] == "TBS")
2907 & (df["UnderlyingTenor"] == "6M")
2908 ]
2910 if df_ins_tbs.empty:
2911 raise ValueError(f"No TBS instruments found for date {date_str} with EUR/EURIBOR.")
2913 print("--------- TBS instruments used ...")
2914 for _, item in df_ins_tbs.iterrows():
2915 print(item["Date"], item["Instrument"], item["Maturity"], item["Quote"], item["UnderlyingTenorShort"], item["UnderlyingTenor"])
2916 logger.info(
2917 f"{item["Date"]}, {item["Instrument"]}, {item["Maturity"]}, {item["Quote"]}, {item["UnderlyingTenorShort"]}, {item["UnderlyingTenor"]}"
2918 )
2920 ins_spec_tbs = sfc.load_specifications_from_pd(df_ins_tbs, refDate, holidays)
2921 # ins_quotes_tbs = df_ins_tbs["Quote"].tolist()
2922 ins_quotes_tbs = (df_ins_tbs["Quote"] / 10000.0).tolist()
2923 logger.debug("instrument specifications created")
2925 curves["basis_curve"] = curve_3m
2927 print("--------------Starting bootstrapper")
2928 logger.debug(f" TBS starting bootstrapper of {len(df_ins_tbs)} instruments")
2929 curve_6m = bootstrap_curve(
2930 ref_date=refDate,
2931 curve_id="OIS_estr",
2932 day_count_convention=df_ins_tbs["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
2933 instruments=ins_spec_tbs,
2934 quotes=ins_quotes_tbs,
2935 curves=curves,
2936 interpolation_type=InterpolationType.LINEAR_LOG,
2937 extrapolation_type=ExtrapolationType.LINEAR_LOG,
2938 # interpolation_type=InterpolationType.HAGAN_DF,
2939 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
2940 )
2942 # print(curve.get_dates())
2943 self.assertIsInstance(curve_6m, DiscountCurve)
2944 self.assertEqual(curve_6m.get_dates()[0], refDate)
2945 logger.debug("bootstrapped curve dates matched")
2946 # the discount curve needs to be able to get the same market quote for the instrument
2947 for i in range(len(ins_spec_tbs)):
2948 model_quote = get_quote(refDate, ins_spec_tbs[i], {"discount_curve": curve, "fixing_curve": curve_6m, "basis_curve": curve_3m})
2949 # compare to given basis points
2950 self.assertAlmostEqual(model_quote, ins_quotes_tbs[i], delta=1e-5) # since the quotes are only to 5 decimals
2951 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
2952 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
2953 # print(i, model_quote, curve.get_df()[i])
2955 logger.debug(f"asserted market quote matched -done")
2956 logger.debug(f"--------------------------------------------------------")
2957 # self.assertEqual(1, 1)
2960class TestReferenceDateDependance(unittest.TestCase):
2961 """Noticed that depending on the stated reference date, which is needed to calculate
2962 the start dates of the instruments, the bootstrapped curve can differ slightly.
2963 This test checks that bootstrapping with different reference dates, but otherwise
2964 identical input data, gives similar curves.
2966 The finer points is because the start dates, and the adjustments to the date can affect
2967 the actual day count fractions, and thus the cash flows, and thus the curve. Especially after applying
2968 conventions like modified following.
2970 Error was determined in the case of OIS, when feeding dates to the scheduler,
2971 if the expiry date is adjusted, when rolling back to get the start date, the start date can
2972 be earlier than the actual start date, causing inconsistencies and uexpected behaviour in the Scheduler.
2973 This is affected by the roll convention used.
2975 The solution implemented was for OIS swaps to calculate the actual expiry including the roll convention
2976 as well as an unadjusted expiry from which the rest of the dates can be calculated from.
2977 """
2979 def setUp(self):
2980 """Set up input file"""
2981 # set directory and file name for Input Quotes
2982 dirName = "./sample_data" # "./"
2983 fileName = "/multi_dates.csv" # "/inputQuotes.csv"
2985 df = pd.read_csv(dirName + fileName, sep=";", decimal=",")
2986 column_names = list(df.columns)
2988 self.quotes_df = df
2989 self.column_names = column_names
2991 def test_date1(self):
2992 """Using 2025 09 24 as a control date, to ensure the proper bootstrapping from Frontmark data example."""
2994 logger.debug(f"--------------------------------------------------------")
2995 logger.debug("test_date dependency 1 start")
2996 # these inputs must be given by user
2997 mon = "09"
2998 day = "24"
2999 year = "2025"
3000 date_str = f"{day}.{mon}.{year}"
3001 refDate = datetime(int(year), int(mon), int(day))
3002 holidays = _ECB()
3004 df = self.quotes_df.copy()
3006 A = df[df["Date"] == date_str]
3008 # df_ins = df[df["Instrument"] == "OIS"]
3009 # dc_df_temp = df[(df["Date"] == selected_date) & (df["Currency"] == selected_currency) & (df["UnderlyingIndex"] == "ESTR") & (df["Instrument"] == "OIS") ]
3010 df_ins = df[(df["Date"] == date_str) & (df["Currency"] == "EUR") & (df["UnderlyingIndex"] == "EONIA") & (df["Instrument"] == "OIS")]
3012 logger.debug("CSV loaded")
3014 min_i = 0
3015 max_i = 18 # internal selection to determine problematic dates
3016 min_i2 = 26
3017 max_i2 = len(df_ins)
3018 # ins_spec = sfc.load_specifications_from_pd(df_ins.iloc[np.r_[min_i:max_i, min_i2:max_i2]], refDate, holidays)
3019 # ins_quotes = df_ins["Quote"].tolist()[min_i:max_i]
3020 # ins_quotes = df_ins["Quote"].tolist()[min_i:max_i] + df_ins["Quote"].tolist()[min_i2:max_i2]
3022 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
3023 ins_quotes = df_ins["Quote"].tolist()
3024 logger.debug("instrument specifications created")
3026 # print("--------------DEBUG PARSING")
3027 # print(ins_quotes[0])
3028 # print(df_ins["DayCountFixed"].tolist()[0])
3029 # print(df_ins.iloc[min_i:max_i].copy())
3030 # print(len(ins_spec), len(ins_quotes))
3031 # print(len(df_ins["Quote"].tolist()))
3032 # print(len(df_ins.iloc[np.r_[min_i:max_i, min_i2:max_i2]]))
3033 # for i in range(len(ins_quotes)):
3034 # print(i, ins_quotes[i])
3036 print("--------------Starting bootstrapper")
3037 logger.debug(f"starting bootstrapper of {len(ins_quotes)} instruments")
3038 curve = bootstrap_curve(
3039 ref_date=refDate,
3040 curve_id="OIS_estr",
3041 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
3042 instruments=ins_spec,
3043 quotes=ins_quotes,
3044 interpolation_type=InterpolationType.LINEAR_LOG,
3045 extrapolation_type=ExtrapolationType.LINEAR_LOG,
3046 # interpolation_type=InterpolationType.HAGAN_DF,
3047 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
3048 )
3049 # print(curve.get_dates())
3050 self.assertIsInstance(curve, DiscountCurve)
3051 self.assertEqual(curve.get_dates()[0], refDate)
3052 logger.debug("bootstrapped curve dates matched")
3053 # the discount curve needs to be able to get the same market quote for the instrument
3054 for i in range(len(ins_spec)):
3055 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve, "fixing_curve": curve})
3056 self.assertAlmostEqual(model_quote, ins_quotes[i], delta=1e-5) # since the quotes are only to 5 decimals
3057 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
3058 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
3059 # print(i, model_quote, curve.get_df()[i])
3061 logger.debug(f"asserted market quote matched -done")
3062 logger.debug(f"--------------------------------------------------------")
3063 # self.assertEqual(1, 1)
3065 def test_date_many(self):
3066 """Test over all OIS, and EUR instruments for every available date in the input data set"""
3068 logger.debug(f"--------------------------------------------------------")
3069 logger.debug("Starting loop, performming bootstrap over all OIS instruments, EUR, for each unique date.")
3070 # these inputs must be given by user
3071 # mon = "09"
3072 # day = "24"
3073 # year = "2025"
3074 # date_str = f"{day}.{mon}.{year}"
3075 # refDate = datetime(int(year), int(mon), int(day))
3076 holidays = _ECB()
3078 df = self.quotes_df.copy()
3080 eur_ois = df[(df["Currency"] == "EUR") & (df["Instrument"] == "OIS")]
3082 for date, subset in eur_ois.groupby("Date"):
3083 logger.debug(f"--------------------------------------------------------")
3084 logger.debug(f"Processing {date}...")
3085 print(subset)
3086 day = date.split(".")[0]
3087 mon = date.split(".")[1]
3088 year = date.split(".")[2]
3089 refDate = datetime(int(year), int(mon), int(day))
3090 logger.debug(f"--------------------------------------------------------")
3092 df_ins = subset
3093 logger.debug("CSV loaded")
3095 ins_spec = sfc.load_specifications_from_pd(df_ins, refDate, holidays)
3096 ins_quotes = df_ins["Quote"].tolist()
3097 logger.debug("instrument specifications created")
3099 # print("--------------DEBUG PARSING")
3100 # print(ins_quotes[0])
3101 # print(df_ins["DayCountFixed"].tolist()[0])
3102 # print(df_ins.iloc[min_i:max_i].copy())
3103 # print(len(ins_spec), len(ins_quotes))
3104 # print(len(df_ins["Quote"].tolist()))
3105 # print(len(df_ins.iloc[np.r_[min_i:max_i, min_i2:max_i2]]))
3106 # for i in range(len(ins_quotes)):
3107 # print(i, ins_quotes[i])
3109 print("--------------Starting bootstrapper")
3110 logger.debug(f"starting bootstrapper of {len(ins_quotes)} instruments")
3111 curve = bootstrap_curve(
3112 ref_date=refDate,
3113 curve_id="OIS_estr",
3114 day_count_convention=df_ins["DayCountFixed"].tolist()[0], # taken the first entry and assume is valid for all other deposits
3115 instruments=ins_spec,
3116 quotes=ins_quotes,
3117 # interpolation_type=InterpolationType.HAGAN_DF,
3118 # extrapolation_type=ExtrapolationType.CONSTANT_DF,
3119 interpolation_type=InterpolationType.LINEAR_LOG,
3120 extrapolation_type=ExtrapolationType.LINEAR_LOG,
3121 )
3122 # print(curve.get_dates())
3123 self.assertIsInstance(curve, DiscountCurve)
3124 self.assertEqual(curve.get_dates()[0], refDate)
3125 logger.debug("bootstrapped curve dates matched")
3126 # the discount curve needs to be able to get the same market quote for the instrument
3127 for i in range(len(ins_spec)):
3128 model_quote = get_quote(refDate, ins_spec[i], {"discount_curve": curve, "fixing_curve": curve})
3129 self.assertAlmostEqual(
3130 model_quote, ins_quotes[i], delta=tolerance_from_quote(ins_quotes[i])
3131 ) # we adjust to check up to decimal fo the given market quote
3132 # per_diff = (model_quote - deposit_quotes[i]) / deposit_quotes[i] * 100
3133 # print(f"model: {model_quote} market: {deposit_quotes[i]} perdiff: {per_diff}")
3135 logger.debug(f"asserted market quote matched -done")
3136 logger.debug(f"--------------------------------------------------------")
3137 # self.assertEqual(1, 1)
3139 logger.debug(f"All unique datets -done")
3140 logger.debug(f"--------------------------------------------------------")
3143if __name__ == "__main__":
3144 # Open a file for capturing output
3145 with open("test_output.txt", "w") as f:
3146 # Save original stdout
3147 original_stdout = sys.stdout
3148 sys.stdout = f
3149 # Run your tests
3150 unittest.main(argv=["first-arg-is-ignored"], exit=False)
3152 # Restore stdout
3153 sys.stdout = original_stdout
3154 # unittest.main()