Coverage for rivapy / instruments / ir_swap_specification.py: 81%
451 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1from abc import abstractmethod as _abstractmethod
2from locale import currency
3from typing import List as _List, Union as _Union, Tuple, Dict
4import numpy as np
5from datetime import datetime, date, timedelta
6from rivapy.tools.holidays_compat import HolidayBase as _HolidayBase, ECB as _ECB
7from rivapy.instruments._logger import logger
9from rivapy.tools.datetools import Period, Schedule, _date_to_datetime, _datetime_to_date_list, _term_to_period, serialize_date
10from rivapy.tools.enums import DayCounterType, RollConvention, SecuritizationLevel, Currency, Rating, Instrument
11from rivapy.tools._validators import _check_positivity, _check_start_before_end, _string_to_calendar, _is_ascending_date_list
12import rivapy.tools.interfaces as interfaces
13from rivapy.tools.datetools import Period, Schedule
15from rivapy.instruments.bond_specifications import BondBaseSpecification
16from rivapy.instruments.components import NotionalStructure, ConstNotionalStructure, VariableNotionalStructure # , ResettingNotionalStructure
17from rivapy.tools.enums import IrLegType
19# Base each swap leg, off of the IRSwapBaseSpecification
20# This IRSwapBaseSpecification is in turn, based off of the BondBaseSpecification
21# Can think about basing the float/fixed leg off of the BondFlaoting/Fixed Note class...
23# WIP
26class IrSwapLegSpecification(interfaces.FactoryObject):
27 """Base interest rate swap leg specification used to define both fixed and floating legs."""
29 def __init__(
30 self,
31 obj_id: str,
32 notional: _Union[float, NotionalStructure],
33 start_dates: _List[datetime],
34 end_dates: _List[datetime],
35 pay_dates: _List[datetime],
36 currency: _Union[Currency, str],
37 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
38 ):
39 """Constructor for the IrSwapLegSpecification class used to define both fixed and floating legs.
41 Args:
42 obj_id (str): obj ID for the instrument.
43 notional (_Union[float, NotionalStructure]): If given a singular float, will convert to ConstNotinalStructure. Contains the notional information.
44 start_dates (_List[datetime]): start date of interest accrual period.
45 end_dates (_List[datetime]): end date of interest accrual period.
46 pay_dates (_List[datetime]): Dates when both legs of the swap exchange cash flows.
47 currency (_Union[Currency, str]): The currency of the swap
48 day_count_convention (_Union[DayCounterType, str], optional): The day count convention used. Defaults to DayCounterType.ThirtyU360.
49 """
50 self.obj_id = obj_id
51 self.notional_structure = notional
52 self.start_dates = start_dates
53 self.end_dates = end_dates
54 self.pay_dates = pay_dates
55 self.currency = Currency.to_string(currency)
56 self.day_count_convention = day_count_convention
58 @property
59 def currency(self) -> str:
60 """The swap leg's currency as a string."""
61 return self._currency
63 @currency.setter
64 def currency(self, value: _Union[Currency, str]):
65 self._currency = Currency.to_string(value)
67 @property
68 def start_dates(self) -> _List[datetime]:
69 """start dates for the interest periods
71 Returns:
72 _List[datetime]: _description_
73 """
74 return self._start_dates
76 @start_dates.setter
77 def start_dates(self, value: _List[datetime]):
78 self._start_dates = value
80 @property
81 def end_dates(self) -> _List[datetime]:
82 """end dates for the interest periods
84 Returns:
85 _List[datetime]: _description_
86 """
87 return self._end_dates
89 @end_dates.setter
90 def end_dates(self, value: _List[datetime]):
91 self._end_dates = value
93 @property
94 def pay_dates(self) -> _List[datetime]:
95 """pay dates for the interest periods
97 Returns:
98 _List[datetime]:
99 """
100 return self._pay_dates
102 @pay_dates.setter
103 def pay_dates(self, value: _List[datetime]):
104 self._pay_dates = value
106 @property
107 def notional_structure(self) -> NotionalStructure:
108 """Return the notionals
110 Returns:
111 NotionalStructure: class object detailing the notionals, start dates, ...
112 """
113 return self._notional_structure
115 @notional_structure.setter
116 def notional_structure(self, value: _Union[float, NotionalStructure]):
117 """If only a float is given, assume a constant notional and create a ConstNotionalStructure.
119 Args:
120 value (_Union[float, NotionalStructure]):
121 """
122 if isinstance(value, (int, float)):
123 self._notional_structure = ConstNotionalStructure(float(value))
124 else:
125 self._notional_structure = value
127 # @abstractmethod
128 # def reset_dates(self) -> _List[datetime]:
129 # """ #TODO brought over from pyvacon, for float leg?
130 # """
131 # pass
133 def _to_dict(self) -> Dict:
134 return_dict = {
135 "obj_id": self.obj_id,
136 "notional": self.notional_structure,
137 "start_dates": self.start_dates,
138 "end_dates": self.end_dates,
139 "pay_dates": self.pay_dates,
140 "currency": self.currency,
141 "day_count_convention": self.day_count_convention,
142 }
143 return return_dict
145 @staticmethod
146 def _create_sample(n_samples: int, seed: int = None):
147 pass # TODO
150class IrFixedLegSpecification(IrSwapLegSpecification):
151 """Specification for a fixed leg for an interest rate swap."""
153 def __init__(
154 self,
155 fixed_rate: float,
156 obj_id: str,
157 notional: _Union[float, NotionalStructure],
158 start_dates: _List[datetime],
159 end_dates: _List[datetime],
160 pay_dates: _List[datetime],
161 currency: _Union[Currency, str],
162 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
163 ):
164 """Constructor for a fixed leg of an interest rate swap.
166 Args:
167 fixed_rate (float): The fixed interest rate defining this leg of the swap.
168 obj_id (str): obj ID for the instrument.
169 notional (_Union[float, NotionalStructure]): If given a singular float, will convert to ConstNotinalStructure. Contains the notional information.
170 start_dates (_List[datetime]): start date of interest accrual period.
171 end_dates (_List[datetime]): end date of interest accrual period.
172 pay_dates (_List[datetime]): Dates when both legs of the swap exchange cash flows.
173 currency (_Union[Currency, str]): The currency of the swap
174 day_count_convention (_Union[DayCounterType, str], optional): The day count convention used. Defaults to DayCounterType.ThirtyU360.
175 """
176 super().__init__(obj_id, notional, start_dates, end_dates, pay_dates, currency, day_count_convention)
177 self.fixed_rate = fixed_rate # _check_positivity(fixed_rate) # TODO is there a need for it to be always positive?
179 # region properties
180 @property
181 def leg_type(self) -> str:
182 return IrLegType.FIXED
184 @property
185 def fixed_rate(self) -> float:
186 return self._fixed_rate
188 @fixed_rate.setter
189 def fixed_rate(self, value: float):
190 self._fixed_rate = value
192 # @property
193 # def reset_dates(self) -> _List[datetime]:
194 # return self.start_dates
196 @property
197 def udl_id(self) -> str:
198 return "" # fixed leg has no underlying
200 def _to_dict(self):
202 return_dict = super()._to_dict()
203 return_dict["fixed_rate"] = self.fixed_rate
204 return return_dict
206 @staticmethod
207 def _create_sample(n_samples: int, seed: int = None):
208 pass # TODO
210 # endregion
212 def get_NotionalStructure(self):
214 return self.notional_structure
217class IrFloatLegSpecification(IrSwapLegSpecification):
218 def __init__(
219 self,
220 obj_id: str,
221 notional: _Union[float, NotionalStructure],
222 reset_dates: _List[datetime],
223 start_dates: _List[datetime],
224 end_dates: _List[datetime],
225 rate_start_dates: _List[datetime], # are these needed here? or are they obtained from the underlying
226 rate_end_dates: _List[datetime],
227 pay_dates: _List[datetime],
228 currency: _Union[Currency, str],
229 udl_id: str,
230 fixing_id: str,
231 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
232 rate_day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
233 spread: float = 0.0,
234 ):
235 """Constructor for a floating leg of an interest rate swap.
237 Args:
238 obj_id (str): obj ID for the instrument.
239 notional (_Union[float, NotionalStructure]): If given a singular float, will convert to ConstNotinalStructure. Contains the notional information.
240 reset_dates (_List[datetime]): Date on which the floating rate (e.g., SOFR, LIBOR) is determined
241 start_dates (_List[datetime]): Date the entire swap begins (effective date)
242 end_dates (_List[datetime]): Date the swap matures
243 rate_start_dates (_List[datetime]): start dates for the determination of the underlying rate
244 rate_end_dates (_List[datetime]): end dates for the determination of the underlying rate
245 pay_dates (_List[datetime]): Dates when both legs of the swap exchange cash flows.
246 currency (_Union[Currency, str]): The currency of the swap
247 udl_id (str): ID of the underlying rate
248 fixing_id (str): fixing id
249 day_count_convention (_Union[DayCounterType, str], optional): The day count convention used.. Defaults to DayCounterType.ThirtyU360.
250 rate_day_count_convention (_Union[DayCounterType, str], optional): The day count convention used for the underlying
251 . Defaults to DayCounterType.ThirtyU360.
252 spread (float, optional): Defaults to 0.0.
253 """
254 super().__init__(obj_id, notional, start_dates, end_dates, pay_dates, currency, day_count_convention)
255 self.reset_dates = reset_dates # TODO: ADD setters to get rid of error notification?
256 self.rate_start_dates = rate_start_dates
257 self.rate_end_dates = rate_end_dates
258 self._spread = spread
259 self.udl_id = udl_id
260 self.fixing_id = fixing_id
261 self.rate_day_count_convention = DayCounterType.to_string(rate_day_count_convention)
263 # TODO to_dict method
265 # region properties
266 @property
267 def leg_type(self) -> str:
268 return IrLegType.FLOAT
270 @property
271 def reset_dates(self) -> _List[datetime]:
272 return self._reset_dates
274 @reset_dates.setter
275 def reset_dates(self, value):
276 self._reset_dates = value
278 @property
279 def udl_id(self) -> str:
280 return self._udl_id
282 @udl_id.setter
283 def udl_id(self, value):
284 self._udl_id = value
286 @property
287 def fixing_id(self) -> str:
288 return self._fixing_id
290 @fixing_id.setter
291 def fixing_id(self, value):
292 self._fixing_id = value
294 @property
295 def spread(self) -> float:
296 return self._spread
298 @spread.setter
299 def spread(self, value):
300 self._spread = value
302 @property
303 def rate_day_count(self) -> str:
304 return self.rate_day_count
306 @property
307 def rate_start_dates(self) -> _List[datetime]:
308 return self._rate_start_dates
310 @rate_start_dates.setter
311 def rate_start_dates(self, value):
312 self._rate_start_dates = value
314 @property
315 def rate_end_dates(self) -> _List[datetime]:
316 return self._rate_end_dates
318 @rate_end_dates.setter
319 def rate_end_dates(self, value):
320 self._rate_end_dates = value
322 def get_underlyings(self) -> Dict[str, str]:
323 return {self.udl_id: self.fixing_id}
325 # endregion
327 def get_NotionalStructure(self):
329 return self.notional_structure
332class IrOISLegSpecification(IrSwapLegSpecification):
333 def __init__(
334 self,
335 obj_id: str,
336 notional: _Union[float, NotionalStructure],
337 rate_reset_dates: _List[datetime],
338 start_dates: _List[datetime],
339 end_dates: _List[datetime],
340 rate_start_dates: _List[datetime], # are these needed here? or are they obtained from the underlying
341 rate_end_dates: _List[datetime],
342 pay_dates: _List[datetime],
343 currency: _Union[Currency, str],
344 udl_id: str,
345 fixing_id: str,
346 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
347 rate_day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
348 spread: float = 0.0,
349 ):
351 # fixingLag?
352 # payLag?
353 # freq?
354 # rate currency vs notional currency?
355 # leg holidays, and rate holidays - again to be dealt with in the scheduler?
356 # FYI scheduler needs, start date, end date, freq, roll convention, holiday...
357 # for the reset dates, will need fixing lag
358 # for the pay dates generation, will need payLag
360 """Constructor for a floating leg of an overnight index swap.
362 Args:
363 obj_id (str): obj ID for the instrument.
364 notional (_Union[float, NotionalStructure]): If given a singular float, will convert to ConstNotinalStructure. Contains the notional information.
365 reset_dates (_List[datetime]): Date on which the floating rate (e.g., SOFR, LIBOR) is determined
366 start_dates (_List[datetime]): Date the entire swap begins (effective date)
367 end_dates (_List[datetime]): Date the swap matures
368 rate_start_dates (_List[datetime]): start dates for the determination of the underlying rate
369 rate_end_dates (_List[datetime]): end dates for the determination of the underlying rate
370 pay_dates (_List[datetime]): Dates when both legs of the swap exchange cash flows.
371 currency (_Union[Currency, str]): The currency of the swap
372 udl_id (str): ID of the underlying rate
373 fixing_id (str): fixing id
374 day_count_convention (_Union[DayCounterType, str], optional): The day count convention used.. Defaults to DayCounterType.ThirtyU360.
375 rate_day_count_convention (_Union[DayCounterType, str], optional): The day count convention used for the underlying
376 . Defaults to DayCounterType.ThirtyU360.
377 spread (float, optional): . Defaults to 0.0.
378 """
379 super().__init__(obj_id, notional, start_dates, end_dates, pay_dates, currency, day_count_convention)
380 self.rate_reset_dates = rate_reset_dates
381 self.rate_start_dates = rate_start_dates
382 self.rate_end_dates = rate_end_dates
383 self._spread = spread
384 self.udl_id = udl_id
385 self.fixing_id = fixing_id
386 self.rate_day_count_convention = DayCounterType.to_string(rate_day_count_convention)
388 # TODO to_dict method
390 # region properties
391 @property
392 def leg_type(self) -> str:
393 return IrLegType.OIS
395 @property
396 def reset_dates(self) -> _List[datetime]:
397 return self._rate_reset_dates
399 @reset_dates.setter
400 def rate_reset_dates(self, value):
401 self._rate_reset_dates = value
403 @property
404 def udl_id(self) -> str:
405 return self._udl_id
407 @udl_id.setter
408 def udl_id(self, value):
409 self._udl_id = value
411 @property
412 def fixing_id(self) -> str:
413 return self._fixing_id
415 @fixing_id.setter
416 def fixing_id(self, value):
417 self._fixing_id = value
419 @property
420 def spread(self) -> float:
421 return self._spread
423 @spread.setter
424 def spread(self, value):
425 self._spread = value
427 @property
428 def rate_day_count(self) -> str:
429 return self.rate_day_count
431 @property
432 def rate_start_dates(self) -> _List[datetime]:
433 return self._rate_start_dates
435 @rate_start_dates.setter
436 def rate_start_dates(self, value):
437 self._rate_start_dates = value
439 @property
440 def rate_end_dates(self) -> _List[datetime]:
441 return self._rate_end_dates
443 @rate_end_dates.setter
444 def rate_end_dates(self, value):
445 self._rate_end_dates = value
447 def get_underlyings(self) -> Dict[str, str]:
448 return {self.udl_id: self.fixing_id}
450 # endregion
452 def get_NotionalStructure(self):
454 return self.notional_structure
456 @staticmethod
457 def ois_scheduler_2D(start_dates: _List[datetime], end_dates: _List[datetime]):
458 """The OIS makeshift scheduler to account for expected 2D array structure of
459 >>> rivapy.instruments.interest_rate_swap_pricer.populate_cashflows_ois
460 Alternatively, the rivapy Scheduler class is used.
461 This is a simplified version for testing purposes.
462 Args:
463 start_dates (_List[datetime]): start dates for a given accrual period, containing the daily start rates inside that accrual period
464 end_dates (_List[datetime]): _description_
466 Returns:
467 _type_: Multi-dimensional lists of datetimes
468 """
470 daily_rate_start_dates = [] # 2D list: coupon i -> list of daily starts
471 daily_rate_end_dates = [] # 2D list: coupon i -> list of daily ends
472 daily_rate_reset_dates = [] # 2D list: coupon i -> list of reset dates
473 pay_dates = [] # 1D list: one pay date per coupon
475 for i in range(len(start_dates)):
477 # for this test we keep it simple and ignore conventions e.g. business day or so. i.e just take every day
478 num_days = (end_dates[i] - start_dates[i]).days
479 daily_schedule = [start_dates[i] + timedelta(days=j) for j in range(num_days)]
481 # Build start/end date pairs for accrual periods
482 starts = daily_schedule[:-1] # all except last
483 ends = daily_schedule[1:] # all except first
485 daily_rate_start_dates.append(starts)
486 daily_rate_end_dates.append(ends)
488 # 4. Compute reset dates (fixing lag applied to each start)
489 # resets = [add_business_days(start, fixingLag, rateHolidays)
490 # for start in starts]
491 # assume simple case reset date is the same as start date
492 resets = starts # reset dates are equal to start dates if spot lag is 0.
493 daily_rate_reset_dates.append(resets)
495 # Compute payment date for the coupon
496 # pay_date = add_business_days(end_dates[i], payLag, holidays)
497 # assume simple case, pay date is end date
498 pay_date = end_dates[i]
499 pay_dates.append(pay_date)
501 return [daily_rate_start_dates, daily_rate_end_dates, daily_rate_reset_dates, pay_dates]
504class InterestRateSwapSpecification(interfaces.FactoryObject):
506 def __init__(
507 self,
508 obj_id: str,
509 notional: _Union[float, NotionalStructure],
510 issue_date: _Union[date, datetime],
511 maturity_date: _Union[date, datetime],
512 pay_leg: _Union[IrFixedLegSpecification, IrFloatLegSpecification, IrOISLegSpecification],
513 receive_leg: _Union[IrFixedLegSpecification, IrFloatLegSpecification, IrOISLegSpecification],
514 currency: _Union[Currency, str] = "EUR",
515 calendar: _Union[_HolidayBase, str] = None,
516 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
517 business_day_convention: _Union[RollConvention, str] = RollConvention.FOLLOWING,
518 issuer: str = None,
519 securitization_level: _Union[SecuritizationLevel, str] = SecuritizationLevel.NONE,
520 rating: _Union[Rating, str] = Rating.NONE,
521 ):
522 """Specification of the entire swap, encapsulating both the pay leg and the receive leg.
524 Args:
525 obj_id (str): _description_
526 notional (_Union[float, NotionalStructure]): _description_
527 issue_date (_Union[date, datetime]): _description_
528 maturity_date (_Union[date, datetime]): _description_
529 pay_leg (_Union[IrFixedLegSpecification, IrFloatLegSpecification]): _description_
530 receive_leg (_Union[IrFixedLegSpecification, IrFloatLegSpecification]): _description_
531 currency (_Union[Currency, str], optional): _description_. Defaults to "EUR".
532 calendar (_Union[_HolidayBase, str], optional): _description_. Defaults to None.
533 day_count_convention (_Union[DayCounterType, str], optional): _description_. Defaults to DayCounterType.ThirtyU360.
534 business_day_convention (_Union[RollConvention, str], optional): _description_. Defaults to RollConvention.FOLLOWING.
535 issuer (str, optional): _description_. Defaults to None.
536 securitization_level (_Union[SecuritizationLevel, str], optional): _description_. Defaults to SecuritizationLevel.NONE.
537 rating (_Union[Rating, str], optional): _description_. Defaults to Rating.NONE.
538 """
539 self.obj_id = obj_id
540 if issuer is not None:
541 self.issuer = issuer
542 if securitization_level is not None:
543 self.securitization_level = securitization_level
544 self.issue_date = issue_date
545 self.maturity_date = maturity_date
546 self.currency = currency
547 self.notional_structure = notional
548 self.rating = Rating.to_string(rating)
549 # validate dates
550 self._validate_derived_issued_instrument()
551 self.pay_leg = pay_leg
552 self.receive_leg = receive_leg
553 self.day_count_convention = day_count_convention
554 self.business_day_convention = RollConvention.to_string(business_day_convention)
555 if calendar is None:
556 self.calendar = _ECB(years=range(issue_date.year, maturity_date.year + 1))
557 else:
558 self.calendar = _string_to_calendar(calendar)
560 @staticmethod # TODO
561 def _create_sample(
562 n_samples: int, seed: int = None, ref_date=None, issuers: _List[str] = None, sec_levels: _List[str] = None, currencies: _List[str] = None
563 ) -> _List["InterestRateSwapSpecification"]:
564 if seed is not None:
565 np.random.seed(seed)
566 if ref_date is None:
567 ref_date = datetime.now()
568 else:
569 ref_date = _date_to_datetime(ref_date)
570 if issuers is None:
571 issuers = ["Issuer_" + str(i) for i in range(int(n_samples / 2))]
572 result = []
573 if currencies is None:
574 currencies = list(Currency)
575 if sec_levels is None:
576 sec_levels = list(SecuritizationLevel)
577 for i in range(n_samples):
578 days = int(15.0 * 365.0 * np.random.beta(2.0, 2.0)) + 1
579 issue_date = ref_date + timedelta(days=np.random.randint(low=-365, high=0))
580 result.append(
581 InterestRateSwapSpecification(
582 obj_id=f"IRS_{i}",
583 issue_date=issue_date,
584 maturity_date=ref_date + timedelta(days=days),
585 currency=np.random.choice(currencies),
586 notional=np.random.choice([100.0, 1000.0, 10_000.0, 100_0000.0]),
587 issuer=np.random.choice(issuers),
588 securitization_level=np.random.choice(sec_levels),
589 )
590 )
591 return result
593 def _validate_derived_issued_instrument(self):
594 self.__issue_date, self.__maturity_date = _check_start_before_end(self.__issue_date, self.__maturity_date)
596 def _to_dict(self) -> dict:
597 result = {
598 "obj_id": self.obj_id,
599 "issuer": self.issuer,
600 "securitization_level": self.securitization_level,
601 "issue_date": serialize_date(self.issue_date),
602 "maturity_date": serialize_date(self.maturity_date),
603 "currency": self.currency,
604 "notional": self.notional_structure,
605 "rating": self.rating,
606 "receive_leg": self.receive_leg,
607 "pay_leg": self.pay_leg,
608 "calendar": getattr(self.calendar, "name", self.calendar.__class__.__name__),
609 "day_count_convention": self.day_count_convention,
610 "business_day_convention": self.business_day_convention,
611 }
612 return result
614 # region properties
615 @property
616 def issuer(self) -> str:
617 """
618 Getter for instrument's issuer.
620 Returns:
621 str: Instrument's issuer.
622 """
623 return self.__issuer
625 @issuer.setter
626 def issuer(self, issuer: str):
627 """
628 Setter for instrument's issuer.
630 Args:
631 issuer(str): Issuer of the instrument.
632 """
633 self.__issuer = issuer
635 @property
636 def rating(self) -> str:
637 return self.__rating
639 @rating.setter
640 def rating(self, rating: _Union[Rating, str]) -> str:
641 self.__rating = Rating.to_string(rating)
643 @property
644 def securitization_level(self) -> str:
645 """
646 Getter for instrument's securitisation level.
648 Returns:
649 str: Instrument's securitisation level.
650 """
651 return self.__securitization_level
653 @securitization_level.setter
654 def securitization_level(self, securitisation_level: _Union[SecuritizationLevel, str]):
655 self.__securitization_level = SecuritizationLevel.to_string(securitisation_level)
657 @property
658 def issue_date(self) -> date:
659 """
660 Getter for IR swap's issue date.
662 Returns:
663 date: IR swap's issue date.
664 """
665 return self.__issue_date
667 @issue_date.setter
668 def issue_date(self, issue_date: _Union[datetime, date]):
669 """
670 Setter for IR swap's issue date.
672 Args:
673 issue_date (Union[datetime, date]): IR swap's issue date.
674 """
675 self.__issue_date = _date_to_datetime(issue_date)
677 @property
678 def maturity_date(self) -> date:
679 """
680 Getter for IR swap's maturity date.
682 Returns:
683 date: IR swap's maturity date.
684 """
685 return self.__maturity_date
687 @maturity_date.setter
688 def maturity_date(self, maturity_date: _Union[datetime, date]):
689 """
690 Setter for IR swap's maturity date.
692 Args:
693 maturity_date (Union[datetime, date]): IR swap's maturity date.
694 """
695 self.__maturity_date = _date_to_datetime(maturity_date)
697 @property
698 def currency(self) -> str:
699 """
700 Getter for IR swap's currency.
702 Returns:
703 str: IR swap's ISO 4217 currency code
704 """
705 return self.__currency
707 @currency.setter
708 def currency(self, currency: str):
709 self.__currency = Currency.to_string(currency)
711 @property
712 def notional_structure(self) -> NotionalStructure:
713 """Return the notionals
715 Returns:
716 NotionalStructure: class object detailing the notionals, start dates, ...
717 """
718 return self._notional_structure
720 @notional_structure.setter
721 def notional_structure(self, value: _Union[float, NotionalStructure]):
722 """If only a float is given, assume a constant notional and create a ConstNotionalStructure.
724 Args:
725 value (_Union[float, NotionalStructure]): _description_
726 """
727 if isinstance(value, (int, float)):
728 self._notional_structure = ConstNotionalStructure(float(value))
729 else:
730 self._notional_structure = value
732 def get_pay_leg(self):
733 return self.pay_leg
735 def get_receive_leg(self):
736 return self.receive_leg
738 def get_fixed_leg(self):
739 """get the fixed leg (only possible for fixed vs. floating swaps -> throws otherwise)"""
741 pay_leg_is_fixed = self.get_pay_leg().leg_type == IrLegType.FIXED
742 receive_leg_is_fixed = self.get_receive_leg().leg_type == IrLegType.FIXED
744 if pay_leg_is_fixed and not receive_leg_is_fixed:
745 leg = self.get_pay_leg()
746 elif receive_leg_is_fixed and not pay_leg_is_fixed:
747 leg = self.get_receive_leg()
748 else:
749 raise ValueError("Swap is not comprised of one fixed leg and one float/OIS leg!")
751 return leg
753 def get_float_leg(self):
754 """get the float leg (only possible for fixed vs. floating swaps -> throws otherwise)"""
756 pay_leg_is_float = self.get_pay_leg().leg_type in [IrLegType.FLOAT, IrLegType.OIS]
757 receive_leg_is_float = self.get_receive_leg().leg_type in [IrLegType.FLOAT, IrLegType.OIS]
759 if pay_leg_is_float and not receive_leg_is_float:
760 leg = self.get_pay_leg()
761 elif receive_leg_is_float and not pay_leg_is_float:
762 leg = self.get_receive_leg()
763 else:
764 raise ValueError("Swap is not comprised of one fixed leg and one float/OIS leg!")
766 return leg
768 def ins_type(self):
769 """Return instrument type
771 Returns:
772 Instrument: Interest Rate Swap
773 """
774 return Instrument.IRS
776 # temp placeholder
777 def get_end_date(self):
778 return self.maturity_date
780 # endregion
783class InterestRateBasisSwapSpecification(interfaces.FactoryObject):
785 def __init__(
786 self,
787 obj_id: str,
788 notional: _Union[float, NotionalStructure],
789 issue_date: _Union[date, datetime],
790 maturity_date: _Union[date, datetime],
791 pay_leg: IrFloatLegSpecification,
792 receive_leg: IrFloatLegSpecification,
793 spread_leg: IrFixedLegSpecification,
794 currency: _Union[Currency, str] = "EUR",
795 calendar: _Union[_HolidayBase, str] = None,
796 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
797 business_day_convention: _Union[RollConvention, str] = RollConvention.FOLLOWING,
798 issuer: str = None,
799 securitization_level: _Union[SecuritizationLevel, str] = SecuritizationLevel.NONE,
800 rating: _Union[Rating, str] = Rating.NONE,
801 ):
802 """Specification of the entire swap, encapsulating both the pay leg and the receive leg.
804 Args:
805 obj_id (str): _description_
806 notional (_Union[float, NotionalStructure]): _description_
807 issue_date (_Union[date, datetime]): _description_
808 maturity_date (_Union[date, datetime]): _description_
809 pay_leg (_Union[IrFixedLegSpecification, IrFloatLegSpecification]): _description_
810 receive_leg (_Union[IrFixedLegSpecification, IrFloatLegSpecification]): _description_
811 currency (_Union[Currency, str], optional): _description_. Defaults to "EUR".
812 calendar (_Union[_HolidayBase, str], optional): _description_. Defaults to None.
813 day_count_convention (_Union[DayCounterType, str], optional): _description_. Defaults to DayCounterType.ThirtyU360.
814 business_day_convention (_Union[RollConvention, str], optional): _description_. Defaults to RollConvention.FOLLOWING.
815 issuer (str, optional): _description_. Defaults to None.
816 securitization_level (_Union[SecuritizationLevel, str], optional): _description_. Defaults to SecuritizationLevel.NONE.
817 rating (_Union[Rating, str], optional): _description_. Defaults to Rating.NONE.
818 """
819 self.obj_id = obj_id
820 if issuer is not None:
821 self.issuer = issuer
822 if securitization_level is not None:
823 self.securitization_level = securitization_level
824 self.issue_date = issue_date
825 self.maturity_date = maturity_date
826 self.currency = currency
827 self.notional_structure = notional
828 self.rating = Rating.to_string(rating)
829 # validate dates
830 self._validate_derived_issued_instrument()
831 self.pay_leg = pay_leg
832 self.receive_leg = receive_leg
833 self.spread_leg = spread_leg
834 self.day_count_convention = day_count_convention
835 self.business_day_convention = RollConvention.to_string(business_day_convention)
836 if calendar is None:
837 self.calendar = _ECB(years=range(issue_date.year, maturity_date.year + 1))
838 else:
839 self.calendar = _string_to_calendar(calendar)
841 @staticmethod # TODO
842 def _create_sample(
843 n_samples: int, seed: int = None, ref_date=None, issuers: _List[str] = None, sec_levels: _List[str] = None, currencies: _List[str] = None
844 ) -> _List["InterestRateBasisSwapSpecification"]:
845 if seed is not None:
846 np.random.seed(seed)
847 if ref_date is None:
848 ref_date = datetime.now()
849 else:
850 ref_date = _date_to_datetime(ref_date)
851 if issuers is None:
852 issuers = ["Issuer_" + str(i) for i in range(int(n_samples / 2))]
853 result = []
854 if currencies is None:
855 currencies = list(Currency)
856 if sec_levels is None:
857 sec_levels = list(SecuritizationLevel)
858 for i in range(n_samples):
859 days = int(15.0 * 365.0 * np.random.beta(2.0, 2.0)) + 1
860 issue_date = ref_date + timedelta(days=np.random.randint(low=-365, high=0))
861 pay_leg = IrFloatLegSpecification() # TODO how to randomly generate?
862 receive_leg = IrFloatLegSpecification()
863 spread_leg = IrFixedLegSpecification()
864 result.append(
865 InterestRateBasisSwapSpecification(
866 obj_id=f"IRS_{i}",
867 issue_date=issue_date,
868 maturity_date=ref_date + timedelta(days=days),
869 pay_leg=pay_leg,
870 receive_leg=receive_leg,
871 spread_leg=spread_leg,
872 currency=np.random.choice(currencies),
873 notional=np.random.choice([100.0, 1000.0, 10_000.0, 100_0000.0]),
874 issuer=np.random.choice(issuers),
875 securitization_level=np.random.choice(sec_levels),
876 )
877 )
878 return result
880 def _validate_derived_issued_instrument(self):
881 self.__issue_date, self.__maturity_date = _check_start_before_end(self.__issue_date, self.__maturity_date)
883 def _to_dict(self) -> dict:
884 result = {
885 "obj_id": self.obj_id,
886 "issuer": self.issuer,
887 "securitization_level": self.securitization_level,
888 "issue_date": serialize_date(self.issue_date),
889 "maturity_date": serialize_date(self.maturity_date),
890 "currency": self.currency,
891 "notional": self.notional_structure,
892 "rating": self.rating,
893 "receive_leg": self.receive_leg,
894 "pay_leg": self.pay_leg,
895 "spread_leg": self.spread_leg,
896 "calendar": getattr(self.calendar, "name", self.calendar.__class__.__name__),
897 "day_count_convention": self.day_count_convention,
898 "business_day_convention": self.business_day_convention,
899 }
900 return result
902 # region properties
903 @property
904 def issuer(self) -> str:
905 """
906 Getter for instrument's issuer.
908 Returns:
909 str: Instrument's issuer.
910 """
911 return self.__issuer
913 @issuer.setter
914 def issuer(self, issuer: str):
915 """
916 Setter for instrument's issuer.
918 Args:
919 issuer(str): Issuer of the instrument.
920 """
921 self.__issuer = issuer
923 @property
924 def rating(self) -> str:
925 return self.__rating
927 @rating.setter
928 def rating(self, rating: _Union[Rating, str]) -> str:
929 self.__rating = Rating.to_string(rating)
931 @property
932 def securitization_level(self) -> str:
933 """
934 Getter for instrument's securitisation level.
936 Returns:
937 str: Instrument's securitisation level.
938 """
939 return self.__securitization_level
941 @securitization_level.setter
942 def securitization_level(self, securitisation_level: _Union[SecuritizationLevel, str]):
943 self.__securitization_level = SecuritizationLevel.to_string(securitisation_level)
945 @property
946 def issue_date(self) -> date:
947 """
948 Getter for IR swap's issue date.
950 Returns:
951 date: IR swap's issue date.
952 """
953 return self.__issue_date
955 @issue_date.setter
956 def issue_date(self, issue_date: _Union[datetime, date]):
957 """
958 Setter for IR swap's issue date.
960 Args:
961 issue_date (Union[datetime, date]): IR swap's issue date.
962 """
963 self.__issue_date = _date_to_datetime(issue_date)
965 @property
966 def maturity_date(self) -> date:
967 """
968 Getter for IR swap's maturity date.
970 Returns:
971 date: IR swap's maturity date.
972 """
973 return self.__maturity_date
975 @maturity_date.setter
976 def maturity_date(self, maturity_date: _Union[datetime, date]):
977 """
978 Setter for IR swap's maturity date.
980 Args:
981 maturity_date (Union[datetime, date]): IR swap's maturity date.
982 """
983 self.__maturity_date = _date_to_datetime(maturity_date)
985 @property
986 def currency(self) -> str:
987 """
988 Getter for IR swap's currency.
990 Returns:
991 str: IR swap's ISO 4217 currency code
992 """
993 return self.__currency
995 @currency.setter
996 def currency(self, currency: str):
997 self.__currency = Currency.to_string(currency)
999 @property
1000 def notional_structure(self) -> NotionalStructure:
1001 """Return the notionals
1003 Returns:
1004 NotionalStructure: class object detailing the notionals, start dates, ...
1005 """
1006 return self._notional_structure
1008 @notional_structure.setter
1009 def notional_structure(self, value: _Union[float, NotionalStructure]):
1010 """If only a float is given, assume a constant notional and create a ConstNotionalStructure.
1012 Args:
1013 value (_Union[float, NotionalStructure]): _description_
1014 """
1015 if isinstance(value, (int, float)):
1016 self._notional_structure = ConstNotionalStructure(float(value))
1017 else:
1018 self._notional_structure = value
1020 def get_pay_leg(self):
1021 return self.pay_leg
1023 def get_receive_leg(self):
1024 return self.receive_leg
1026 def get_spread_leg(self):
1027 return self.spread_leg
1029 def get_fixed_leg(self):
1030 """get the fixed leg (only possible for fixed vs. floating swaps -> throws otherwise)"""
1032 pay_leg_is_fixed = self.get_pay_leg().leg_type == IrLegType.FIXED
1033 receive_leg_is_fixed = self.get_receive_leg().leg_type == IrLegType.FIXED
1035 if pay_leg_is_fixed and not receive_leg_is_fixed:
1036 leg = self.get_pay_leg()
1037 elif receive_leg_is_fixed and not pay_leg_is_fixed:
1038 leg = self.get_receive_leg()
1039 else:
1040 raise ValueError("Swap is not comprised of one fixed leg and one float/OIS leg!")
1042 return leg
1044 def get_float_leg(self):
1045 """get the float leg (only possible for fixed vs. floating swaps -> throws otherwise)"""
1047 pay_leg_is_float = self.get_pay_leg().leg_type in [IrLegType.FLOAT, IrLegType.OIS]
1048 receive_leg_is_float = self.get_receive_leg().leg_type in [IrLegType.FLOAT, IrLegType.OIS]
1050 if pay_leg_is_float and not receive_leg_is_float:
1051 leg = self.get_pay_leg()
1052 elif receive_leg_is_float and not pay_leg_is_float:
1053 leg = self.get_receive_leg()
1054 else:
1055 raise ValueError("Swap is not comprised of one fixed leg and one float/OIS leg!")
1057 return leg
1059 def ins_type(self):
1060 """Return instrument type
1062 Returns:
1063 Instrument: Basis Swap
1064 """
1065 return Instrument.BS
1067 # temp placeholder
1068 def get_end_date(self):
1069 return self.maturity_date
1071 # endregion