Coverage for rivapy / instruments / fra_specifications.py: 88%
217 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1from rivapy.instruments._logger import logger
2from abc import abstractmethod as _abstractmethod
3from typing import List as _List, Union as _Union, Tuple, Optional as _Optional
4import numpy as np
5from datetime import datetime, date, timedelta
6from dateutil.relativedelta import relativedelta
7from rivapy.tools.holidays_compat import HolidayBase as _HolidayBase, ECB as _ECB
8from rivapy.instruments.components import Issuer
9from rivapy.tools.datetools import (
10 Period,
11 Schedule,
12 _date_to_datetime,
13 _datetime_to_date_list,
14 _term_to_period,
15 roll_day,
16 calc_start_day,
17 serialize_date,
18)
19from rivapy.tools.enums import (
20 DayCounterType,
21 RollConvention,
22 SecuritizationLevel,
23 Currency,
24 Rating,
25 Instrument,
26 InterestRateIndex,
27 get_index_by_alias,
28)
29from rivapy.tools._validators import _check_positivity, _check_start_before_end, _string_to_calendar, _is_ascending_date_list
30import rivapy.tools.interfaces as interfaces
33class ForwardRateAgreementSpecification(interfaces.FactoryObject):
35 def __init__(
36 self,
37 obj_id: str,
38 trade_date: _Union[date, datetime],
39 notional: float,
40 rate: float,
41 start_date: _Union[date, datetime],
42 end_date: _Union[date, datetime],
43 udlID: str,
44 rate_start_date: _Union[date, datetime],
45 rate_end_date: _Union[date, datetime],
46 maturity_date: _Union[date, datetime] = None,
47 day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
48 business_day_convention: _Union[RollConvention, str] = RollConvention.FOLLOWING,
49 rate_day_count_convention: _Union[DayCounterType, str] = DayCounterType.ThirtyU360,
50 rate_business_day_convention: _Union[RollConvention, str] = RollConvention.FOLLOWING,
51 calendar: _Union[_HolidayBase, str] = None,
52 currency: _Union[Currency, str] = "EUR",
53 # ex_settle: int =0,
54 payment_days: int = 0,
55 spot_days: int = 2,
56 start_period: int = None,
57 # _Optional[_Union[Period, str]] = None,
58 end_period: int = None,
59 ir_index: str = None,
60 issuer: _Optional[_Union[Issuer, str]] = None,
61 securitization_level: _Union[SecuritizationLevel, str] = SecuritizationLevel.NONE,
62 rating: _Union[Rating, str] = Rating.NONE,
63 ):
64 """Constructor for Forward Rate Agreement specification.
66 Args:
67 obj_id (str): (Preferably) Unique label of the FRA
68 trade_date (_Union[date, datetime]): FRA Trade date.
69 maturity_date (_Union[date, datetime]): FRA's maturity/expiry date. Must lie after the trade_date.
70 notional (float, optional): Fra's notional/face value. Must be positive.
71 rate (float): Agreed upon forward rate, a.k.a. FRA rate.
72 start_date (_Union[date, datetime]): start date of the interest rate (FRA_rate) reference period from which interest is accrued.
73 end_date (_Union[date, datetime]): end date of the interest rate (FRA_rate) reference period from which interest is accrued.
74 udlID (str): ID of the underlying Index rate used for the floating rate for fixing.
75 rate_start_date (_Union[date, datetime]): start date of fixing period for the floating rate
76 rate_end_date (_Union[date, datetime]): end date of fixing period for the floating rate
77 day_count_convention (Union[DayCounter, str], optional): Day count convention for determining period
78 length. Defaults to DayCounter.ThirtyU360.
79 business_day_convention (Union[RollConvention, str], optional): Set of rules defining the adjustment of
80 days to ensure each date being a business
81 day with respect to a given holiday
82 calendar. Defaults to
83 RollConvention.FOLLOWING
84 rate_day_count_convention (Union[DayCounter, str], optional): Day count convention for determining period
85 length. Defaults to DayCounter.ThirtyU360.
86 rate_business_day_convention (Union[RollConvention, str], optional): Set of rules defining the adjustment of
87 days to ensure each date being a business
88 day with respect to a given holiday
89 calendar. Defaults to
90 RollConvention.FOLLOWING
91 calendar (Union[HolidayBase, str], optional): Holiday calendar defining the bank holidays of a country or
92 province (but not all non-business days as for example
93 Saturdays and Sundays).
94 Defaults (through constructor) to holidays.ECB
95 (= Target2 calendar) between start_day and end_day.
96 currency (str, optional): Currency as alphabetic, Defaults to 'EUR'.
97 payment_days (int): Number of days for payment after the start date. Defaults to 0.
98 spot_days (int): time difference between fixing date and start dategiven in days.
99 start_period (int): forward start period given in months e.g. 1 from 1Mx4M
100 end_period (int): forward end period given in months e.g. 4 from 1Mx4M
101 ir_index (str): ID of the underlying Index rate used for the floating rate for fixing.
102 issuer (str, optional): Name/id of issuer. Defaults to None.
103 securitization_level (_Union[SecuritizationLevel, str], optional): Securitization level. Defaults to None.
104 rating (_Union[Rating, str]): Paper rating.
105 """
106 # positional arguments
107 self.obj_id = obj_id
108 self._trade_date = trade_date
109 self._notional = notional
110 self._rate = rate
111 self._start_date = start_date
112 self._end_date = end_date
113 if maturity_date is None:
114 self._maturity_date = self._end_date
115 else:
116 self._maturity_date = maturity_date
117 self._udlID = udlID
118 self._rate_start_date = rate_start_date
119 self._rate_end_date = rate_end_date
120 # optional arguments
121 self._day_count_convention = day_count_convention # TODO: correct syntax with setter?? HN
122 self._business_day_convention = RollConvention.to_string(business_day_convention)
123 self._rate_day_count_convention = rate_day_count_convention
124 self._rate_business_day_convention = RollConvention.to_string(rate_business_day_convention)
125 if calendar is None:
126 self._calendar = _ECB(years=range(trade_date.year, end_date.year + 1))
127 else:
128 self._calendar = _string_to_calendar(calendar)
129 self._currency = currency
130 # self.ex_settle = ex_settle
131 # self.trade_settle = trade_settle
132 self._fixing_date = calc_start_day(self._start_date, f"{spot_days}D", self._business_day_convention, self._calendar)
133 self._spot_days = spot_days
135 # if start_period is not None:
136 self.start_period = start_period
138 # if end_period is not None:
139 self.end_period = end_period
141 if ir_index is not None:
142 self._ir_index = ir_index
143 self._index = get_index_by_alias(ir_index)
144 self._indexdata = self._index.value
145 if issuer is not None:
146 self._issuer = issuer
147 if securitization_level is not None:
148 self.securitization_level = securitization_level
149 self._rating = Rating.to_string(rating)
150 self._payment_days = payment_days
152 # give dates where applicable as optional, if not given, calculate based on spot lag, index spot lag, and forward period YMxZM (e.g. 1Mx4M)
153 # e.g. for trade date D1 and spotLag, S1, and start_period = 1Mx4M
154 # start_date = D1 + S1 + 1Month # this is the date it starts accruing interest
155 # but how much interest? -> the pre-agreed FRA rate, fixed
156 # how is it settled? -> at settledate=start date, and using
157 # The floating rate index (e.g., LIBOR, SOFR, EURIBOR) used to determine the settlement amoun
158 # This is determined at the fixing_date ( usually spot lag before, e.g. 2 days)
160 # if trade date, spotlag, startperiod,endperiod give, then recalcualte start_datet etc...
161 if trade_date and spot_days and start_period and end_period:
162 spot_date = roll_day(
163 day=trade_date + timedelta(days=spot_days), # need holiday
164 calendar=self.calendar,
165 business_day_convention=self.rate_business_day_convention,
166 start_day=None,
167 )
169 self._start_date = roll_day(
170 day=spot_date + relativedelta(months=start_period), # need holiday
171 calendar=self.calendar,
172 business_day_convention=self.rate_business_day_convention,
173 start_day=None,
174 ) # spot_date + start_period #need roll convention: ddc, bdc, holiday, date
175 self._end_date = roll_day(
176 day=self.start_date + relativedelta(months=end_period - start_period), # need holiday
177 calendar=self.calendar,
178 business_day_convention=self.rate_business_day_convention,
179 start_day=None,
180 ) # start_date + end_period #need roll convention: ddc, bdc, holiday, date
182 # VALIDATE DATES
183 # TODO: self._validate_derived_issued_instrument()
185 @staticmethod
186 def _create_sample(
187 n_samples: int, seed: int = None, ref_date=None, issuers: _List[str] = None, sec_levels: _List[str] = None, currencies: _List[str] = None
188 ) -> _List["ForwardRateAgreementSpecification"]:
189 """Create a random sample of multiple instruments of this type with varied specification parameters.
191 Args:
192 n_samples (int): The number of desired sample objects
193 seed (int, optional): Seed number to allow repeated result. Defaults to None.
194 ref_date (_type_, optional): Reference date . Defaults to None.
195 issuers (_List[str], optional): list of issuers. Defaults to None.
196 sec_levels (_List[str], optional): list of possible securitization levels. Defaults to None.
197 currencies (_List[str], optional): list of possible currencies used. Defaults to None.
199 Returns:
200 _List[ForwardRateAgreementSpecification]: where each entry is a dict representing with the information needed to specify an instrument.
201 """
202 if seed is not None:
203 np.random.seed(seed)
204 if ref_date is None:
205 ref_date = datetime.now()
206 else:
207 ref_date = _date_to_datetime(ref_date)
208 if issuers is None:
209 issuers = ["Issuer_" + str(i) for i in range(int(n_samples / 2))]
210 result = []
211 if currencies is None:
212 currencies = list(Currency)
213 if sec_levels is None:
214 sec_levels = list(SecuritizationLevel)
215 for i in range(n_samples):
216 days = int(15.0 * 365.0 * np.random.beta(2.0, 2.0)) + 1
217 trade_date = ref_date + timedelta(days=np.random.randint(low=-365, high=0))
218 maturity_date = ref_date + timedelta(days=days)
219 start_date = ref_date + relativedelta(months=np.random.randint(low=1, high=3))
220 end_date = start_date + relativedelta(months=np.random.choice([3, 6]))
221 # spot_days=2, fixing pre_lag =2
222 result.append(
223 ForwardRateAgreementSpecification(
224 obj_id=f"Deposit_{i}",
225 trade_date=trade_date,
226 maturity_date=maturity_date,
227 notional=np.random.choice([100.0, 1000.0, 10_000.0, 100_0000.0]),
228 rate=np.random.choice([0.01, 0.02, 0.03, 0.04, 0.05]),
229 start_date=start_date,
230 end_date=end_date,
231 udlID="dummy_underlying_index",
232 rate_start_date=start_date - timedelta(days=2),
233 rate_end_date=end_date - timedelta(days=2),
234 # "day_count_convention": self.day_count_convention, #TODO
235 # "business_day_convention": self.business_day_convention,
236 # "rate_day_count_convention": self.rate_day_count_convention,
237 # "rate_business_day_convention": self.rate_business_day_convention,
238 calendar=_ECB(years=range(trade_date.year, maturity_date.year + 1)),
239 currency=np.random.choice(currencies),
240 # "spot_days": self.spot_days, # not needed if start dates given
241 # "start_period": self.start_period,
242 # "end_period": self.end_period,
243 issuer=np.random.choice(issuers),
244 securitization_level=np.random.choice(sec_levels),
245 )
246 )
247 return result
249 def _validate_derived_issued_instrument(self):
250 self.__trade_date, self.__maturity_date = _check_start_before_end(self.__trade_date, self.__maturity_date)
252 def _to_dict(self) -> dict:
254 result = {
255 "obj_id": self.obj_id,
256 "trade_date": serialize_date(self.trade_date),
257 "maturity_date": serialize_date(self.maturity_date),
258 "notional": self.notional,
259 "rate": self.rate,
260 "start_date": serialize_date(self.start_date),
261 "end_date": serialize_date(self.end_date),
262 "udlID": self.udlID,
263 "rate_start_date": serialize_date(self.rate_start_date),
264 "rate_end_date": serialize_date(self.rate_end_date),
265 "day_count_convention": self.day_count_convention,
266 "business_day_convention": self.business_day_convention,
267 "rate_day_count_convention": self.rate_day_count_convention,
268 "rate_business_day_convention": self.rate_business_day_convention,
269 "calendar": getattr(self.calendar, "name", self.calendar.__class__.__name__),
270 "currency": self.currency,
271 "payment_days": self.payment_days,
272 "spot_days": self.spot_days,
273 "start_period": self.start_period,
274 "end_period": self.end_period,
275 "issuer": self.issuer,
276 "securitization_level": self.securitization_level,
277 "rating": self.rating,
278 }
279 return result
281 def get_schedule(self) -> Schedule:
282 """Returns the schedule of the accrual periods of the instrument."""
283 return Schedule(
284 start_day=self._start_date,
285 end_day=self._end_date,
286 time_period=self._frequency,
287 backwards=True,
288 stub_type_is_Long=True,
289 business_day_convention=self.business_day_convention,
290 roll_convention=None,
291 calendar=self._calendar,
292 )
293 # region properties
295 @property
296 def calendar(self):
297 """Calender used for this instrument
299 Returns:
300 _type_: _description_
301 """
302 return self._calendar
304 @property
305 def issuer(self) -> str:
306 """
307 Getter for instrument's issuer.
309 Returns:
310 str: Instrument's issuer.
311 """
312 return self._issuer
314 @issuer.setter
315 def issuer(self, issuer: str):
316 """
317 Setter for instrument's issuer.
319 Args:
320 issuer(str): Issuer of the instrument.
321 """
322 self._issuer = issuer
324 @property
325 def udlID(self) -> str:
326 """
327 Getter for ID of the instruments underlying.
329 Returns:
330 str: Instrument's udlID.
331 """
332 return self._udlID
334 @udlID.setter
335 def udlID(self, udlID: str):
336 """
337 Setter for ID of the instruments underlying.
339 Args:
340 udlID(str): udlID of the instrument.
341 """
342 self._udlID = udlID
344 @property
345 def rating(self) -> str:
346 """Getter for instrument's rating.
348 Returns:
349 str: instrument's rating
350 """
351 return self._rating
353 @rating.setter
354 def rating(self, rating: _Union[Rating, str]) -> str:
355 self._rating = Rating.to_string(rating)
357 @property
358 def securitization_level(self) -> str:
359 """
360 Getter for instrument's securitisation level.
362 Returns:
363 str: Instrument's securitisation level.
364 """
365 return self._securitization_level
367 @securitization_level.setter
368 def securitization_level(self, securitisation_level: _Union[SecuritizationLevel, str]):
369 self._securitization_level = SecuritizationLevel.to_string(securitisation_level)
371 @property
372 def rate(self) -> float:
373 """
374 Getter for instrument's rate.
376 Returns:
377 float: Instrument's rate.
378 """
379 return self._rate
381 @rate.setter
382 def rate(self, rate: float):
383 """
384 Setter for instrument's rate.
386 Args:
387 (float): interest rate of the instrument.
388 """
389 self._rate = rate
391 @property
392 def trade_date(self) -> date:
393 """
394 Getter for FRA's issue date.
396 Returns:
397 date: FRA's issue date.
398 """
399 return self._trade_date
401 @trade_date.setter
402 def trade_date(self, trade_date: _Union[datetime, date]):
403 """
404 Setter for FRA's issue date.
406 Args:
407 issue (Union[datetime, date]): FRA's issue date.
408 """
409 self._trade_date = _date_to_datetime(trade_date)
411 @property
412 def start_date(self) -> date:
413 """
414 Getter for FRA's start date.
416 Returns:
417 date: FRA's start date.
418 """
419 return self._start_date
421 @start_date.setter
422 def start_date(self, start_date: _Union[datetime, date]):
423 """
424 Setter for FRA's start date.
426 Args:
427 start_date (Union[datetime, date]): FRA's start date.
428 """
429 self._start_date = _date_to_datetime(start_date)
431 @property
432 def end_date(self) -> date:
433 """
434 Getter for FRA's end date.
436 Returns:
437 date: FRA's end date.
438 """
439 return self._end_date
441 @end_date.setter
442 def end_date(self, end_date: _Union[datetime, date]):
443 """
444 Setter for FRA's end date.
446 Args:
447 end_date (Union[datetime, date]): FRA's end date.
448 """
449 self._end_date = _date_to_datetime(end_date)
451 @property
452 def rate_start_date(self) -> date:
453 """
454 Getter for FRA's underlying rate start date.
456 Returns:
457 date: FRA's underlaying rate start date.
458 """
459 return self._rate_start_date
461 @start_date.setter
462 def rate_start_date(self, start_date: _Union[datetime, date]):
463 """
464 Setter for FRA's underlying rate start date.
466 Args:
467 start_date (Union[datetime, date]): FRA's underlaying rate start date.
468 """
469 self._rate_start_date = _date_to_datetime(start_date)
471 @property
472 def rate_end_date(self) -> date:
473 """
474 Getter for FRA's underlying rate end date.
476 Returns:
477 date: FRA's underlying rate end date.
478 """
479 return self._rate_end_date
481 @end_date.setter
482 def rate_end_date(self, end_date: _Union[datetime, date]):
483 """
484 Setter for FRA's underlying rate end date.
486 Args:
487 end_date (Union[datetime, date]): FRA's underlying rate end date.
488 """
489 self._rate_end_date = _date_to_datetime(end_date)
491 @property
492 def maturity_date(self) -> date:
493 """
494 Getter for FRA's maturity date.
496 Returns:
497 date: FRA's maturity date.
498 """
499 return self._maturity_date
501 @maturity_date.setter
502 def maturity_date(self, maturity_date: _Union[datetime, date]):
503 """
504 Setter for FRA's maturity date.
506 Args:
507 maturity_date (Union[datetime, date]): FRA's maturity date.
508 """
509 self._maturity_date = _date_to_datetime(maturity_date)
511 @property
512 def currency(self) -> str:
513 """
514 Getter for FRA's currency.
516 Returns:
517 str: FRA's currency code
518 """
519 return self._currency
521 @currency.setter
522 def currency(self, currency: str):
523 self._currency = Currency.to_string(currency)
525 @property
526 def notional(self) -> float:
527 """
528 Getter for FRA's face value.
530 Returns:
531 float: FRA's face value.
532 """
533 return self._notional
535 @notional.setter
536 def notional(self, notional):
537 self._notional = _check_positivity(notional)
539 @property
540 def day_count_convention(self) -> str:
541 """
542 Getter for FRA's day count convention.
544 Returns:
545 str: FRA's day count convention.
546 """
547 return self._day_count_convention
549 @day_count_convention.setter
550 def day_count_convention(self, day_count_convention: _Union[DayCounterType, str]) -> str:
551 self._day_count_convention = DayCounterType.to_string(day_count_convention)
553 @property
554 def rate_day_count_convention(self) -> str:
555 """
556 Getter for FRA's underlying rate's day count convention.
558 Returns:
559 str: FRA's underlying rate's day count convention.
560 """
561 return self._rate_day_count_convention
563 @rate_day_count_convention.setter
564 def rate_day_count_convention(self, rate_day_count_convention: _Union[DayCounterType, str]) -> str:
565 self._rate_day_count_convention = DayCounterType.to_string(rate_day_count_convention)
567 @property
568 def business_day_convention(self) -> str:
569 """
570 Getter for FRA's underlying rate's business_day_convention.
572 Returns:
573 str: FRA's underlying rate's business_day_convention.
574 """
575 return self._business_day_convention
577 @business_day_convention.setter
578 def business_day_convention(self, business_day_convention: _Union[DayCounterType, str]) -> str:
579 self._business_day_convention = DayCounterType.to_string(business_day_convention)
581 @property
582 def rate_business_day_convention(self) -> str:
583 """
584 Getter for FRA's underlying rate's business_day_convention.
586 Returns:
587 str: FRA's underlying rate's business_day_convention.
588 """
589 return self._rate_business_day_convention
591 @rate_business_day_convention.setter
592 def rate_business_day_convention(self, business_day_convention: _Union[DayCounterType, str]) -> str:
593 """Setter for FRA's underlying rate's business_day_convention."""
594 self._rate_business_day_convention = DayCounterType.to_string(business_day_convention)
596 @property
597 def spot_days(self) -> int:
598 """Getter for the spot lag given in days
600 Returns:
601 float: _description_
602 """
603 return self._spot_days
605 @property
606 def start_period(self) -> int:
607 """Getter for the start period, given in Months
609 Returns:
610 float: _description_
611 """
612 return self._start_period
614 @start_period.setter
615 def start_period(self, start_period) -> int:
616 """setter for the start period, given in Months
618 Returns:
619 float: _description_
620 """
621 self._start_period = start_period
623 @property
624 def end_period(self) -> int:
625 """Getter for the spot lag
627 Returns:
628 float: _description_
629 """
630 return self._end_period
632 @end_period.setter
633 def end_period(self, end_period) -> int:
634 """setter for the end period, given in Months
636 Returns:
637 float: _description_
638 """
639 self._end_period = end_period
641 @property
642 def index(self) -> str:
643 """Getter for the underlying Index rate used for the floating rate for fixing.
645 Returns:
646 str: _description_
647 """
648 return self._index
650 @index.setter
651 def index(self, index: str):
652 self._index = index
654 def ins_type(self):
655 """Return instrument type
657 Returns:
658 Instrument: Forward rate agreement
659 """
660 return Instrument.FRA
662 @property
663 def payment_days(self) -> int:
664 """Getter for the number of settlement days.
666 Returns:
667 int: Number of settlement days.
668 """
669 return self._payment_days
671 @payment_days.setter
672 def payment_days(self, payment_days: int):
673 self._payment_days = payment_days
675 # temp placeholder
676 def get_end_date(self):
677 return self.maturity_date
679 # endregion