Coverage for rivapy / tools / datetools.py: 91%
517 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1# -*- coding: utf-8 -*-
2import re
3from typing import Callable, Dict, Any, Optional as _Optional
4from datetime import datetime, date
5from dateutil.relativedelta import relativedelta
6from dateutil.rrule import WE
7from calendar import monthrange, isleap
8from typing import List as _List, Union as _Union, Callable
9from rivapy.tools.holidays_compat import HolidayBase as _HolidayBase
11# from holidays import DE
12from holidays.financial.european_central_bank import ECB as _ECB
13from rivapy.tools.enums import RollConvention, DayCounterType, RollRule
14from rivapy.tools._validators import _string_to_calendar
17# TODO: Switch to locally configured logger.
18from rivapy.tools._logger import logger
21class DayCounter:
23 def __init__(self, daycounter: _Union[str, DayCounterType]):
24 self._dc = DayCounterType.to_string(daycounter)
25 self._yf = DayCounter.get(self._dc)
27 def yf(
28 self,
29 d1: _Union[date, datetime],
30 d2: _Union[_Union[date, datetime], _List[_Union[date, datetime]]],
31 coupon_schedule: _List[_Union[date, datetime]] = None, # Added optional argument
32 coupon_frequency: float = None, # Added optional argument
33 ) -> _Union[float, _List[float]]:
35 if self._dc == DayCounterType.ActActICMA.value:
36 if coupon_schedule is None or coupon_frequency is None:
37 raise ValueError("For ActActICMA, 'coupon_schedule' and 'coupon_frequency' must be provided.")
38 if isinstance(d2, list):
39 return [self._yf(d1, d2_, coupon_schedule, coupon_frequency) for d2_ in d2]
40 else:
41 return self._yf(d1, d2, coupon_schedule, coupon_frequency)
42 else:
43 if isinstance(d2, list):
44 return [self._yf(d1, d2_) for d2_ in d2]
45 else:
46 return self._yf(d1, d2)
48 @staticmethod
49 def get(daycounter: _Union[str, DayCounterType]) -> Callable[[_Union[date, datetime], _Union[date, datetime]], float]:
50 dc = DayCounterType.to_string(daycounter)
52 mapping = {
53 DayCounterType.Act365Fixed.value: DayCounter.yf_Act365Fixed,
54 DayCounterType.ACT_ACT.value: DayCounter.yf_ActAct,
55 DayCounterType.ACT360.value: DayCounter.yf_Act360,
56 DayCounterType.ThirtyU360.value: DayCounter.yf_30U360,
57 DayCounterType.ThirtyE360.value: DayCounter.yf_30E360,
58 DayCounterType.Thirty360ISDA.value: DayCounter.yf_30360ISDA,
59 DayCounterType.ActActICMA.value: DayCounter.yf_ActActICMA,
60 }
62 if dc in mapping:
63 return mapping[dc]
64 else:
65 raise NotImplementedError(f"{dc} not yet implemented.")
67 @staticmethod
68 def yf_ActActICMA(
69 d1: _Union[date, datetime], d2: _Union[date, datetime], coupon_schedule: _List[_Union[date, datetime]], coupon_frequency: _Union[int, float]
70 ) -> float:
71 """This method implements the Act/Act ICMA day count convention which is used for Bonds.
73 Args:
74 d1 (_Union[date, datetime]): start date of the period for which the year fraction is calculated.
75 d2 (_Union[date, datetime]): end date of the period for which the year fraction is calculated.
76 coupon_schedule (_List[_Union[date, datetime]]): Sorted list of all coupon payment days.
77 coupon_frequency (int): Number of coupon payments per year (e.g., 1 for annual, 2 for semi-annual)
79 Returns:
80 float: year fraction
81 """
82 if coupon_frequency == 0:
83 raise ValueError("Coupon frequency must be greater than 0.")
84 d1_dt = _date_to_datetime(d1)
85 d2_dt = _date_to_datetime(d2)
86 coupon_schedule_dt = [_date_to_datetime(cs_date) for cs_date in coupon_schedule]
88 yf = 0.0
89 for i in range(len(coupon_schedule_dt) - 1):
90 cp_start_dt = coupon_schedule_dt[i]
91 cp_end_dt = coupon_schedule_dt[i + 1]
93 # consider overlapping periods only
94 if d1_dt <= cp_end_dt and d2_dt >= cp_start_dt:
95 fraction_period_start_dt = max(d1_dt, cp_start_dt)
96 fraction_period_end_dt = min(d2_dt, cp_end_dt)
98 days_cp = (cp_end_dt - cp_start_dt).days
99 days_fraction = (fraction_period_end_dt - fraction_period_start_dt).days
101 yf += days_fraction / (days_cp * coupon_frequency)
103 return yf
105 @staticmethod
106 def yf_Act365Fixed(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
107 """This method implements the Act/365f day count convention.
108 The actual number of days between d2 and d1 is divided by 365.
110 Args:
111 d1 (_Union[date, datetime]): start date
112 d2 (_Union[date, datetime]): end date
114 Returns:
115 float: year fraction
116 """
117 d1_dt = _date_to_datetime(d1)
118 d2_dt = _date_to_datetime(d2)
119 return (d2_dt - d1_dt).total_seconds() / (365.0 * 24 * 60 * 60)
121 @staticmethod
122 def yf_ActAct(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
123 """This method implements the Act/Act ISDA day count convention.
124 The acutal number of days between d2 and d1 is divded by the acutal number of days in the respective year.
125 In cases where d2 and d1 are located in different years, the period is split into sub periods and the year fraction is calculated on each sub period with its respective
126 number of days in that year. This is especially important if d1 is located in a regular year and d2 is located in a leap year.
128 Args:
129 d1 (_Union[date, datetime]): start date
130 d2 (_Union[date, datetime]): end date
132 Returns:
133 float: year fraction
134 """
135 d1_dt = _date_to_datetime(d1)
136 d2_dt = _date_to_datetime(d2)
138 if d1_dt > d2_dt:
139 raise ValueError("d1 must be before d2")
141 # Calculate the fraction for each year the period spans
142 current_date_dt = d1_dt
143 year_fraction = 0.0
145 while current_date_dt < d2_dt:
146 # Ensure year_end_dt and start_of_year_dt are datetime, preserving tzinfo if present
147 year_end_dt = datetime(current_date_dt.year, 12, 31, tzinfo=current_date_dt.tzinfo)
148 start_of_year_dt = datetime(current_date_dt.year, 1, 1, tzinfo=current_date_dt.tzinfo)
149 days_in_year = (year_end_dt - start_of_year_dt).days + 1 # Actual days in the year
151 # If the period ends within the same year
152 if d2_dt.year == current_date_dt.year:
153 year_fraction += (d2_dt - current_date_dt).days / days_in_year
154 break
156 # Add the fraction for the remaining days in the current year
157 year_fraction += ((year_end_dt - current_date_dt).days + 1) / days_in_year
158 # Move to the start of the next year
159 current_date_dt = datetime(current_date_dt.year + 1, 1, 1, tzinfo=current_date_dt.tzinfo)
161 return year_fraction
163 @staticmethod
164 def yf_Act360(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
165 """This method implements the Act/360 day count convention.
166 Here the actual number of days between d2 and d1 is computed and divided by 360, since this day count convention assumes that each year contains 360 days.
168 Args:
169 d1 (_Union[date, datetime]): start date
170 d2 (_Union[date, datetime]): end date
172 Returns:
173 float: _description_
174 """
175 d1_dt = _date_to_datetime(d1)
176 d2_dt = _date_to_datetime(d2)
177 return ((d2_dt - d1_dt).days) / 360.0 # Ensure float division
179 # @staticmethod
180 # def yf_Bus252(d1: _Union[date, datetime], d2: _Union[date, datetime])->float:
181 # """This method implements the Bus/252 day count convention.
183 # Args:
184 # d1 (_Union[date, datetime]): start date
185 # d2 (_Union[date, datetime]): end date
187 # Returns:
188 # float: _description_
189 # """
190 # return ((d2 - d1).days)/252
192 @staticmethod
193 def yf_30U360(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
194 """This method implements the 30U360 convention.
195 The following logic is applied:
197 1. If d2.day == 31 and d1.day >= 30 -> d2.day = 30
198 2. If d1.day == 31 -> d1.day = 30
199 3. If (d1.day == EndOfMonth(Feb) and d1.month==2) and (d2.day == EndOfMonth(Feb) and d2.month==2) -> d2.day = 30
200 4. If (d1.day == EndOfMonth(Feb) and d1.month==2) -> d1.day = 30
202 Args:
203 d1 (_Union[date, datetime]): start date
204 d2 (_Union[date, datetime]): end date
206 Returns:
207 float: year fraction
208 """
209 d1_dt = _date_to_datetime(d1)
210 d2_dt = _date_to_datetime(d2)
212 m_range1 = monthrange(d1_dt.year, d1_dt.month)
213 m_range2 = monthrange(d2_dt.year, d2_dt.month)
215 day1 = d1_dt.day
216 day2 = d2_dt.day
218 if (d2_dt.day == 31) and (d1_dt.day >= 30):
219 day2 = 30
221 if d1_dt.day == 31:
222 day1 = 30
224 if (d1_dt.day == m_range1[-1] and d1_dt.month == 2) and (d2_dt.day == m_range2[-1] and d2_dt.month == 2): # Corrected d1.month to d2_dt.month
225 day2 = 30
227 if d1_dt.day == m_range1[-1] and d1_dt.month == 2:
228 day1 = 30
229 return (d2_dt.year - d1_dt.year) + (d2_dt.month - d1_dt.month) / 12.0 + (day2 - day1) / 360.0
231 @staticmethod
232 def yf_30360ISDA(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
233 """This method implements the 30/360 ISDA (Bond Basis) day count convention.
234 The following logic is applied:
236 1. If d2.day == 31 and d1.day >= 30 -> d2.day = 30
237 2. If d1.day == 31 -> d1.day = 30
239 Args:
240 d1 (_Union[date, datetime]): start date
241 d2 (_Union[date, datetime]): end date
243 Returns:
244 float: year fraction
245 """
246 d1_dt = _date_to_datetime(d1)
247 d2_dt = _date_to_datetime(d2)
249 day1 = d1_dt.day
250 day2 = d2_dt.day
252 if (d2_dt.day == 31) and (d1_dt.day >= 30): # Original logic used d1.day here
253 day2 = 30
255 if d1_dt.day == 31:
256 day1 = 30
257 return (d2_dt.year - d1_dt.year) + (d2_dt.month - d1_dt.month) / 12.0 + (day2 - day1) / 360.0
259 @staticmethod
260 def yf_30E360(d1: _Union[date, datetime], d2: _Union[date, datetime]) -> float:
261 """This day count convention implements the Eurobond Basis day count convention.
262 The following logic is applied:
264 1. If d1.day >= 30 -> d1.day = 30
265 2. If d2.day >= 30 -> d2.day = 30
267 Args:
268 d1 (_Union[date, datetime]): start date
269 d2 (_Union[date, datetime]): end date
271 Returns:
272 float: year fraction
273 """
274 d1_dt = _date_to_datetime(d1)
275 d2_dt = _date_to_datetime(d2)
277 def _adjust_day(day: int):
278 if day >= 30:
279 return 30
280 return day
282 day1 = _adjust_day(d1_dt.day)
283 day2 = _adjust_day(d2_dt.day)
285 return (d2_dt.year - d1_dt.year) + (d2_dt.month - d1_dt.month) / 12.0 + (day2 - day1) / 360.0
288class Period:
289 def __init__(self, years: int = 0, months: int = 0, days: int = 0):
290 """
291 Time Period expressed in years, months and days.
293 Args:
294 years (int, optional): Number of years in time period. Defaults to 0.
295 months (int, optional): Number of months in time period. Defaults to 0.
296 days (int, optional): Number of days in time period. Defaults to 0.
297 """
298 self.years = years
299 self.months = months
300 self.days = days
302 @staticmethod
303 def from_string(period: str):
304 """Creates a Period from a string
306 Args:
307 period (str): The string defining the period. The string must be defined by the number of days/months/years followed by one of the letters 'Y'/'M'/'D', i.e. '6M' means 6 months, or 'O/N', or 'T/N'.
309 Returns:
310 Period: The resulting period
312 Examples:
313 .. code-block:: python
315 >>> p = Period('6M') # period of 6 months
316 >>> p = Period('1Y') #period of 1 year
317 """
318 if period == "T/N" or period == "O/N":
319 return Period(days=1)
320 else:
321 period_length = int(period[:-1])
322 period_type = period[1]
323 if period_type == "Y":
324 return Period(years=period_length)
325 elif period_type == "M":
326 return Period(months=period_length)
327 elif period_type == "D":
328 return Period(days=period_length)
329 raise Exception(
330 period + " is not a valid period string. See documentation of tools.datetools.Period for deocumentation of valid strings."
331 )
333 @property
334 def years(self) -> int:
335 """
336 Getter for years of period.
338 Returns:
339 int: Number of years for specified time period.
340 """
341 return self.__years
343 @years.setter
344 def years(self, years: int):
345 """
346 Setter for years of period.
348 Args:
349 years(int): Number of years.
350 """
351 self.__years = years
353 @property
354 def months(self) -> int:
355 """
356 Getter for months of period.
358 Returns:
359 int: Number of months for specified time period.
360 """
361 return self.__months
363 @months.setter
364 def months(self, months: int):
365 """
366 Setter for months of period.
368 Args:
369 months(int): Number of months.
370 """
371 self.__months = months
373 @property
374 def days(self) -> int:
375 """
376 Getter for number of days in time period.
378 Returns:
379 int: Number of days for specified time period.
380 """
381 return self.__days
383 @days.setter
384 def days(self, days: int):
385 """
386 Setter for days of period.
388 Args:
389 days(int): Number of days.
390 """
391 self.__days = days
393 def __eq__(self, other: "Period"):
394 return self.years == other.years and self.months == other.months and self.days == other.days
397class Schedule:
398 def __init__(
399 self,
400 start_day: _Union[date, datetime],
401 end_day: _Union[date, datetime],
402 time_period: _Union[Period, str],
403 backwards: bool = True,
404 # stub_mode: str = "automatic", # could alternatively be "force" or "none" (i.e. force a stub period even if not necessary, or do not allow stub periods at all)
405 stub_type_is_Long: bool = True,
406 # stub_placement: str = "ending", # could alternatively be "beginning" (i.e. place stub period at the end, at the beginning)
407 business_day_convention: _Union[RollConvention, str] = RollConvention.MODIFIED_FOLLOWING,
408 calendar: _Optional[_Union[_HolidayBase, str]] = None,
409 roll_convention: _Union[RollRule, str] = RollRule.NONE,
410 settle_days: int = 0,
411 ref_date: _Optional[_Union[date, datetime]] = None,
412 ):
413 """
414 A schedule is a list of dates, e.g. of coupon payments, fixings, etc., which is defined by its first (= start
415 day) and last (= end day) day, by its distance between two consecutive dates (= time period) and by the
416 procedure for rolling out the schedule, more precisely by the direction (backwards/forwards) and the dealing
417 with incomplete periods (stubs). Moreover, the schedule ensures to comply to business day conventions with
418 respect to a specified holiday calendar.
420 Args:
421 start_day (_Union[date, datetime]): Schedule's first day - beginning of the schedule.
422 end_day (_Union[date, datetime]): Schedule's last day - end of the schedule.
423 time_period (_Union[Period, str]): Time distance between two consecutive dates.
424 backwards (bool, optional): Defines direction for rolling out the schedule. True means the schedule will be
425 rolled out (backwards) from end day to start day. Defaults to True.
426 stub_type_is_Long (bool, optional): Defines if a stub period is accepted (False) to be shorter than
427 the others, or if its remaining days are added to the neighbouring period (True).
428 Defaults to True.
429 business_day_convention (_Union[RollConvention, str], optional): Set of rules defining the adjustment of
430 days to ensure each date being a business
431 day with respect to a given holiday
432 calendar. Defaults to
433 RollConvention.MODIFIED_FOLLOWING
434 calendar (_Union[_HolidayBase, str], optional): Holiday calendar defining the bank holidays of a country or
435 province (but not all non-business days as for example
436 Saturdays and Sundays).
437 Defaults (through constructor) to holidays.ECB
438 (= Target2 calendar) between start_day and end_day.
439 roll_convention (_Union[RollRule, str], optional): Defines the roll convention for the schedule.
440 settle_days (int, optional): Number of days for settlement. Defaults to 0.
441 ref_date (_Optional[_Union[date, datetime]]): Reference date for the schedule. If provided, the schedule will be shortened and include the dates that are after the reference date plus the immediate date before it.
443 Examples:
445 .. code-block:: python
447 >>> from datetime import date
448 >>> from rivapy.tools import Schedule
449 >>> schedule = Schedule(date(2020, 8, 21), date(2021, 8, 21), Period(0, 3, 0), True, False, RollConvention.UNADJUSTED, holidays_de).generate_dates(False),
450 [date(2020, 8, 21), date(2020, 11, 21), date(2021, 2, 21), date(2021, 5, 21), date(2021, 8, 21)])
451 """
452 self.start_day = start_day
453 self.end_day = end_day
454 self.time_period = time_period
455 self.backwards = backwards
456 self.stub_type_is_Long = stub_type_is_Long
457 self.business_day_convention = business_day_convention
458 self.calendar = calendar
459 self.roll_convention = roll_convention
460 self.settle_days = settle_days
461 self.ref_date = ref_date
463 @property
464 def start_day(self):
465 """
466 Getter for schedule's start date.
468 Returns:
469 Start date of specified schedule.
470 """
471 return self.__start_day
473 @start_day.setter
474 def start_day(self, start_day: _Union[date, datetime]):
475 self.__start_day = _date_to_datetime(start_day)
477 @property
478 def end_day(self):
479 """
480 Getter for schedule's end date.
482 Returns:
483 End date of specified schedule.
484 """
485 return self.__end_day
487 @end_day.setter
488 def end_day(self, end_day: _Union[date, datetime]):
489 self.__end_day = _date_to_datetime(end_day)
491 @property
492 def time_period(self):
493 """
494 Getter for schedule's time period.
496 Returns:
497 Time period of specified schedule.
498 """
499 return self.__time_period
501 @time_period.setter
502 def time_period(self, time_period: _Union[Period, str]):
503 self.__time_period = _term_to_period(time_period)
505 @property
506 def backwards(self):
507 """
508 Getter for schedule's roll out direction.
510 Returns:
511 True, if rolled out from end day to start day.
512 False, if rolled out from start day to end day.
513 """
514 return self.__backwards
516 @backwards.setter
517 def backwards(self, backwards: bool):
518 self.__backwards = backwards
520 @property
521 def stub_type_is_Long(self):
522 """
523 Getter for potential existence of shlong periods (stub_type_is_long).
525 Returns:
526 True, if a shorter period is allowed.
527 False, if only a longer period is allowed.
528 """
529 return self.stub_type_is_Long
531 @stub_type_is_Long.setter
532 def stub_type_is_Long(self, stub_type_is_Long: bool):
533 self.__stub_type_is_Long = stub_type_is_Long
535 @property
536 def business_day_convention(self):
537 """
538 Getter for schedule's business day convention.
540 Returns:
541 Business day convention of specified schedule.
542 """
543 return self.__business_day_convention
545 @business_day_convention.setter
546 def business_day_convention(self, business_day_convention: _Union[RollConvention, str]):
547 self.__business_day_convention = RollConvention.to_string(business_day_convention)
549 @property
550 def calendar(self):
551 """
552 Getter for schedule's holiday calendar.
554 Returns:
555 Holiday calendar of specified schedule.
556 """
557 return self.__calendar
559 @calendar.setter
560 def calendar(self, calendar: _Union[_HolidayBase, str]):
561 if calendar is None:
562 self.__calendar: _HolidayBase = _ECB(years=range(self.__start_day.year, self.__end_day.year + 1))
563 else:
564 self.__calendar: _HolidayBase = _string_to_calendar(calendar)
566 @property
567 def roll_convention(self):
568 """
569 Getter for schedule's roll convention.
571 Returns:
572 Roll convention of specified schedule.
573 """
574 return self.__roll_convention
576 @roll_convention.setter
577 def roll_convention(self, roll_convention: _Union[RollRule, str]):
578 """
579 Setter for schedule's roll convention.
581 Args:
582 roll_convention (Union[RollRule, str]): Roll convention of specified schedule.
583 """
584 if isinstance(roll_convention, str):
585 roll_convention = RollRule[roll_convention.upper()]
586 self.__roll_convention = roll_convention.value
588 @staticmethod
589 def _generate_eom_dates(from_, to_, term, direction, backwards) -> _List[date]:
590 dates = []
591 if _is_ambiguous_date(from_):
592 from_ = datetime(from_.year, from_.month, monthrange(from_.year, from_.month)[-1])
593 if _is_ambiguous_date(to_):
594 to_ = datetime(to_.year, to_.month, monthrange(to_.year, to_.month)[-1])
595 while ((not backwards) & (from_ <= to_)) | (backwards & (to_ <= from_)):
596 dates.append(from_)
597 from_ += direction * relativedelta(years=term.years, months=term.months, day=31)
598 return dates
600 @staticmethod
601 def _generate_none_dates(from_, to_, term, direction, backwards) -> _List[date]:
602 dates = []
603 while ((not backwards) & (from_ <= to_)) | (backwards & (to_ <= from_)):
604 dates.append(from_)
605 from_ += direction * relativedelta(years=term.years, months=term.months, days=term.days)
606 return dates
608 @staticmethod
609 def _generate_dom_dates(from_, to_, term, direction, backwards) -> _List[date]:
610 dates = []
611 date = from_
612 i = 0
613 days = from_.day
614 while ((not backwards) & (date <= to_)) | (backwards & (to_ <= date)):
615 dates.append(date)
616 i += 1
617 date = from_ + direction * relativedelta(years=term.years, months=term.months * i, day=days)
618 return dates
620 @staticmethod
621 def _generate_imm_dates(from_, to_, term, direction, backwards) -> _List[date]:
622 dates = []
623 from_new = from_
624 # shift to next IMM if necessary, i.e. from_ may not be part of the generated dates
625 if not _is_IMM_date(from_):
626 from_new = _date_to_datetime(next_IMM_date(from_))
627 if backwards:
628 from_new = _date_to_datetime(next_IMM_date(from_ - relativedelta(months=3)))
629 # adjust term if not a multiple of 3 months
630 term_new = term
631 if term.months % 3 != 0:
632 term_new = next_IMM_period(term)
633 while ((not backwards) & (from_new <= to_)) | (backwards & (to_ <= from_new)):
634 dates.append(from_new)
635 from_new += direction * relativedelta(years=term_new.years, months=term_new.months, day=1, weekday=WE(3))
636 return dates
638 @staticmethod
639 def _generate_dates_by_roll_convention(roll_convention_, from_, to_, term, direction, backwards) -> _List[date]:
640 # Ensure roll_convention_ is an enum instance
641 if isinstance(roll_convention_, str):
642 roll_convention_ = RollRule[roll_convention_.upper()]
643 elif not isinstance(roll_convention_, RollRule):
644 raise Exception(f"Invalid roll convention type: {type(roll_convention_)}")
646 RollConventionMap = {
647 RollRule.EOM: Schedule._generate_eom_dates,
648 RollRule.NONE: Schedule._generate_none_dates,
649 RollRule.DOM: Schedule._generate_dom_dates,
650 RollRule.IMM: Schedule._generate_imm_dates,
651 }
652 if roll_convention_ not in RollConventionMap:
653 raise Exception(f"Unknown roll convention '{roll_convention_}'! Must be one of {list(RollConventionMap.keys())}")
654 return RollConventionMap[roll_convention_](from_, to_, term, direction, backwards)
656 # ToDo: clarify what is done here --> automatic stub, allow_stub control if long or short, always at the end when rolling forward, at the beginning when rolling backwards
657 # ToDo: add tests, check out deposits and FRAs
658 @staticmethod
659 def _roll_out(
660 from_: _Union[date, datetime],
661 to_: _Union[date, datetime],
662 term: _Union[Period, str],
663 backwards: bool = False,
664 long_stub: bool = True,
665 roll_convention_: _Union[RollRule, str] = "NONE",
666 ref_date: _Optional[_Union[date, datetime]] = None,
667 ) -> _List[date]:
668 """
669 Rolls out dates from from_ to to_ in the specified direction applying the given term under consideration of the
670 specification for allowing shorter periods.
672 Args:
673 from_ (_Union[date, datetime]): Beginning of the roll out mechanism.
674 to_ (_Union[date, datetime]): End of the roll out mechanism.
675 term (Period): Difference between rolled out dates.
676 backwards (bool): Direction of roll out mechanism: backwards if True, forwards if False.
677 long_stub (bool): Defines if periods longer than term are allowed.
679 Returns:
680 Date schedule not adjusted to business days.
681 """
682 if isinstance(term, str):
683 term = _term_to_period(term)
684 if isinstance(roll_convention_, str):
685 roll_convention_ = RollRule[roll_convention_.upper()]
686 # convert datetime to date (if necessary):
687 from_ = _date_to_datetime(from_)
688 to_ = _date_to_datetime(to_)
689 # check input consistency:
690 if (not backwards) & (from_ < to_):
691 direction = +1
692 elif backwards & (from_ > to_):
693 direction = -1
694 else:
695 raise Exception(
696 "From-date '"
697 + str(from_)
698 + "' and to-date '"
699 + str(to_)
700 + "' are not consistent with roll direction (backwards = '"
701 + str(backwards)
702 + "')!"
703 )
704 # generates a list of dates ...
705 dates = Schedule._generate_dates_by_roll_convention(roll_convention_, from_, to_, term, direction, backwards)
706 # return empty list if no dates were generated
707 if dates == []:
708 logger.info("No dates were generated!")
709 return dates
710 if roll_convention_ == RollRule.EOM and _is_ambiguous_date(from_):
711 from_ = datetime(from_.year, from_.month, monthrange(from_.year, from_.month)[-1])
712 if roll_convention_ == RollRule.EOM and _is_ambiguous_date(to_):
713 to_ = datetime(to_.year, to_.month, monthrange(to_.year, to_.month)[-1])
714 if _date_to_datetime(dates[-1]) != to_:
715 # ... by adding a short stub or ...
716 if not long_stub or len(dates) == 1: # 2025 HN
717 dates.append(to_)
718 # ... by extending last period.
719 else:
720 dates[-1] = to_
722 if ref_date is not None:
723 dates = [
724 d for d in dates if d >= calc_start_day(ref_date, term, roll_convention=roll_convention_)
725 ] # Keep only dates after the reference date plus the last date before the reference date.
726 if backwards:
727 dates.reverse()
728 return dates
730 def generate_dates(self, ends_only: bool) -> _List[date]:
731 """
732 Generate list of schedule days according to the schedule specification, in particular with regards to business
733 day convention, roll_convention and calendar given.
735 Args:
736 ends_only (bool): Flag to indicate if period beginnings shall be included, e.g. for defining accrual
737 periods: True, if only period ends shall be included, e.g. for defining payment dates.
739 Returns:
740 List[date]: List of schedule dates (including start and end date) adjusted to rolling convention.
741 """
742 # roll out dates ignoring any business day issues
743 if self.__backwards:
744 schedule_dates = Schedule._roll_out(
745 self.__end_day, self.__start_day, self.__time_period, True, self.__stub_type_is_Long, self.__roll_convention
746 )
747 # schedule_dates.reverse()
748 else:
749 schedule_dates = Schedule._roll_out(
750 self.__start_day, self.__end_day, self.__time_period, False, self.__stub_type_is_Long, self.__roll_convention
751 )
753 # adjust according to business day convention
754 rolled_schedule_dates = [roll_day(schedule_dates[0], self.__calendar, self.__business_day_convention, schedule_dates[0], self.settle_days)]
755 for i in range(1, len(schedule_dates)):
756 rolled_schedule_dates.append(
757 roll_day(schedule_dates[i], self.__calendar, self.__business_day_convention, rolled_schedule_dates[i - 1], self.settle_days)
758 )
759 if ends_only:
760 rolled_schedule_dates.pop(0)
762 logger.debug(
763 "Schedule dates successfully calculated from '"
764 + str(self.__start_day)
765 + "' to '"
766 + str(self.__end_day)
767 + "' adjusted by business day convention and settlement days."
768 )
769 return rolled_schedule_dates
772def _date_to_datetime(date_time: _Union[datetime, date]) -> datetime:
773 """
774 Converts a date to a datetime or leaves it unchanged if it is already of type datetime.
776 Args:
777 date_time (_Union[datetime, date]): Date(time) to be converted.
779 Returns:
780 datetime: (Potentially) Converted datetime.
781 """
782 if isinstance(date_time, datetime):
783 return date_time
784 elif isinstance(date_time, date):
785 return datetime.combine(date_time, datetime.min.time())
786 else:
787 raise TypeError("'" + str(date_time) + "' must be of type datetime or date!")
790def _datetime_to_date_list(date_times: _Union[_List[datetime], _List[date]]) -> _List[date]:
791 """
792 Converts types of date list from datetime to date or leaves it unchanged if they are already of type date.
794 Args:
795 date_times (_Union[List[datetime], List[date]]): List of date(time)s to be converted.
797 Returns:
798 List[date]: List of (potentially) converted date(time)s.
799 """
800 if isinstance(date_times, list):
801 return [_date_to_datetime(date_time) for date_time in date_times]
802 else:
803 raise TypeError("'" + str(date_times) + "' must be a list of type datetime or date!")
806def _string_to_period(term: str) -> Period:
807 """
808 Converts terms, e.g. 1D, 3M, and 5Y, into periods, i.e. Period(0, 0, 1), Period(0, 3, 0), and Period(5, 0, 0),
809 respectively.
811 Args:
812 term (str): Term to be converted into a period.
814 Returns:
815 Period: Period corresponding to the term specified.
816 """
817 if term == "T/N" or term == "O/N":
818 return Period(days=1)
819 else:
820 unit = term[-1]
821 try:
822 measure = int(term[:-1])
823 except ValueError:
824 measure = 0
825 if unit.upper() == "D":
826 period = Period(0, 0, measure)
827 elif unit.upper() == "M":
828 period = Period(0, measure, 0)
829 elif unit.upper() == "Y":
830 period = Period(measure, 0, 0)
831 else:
832 raise Exception("Unknown term! Please use: 'D', 'M', or 'Y'.")
833 return period
836def _term_to_period(term: _Union[Period, str]) -> Period:
837 """
838 Converts a term provided as period or string into period format if necessary.
840 Args:
841 term (_Union[Period, str]): Tenor to be converted if provided as string.
843 Returns:
844 Period: Tenor (potentially converted) in(to) period format.
845 """
846 if isinstance(term, Period):
847 return term
848 elif isinstance(term, str):
849 return _string_to_period(term)
850 else:
851 raise TypeError("The term '" + str(term) + "' must be provided as Period or string!")
854def _period_to_string(period: _Union[Period, str]) -> str:
855 """
856 Converts a period into string format.
858 Args:
859 period (Period): Period to be converted.
861 Returns:
862 str: Period in string format.
863 """
864 if isinstance(period, Period):
865 if period.years > 0 and period.months == 0 and period.days == 0:
866 return str(period.years) + "Y"
867 elif period.months > 0 and period.years == 0 and period.days == 0:
868 return str(period.months) + "M"
869 elif period.days > 0 and period.years == 0 and period.months == 0:
870 return str(period.days) + "D"
871 else:
872 raise Exception("The period '" + str(period) + "' cannot be converted to string format!")
873 elif isinstance(period, str):
874 return period
877def _is_ambiguous_date(day: _Union[date, datetime]) -> bool:
878 """
879 Checks if a given day is an ambiguous date, i.e. 30th of January, March, May, July, August, October or December.
881 Args:
882 day (_Union[date, datetime]): Day to be checked.
884 Returns:
885 bool: True if day is ambiguous date, False otherwise.
886 """
887 return ((day.day == 30) and (day.month in [1, 3, 5, 7, 8, 10, 12])) or ((day.day == 28 or day.day == 29) and day.month == 2)
890def _is_IMM_date(day: _Union[date, datetime]) -> bool:
891 """
892 Checks if a given day is an IMM date, i.e. the third Wednesday of March, June, September or December.
894 Args:
895 day (_Union[date, datetime]): Day to be checked.
897 Returns:
898 bool: True if day is IMM date, False otherwise.
899 """
900 return (day.month in [3, 6, 9, 12]) and (day.weekday() == 2) and (day.day >= 15) and (day.day <= 21)
903def next_IMM_date(from_date: _Union[date, datetime]) -> date:
904 """
905 Calculates the next IMM date (3rd Wednesday of March, June, September, December) on or after the given date.
907 Args:
908 from_date (_Union[date, datetime]): The date from which to find the next IMM date.
910 Returns:
911 date: The next IMM date on or after the given date.
912 """
913 from_date_dt = _date_to_datetime(from_date + relativedelta(days=1))
914 year = from_date_dt.year
915 month = from_date_dt.month
917 # Determine the next IMM month
918 if month <= 3:
919 imm_month = 3
920 elif month <= 6:
921 imm_month = 6
922 elif month <= 9:
923 imm_month = 9
924 else:
925 imm_month = 12
927 # Calculate the third Wednesday of the IMM month
928 first_day_of_imm_month = datetime(year, imm_month, 1)
929 first_wednesday = first_day_of_imm_month + relativedelta(weekday=WE(1))
930 third_wednesday = first_wednesday + relativedelta(weeks=2)
932 # If the calculated IMM date is before the from_date, move to the next IMM date
933 if third_wednesday < from_date_dt:
934 if imm_month == 12:
935 imm_month = 3
936 year += 1
937 else:
938 imm_month += 3
940 first_day_of_imm_month = datetime(year, imm_month, 1)
941 first_wednesday = first_day_of_imm_month + relativedelta(weekday=WE(1))
942 third_wednesday = first_wednesday + relativedelta(weeks=2)
943 logger.warning("Next IMM date from " + str(from_date) + " to next IMM date " + str(third_wednesday))
944 return third_wednesday.date()
947def next_IMM_period(period: Period) -> Period:
948 """
949 Adjusts the given period to the next multiple of 3 months, as IMM dates occur every 3 months.
951 Args:
952 period (Period): The original period.
954 Returns:
955 Period: The adjusted period.
956 """
958 months = period.months + (period.years * 12)
959 if period.days > 0:
960 months += 1 # If there are extra days, round up to the next month
961 months = ((months + 2) // 3) * 3 # Round up to next multiple of 3
962 return Period(0, months, 0)
965def calc_end_day(
966 start_day: _Union[date, datetime],
967 term: str,
968 business_day_convention: _Union[RollConvention, str] = RollConvention.UNADJUSTED,
969 calendar: _Union[_HolidayBase, str] = _ECB(),
970 roll_convention: _Union[RollRule, str] = RollRule.NONE,
971) -> date:
972 """
973 Derives the end date of a time period based on the start day the the term given as string, e.g. 1D, 3M, or 5Y.
974 If business day convention, corresponding calendar, and roll convention are provided the end date is additionally rolled accordingly.
976 Args:
977 start_day (_Union[date, datetime): Beginning of the time period with length term.
978 term (str): Term defining the period from start to end date.
979 business_day_convention (_Union[RollConvention, str], optional): Set of rules defining how to adjust
980 non-business days. Defaults to None.
981 calendar (_Union[_HolidayBase, str], optional): Holiday calender defining non-business days
982 (but not Saturdays and Sundays).
983 Defaults to None.
984 roll_convention (_Union[RollRule, str], optional): Convention for rolling dates, e.g. "EOM" for end of month.
986 Returns:
987 date: End date potentially adjusted according to the specified business day convention with respect to the given
988 calendar.
989 """
990 start_date = _date_to_datetime(start_day)
991 period = _term_to_period(term)
992 # Convert string roll_convention to enum if needed
993 roll_conv = RollRule.to_string(roll_convention) if roll_convention is not None else "NONE"
995 if roll_conv == RollRule.EOM.value and _is_ambiguous_date(start_date): # add ambiguous dates, i.e. 30 of Jan, Mar, May, Jul, Aug, Oct, Dec
996 end_date = start_date + relativedelta(years=period.years, months=period.months, day=31)
997 elif roll_conv == RollRule.IMM.value: # add IMM dates, i.e. 3rd Wednesday of Mar, Jun, Sep, Dec
998 period_new = next_IMM_period(period)
999 if _is_IMM_date(start_date):
1000 end_date = start_date + relativedelta(years=period_new.years, months=period_new.months, day=1, weekday=WE(3))
1001 else:
1002 end_date = next_IMM_date(start_date) + relativedelta(years=period_new.years, months=period_new.months, day=1, weekday=WE(3))
1003 elif roll_conv == RollRule.EOM.value or roll_conv == RollRule.IMM.value or roll_conv == RollRule.NONE.value or roll_conv == RollRule.DOM.value:
1004 end_date = start_date + relativedelta(years=period.years, months=period.months, days=period.days)
1005 else:
1006 raise Exception(
1007 "Unknown roll convention '" + str(roll_convention) + "'! Please use RollRule.NONE, RollRule.EOM, RollRule.DOM, or RollRule.IMM."
1008 )
1009 if (business_day_convention is not None) & (calendar is not None):
1010 end_date = roll_day(end_date, calendar, business_day_convention, start_date)
1012 return end_date
1015def calc_start_day(
1016 end_day: _Union[date, datetime],
1017 term: _Union[Period, str],
1018 business_day_convention: _Union[RollConvention, str] = "Unadjusted",
1019 calendar: _Union[_HolidayBase, str] = _ECB(),
1020 roll_convention: _Union[RollRule, str] = "NONE",
1021 max_iter: int = 10,
1022) -> date:
1023 """
1024 Derives the start date of a time period based on the end day, term, business day convention, calendar, and roll_convention.
1025 The start date may be a business day or not, depending on the business day convention provided.
1026 The function ensures that applying calc_end_day to the resulting start date with the same parameters returns the original end_day.
1027 Depending on the combination of the input parameters, a start date may not exist. In such cases, the function returns None and logs a warning.
1028 For other combinations, the start may not be unique. In such cases, the function returns the latest business day for which the end date is matched.
1030 Args:
1031 end_day (_Union[date, datetime]): End of the time period with length term.
1032 term (str): Term defining the period from start to end date.
1033 business_day_convention (_Union[RollConvention, str], optional): Set of rules defining how to adjust
1034 non-business days. Defaults to "Unadjusted".
1035 calendar (_Union[_HolidayBase, str], optional): Holiday calendar defining non-business days.
1036 roll_convention (_Union[RollRule, str], optional): Convention for rolling dates, e.g. "EOM" for end of month.
1037 max_iter (int, optional): Maximum number of iterations for the search. Defaults to 10.
1039 Returns:
1040 date: Start date such that calc_end_day(start_date, ...) == end_day.
1041 """
1042 end_date = _date_to_datetime(end_day)
1043 period = _term_to_period(term)
1044 if not (business_day_convention == "Unadjusted" or business_day_convention == RollConvention.UNADJUSTED) and not is_business_day(
1045 end_date, calendar
1046 ):
1047 logger.warning(
1048 f"Cannot not find a start date such that calc_end_day(start_date, ...) == end_day given combination of business day convention, calendar, and end_day."
1049 )
1050 return None
1051 start_date = next_or_previous_business_day(end_date - relativedelta(years=period.years, months=period.months, days=period.days), calendar, False)
1052 # Try to find the correct start_date such that calc_end_day(start_date, ...) == end_day
1053 for i in range(max_iter):
1054 candidate_end = calc_end_day(
1055 start_date,
1056 term,
1057 business_day_convention=business_day_convention,
1058 calendar=calendar,
1059 roll_convention=roll_convention,
1060 )
1061 if candidate_end == end_date:
1062 if not is_business_day(start_date, calendar):
1063 logger.warning(
1064 f"Found start date {start_date} such that calc_end_day(start_date, ...) == end_day, but start date is not a business day in the given calendar."
1065 )
1066 return start_date
1067 # Adjust start_date by one day if not matching
1068 # If candidate_end < end_date, move start_date back; else, move forward
1069 delta = (_date_to_datetime(candidate_end) - end_date).days
1070 start_date -= relativedelta(days=delta if delta != 0 else 1)
1071 # If not found, handle error internally
1072 logger.error(f"Could not find a start date such that calc_end_day(start_date, ...) == end_day after {max_iter} iterations.")
1073 return None
1076def last_day_of_month(day: _Union[date, datetime]) -> date:
1077 """
1078 Derives last day of the month corresponding to the given day.
1080 Args:
1081 day (_Union[date, datetime]): Day defining month and year for derivation of month's last day.
1083 Returns:
1084 date: Date of last day of the corresponding month.
1085 """
1086 return date(day.year, day.month, monthrange(day.year, day.month)[1])
1089def is_last_day_of_month(day: _Union[date, datetime]) -> bool:
1090 """
1091 Checks if a given day is the last day of the corresponding month.
1093 Args:
1094 day (_Union[date, datetime]): Day to be checked.
1096 Returns:
1097 bool: True, if day is last day of the month, False otherwise.
1098 """
1099 return _date_to_datetime(day) == _date_to_datetime(last_day_of_month(day))
1102def is_business_day(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> bool:
1103 """
1104 Checks if a given day is a business day in a given calendar.
1106 Args:
1107 day (_Union[date, datetime]): Day to be checked.
1108 calendar (_Union[_HolidayBase, str]): List of holidays defined by the given calendar.
1110 Returns:
1111 bool: True, if day is a business day, False otherwise.
1112 """
1113 # TODO: adjust for countries with weekend not on Saturday/Sunday (http://worldmap.workingdays.org/)
1114 return (day.isoweekday() < 6) & (day not in _string_to_calendar(calendar))
1117def last_business_day_of_month(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1118 """
1119 Derives the last business day of a month corresponding to a given day based on the holidays set in the calendar.
1121 Args:
1122 day (_Union[date, datetime]): Day defining month and year for deriving the month's last business day.
1123 calendar (_Union[_HolidayBase, str]): List of holidays defined by the given calendar.
1125 Returns:
1126 date: Date of last business day of the corresponding month.
1127 """
1128 check_day = date(day.year, day.month, monthrange(day.year, day.month)[1])
1129 while not (is_business_day(check_day, calendar)):
1130 check_day -= relativedelta(days=1)
1131 return check_day
1134def is_last_business_day_of_month(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> bool:
1135 """
1136 Checks it the given day is the last business day of the corresponding month.
1138 Args:
1139 day (_Union[date, datetime]): day to be checked
1140 calendar (_Union[_HolidayBase, str]): list of holidays defined by the given calendar
1142 Returns:
1143 bool: True if day is last business day of the corresponding month, False otherwise.
1144 """
1145 return _date_to_datetime(day) == _date_to_datetime(last_business_day_of_month(day, calendar))
1148def nearest_business_day(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str], following_first: bool = True) -> date:
1149 """
1150 Derives nearest business day from given day for a given calendar. If there are equally near days preceding and
1151 following the flag following_first determines if the following day is preferred to the preceding one.
1153 Args:
1154 day (_Union[date, datetime]): Day for which the nearest business day is to be found.
1155 calendar (_Union[_HolidayBase, str]): List of holidays given by calendar.
1156 following_first (bool): Flag for deciding if following days are preferred to an equally near preceding day.
1157 Default value is True.
1159 Returns:
1160 date: Nearest business day to given day according to given calendar.
1161 """
1162 distance = 0
1163 if following_first:
1164 direction = -1
1165 else:
1166 direction = +1
1168 day = _date_to_datetime(day)
1169 while not is_business_day(day, calendar):
1170 distance += 1
1171 direction *= -1
1172 day += direction * relativedelta(days=distance)
1173 return day
1176def nearest_last_business_day_of_month(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str], following_first: bool = True) -> date:
1177 """
1178 Derives nearest last business day of a month from given day for a given calendar. If there are equally near days
1179 preceding and following the flag following_first determines if the following day is preferred to the preceding one.
1181 Args:
1182 day (_Union[date, datetime]): Day for which the nearest last business day of the month is to be found.
1183 calendar (_Union[_HolidayBase, str]): List of holidays given by calendar.
1184 following_first (bool, optional): Flag for deciding if following days are preferred to an equally near preceding
1185 day. Defaults to True.
1187 Returns:
1188 date: Nearest last business day of a month to given day according to given calendar.
1189 """
1190 distance = 0
1191 if following_first:
1192 direction = -1
1193 else:
1194 direction = +1
1196 day = _date_to_datetime(day)
1197 while not is_last_business_day_of_month(day, calendar):
1198 distance += 1
1199 direction *= -1
1200 day += direction * relativedelta(days=distance)
1201 return day
1204def next_or_previous_business_day(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str], following_first: bool) -> date:
1205 """
1206 Derives the preceding or following business day to a given day according to a given calendar depending on the flag
1207 following_first. If the day is already a business day the function directly returns the day.
1209 Args:
1210 day (_Union[date, datetime]): Day for which the preceding or following business day is to be found.
1211 calendar (_HolidayBase): List of holidays defined by the calendar.
1212 following_first (bool): Flag to determine in the following (True) or preceding (False) business day is to be
1213 found.
1215 Returns:
1216 date: Preceding or following business day, respectively, or day itself if it is a business day.
1217 """
1218 if following_first:
1219 direction = +1
1220 else:
1221 direction = -1
1223 day = _date_to_datetime(day)
1224 while not is_business_day(day, calendar):
1225 day += direction * relativedelta(days=1)
1227 return day
1230def following(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1231 """
1232 Derives the (potentially) adjusted business day according to the business day convention 'Following' for a specified
1233 day with respect to a specific calendar: The adjusted date is the following good business day.
1235 Args:
1236 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1237 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1239 Returns:
1240 date: Adjusted business day according to the roll convention 'Following' with respect to calendar if the day is
1241 not already a business day. Otherwise the (unadjusted) day is returned.
1242 """
1243 return next_or_previous_business_day(day, calendar, True)
1246def preceding(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1247 """
1248 Derives the (potentially) adjusted business day according to the business day convention 'Preceding' for a specified
1249 day with respect to a specific calendar: The adjusted date is the preceding good business day.
1251 Args:
1252 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1253 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1255 Returns:
1256 date: Adjusted business day according to the roll convention 'Preceding' with respect to calendar if the day is
1257 not already a business day. Otherwise the (unadjusted) day is returned.
1258 """
1259 return next_or_previous_business_day(day, calendar, False)
1262def modified_following(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1263 """
1264 Derives the (potentially) adjusted business day according to the business day convention 'Modified Following' for a
1265 specified day with respect to a specific calendar: The adjusted date is the following good business day unless the
1266 day is in the next calendar month, in which case the adjusted date is the preceding good business day.
1268 Args:
1269 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1270 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1272 Returns:
1273 date: Adjusted business day according to the roll convention 'Modified Following' with respect to calendar if
1274 the day is not already a business day. Otherwise the (unadjusted) day is returned.
1275 """
1276 next_day = next_or_previous_business_day(day, calendar, True)
1277 if next_day.month != day.month:
1278 return preceding(day, calendar)
1279 else:
1280 return next_day
1283def modified_following_eom(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str], start_day: _Union[date, datetime]) -> date:
1284 """
1285 Derives the (potentially) adjusted business day according to the business day convention 'End of Month' for a
1286 specified day with respect to a specific calendar: Where the start date of a period is on the final business day of
1287 a particular calendar month, the end date is on the final business day of the end month (not necessarily the
1288 corresponding date in the end month).
1290 Args:
1291 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1292 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1293 start_day (_Union[date, datetime]): Day at which the period under consideration begins.
1295 Returns:
1296 date: Adjusted business day according to the roll convention 'End of Month' with respect to calendar.
1297 """
1298 if isinstance(start_day, date) | isinstance(start_day, datetime):
1299 if is_last_business_day_of_month(start_day, calendar):
1300 return nearest_last_business_day_of_month(day, calendar)
1301 else:
1302 return modified_following(day, calendar)
1303 else:
1304 raise Exception("The roll convention " + str(RollConvention.MODIFIED_FOLLOWING_EOM) + " cannot be evaluated without a start_day")
1307def modified_following_bimonthly(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1308 """
1309 Derives the (potentially) adjusted business day according to the business day convention 'Modified Following
1310 Bimonthly' for a specified day with respect to a specific calendar: The adjusted date is the following good business
1311 day unless that day crosses the mid-month (15th) or end of a month, in which case the adjusted date is the preceding
1312 good business day.
1314 Args:
1315 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1316 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1318 Returns:
1319 date: Adjusted business day according to the roll convention 'Modified Following Bimonthly' with respect to
1320 calendar if the day is not already a business day. Otherwise the (unadjusted) day is returned.
1321 """
1322 next_day = next_or_previous_business_day(day, calendar, True)
1323 if (next_day.month != day.month) | ((next_day.day > 15) & (day.day <= 15)):
1324 return preceding(day, calendar)
1325 else:
1326 return next_day
1329def modified_preceding(day: _Union[date, datetime], calendar: _Union[_HolidayBase, str]) -> date:
1330 """
1331 Derives the (potentially) adjusted business day according to the business day convention 'Modified Preceding' for a
1332 specified day with respect to a specific calendar: The adjusted date is the preceding good business day unless the
1333 day is in the previous calendar month, in which case the adjusted date is the following good business day.
1335 Args:
1336 day (_Union[date, datetime]): Day to be adjusted according to the roll convention if not already a business day.
1337 calendar (_Union[_HolidayBase, str]): Calendar defining holidays additional to weekends.
1339 Returns:
1340 date: Adjusted business day according to the roll convention 'Modified Preceding' with respect to calendar if
1341 the day is not already a business day. Otherwise the (unadjusted) day is returned.
1342 """
1343 prev_day = next_or_previous_business_day(day, calendar, False)
1344 if prev_day.month != day.month:
1345 return following(day, calendar)
1346 else:
1347 return prev_day
1350# to be used in the switcher (identical argument list)
1351def unadjusted(day: _Union[date, datetime], _) -> date:
1352 """
1353 Leaves the day unchanged independent from the fact if it is already a business day.
1355 Args:
1356 day (_Union[date, datetime]): Day to be adjusted according to the roll convention.
1357 _: Placeholder for calendar argument.
1359 Returns:
1360 date: Unadjusted day.
1361 """
1362 return _date_to_datetime(day)
1365def roll_day(
1366 day: _Union[date, datetime],
1367 calendar: _Union[_HolidayBase, str],
1368 business_day_convention: _Union[RollConvention, str],
1369 start_day: _Optional[_Union[date, datetime]] = None,
1370 settle_days: int = 0,
1371) -> date:
1372 """
1373 Adjusts a given day according to the specified business day convention with respect to a given calendar or if the
1374 given day falls on a Saturday or Sunday. For some roll conventions not only the (end) day to be adjusted but also
1375 the start day of a period is relevant for the adjustment of the given (end) day.
1377 Args:
1378 day (_Union[date, datetime]): Day to be adjusted if it is a non-business day.
1379 calendar (_Union[_HolidayBase, str]): Holiday calendar defining non-business days (but not weekends).
1380 business_day_convention (_Union[RollConvention, str]): Set of rules defining how to adjust non-business days.
1381 start_day (_Union[date, datetime], optional): Period's start day that may influence the rolling of the end day.
1382 Defaults to None.
1384 Returns:
1385 date: Adjusted day.
1386 """
1387 roll_convention = RollConvention.to_string(business_day_convention)
1388 # if start_day is not None:
1389 # start_day = _date_to_datetime(start_day)
1391 switcher: Dict[str, Callable[..., date]] = {
1392 "Unadjusted": unadjusted,
1393 "Following": following,
1394 "ModifiedFollowing": modified_following,
1395 "ModifiedFollowingEOM": modified_following_eom,
1396 "ModifiedFollowingBimonthly": modified_following_bimonthly,
1397 "Nearest": nearest_business_day,
1398 "Preceding": preceding,
1399 "ModifiedPreceding": modified_preceding,
1400 }
1401 # Get the appropriate roll function from switcher dictionary
1403 roll_func = switcher.get(roll_convention)
1405 if roll_func is None:
1406 raise ValueError(f"Business day convention '{business_day_convention}' is not known!")
1408 # Check if roll_func expects three arguments (including self for methods)
1409 import inspect
1411 params = inspect.signature(roll_func).parameters
1412 if "start_day" in params and settle_days == 0:
1413 return roll_func(day, calendar, start_day)
1414 elif "start_day" in params and settle_days > 0:
1415 with_settlement = roll_func(day, calendar, start_day) + relativedelta(days=settle_days)
1416 return roll_func(with_settlement, calendar, start_day)
1417 elif "start_day" not in params and settle_days > 0:
1418 with_settlement = roll_func(day, calendar) + relativedelta(days=settle_days)
1419 return roll_func(with_settlement, calendar)
1420 else:
1421 return roll_func(day, calendar)
1424def serialize_date(val):
1425 if isinstance(val, (datetime, date)):
1426 return val.isoformat()
1427 return val
1430# class PowerSchedule:
1431# def __init__(self,
1432# start_day: _Union[date, datetime],
1433# end_day: _Union[date, datetime],
1434# time_period: _Union[Period, str],
1435# backwards: bool = True,
1436# business_day_convention: _Union[RollConvention, str] = RollConvention.MODIFIED_FOLLOWING,
1437# calendar: _Union[_HolidayBase, str] = None):
1438# """
1440# Args:
1441# start_day (_Union[date, datetime]): Schedule's first day - beginning of the schedule.
1442# end_day (_Union[date, datetime]): Schedule's last day - end of the schedule.
1443# time_period (_Union[Period, str]): Time distance between two consecutive dates.
1444# backwards (bool, optional): Defines direction for rolling out the schedule. True means the schedule will be
1445# rolled out (backwards) from end day to start day. Defaults to True.
1446# stub (bool, optional): Defines if the first/last period is accepted (True), even though it is shorter than
1447# the others, or if it remaining days are added to the neighbouring period (False).
1448# Defaults to True.
1449# business_day_convention (_Union[RollConvention, str], optional): Set of rules defining the adjustment of
1450# days to ensure each date being a business
1451# day with respect to a given holiday
1452# calendar. Defaults to
1453# RollConvention.MODIFIED_FOLLOWING
1454# calendar (_Union[_HolidayBase, str], optional): Holiday calendar defining the bank holidays of a country or
1455# province (but not all non-business days as for example
1456# Saturdays and Sundays).
1457# Defaults (through constructor) to holidays.ECB
1458# (= Target2 calendar) between start_day and end_day.
1460# Examples:
1462# .. code-block:: python
1464# >>> from datetime import date
1465# >>> from rivapy.tools import schedule
1466# >>> schedule = Schedule(date(2020, 8, 21), date(2021, 8, 21), Period(0, 3, 0), True, False, RollConvention.UNADJUSTED, holidays_de).generate_dates(False),
1467# [date(2020, 8, 21), date(2020, 11, 21), date(2021, 2, 21), date(2021, 5, 21), date(2021, 8, 21)])
1468# """
1469# self.start_day = start_day
1470# self.end_day = end_day
1471# self.time_period = time_period
1472# self.backwards = backwards
1473# self.business_day_convention = business_day_convention
1474# self.calendar = calendar
1477# @property
1478# def start_day(self):
1479# """
1480# Getter for schedule's start date.
1482# Returns:
1483# Start date of specified schedule.
1484# """
1485# return self.__start_day
1487# @start_day.setter
1488# def start_day(self, start_day: _Union[date, datetime]):
1489# self.__start_day = _date_to_datetime(start_day)
1491# @property
1492# def end_day(self):
1493# """
1494# Getter for schedule's end date.
1496# Returns:
1497# End date of specified schedule.
1498# """
1499# return self.__end_day
1501# @end_day.setter
1502# def end_day(self, end_day: _Union[date, datetime]):
1503# self.__end_day = _date_to_datetime(end_day)
1505# @property
1506# def time_period(self):
1507# """
1508# Getter for schedule's time period.
1510# Returns:
1511# Time period of specified schedule.
1512# """
1513# return self.__time_period
1515# @time_period.setter
1516# def time_period(self, time_period: _Union[Period, str]):
1517# self.__time_period = _term_to_period(time_period)
1519# @property
1520# def backwards(self):
1521# """
1522# Getter for schedule's roll out direction.
1524# Returns:
1525# True, if rolled out from end day to start day.
1526# False, if rolled out from start day to end day.
1527# """
1528# return self.__backwards
1530# @backwards.setter
1531# def backwards(self, backwards: bool):
1532# self.__backwards = backwards
1534# @property
1535# def stub(self):
1536# """
1537# Getter for potential existence of short periods (stubs).
1539# Returns:
1540# True, if a shorter period is allowed.
1541# False, if only a longer period is allowed.
1542# """
1543# return self.__stub
1545# @stub.setter
1546# def stub(self, stub: bool):
1547# self.__stub = stub
1549# @property
1550# def business_day_convention(self):
1551# """
1552# Getter for schedule's business day convention.
1554# Returns:
1555# Business day convention of specified schedule.
1556# """
1557# return self.__business_day_convention
1559# @business_day_convention.setter
1560# def business_day_convention(self, business_day_convention: _Union[RollConvention, str]):
1561# self.__business_day_convention = RollConvention.to_string(business_day_convention)
1563# @property
1564# def calendar(self):
1565# """
1566# Getter for schedule's holiday calendar.
1568# Returns:
1569# Holiday calendar of specified schedule.
1570# """
1571# return self.__calendar
1573# @calendar.setter
1574# def calendar(self, calendar: _Union[_HolidayBase, str]):
1575# if calendar is None:
1576# self.__calendar = _ECB(years=range(self.__start_day.year, self.__end_day.year + 1))
1577# else:
1578# self.__calendar = _string_to_calendar(calendar)
1580# @staticmethod
1581# def _roll_out(from_: _Union[date, datetime], to_: _Union[date, datetime], term: Period, backwards: bool,
1582# allow_stub: bool) -> _List[date]:
1583# """
1584# Rolls out dates from from_ to to_ in the specified direction applying the given term under consideration of the
1585# specification for allowing shorter periods.
1587# Args:
1588# from_ (_Union[date, datetime]): Beginning of the roll out mechanism.
1589# to_ (_Union[date, datetime]): End of the roll out mechanism.
1590# term (Period): Difference between rolled out dates.
1591# backwards (bool): Direction of roll out mechanism: backwards if True, forwards if False.
1592# allow_stub (bool): Defines if periods shorter than term are allowed.
1594# Returns:
1595# Date schedule not yet adjusted to any business day convention.
1596# """
1597# # convert datetime to date (if necessary):
1598# from_ = _date_to_datetime(from_)
1599# to_ = _date_to_datetime(to_)
1601# # check input consistency:
1602# if (~backwards) & (from_ < to_):
1603# direction = +1
1604# elif backwards & (from_ > to_):
1605# direction = -1
1606# else:
1607# raise Exception("From-date '" + str(from_) + "' and to-date '" + str(to_) +
1608# "' are not consistent with roll direction (backwards = '" + str(backwards) + "')!")
1610# # generates a list of dates ...
1611# dates = []
1612# # ... for forward rolling case or backward rolling case ...
1613# while ((~backwards) & (from_ <= to_)) | (backwards & (to_ <= from_)):
1614# dates.append(from_)
1615# from_ += direction * relativedelta(years=term.years, months=term.months, days=term.days)
1616# # ... and compete list for fractional periods ...
1617# if dates[-1] != to_:
1618# # ... by adding stub or ...
1619# if allow_stub:
1620# dates.append(to_)
1621# # ... by extending last period.
1622# else:
1623# dates[-1] = to_
1624# return dates
1626# def generate_dates(self, ends_only: bool) -> _List[date]:
1627# """
1628# Generate list of schedule days according to the schedule specification, in particular with regards to business
1629# day convention and calendar given.
1631# Args:
1632# ends_only (bool): Flag to indicate if period beginnings shall be included, e.g. for defining accrual
1633# periods: True, if only period ends shall be included, e.g. for defining payment dates.
1635# Returns:
1636# List[date]: List of schedule dates (including start and end date) adjusted to rolling convention.
1637# """
1638# # roll out dates ignoring any business day issues
1639# if self.__backwards:
1640# schedule_dates = Schedule._roll_out(self.__end_day, self.__start_day, self.__time_period,
1641# True, self.__stub)
1642# schedule_dates.reverse()
1643# else:
1644# schedule_dates = Schedule._roll_out(self.__start_day, self.__end_day, self.__time_period,
1645# False, self.__stub)
1647# # adjust according to business day convention
1648# rolled_schedule_dates = [roll_day(schedule_dates[0], self.__calendar, self.__business_day_convention,
1649# schedule_dates[0])]
1650# [rolled_schedule_dates.append(roll_day(schedule_dates[i], self.__calendar, self.__business_day_convention,
1651# rolled_schedule_dates[i - 1])) for i in range(1, len(schedule_dates))]
1653# if ends_only:
1654# rolled_schedule_dates.pop(0)
1656# logger.debug("Schedule dates successfully calculated from '"
1657# + str(self.__start_day) + "' to '" + str(self.__end_day) + "'.")
1658# return rolled_schedule_dates