From 3f7012d3ee36ab5954d2ee359a5e028c70b7aacc Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Fri, 3 Jan 2025 18:08:28 +0100 Subject: [PATCH 1/6] TYP: legs.py (#595) Co-authored-by: JHM Darbyshire (win11) --- python/rateslib/legs.py | 428 ++++++++++++++++++++++--------------- python/rateslib/periods.py | 18 +- 2 files changed, 265 insertions(+), 181 deletions(-) diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index 424d1a41..c6cb2622 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -25,17 +25,16 @@ import warnings from abc import ABCMeta, abstractmethod from datetime import datetime -from typing import NoReturn +from typing import Any, NoReturn import pandas as pd from pandas import DataFrame, Series -from pandas.tseries.offsets import CustomBusinessDay from rateslib import defaults -from rateslib.calendars import add_tenor +from rateslib.calendars import CalInput, add_tenor from rateslib.curves import Curve, index_left from rateslib.default import NoInput, _drb -from rateslib.dual import Dual, Dual2, DualTypes, gradient, set_order +from rateslib.dual import Dual, Dual2, DualTypes from rateslib.fx import FXForwards, FXRates from rateslib.periods import ( Cashflow, @@ -167,7 +166,7 @@ def __init__( roll: str | int | NoInput = NoInput(0), eom: bool | NoInput = NoInput(0), modifier: str | NoInput = NoInput(0), - calendar: CustomBusinessDay | str | NoInput = NoInput(0), + calendar: CalInput = NoInput(0), payment_lag: int | NoInput = NoInput(0), notional: float | NoInput = NoInput(0), currency: str | NoInput = NoInput(0), @@ -190,39 +189,33 @@ def __init__( calendar, payment_lag, ) - self.convention = defaults.convention if convention is NoInput.blank else convention - self.currency = defaults.base_currency if currency is NoInput.blank else currency.lower() - - self.payment_lag_exchange = ( - defaults.payment_lag_exchange - if payment_lag_exchange is NoInput.blank - else payment_lag_exchange - ) - self.initial_exchange = initial_exchange - self.final_exchange = final_exchange - - self._notional = defaults.notional if notional is NoInput.blank else notional - self._amortization = 0 if amortization is NoInput.blank else amortization + self.convention: str = _drb(defaults.convention, convention) + self.currency: str = _drb(defaults.base_currency, currency).lower() + self.payment_lag_exchange: int = _drb(defaults.payment_lag_exchange, payment_lag_exchange) + self.initial_exchange: bool = initial_exchange + self.final_exchange: bool = final_exchange + self._notional: float = _drb(defaults.notional, notional) + self._amortization: float = _drb(0.0, amortization) if getattr(self, "_delay_set_periods", False): pass else: self._set_periods() @property - def notional(self): + def notional(self) -> float: return self._notional @notional.setter - def notional(self, value): + def notional(self, value: float) -> None: self._notional = value self._set_periods() @property - def amortization(self): + def amortization(self) -> float: return self._amortization @amortization.setter - def amortization(self, value): + def amortization(self, value: float) -> None: self._amortization = value self._set_periods() @@ -255,7 +248,7 @@ def _set_periods(self) -> None: notional=self.notional - self.amortization * i, iterator=i, ) - for i, period in self.schedule.table.to_dict(orient="index").items() + for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) ] if self.final_exchange and self.amortization != 0: amortization = [ @@ -294,15 +287,19 @@ def _set_periods(self) -> None: ), ) - # @abstractmethod - # def _regular_period(self): - # pass # pragma: no cover + @abstractmethod + def _regular_period(self, *args: Any, **kwargs: Any) -> Any: + # implemented by individual legs to satify generic `set_periods` methods + pass # pragma: no cover # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International # Commercial use of this code, and/or copying and redistribution is prohibited. # Contact rateslib at gmail.com if this code is observed outside its intended sphere. - def analytic_delta(self, *args, **kwargs): + # def _regular_period(self, *args: Any, **kwargs: Any) -> Any: + # pass + + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *Leg* via summing all periods. @@ -312,7 +309,7 @@ def analytic_delta(self, *args, **kwargs): _ = (period.analytic_delta(*args, **kwargs) for period in self.periods) return sum(_) - def cashflows(self, *args, **kwargs) -> DataFrame: + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *Leg* used in calculating cashflows. @@ -322,140 +319,44 @@ def cashflows(self, *args, **kwargs) -> DataFrame: seq = [period.cashflows(*args, **kwargs) for period in self.periods] return DataFrame.from_records(seq) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *Leg* via summing all periods. For arguments see :meth:`BasePeriod.npv()`. """ - _is_local = (len(args) == 5 and args[4]) or kwargs.get("local", False) + _is_local = (len(args) >= 5 and args[4]) or kwargs.get("local", False) if _is_local: - _ = (period.npv(*args, **kwargs)[self.currency] for period in self.periods) + _ = (period.npv(*args, **kwargs)[self.currency] for period in self.periods) # type: ignore[index] return {self.currency: sum(_)} else: _ = (period.npv(*args, **kwargs) for period in self.periods) return sum(_) - @property - def _is_linear(self): - """ - Tests if analytic delta spread is a linear function affecting NPV. - - This is non-linear if the spread is itself compounded, which only occurs - on RFR trades with *"isda_compounding"* or *"isda_flat_compounding"*, which - should typically be avoided anyway. - - Returns - ------- - bool - """ - # ruff: noqa: SIM103 - if ( - "Float" in type(self).__name__ - and "rfr" in self.fixing_method - and self.spread_compound_method != "none_simple" - ): - return False - return True - - def _spread_isda_approximated_rate(self, target_npv, fore_curve, disc_curve): - """ - Use approximated derivatives through geometric averaged 1day rates to derive the - spread - """ - a, b = 0.0, 0.0 - for period in self.periods: - try: - a_, b_ = period._get_analytic_delta_quadratic_coeffs(fore_curve, disc_curve) - a += a_ - b += b_ - except AttributeError: # the period might be of wrong kind: TODO: better filter - pass - c = -target_npv - - # perform the quadratic solution - _1 = -c / b - if abs(a) > 1e-14: - _2a = (-b - (b**2 - 4 * a * c) ** 0.5) / (2 * a) - _2b = (-b + (b**2 - 4 * a * c) ** 0.5) / (2 * a) # alt quadratic soln - if abs(_1 - _2a) < abs(_1 - _2b): - _ = _2a - else: - _ = _2b # select quadratic soln - else: - # this is to avoid divide by zero errors and return an approximation - # also isda_flat_compounding has a=0 - _ = _1 - - return _ + # @property + # def _is_linear(self) -> bool: + # """ + # Tests if analytic delta spread is a linear function affecting NPV. + # + # This is non-linear if the spread is itself compounded, which only occurs + # on RFR trades with *"isda_compounding"* or *"isda_flat_compounding"*, which + # should typically be avoided anyway. + # + # Returns + # ------- + # bool + # """ + # # ruff: noqa: SIM103 + # return True - def _spread_isda_dual2( + def _spread( self, - target_npv, - fore_curve, - disc_curve, - fx=NoInput(0), - ): # pragma: no cover - # This method is unused and untested, superseded by _spread_isda_approx_rate - - # This method creates a dual2 variable for float spread + obtains derivatives automatically - _fs = self.float_spread - self.float_spread = Dual2(0.0 if _fs is None else float(_fs), "spread_z") - - # This method uses ad-hoc AD to solve a specific problem for which - # there is no closed form solution. Calculating NPV is very inefficient - # so, we only do this once as opposed to using a root solver algo - # which would otherwise converge to the exact solution but is - # practically not workable. - - # This method is more accurate than the 'spread through approximated - # derivatives' method, but it is a more costly and less robust method - # due to its need to work in second order mode. - - fore_ad = fore_curve.ad - fore_curve._set_ad_order(2) - - disc_ad = disc_curve.ad - disc_curve._set_ad_order(2) - - if isinstance(fx, FXRates | FXForwards): - _fx = None if fx is None else fx._ad - fx._set_ad_order(2) - - npv = self.npv(fore_curve, disc_curve, fx, self.currency) - b = gradient(npv, "spread_z", order=1)[0] - a = 0.5 * gradient(npv, "spread_z", order=2)[0][0] - c = -target_npv - - # Perform quadratic solution - _1 = -c / b - if abs(a) > 1e-14: - _2a = (-b - (b**2 - 4 * a * c) ** 0.5) / (2 * a) - _2b = (-b + (b**2 - 4 * a * c) ** 0.5) / (2 * a) # alt quadratic soln - if abs(_1 - _2a) < abs(_1 - _2b): - _ = _2a - else: - _ = _2b # select quadratic soln - else: # pragma: no cover - # this is to avoid divide by zero errors and return an approximation - _ = _1 - warnings.warn( - "Divide by zero encountered and the spread is approximated to " "first order.", - UserWarning, - ) - - # This is required by the Dual2 AD approach to revert to original order. - self.float_spread = _fs - fore_curve._set_ad_order(fore_ad) - disc_curve._set_ad_order(disc_ad) - if isinstance(fx, FXRates | FXForwards): - fx._set_ad_order(_fx) - _ = set_order(_, disc_ad) # use disc_ad: credit spread from disc curve - - return _ - - def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): + target_npv: DualTypes, + fore_curve: Curve, + disc_curve: Curve, + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), + ) -> DualTypes: """ Calculates an adjustment to the ``fixed_rate`` or ``float_spread`` to match a specific target NPV. @@ -492,15 +393,10 @@ def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): Examples -------- """ - if self._is_linear: - a_delta = self.analytic_delta(fore_curve, disc_curve, fx, self.currency) - return -target_npv / a_delta - else: - _ = self._spread_isda_approximated_rate(target_npv, fore_curve, disc_curve) - # _ = self._spread_isda_dual2(target_npv, fore_curve, disc_curve, fx) - return _ + a_delta = self.analytic_delta(fore_curve, disc_curve, fx, self.currency) + return -target_npv / a_delta - def __repr__(self): + def __repr__(self) -> str: return f"" @@ -510,8 +406,13 @@ class _FixedLegMixin: :class:`~rateslib.periods.FixedPeriod` s. """ + convention: str + schedule: Schedule + currency: str + _fixed_rate: DualTypes | NoInput + @property - def fixed_rate(self): + def fixed_rate(self) -> DualTypes | NoInput: """ float or NoInput : If set will also set the ``fixed_rate`` of contained :class:`FixedPeriod` s. @@ -519,7 +420,7 @@ def fixed_rate(self): return self._fixed_rate @fixed_rate.setter - def fixed_rate(self, value): + def fixed_rate(self, value: DualTypes) -> None: self._fixed_rate = value for period in getattr(self, "periods", []): if isinstance(period, FixedPeriod | CreditPremiumPeriod): @@ -533,7 +434,7 @@ def _regular_period( notional: float, stub: bool, iterator: int, - ): + ) -> FixedPeriod: return FixedPeriod( fixed_rate=self.fixed_rate, start=start, @@ -550,7 +451,7 @@ def _regular_period( ) -class FixedLeg(BaseLeg, _FixedLegMixin): +class FixedLeg(_FixedLegMixin, BaseLeg): """ Create a fixed leg composed of :class:`~rateslib.periods.FixedPeriod` s. @@ -596,12 +497,14 @@ class FixedLeg(BaseLeg, _FixedLegMixin): fixed_leg_exch.npv(curve) """ # noqa: E501 - def __init__(self, *args, fixed_rate: float | NoInput = NoInput(0), **kwargs): + def __init__( + self, *args: Any, fixed_rate: DualTypes | NoInput = NoInput(0), **kwargs: Any + ) -> None: self._fixed_rate = fixed_rate super().__init__(*args, **kwargs) self._set_periods() - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *FixedLeg* via summing all periods. @@ -610,7 +513,7 @@ def analytic_delta(self, *args, **kwargs): """ return super().analytic_delta(*args, **kwargs) - def cashflows(self, *args, **kwargs) -> DataFrame: + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *FixedLeg* used in calculating cashflows. @@ -619,7 +522,7 @@ def cashflows(self, *args, **kwargs) -> DataFrame: """ return super().cashflows(*args, **kwargs) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *FixedLeg* via summing all periods. @@ -644,7 +547,16 @@ class _FloatLegMixin: :meth:`~rateslib.periods.FloatPeriod.fixings_table`. """ - def _get_fixings_from_series(self, ser: Series, ini_period: int = 0) -> list: + convention: str + schedule: Schedule + currency: str + _float_spread: DualTypes + fixing_method: str + method_param: int + + def _get_fixings_from_series( + self, ser: Series[DualTypes], ini_period: int = 0 + ) -> list[Series[DualTypes] | NoInput]: # type: ignore[type-var] """ Determine which fixings can be set for Periods with the given Series. """ @@ -662,7 +574,7 @@ def _get_fixings_from_series(self, ser: Series, ini_period: int = 0) -> list: add_tenor( self.schedule.aschedule[i], f"-{adj_days}B", - None, + "NONE", self.schedule.calendar, ) for i in range(ini_period, self.schedule.n_periods) @@ -691,6 +603,107 @@ def _set_fixings( self.fixings = fixings_ + [NoInput(0)] * (self.schedule.n_periods - len(fixings_)) + # def _spread_isda_dual2( + # self, + # target_npv, + # fore_curve, + # disc_curve, + # fx=NoInput(0), + # ): # pragma: no cover + # # This method is unused and untested, superseded by _spread_isda_approx_rate + # + # # This method creates a dual2 variable for float spread + obtains derivativs automatically + # _fs = self.float_spread + # self.float_spread = Dual2(0.0 if _fs is None else float(_fs), "spread_z") + # + # # This method uses ad-hoc AD to solve a specific problem for which + # # there is no closed form solution. Calculating NPV is very inefficient + # # so, we only do this once as opposed to using a root solver algo + # # which would otherwise converge to the exact solution but is + # # practically not workable. + # + # # This method is more accurate than the 'spread through approximated + # # derivatives' method, but it is a more costly and less robust method + # # due to its need to work in second order mode. + # + # fore_ad = fore_curve.ad + # fore_curve._set_ad_order(2) + # + # disc_ad = disc_curve.ad + # disc_curve._set_ad_order(2) + # + # if isinstance(fx, FXRates | FXForwards): + # _fx = None if fx is None else fx._ad + # fx._set_ad_order(2) + # + # npv = self.npv(fore_curve, disc_curve, fx, self.currency) + # b = gradient(npv, "spread_z", order=1)[0] + # a = 0.5 * gradient(npv, "spread_z", order=2)[0][0] + # c = -target_npv + # + # # Perform quadratic solution + # _1 = -c / b + # if abs(a) > 1e-14: + # _2a = (-b - (b**2 - 4 * a * c) ** 0.5) / (2 * a) + # _2b = (-b + (b**2 - 4 * a * c) ** 0.5) / (2 * a) # alt quadratic soln + # if abs(_1 - _2a) < abs(_1 - _2b): + # _ = _2a + # else: + # _ = _2b # select quadratic soln + # else: # pragma: no cover + # # this is to avoid divide by zero errors and return an approximation + # _ = _1 + # warnings.warn( + # "Divide by zero encountered and the spread is approximated to " "first order.", + # UserWarning, + # ) + # + # # This is required by the Dual2 AD approach to revert to original order. + # self.float_spread = _fs + # fore_curve._set_ad_order(fore_ad) + # disc_curve._set_ad_order(disc_ad) + # if isinstance(fx, FXRates | FXForwards): + # fx._set_ad_order(_fx) + # _ = set_order(_, disc_ad) # use disc_ad: credit spread from disc curve + # + # return _ + + def _spread_isda_approximated_rate( + self, + target_npv: DualTypes, + fore_curve: Curve, + disc_curve: Curve, + ) -> DualTypes: + """ + Use approximated derivatives through geometric averaged 1day rates to derive the + spread + """ + a, b = 0.0, 0.0 + for period in self.periods: + try: + a_, b_ = period._get_analytic_delta_quadratic_coeffs(fore_curve, disc_curve) + a += a_ + b += b_ + except AttributeError: # the period might be of wrong kind: TODO: better filter + pass + c = -target_npv + + # perform the quadratic solution + _1 = -c / b + if abs(a) > 1e-14: + _2a = (-b - (b**2 - 4 * a * c) ** 0.5) / (2 * a) + _2b = (-b + (b**2 - 4 * a * c) ** 0.5) / (2 * a) # alt quadratic soln + if abs(_1 - _2a) < abs(_1 - _2b): + _: DualTypes = _2a + else: + _ = _2b # select quadratic soln + else: + # this is to avoid divide by zero errors and return an approximation + # also isda_flat_compounding has a=0 + _ = _1 + + return _ + @property def float_spread(self): """ @@ -784,8 +797,71 @@ def _regular_period( spread_compound_method=self.spread_compound_method, ) + @property + def _is_linear(self) -> bool: + """ + Tests if analytic delta spread is a linear function affecting NPV. + + This is non-linear if the spread is itself compounded, which only occurs + on RFR trades with *"isda_compounding"* or *"isda_flat_compounding"*, which + should typically be avoided anyway. + + Returns + ------- + bool + """ + # ruff: noqa: SIM103 + if "rfr" in self.fixing_method and self.spread_compound_method != "none_simple": + return False + return True + + def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): + """ + Calculates an adjustment to the ``fixed_rate`` or ``float_spread`` to match + a specific target NPV. + + Parameters + ---------- + target_npv : float, Dual or Dual2 + The target NPV that an adjustment to the parameter will achieve. **Must + be in local currency of the leg.** + fore_curve : Curve or LineCurve + The forecast curve passed to analytic delta calculation. + disc_curve : Curve + The discounting curve passed to analytic delta calculation. + fx : FXForwards, optional + Required for multi-currency legs which are MTM exchanged. + + Returns + ------- + float, Dual, Dual2 -class FloatLeg(BaseLeg, _FloatLegMixin): + Notes + ----- + ``FixedLeg`` and ``FloatLeg`` with a *"none_simple"* spread compound method have + linear sensitivity to the spread. This can be calculated directly and + exactly using an analytic delta calculation. + + *"isda_compounding"* and *"isda_flat_compounding"* spread compound methods + have non-linear sensitivity to the spread. This requires a root finding, + iterative algorithm, which, coupled with very poor performance of calculating + period rates under this method is exceptionally slow. We approximate this + using first and second order AD and extrapolate a solution as a Taylor + expansion. This results in approximation error. + + Examples + -------- + """ + if self._is_linear: + a_delta = self.analytic_delta(fore_curve, disc_curve, fx, self.currency) + return -target_npv / a_delta + else: + _ = self._spread_isda_approximated_rate(target_npv, fore_curve, disc_curve) + # _ = self._spread_isda_dual2(target_npv, fore_curve, disc_curve, fx) + return _ + + +class FloatLeg(_FloatLegMixin, BaseLeg): """ Create a floating leg composed of :class:`~rateslib.periods.FloatPeriod` s. @@ -1131,8 +1207,11 @@ def index_base(self, value: DualTypes | Series[DualTypes] | NoInput) -> None: if isinstance(period, IndexFixedPeriod | IndexCashflow): period.index_base = self._index_base + def _regular_period(self) -> None: + pass + -class ZeroFloatLeg(BaseLeg, _FloatLegMixin): +class ZeroFloatLeg(_FloatLegMixin, BaseLeg): """ Create a zero coupon floating leg composed of :class:`~rateslib.periods.FloatPeriod` s. @@ -1450,7 +1529,7 @@ def cashflows( return DataFrame.from_records(seq) -class ZeroFixedLeg(BaseLeg, _FixedLegMixin): +class ZeroFixedLeg(_FixedLegMixin, BaseLeg): """ Create a zero coupon fixed leg composed of a single :class:`~rateslib.periods.FixedPeriod` . @@ -1665,7 +1744,7 @@ def npv(self, *args, **kwargs): return super().npv(*args, **kwargs) -class ZeroIndexLeg(BaseLeg, _IndexLegMixin): +class ZeroIndexLeg(_IndexLegMixin, BaseLeg): """ Create a zero coupon index leg composed of a single :class:`~rateslib.periods.IndexFixedPeriod` and @@ -1817,7 +1896,7 @@ def npv(self, *args, **kwargs): return super().npv(*args, **kwargs) -class CreditPremiumLeg(BaseLeg, _FixedLegMixin): +class CreditPremiumLeg(_FixedLegMixin, BaseLeg): """ Create a credit premium leg composed of :class:`~rateslib.periods.CreditPremiumPeriod` s. @@ -2588,7 +2667,7 @@ def analytic_delta( return ret -class FixedLegMtm(BaseLegMtm, _FixedLegMixin): +class FixedLegMtm(_FixedLegMixin, BaseLegMtm): """ Create a leg of :class:`~rateslib.periods.FixedPeriod` s and initial, mtm and final :class:`~rateslib.periods.Cashflow` s. @@ -2650,7 +2729,7 @@ def __init__( # Contact rateslib at gmail.com if this code is observed outside its intended sphere. -class FloatLegMtm(BaseLegMtm, _FloatLegMixin): +class FloatLegMtm(_FloatLegMixin, BaseLegMtm): """ Create a leg of :class:`~rateslib.periods.FloatPeriod` s and initial, mtm and final :class:`~rateslib.periods.Cashflow` s. @@ -2824,6 +2903,9 @@ def analytic_delta(self, *args, **kwargs): """ return super().analytic_delta(*args, **kwargs) + def _regular_period(self, *args: Any, **kwargs: Any) -> Any: + pass + # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International # Commercial use of this code, and/or copying and redistribution is prohibited. diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index be0e6a00..c3d903de 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -509,8 +509,10 @@ class FixedPeriod(BasePeriod): """ - def __init__(self, *args: Any, fixed_rate: float | NoInput = NoInput(0), **kwargs: Any) -> None: - self.fixed_rate = fixed_rate + def __init__( + self, *args: Any, fixed_rate: DualTypes | NoInput = NoInput(0), **kwargs: Any + ) -> None: + self.fixed_rate: DualTypes | NoInput = fixed_rate super().__init__(*args, **kwargs) def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: @@ -2268,26 +2270,26 @@ class CreditPremiumPeriod(BasePeriod): def __init__( self, *args: Any, - fixed_rate: float | NoInput = NoInput(0), + fixed_rate: DualTypes | NoInput = NoInput(0), premium_accrued: bool | NoInput = NoInput(0), **kwargs: Any, ) -> None: - self.premium_accrued = _drb(defaults.cds_premium_accrued, premium_accrued) - self.fixed_rate = fixed_rate + self.premium_accrued: bool = _drb(defaults.cds_premium_accrued, premium_accrued) + self.fixed_rate: DualTypes | NoInput = fixed_rate super().__init__(*args, **kwargs) @property - def cashflow(self) -> float | None: + def cashflow(self) -> DualTypes | None: """ float, Dual or Dual2 : The calculated value from rate, dcf and notional. """ if isinstance(self.fixed_rate, NoInput): return None else: - _: float = -self.notional * self.dcf * self.fixed_rate * 0.01 + _: DualTypes = -self.notional * self.dcf * self.fixed_rate * 0.01 return _ - def accrued(self, settlement: datetime) -> float | None: + def accrued(self, settlement: datetime) -> DualTypes | None: """ Calculate the amount of premium accrued until a specific date within the *Period*. From 6593bfc30c3342c07d9c2358832721a085456554 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Fri, 3 Jan 2025 19:31:33 +0100 Subject: [PATCH 2/6] TST: CustomLeg allows additional Periods (#596) --- docs/source/i_whatsnew.rst | 5 +++ python/rateslib/legs.py | 33 +++++++++++------ python/tests/test_legs.py | 76 ++++++++++++++++++++++++++++++++++++-- 3 files changed, 100 insertions(+), 14 deletions(-) diff --git a/docs/source/i_whatsnew.rst b/docs/source/i_whatsnew.rst index 027e5e22..4367e5e8 100644 --- a/docs/source/i_whatsnew.rst +++ b/docs/source/i_whatsnew.rst @@ -55,6 +55,11 @@ email contact, see `rateslib `_. (`532 `_) (`535 `_) (`536 `_) + * - Bug + - :meth:`~rateslib.legs.CustomLeg` now allows construction from recently constructed + *Period* types including *CreditProtectionPeriod*, *CreditPremiumPeriod*, + *IndexCashflow* and *IndexFixedPeriod*. + (`596 `_) * - Dependencies - Drop support for Python 3.9, only versions 3.10 - 3.13 now supported. * - Refactor diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index c6cb2622..2264d160 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -57,6 +57,17 @@ # Contact rateslib at gmail.com if this code is observed outside its intended sphere. +Period = ( + FixedPeriod + | FloatPeriod + | Cashflow + | IndexFixedPeriod + | IndexCashflow + | CreditPremiumPeriod + | CreditProtectionPeriod +) + + class BaseLeg(metaclass=ABCMeta): """ Abstract base class with common parameters for all ``Leg`` subclasses. @@ -288,7 +299,7 @@ def _set_periods(self) -> None: ) @abstractmethod - def _regular_period(self, *args: Any, **kwargs: Any) -> Any: + def _regular_period(self, *args: Any, **kwargs: Any) -> Period: # implemented by individual legs to satify generic `set_periods` methods pass # pragma: no cover @@ -705,7 +716,7 @@ def _spread_isda_approximated_rate( return _ @property - def float_spread(self): + def float_spread(self) -> DualTypes: """ float or NoInput : If set will also set the ``float_spread`` of contained :class:`~rateslib.periods.FloatPeriod` s. @@ -713,7 +724,7 @@ def float_spread(self): return self._float_spread @float_spread.setter - def float_spread(self, value): + def float_spread(self, value: DualTypes) -> None: self._float_spread = value if value is NoInput(0): _ = 0.0 @@ -749,7 +760,7 @@ def float_spread(self, value): # df = pd.concat([df, self.periods[i].fixings_table(curve)]) # return df - def _fixings_table(self, *args, **kwargs): + def _fixings_table(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return a DataFrame of fixing exposures on a :class:`~rateslib.legs.FloatLeg`. @@ -777,7 +788,7 @@ def _regular_period( notional: float, stub: bool, iterator: int, - ): + ) -> FloatPeriod: return FloatPeriod( float_spread=self.float_spread, start=start, @@ -2867,16 +2878,16 @@ class CustomLeg(BaseLeg): """ # noqa: E501 def __init__(self, periods): - if not all(isinstance(p, FixedPeriod | FloatPeriod | Cashflow) for p in periods): + if not all(isinstance(p, Period) for p in periods): raise ValueError( "Each object in `periods` must be of type {FixedPeriod, FloatPeriod, " "Cashflow}.", ) self._set_periods(periods) - def _set_periods(self, periods): - self.periods = periods + def _set_periods(self, periods: list[Any]) -> None: + self.periods: list[Any] = periods - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *CustomLeg* via summing all periods. @@ -2885,7 +2896,7 @@ def npv(self, *args, **kwargs): """ return super().npv(*args, **kwargs) - def cashflows(self, *args, **kwargs): + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *CustomLeg* used in calculating cashflows. @@ -2894,7 +2905,7 @@ def cashflows(self, *args, **kwargs): """ return super().cashflows(*args, **kwargs) - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *CustomLeg* via summing all periods. diff --git a/python/tests/test_legs.py b/python/tests/test_legs.py index 00639527..a5032201 100644 --- a/python/tests/test_legs.py +++ b/python/tests/test_legs.py @@ -10,21 +10,27 @@ from rateslib.dual import Dual from rateslib.fx import FXForwards, FXRates from rateslib.legs import ( - Cashflow, CreditPremiumLeg, CreditProtectionLeg, CustomLeg, FixedLeg, FixedLegMtm, - FixedPeriod, FloatLeg, FloatLegMtm, - FloatPeriod, IndexFixedLeg, ZeroFixedLeg, ZeroFloatLeg, ZeroIndexLeg, ) +from rateslib.periods import ( + Cashflow, + CreditPremiumPeriod, + CreditProtectionPeriod, + FixedPeriod, + FloatPeriod, + IndexCashflow, + IndexFixedPeriod, +) @pytest.fixture @@ -1914,6 +1920,70 @@ def test_mtm_raises_alt(self) -> None: class TestCustomLeg: + @pytest.mark.parametrize( + "period", + [ + FixedPeriod( + start=dt(2022, 1, 1), + end=dt(2023, 1, 1), + payment=dt(2023, 1, 9), + frequency="A", + fixed_rate=1.0, + ), + FloatPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + float_spread=10.0, + ), + CreditPremiumPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + fixed_rate=4.0, + currency="usd", + ), + CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ), + Cashflow(notional=1e9, payment=dt(2022, 4, 3)), + IndexFixedPeriod( + start=dt(2022, 1, 3), + end=dt(2022, 4, 3), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 3), + frequency="Q", + fixed_rate=4.00, + currency="usd", + index_base=100.0, + ), + IndexCashflow( + notional=200.0, + payment=dt(2022, 2, 1), + index_base=100.0, + ), + ], + ) + def test_init(self, curve, period) -> None: + CustomLeg(periods=[period, period]) + def test_npv(self, curve) -> None: cl = CustomLeg( periods=[ From d1544cc2bfe82f44b89115071afd14f2ed298975 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Sat, 4 Jan 2025 14:22:53 +0100 Subject: [PATCH 3/6] TYP: legs.py (#597) --- python/rateslib/legs.py | 138 ++++++++++++++++++++++--------------- python/rateslib/periods.py | 9 +-- 2 files changed, 87 insertions(+), 60 deletions(-) diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index 2264d160..2ef631e2 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -162,7 +162,8 @@ class BaseLeg(metaclass=ABCMeta): CustomLeg : Create a leg composed of user specified periods. """ - _is_mtm = False + _is_mtm: bool = False + periods: list[Period] @abc.abstractmethod def __init__( @@ -561,12 +562,17 @@ class _FloatLegMixin: convention: str schedule: Schedule currency: str - _float_spread: DualTypes + _float_spread: DualTypes | NoInput fixing_method: str + spread_compound_method: str method_param: int + periods: list[Period] + fixings: list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] # type: ignore[type-var] def _get_fixings_from_series( - self, ser: Series[DualTypes], ini_period: int = 0 + self, + ser: Series[DualTypes], # type: ignore[type-var] + ini_period: int = 0, ) -> list[Series[DualTypes] | NoInput]: # type: ignore[type-var] """ Determine which fixings can be set for Periods with the given Series. @@ -594,19 +600,25 @@ def _get_fixings_from_series( def _set_fixings( self, - fixings, - ): + fixings: Series[DualTypes] # type: ignore[type-var] + | list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] + | tuple[DualTypes, Series[DualTypes]] + | DualTypes + | NoInput, + ) -> None: """ Re-organises the fixings input to list structure for each period. Requires a ``schedule`` object and ``float_args``. """ - if fixings is NoInput.blank: - fixings_ = [] + if isinstance(fixings, NoInput): + fixings_: list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] = [] # type: ignore[type-var] elif isinstance(fixings, Series): - fixings_ = fixings.sort_index() # oldest fixing at index 0: latest -1 - fixings_ = self._get_fixings_from_series(fixings_) + # oldest fixing at index 0: latest -1 + sorted_fixings: Series[DualTypes] = fixings.sort_index() # type: ignore[attr-defined, type-var] + fixings_ = self._get_fixings_from_series(sorted_fixings) # type: ignore[assignment] elif isinstance(fixings, tuple): - fixings_ = [fixings[0]] + self._get_fixings_from_series(fixings[1], 1) + fixings_ = [fixings[0]] + fixings_.extend(self._get_fixings_from_series(fixings[1], 1)) elif not isinstance(fixings, list): fixings_ = [fixings] else: # fixings as a list should be remaining @@ -689,14 +701,13 @@ def _spread_isda_approximated_rate( Use approximated derivatives through geometric averaged 1day rates to derive the spread """ - a, b = 0.0, 0.0 - for period in self.periods: - try: - a_, b_ = period._get_analytic_delta_quadratic_coeffs(fore_curve, disc_curve) - a += a_ - b += b_ - except AttributeError: # the period might be of wrong kind: TODO: better filter - pass + a: DualTypes = 0.0 + b: DualTypes = 0.0 + for period in [_ for _ in self.periods if isinstance(_, FloatPeriod)]: + a_, b_ = period._get_analytic_delta_quadratic_coeffs(fore_curve, disc_curve) + a += a_ + b += b_ + c = -target_npv # perform the quadratic solution @@ -716,7 +727,7 @@ def _spread_isda_approximated_rate( return _ @property - def float_spread(self) -> DualTypes: + def float_spread(self) -> DualTypes | NoInput: """ float or NoInput : If set will also set the ``float_spread`` of contained :class:`~rateslib.periods.FloatPeriod` s. @@ -826,7 +837,13 @@ def _is_linear(self) -> bool: return False return True - def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): + def _spread( + self, + target_npv: DualTypes, + fore_curve: Curve, + disc_curve: Curve, + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), + ) -> DualTypes: """ Calculates an adjustment to the ``fixed_rate`` or ``float_spread`` to match a specific target NPV. @@ -864,12 +881,11 @@ def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): -------- """ if self._is_linear: - a_delta = self.analytic_delta(fore_curve, disc_curve, fx, self.currency) + a_delta: DualTypes = self.analytic_delta(fore_curve, disc_curve, fx, self.currency) # type: ignore[attr-defined] return -target_npv / a_delta else: - _ = self._spread_isda_approximated_rate(target_npv, fore_curve, disc_curve) + return self._spread_isda_approximated_rate(target_npv, fore_curve, disc_curve) # _ = self._spread_isda_dual2(target_npv, fore_curve, disc_curve, fx) - return _ class FloatLeg(_FloatLegMixin, BaseLeg): @@ -1007,14 +1023,18 @@ class FloatLeg(_FloatLegMixin, BaseLeg): def __init__( self, - *args, - float_spread: float | NoInput = NoInput(0), - fixings: float | list | Series | tuple | NoInput = NoInput(0), + *args: Any, + float_spread: DualTypes | NoInput = NoInput(0), + fixings: Series[DualTypes] # type: ignore[type-var] + | list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] + | tuple[DualTypes, Series[DualTypes]] + | DualTypes + | NoInput = NoInput(0), fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: self._float_spread = float_spread ( self.fixing_method, @@ -1027,7 +1047,7 @@ def __init__( self._set_fixings(fixings) self._set_periods() - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *FloatLeg* via summing all periods. @@ -1036,7 +1056,7 @@ def analytic_delta(self, *args, **kwargs): """ return super().analytic_delta(*args, **kwargs) - def cashflows(self, *args, **kwargs) -> DataFrame: + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *FloatLeg* used in calculating cashflows. @@ -1045,7 +1065,7 @@ def cashflows(self, *args, **kwargs) -> DataFrame: """ return super().cashflows(*args, **kwargs) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *FloatLeg* via summing all periods. @@ -1058,7 +1078,7 @@ def fixings_table( self, curve: Curve, disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), approximate: bool = False, right: datetime | NoInput = NoInput(0), @@ -1127,10 +1147,12 @@ def _set_periods(self) -> None: class _IndexLegMixin: - schedule = None - index_method = None + schedule: Schedule + index_method: str _index_fixings = None - _index_base = None + _index_base: DualTypes | NoInput = NoInput(0) + periods: list[Period] + index_lag: int # def _set_index_fixings_on_periods(self): # """ @@ -1193,7 +1215,7 @@ def index_fixings(self, value): period.index_fixings = _ @property - def index_base(self): + def index_base(self) -> DualTypes | NoInput: return self._index_base @index_base.setter @@ -1293,14 +1315,18 @@ class ZeroFloatLeg(_FloatLegMixin, BaseLeg): def __init__( self, - *args, - float_spread: float | NoInput = NoInput(0), - fixings: float | list | Series | NoInput = NoInput(0), + *args: Any, + float_spread: DualTypes | NoInput = NoInput(0), + fixings: Series[DualTypes] # type: ignore[type-var] + | list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] + | tuple[DualTypes, Series[DualTypes]] + | DualTypes + | NoInput = NoInput(0), fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: self._float_spread = float_spread ( self.fixing_method, @@ -1321,8 +1347,8 @@ def __init__( self._set_fixings(fixings) self._set_periods() - def _set_periods(self): - self.periods = [ + def _set_periods(self) -> None: + self.periods: list[FloatPeriod] = [ # type: ignore[assignment] FloatPeriod( float_spread=self.float_spread, start=period[defaults.headers["a_acc_start"]], @@ -1341,15 +1367,15 @@ def _set_periods(self): method_param=self.method_param, spread_compound_method=self.spread_compound_method, ) - for i, period in self.schedule.table.to_dict(orient="index").items() + for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) ] @property - def dcf(self): + def dcf(self) -> float: _ = [period.dcf for period in self.periods] return sum(_) - def rate(self, curve): + def rate(self, curve: Curve | NoInput) -> DualTypes: """ Calculate a simple period type floating rate for the zero coupon leg. @@ -1362,16 +1388,16 @@ def rate(self, curve): ------- float, Dual, Dual2 """ - compounded_rate = 1.0 + compounded_rate: DualTypes = 1.0 for period in self.periods: - compounded_rate *= 1 + period.dcf * period.rate(curve) / 100 + compounded_rate *= 1.0 + period.dcf * period.rate(curve) / 100 return 100 * (compounded_rate - 1.0) / self.dcf def npv( self, curve: Curve, disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), local: bool = False, ) -> dict[str, DualTypes] | DualTypes: @@ -1399,7 +1425,7 @@ def fixings_table( self, curve: Curve, disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), approximate: bool = False, right: datetime | NoInput = NoInput(0), @@ -1464,9 +1490,9 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ): + ) -> DualTypes: """ Return the analytic delta of the *ZeroFloatLeg* from all periods. @@ -1491,7 +1517,7 @@ def cashflows( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ): """ @@ -1824,7 +1850,7 @@ class ZeroIndexLeg(_IndexLegMixin, BaseLeg): def __init__( self, *args, - index_base: float | Series | NoInput = NoInput(0), + index_base: DualTypes | Series[DualTypes] | NoInput = NoInput(0), index_fixings: float | Series | NoInput = NoInput(0), index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), @@ -2800,8 +2826,8 @@ class FloatLegMtm(_FloatLegMixin, BaseLegMtm): def __init__( self, *args, - float_spread: float | NoInput = NoInput(0), - fixings: float | list | NoInput = NoInput(0), + float_spread: DualTypes | NoInput = NoInput(0), + fixings: DualTypes | list[DualTypes] | NoInput = NoInput(0), fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index c3d903de..c886f031 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -900,7 +900,7 @@ def __init__( self, *args: Any, float_spread: DualTypes | NoInput = NoInput(0), - fixings: float | list[float] | Series[float] | NoInput = NoInput(0), + fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), @@ -1121,7 +1121,8 @@ def _rate_ibor(self, curve: Curve | dict[str, Curve] | NoInput) -> DualTypes: cal_, _ = self._maybe_get_cal_and_conv_from_curve(curve) fixing_date = cal_.lag(self.start, -self.method_param, False) try: - return self.fixings[fixing_date] + self.float_spread / 100 + fixing: DualTypes = self.fixings[fixing_date] + self.float_spread / 100 # type: ignore[index] + return fixing except KeyError: warnings.warn( "A FloatPeriod `fixing date` was not found in the given `fixings` Series.\n" @@ -1361,13 +1362,13 @@ def _rfr_get_series_with_populated_fixings( if isinstance(self.fixings, list): rates.iloc[: len(self.fixings)] = self.fixings elif isinstance(self.fixings, Series): - if not self.fixings.index.is_monotonic_increasing: + if not self.fixings.index.is_monotonic_increasing: # type: ignore[attr-defined] raise ValueError( "`fixings` as a Series must have a monotonically increasing " "datetimeindex.", ) # [-2] is used because the last rfr fixing is 1 day before the end - fixing_rates = self.fixings.loc[obs_dates.iloc[0] : obs_dates.iloc[-2]] # type: ignore[misc] + fixing_rates = self.fixings.loc[obs_dates.iloc[0] : obs_dates.iloc[-2]] # type: ignore[attr-defined, misc] try: rates.loc[fixing_rates.index] = fixing_rates From 54e94333f5646ef72bf245a1013510d42d461798 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Sat, 4 Jan 2025 20:23:34 +0100 Subject: [PATCH 4/6] TYP: legs.py (#598) --- python/rateslib/legs.py | 195 ++++++++++++++++++++----------------- python/rateslib/periods.py | 12 +-- 2 files changed, 110 insertions(+), 97 deletions(-) diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index 2ef631e2..a53046df 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -25,7 +25,7 @@ import warnings from abc import ABCMeta, abstractmethod from datetime import datetime -from typing import Any, NoReturn +from typing import Any import pandas as pd from pandas import DataFrame, Series @@ -34,7 +34,7 @@ from rateslib.calendars import CalInput, add_tenor from rateslib.curves import Curve, index_left from rateslib.default import NoInput, _drb -from rateslib.dual import Dual, Dual2, DualTypes +from rateslib.dual import Dual, Dual2, DualTypes, _dual_float from rateslib.fx import FXForwards, FXRates from rateslib.periods import ( Cashflow, @@ -301,7 +301,7 @@ def _set_periods(self) -> None: @abstractmethod def _regular_period(self, *args: Any, **kwargs: Any) -> Period: - # implemented by individual legs to satify generic `set_periods` methods + # implemented by individual legs to satisfy generic `set_periods` methods pass # pragma: no cover # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International @@ -1149,7 +1149,7 @@ def _set_periods(self) -> None: class _IndexLegMixin: schedule: Schedule index_method: str - _index_fixings = None + _index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0) # type: ignore[type-var] _index_base: DualTypes | NoInput = NoInput(0) periods: list[Period] index_lag: int @@ -1184,11 +1184,13 @@ class _IndexLegMixin: # # TODO index_fixings as a list cannot handle amortization. Use a Series. @property - def index_fixings(self): + def index_fixings(self) -> DualTypes | list[DualTypes] | Series[DualTypes] | NoInput: # type: ignore[type-var] return self._index_fixings @index_fixings.setter - def index_fixings(self, value): + def index_fixings( + self, value: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput # type: ignore[type-var] + ) -> None: self._index_fixings = value for i, period in enumerate(self.periods): if isinstance(period, IndexFixedPeriod | IndexCashflow): @@ -1219,7 +1221,7 @@ def index_base(self) -> DualTypes | NoInput: return self._index_base @index_base.setter - def index_base(self, value: DualTypes | Series[DualTypes] | NoInput) -> None: + def index_base(self, value: DualTypes | Series[DualTypes] | NoInput) -> None: # type: ignore[type-var] if isinstance(value, Series): _: DualTypes | None = IndexMixin._index_value( i_fixings=value, @@ -1240,8 +1242,8 @@ def index_base(self, value: DualTypes | Series[DualTypes] | NoInput) -> None: if isinstance(period, IndexFixedPeriod | IndexCashflow): period.index_base = self._index_base - def _regular_period(self) -> None: - pass + def _regular_period(self, *args: Any, **kwargs: Any) -> Period: # type: ignore[empty-body] + pass # pragma: no cover class ZeroFloatLeg(_FloatLegMixin, BaseLeg): @@ -1342,7 +1344,7 @@ def __init__( "construction. Set the `frequency` equal to the compounding frequency of the " "expressed fixed rate, e.g. 'S' for semi-annual compounding.", ) - if abs(float(self.amortization)) > 1e-8: + if abs(_dual_float(self.amortization)) > 1e-8: raise ValueError("`ZeroFloatLeg` cannot be defined with `amortization`.") self._set_fixings(fixings) self._set_periods() @@ -1429,7 +1431,7 @@ def fixings_table( base: str | NoInput = NoInput(0), approximate: bool = False, right: datetime | NoInput = NoInput(0), - ) -> NoReturn: # pragma: no cover + ) -> DataFrame: """ Return a DataFrame of fixing exposures on a :class:`~rateslib.legs.ZeroFloatLeg`. @@ -1453,9 +1455,9 @@ def fixings_table( ------- DataFrame """ - if disc_curve is NoInput.blank and isinstance(curve, dict): + if isinstance(disc_curve, NoInput) and isinstance(curve, dict): raise ValueError("Cannot infer `disc_curve` from a dict of curves.") - elif disc_curve is NoInput.blank: + elif isinstance(disc_curve, NoInput): if curve._base_type == "dfs": disc_curve = curve else: @@ -1477,8 +1479,8 @@ def fixings_table( for period in self.periods: df = period.fixings_table(curve, approximate, disc_curve) scalar = prod / (1 + period.dcf * period.rate(curve) / 100.0) - df[(curve.id, "risk")] *= scalar - df[(curve.id, "notional")] *= scalar + df[(curve.id, "risk")] *= scalar # type: ignore[operator] + df[(curve.id, "notional")] *= scalar # type: ignore[operator] dfs.append(df) with warnings.catch_warnings(): @@ -1499,18 +1501,18 @@ def analytic_delta( For arguments see :meth:`BasePeriod.analytic_delta()`. """ - disc_curve_: Curve | NoInput = _disc_maybe_from_curve(curve, disc_curve) - fx, base = _get_fx_and_base(self.currency, fx, base) - compounded_rate = 1.0 + disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve) + fx_, base = _get_fx_and_base(self.currency, fx, base) + compounded_rate: DualTypes = 1.0 for period in self.periods: compounded_rate *= 1 + period.dcf * period.rate(curve) / 100 - a_sum = 0.0 + a_sum: DualTypes = 0.0 for period in self.periods: - _ = period.analytic_delta(curve, disc_curve_, fx, base) / disc_curve_[period.payment] + _ = period.analytic_delta(curve, disc_curve_, fx_, base) / disc_curve_[period.payment] _ *= compounded_rate / (1 + period.dcf * period.rate(curve) / 100) a_sum += _ - a_sum *= disc_curve_[self.schedule.pschedule[-1]] * fx + a_sum *= disc_curve_[self.schedule.pschedule[-1]] * fx_ return a_sum def cashflows( @@ -1519,7 +1521,7 @@ def cashflows( disc_curve: Curve | NoInput = NoInput(0), fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ): + ) -> DataFrame: """ Return the properties of the *ZeroFloatLeg* used in calculating cashflows. @@ -1529,19 +1531,21 @@ def cashflows( disc_curve_: Curve | NoInput = _disc_maybe_from_curve(curve, disc_curve) fx, base = _get_fx_and_base(self.currency, fx, base) - if curve is NoInput.blank: + if isinstance(curve, NoInput): rate, cashflow = None, None - if disc_curve_ is NoInput.blank: - npv, npv_fx, df, collateral = None, None, None, None + npv, npv_fx, df, collateral = None, None, None, None else: - rate = float(self.rate(curve)) - cashflow = -float(self.notional * self.dcf * rate / 100) - npv = float(self.npv(curve, disc_curve_)) - npv_fx = npv * float(fx) - df = float(disc_curve_[self.schedule.pschedule[-1]]) - collateral = disc_curve_.collateral + rate = _dual_float(self.rate(curve)) + cashflow = -_dual_float(self.notional * self.dcf * rate / 100) + if not isinstance(disc_curve_, NoInput): + npv = _dual_float(self.npv(curve, disc_curve_)) # type: ignore[arg-type] + npv_fx = npv * _dual_float(fx) + df = _dual_float(disc_curve_[self.schedule.pschedule[-1]]) + collateral = disc_curve_.collateral + else: + npv, npv_fx, df, collateral = None, None, None, None - spread = 0.0 if self.float_spread is NoInput.blank else float(self.float_spread) + spread = 0.0 if isinstance(self.float_spread, NoInput) else _dual_float(self.float_spread) seq = [ { defaults.headers["type"]: type(self).__name__, @@ -1552,13 +1556,13 @@ def cashflows( defaults.headers["payment"]: self.schedule.pschedule[-1], defaults.headers["convention"]: self.convention, defaults.headers["dcf"]: self.dcf, - defaults.headers["notional"]: float(self.notional), + defaults.headers["notional"]: _dual_float(self.notional), defaults.headers["df"]: df, defaults.headers["rate"]: rate, defaults.headers["spread"]: spread, defaults.headers["cashflow"]: cashflow, defaults.headers["npv"]: npv, - defaults.headers["fx"]: float(fx), + defaults.headers["fx"]: _dual_float(fx), defaults.headers["npv_fx"]: npv_fx, defaults.headers["collateral"]: collateral, }, @@ -1618,8 +1622,11 @@ class ZeroFixedLeg(_FixedLegMixin, BaseLeg): zfl.cashflows(curve) """ + periods: list[FixedPeriod] # type: ignore[assignment] - def __init__(self, *args, fixed_rate: float | NoInput = NoInput(0), **kwargs): + def __init__( + self, *args: Any, fixed_rate: DualTypes | NoInput = NoInput(0), **kwargs: Any + ) -> None: super().__init__(*args, **kwargs) self.fixed_rate = fixed_rate if self.schedule.frequency == "Z": @@ -1628,10 +1635,10 @@ def __init__(self, *args, fixed_rate: float | NoInput = NoInput(0), **kwargs): "construction. Set the `frequency` equal to the compounding frequency of the " "expressed fixed rate, e.g. 'S' for semi-annual compounding.", ) - if abs(float(self.amortization)) > 1e-8: + if abs(_dual_float(self.amortization)) > 1e-8: raise ValueError("`ZeroFixedLeg` cannot be defined with `amortization`.") - def _set_periods(self): + def _set_periods(self) -> None: self.periods = [ FixedPeriod( fixed_rate=NoInput(0), @@ -1650,7 +1657,7 @@ def _set_periods(self): ] @property - def fixed_rate(self): + def fixed_rate(self) -> DualTypes | NoInput: """ float or None : If set will also set the ``fixed_rate`` of contained :class:`FixedPeriod` s. @@ -1658,13 +1665,13 @@ def fixed_rate(self): return self._fixed_rate @fixed_rate.setter - def fixed_rate(self, value): + def fixed_rate(self, value: DualTypes | NoInput) -> None: # overload the setter for a zero coupon to convert from IRR to period rate. # the headline fixed_rate is the IRR rate but the rate attached to Periods is a simple # rate in order to determine cashflows according to the normal cashflow logic. self._fixed_rate = value f = 12 / defaults.frequency_months[self.schedule.frequency] - if value is not NoInput.blank: + if not isinstance(value, NoInput): period_rate = 100 * (1 / self.dcf) * ((1 + value / (100 * f)) ** (self.dcf * f) - 1) else: period_rate = NoInput(0) @@ -1674,7 +1681,7 @@ def fixed_rate(self, value): period.fixed_rate = period_rate @property - def dcf(self): + def dcf(self) -> float: """ The DCF of a *ZeroFixedLeg* is defined as DCF of the single *FixedPeriod* spanning the *Leg*. @@ -1686,9 +1693,9 @@ def cashflows( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ): + ) -> DataFrame: """ Return the cashflows of the *ZeroFixedLeg* from all periods. @@ -1696,16 +1703,16 @@ def cashflows( :meth:`BasePeriod.cashflows()`. """ disc_curve_: Curve | NoInput = _disc_maybe_from_curve(curve, disc_curve) - fx, base = _get_fx_and_base(self.currency, fx, base) + fx_, base = _get_fx_and_base(self.currency, fx, base) rate = self.fixed_rate cashflow = self.periods[0].cashflow - if disc_curve is NoInput.blank or rate is NoInput.blank: + if isinstance(disc_curve_, NoInput) or isinstance(rate, NoInput): npv, npv_fx, df, collateral = None, None, None, None else: - npv = float(self.npv(curve, disc_curve_)) - npv_fx = npv * float(fx) - df = float(disc_curve_[self.schedule.pschedule[-1]]) + npv = _dual_float(self.npv(curve, disc_curve_)) # type: ignore[arg-type] + npv_fx = npv * _dual_float(fx_) + df = _dual_float(disc_curve_[self.schedule.pschedule[-1]]) collateral = disc_curve_.collateral seq = [ @@ -1718,13 +1725,13 @@ def cashflows( defaults.headers["payment"]: self.schedule.pschedule[-1], defaults.headers["convention"]: self.convention, defaults.headers["dcf"]: self.dcf, - defaults.headers["notional"]: float(self.notional), + defaults.headers["notional"]: _dual_float(self.notional), defaults.headers["df"]: df, defaults.headers["rate"]: self.fixed_rate, defaults.headers["spread"]: None, defaults.headers["cashflow"]: cashflow, defaults.headers["npv"]: npv, - defaults.headers["fx"]: float(fx), + defaults.headers["fx"]: _dual_float(fx_), defaults.headers["npv_fx"]: npv_fx, defaults.headers["collateral"]: collateral, }, @@ -1735,7 +1742,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ @@ -1744,34 +1751,40 @@ def analytic_delta( For arguments see :meth:`BasePeriod.analytic_delta()`. """ - disc_curve_: Curve | NoInput = _disc_maybe_from_curve(curve, disc_curve) + disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve) fx, base = _get_fx_and_base(self.currency, fx, base) - if self.fixed_rate is NoInput.blank: - return None + if isinstance(self.fixed_rate, NoInput): + raise ValueError("Must have `fixed_rate` on ZeroFixedLeg for analytic delta.") f = 12 / defaults.frequency_months[self.schedule.frequency] - _ = self.notional * self.dcf * disc_curve_[self.periods[0].payment] + _: DualTypes = self.notional * self.dcf * disc_curve_[self.periods[0].payment] _ *= (1 + self.fixed_rate / (100 * f)) ** (self.dcf * f - 1) return _ / 10000 * fx - def _analytic_delta(self, *args, **kwargs) -> DualTypes: + def _analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Analytic delta based on period rate and not IRR. """ _ = [period.analytic_delta(*args, **kwargs) for period in self.periods] return sum(_) - def _spread(self, target_npv, fore_curve, disc_curve, fx=NoInput(0)): + def _spread( + self, + target_npv: DualTypes, + fore_curve: Curve, + disc_curve: Curve, + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0) + ) -> DualTypes: """ Overload the _spread calc to use analytic delta based on period rate """ a_delta = self._analytic_delta(fore_curve, disc_curve, fx, self.currency) period_rate = -target_npv / (a_delta * 100) f = 12 / defaults.frequency_months[self.schedule.frequency] - _ = f * ((1 + period_rate * self.dcf / 100) ** (1 / (self.dcf * f)) - 1) + _: DualTypes = f * ((1 + period_rate * self.dcf / 100) ** (1 / (self.dcf * f)) - 1) return _ * 10000 - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *ZeroFixedLeg* via summing all periods. @@ -1849,13 +1862,13 @@ class ZeroIndexLeg(_IndexLegMixin, BaseLeg): def __init__( self, - *args, + *args: Any, index_base: DualTypes | Series[DualTypes] | NoInput = NoInput(0), index_fixings: float | Series | NoInput = NoInput(0), index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: self.index_method = ( defaults.index_method if index_method is NoInput.blank else index_method.lower() ) @@ -1864,7 +1877,7 @@ def __init__( self.index_fixings = index_fixings # set index fixings after periods init self.index_base = index_base # set after periods initialised - def _set_periods(self): + def _set_periods(self) -> None: self.periods = [ IndexFixedPeriod( fixed_rate=100.0, @@ -1893,13 +1906,13 @@ def _set_periods(self): ), ] - def cashflow(self, curve: Curve | None = None): + def cashflow(self, curve: Curve | NoInput = NoInput(0)) -> DualTypes: """Aggregate the cashflows on the *IndexFixedPeriod* and *Cashflow* period using a *Curve*.""" - _ = self.periods[0].cashflow(curve) + self.periods[1].cashflow + _: DualTypes = self.periods[0].cashflow(curve) + self.periods[1].cashflow return _ - def cashflows(self, *args, **kwargs): + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *ZeroIndexLeg* used in calculating cashflows. @@ -1914,7 +1927,7 @@ def cashflows(self, *args, **kwargs): _["Period"] = None return _ - def analytic_delta(self, *args, **kwargs) -> float: + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *ZeroIndexLeg* via summing all periods. @@ -1923,7 +1936,7 @@ def analytic_delta(self, *args, **kwargs) -> float: """ return 0.0 - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *ZeroIndexLeg* via summing all periods. @@ -2451,17 +2464,17 @@ class BaseLegMtm(BaseLeg, metaclass=ABCMeta): FloatLegExchangeMtm : Create a floating leg with notional and Mtm exchanges. """ - _do_not_repeat_set_periods = False - _is_mtm = True + _do_not_repeat_set_periods: bool = False + _is_mtm: bool = True def __init__( self, - *args, + *args: Any, fx_fixings: list | float | Dual | Dual2 | NoInput = NoInput(0), alt_currency: str | NoInput = NoInput(0), alt_notional: float | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: if alt_currency is NoInput.blank: raise ValueError("`alt_currency` and `currency` must be supplied for MtmLeg.") self.alt_currency = alt_currency.lower() @@ -2476,11 +2489,11 @@ def __init__( self.fx_fixings = fx_fixings # calls the setter @property - def notional(self): + def notional(self) -> DualTypes: return self._notional @notional.setter - def notional(self, value): + def notional(self, value: DualTypes): self._notional = value @property @@ -2626,7 +2639,7 @@ def _set_periods(self, fx): notional=notionals[i], iterator=i, ) - for i, period in self.schedule.table.to_dict(orient="index").items() + for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) ] mtm_flows = [ Cashflow( @@ -2670,7 +2683,7 @@ def npv( fx: float | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), local: bool = False, - ): + ) -> DualTypes | dict[str, DualTypes]: if not self._do_not_repeat_set_periods: self._set_periods(fx) ret = super().npv(curve, disc_curve, fx, base, local) @@ -2681,9 +2694,9 @@ def cashflows( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ): + ) -> DataFrame: if not self._do_not_repeat_set_periods: self._set_periods(fx) ret = super().cashflows(curve, disc_curve, fx, base) @@ -2694,9 +2707,9 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ): + ) -> DualTypes: if not self._do_not_repeat_set_periods: self._set_periods(fx) ret = super().analytic_delta(curve, disc_curve, fx, base) @@ -2751,10 +2764,10 @@ class FixedLegMtm(_FixedLegMixin, BaseLegMtm): def __init__( self, - *args, - fixed_rate: float | NoInput = NoInput(0), - **kwargs, - ): + *args: Any, + fixed_rate: DualTypes | NoInput = NoInput(0), + **kwargs: Any, + ) -> None: self._fixed_rate = fixed_rate super().__init__( *args, @@ -2825,14 +2838,14 @@ class FloatLegMtm(_FloatLegMixin, BaseLegMtm): def __init__( self, - *args, + *args: Any, float_spread: DualTypes | NoInput = NoInput(0), fixings: DualTypes | list[DualTypes] | NoInput = NoInput(0), fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: self._float_spread = float_spread ( self.fixing_method, @@ -2852,11 +2865,11 @@ def fixings_table( self, curve: Curve, disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), approximate: bool = False, right: datetime | NoInput = NoInput(0), - ): + ) -> DataFrame: """ Return a DataFrame of fixing exposures on a :class:`~rateslib.legs.FloatLegMtm`. @@ -2910,7 +2923,7 @@ def __init__(self, periods): ) self._set_periods(periods) - def _set_periods(self, periods: list[Any]) -> None: + def _set_periods(self, periods: list[Period]) -> None: self.periods: list[Any] = periods def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index c886f031..3fc1cc38 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -155,7 +155,7 @@ def _maybe_local( def _maybe_fx_converted( value: DualTypes, currency: str, - fx: float | FXRates | FXForwards | NoInput, + fx: DualTypes | FXRates | FXForwards | NoInput, base: str | NoInput, ) -> DualTypes: fx_, _ = _get_fx_and_base(currency, fx, base) @@ -289,7 +289,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ @@ -930,7 +930,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ @@ -2362,7 +2362,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ @@ -2547,7 +2547,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ @@ -3074,7 +3074,7 @@ def analytic_delta( self, curve: Curve | NoInput = NoInput(0), disc_curve: Curve | NoInput = NoInput(0), - fx: float | FXRates | FXForwards | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), ) -> DualTypes: """ From e18239ea3e7a4001643583828f9b3dd9b17a3861 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Mon, 6 Jan 2025 13:40:06 +0100 Subject: [PATCH 5/6] TYP: legs.py (#599) Co-authored-by: JHM Darbyshire (M1) --- python/rateslib/legs.py | 335 ++++++++++++++++++++++--------------- python/rateslib/periods.py | 24 +-- python/tests/test_legs.py | 19 ++- 3 files changed, 227 insertions(+), 151 deletions(-) diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index a53046df..8d127819 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -25,6 +25,7 @@ import warnings from abc import ABCMeta, abstractmethod from datetime import datetime +from math import prod from typing import Any import pandas as pd @@ -34,7 +35,7 @@ from rateslib.calendars import CalInput, add_tenor from rateslib.curves import Curve, index_left from rateslib.default import NoInput, _drb -from rateslib.dual import Dual, Dual2, DualTypes, _dual_float +from rateslib.dual import Dual, Dual2, DualTypes, Variable, _dual_float from rateslib.fx import FXForwards, FXRates from rateslib.periods import ( Cashflow, @@ -180,9 +181,9 @@ def __init__( modifier: str | NoInput = NoInput(0), calendar: CalInput = NoInput(0), payment_lag: int | NoInput = NoInput(0), - notional: float | NoInput = NoInput(0), + notional: DualTypes | NoInput = NoInput(0), currency: str | NoInput = NoInput(0), - amortization: float | NoInput = NoInput(0), + amortization: DualTypes | NoInput = NoInput(0), convention: str | NoInput = NoInput(0), payment_lag_exchange: int | NoInput = NoInput(0), initial_exchange: bool = False, @@ -206,7 +207,7 @@ def __init__( self.payment_lag_exchange: int = _drb(defaults.payment_lag_exchange, payment_lag_exchange) self.initial_exchange: bool = initial_exchange self.final_exchange: bool = final_exchange - self._notional: float = _drb(defaults.notional, notional) + self._notional: DualTypes = _drb(defaults.notional, notional) self._amortization: float = _drb(0.0, amortization) if getattr(self, "_delay_set_periods", False): pass @@ -214,11 +215,11 @@ def __init__( self._set_periods() @property - def notional(self) -> float: + def notional(self) -> DualTypes: return self._notional @notional.setter - def notional(self, value: float) -> None: + def notional(self, value: DualTypes) -> None: self._notional = value self._set_periods() @@ -300,7 +301,15 @@ def _set_periods(self) -> None: ) @abstractmethod - def _regular_period(self, *args: Any, **kwargs: Any) -> Period: + def _regular_period( + self, + start: datetime, + end: datetime, + payment: datetime, + stub: bool, + notional: DualTypes, + iterator: int, + ) -> Period: # implemented by individual legs to satisfy generic `set_periods` methods pass # pragma: no cover @@ -443,7 +452,7 @@ def _regular_period( start: datetime, end: datetime, payment: datetime, - notional: float, + notional: DualTypes, stub: bool, iterator: int, ) -> FixedPeriod: @@ -463,7 +472,7 @@ def _regular_period( ) -class FixedLeg(_FixedLegMixin, BaseLeg): +class FixedLeg(_FixedLegMixin, BaseLeg): # type: ignore[misc] """ Create a fixed leg composed of :class:`~rateslib.periods.FixedPeriod` s. @@ -796,8 +805,8 @@ def _regular_period( start: datetime, end: datetime, payment: datetime, - notional: float, stub: bool, + notional: DualTypes, iterator: int, ) -> FloatPeriod: return FloatPeriod( @@ -1021,6 +1030,8 @@ class FloatLeg(_FloatLegMixin, BaseLeg): float_leg.fixings_table(swestr_curve)[dt(2022,12,28):dt(2023,1,4)] """ # noqa: E501 + _delay_set_periods: bool = True # do this to set fixings first + def __init__( self, *args: Any, @@ -1042,7 +1053,6 @@ def __init__( self.spread_compound_method, ) = _validate_float_args(fixing_method, method_param, spread_compound_method) - self._delay_set_periods = True # do this to set fixings first super().__init__(*args, **kwargs) self._set_fixings(fixings) self._set_periods() @@ -1111,7 +1121,7 @@ def fixings_table( ) def _set_periods(self) -> None: - return super()._set_periods() + return super(_FloatLegMixin, self)._set_periods() # @property # def _is_complex(self): @@ -1189,7 +1199,8 @@ def index_fixings(self) -> DualTypes | list[DualTypes] | Series[DualTypes] | NoI @index_fixings.setter def index_fixings( - self, value: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput # type: ignore[type-var] + self, + value: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput, # type: ignore[type-var] ) -> None: self._index_fixings = value for i, period in enumerate(self.periods): @@ -1242,8 +1253,8 @@ def index_base(self, value: DualTypes | Series[DualTypes] | NoInput) -> None: # if isinstance(period, IndexFixedPeriod | IndexCashflow): period.index_base = self._index_base - def _regular_period(self, *args: Any, **kwargs: Any) -> Period: # type: ignore[empty-body] - pass # pragma: no cover + # def _regular_period(self, *args: Any, **kwargs: Any) -> Period: # type: ignore[empty-body] + # pass # pragma: no cover class ZeroFloatLeg(_FloatLegMixin, BaseLeg): @@ -1315,6 +1326,8 @@ class ZeroFloatLeg(_FloatLegMixin, BaseLeg): zfl.cashflows(curve) """ # noqa: E501 + _delay_set_periods: bool = True + def __init__( self, *args: Any, @@ -1336,7 +1349,6 @@ def __init__( self.spread_compound_method, ) = _validate_float_args(fixing_method, method_param, spread_compound_method) - self._delay_set_periods = True super().__init__(*args, **kwargs) if self.schedule.frequency == "Z": raise ValueError( @@ -1346,35 +1358,35 @@ def __init__( ) if abs(_dual_float(self.amortization)) > 1e-8: raise ValueError("`ZeroFloatLeg` cannot be defined with `amortization`.") + if self.initial_exchange or self.final_exchange: + raise ValueError("`initial_exchange` or `final_exchange` not allowed on ZeroFloatLeg.") self._set_fixings(fixings) self._set_periods() + def _regular_period( + self, + start: datetime, + end: datetime, + payment: datetime, + stub: bool, + notional: DualTypes, + iterator: int, + ) -> FloatPeriod: + return super()._regular_period( + start=start, + end=end, + payment=self.schedule.pschedule[-1], + notional=notional, + stub=stub, + iterator=iterator, + ) + def _set_periods(self) -> None: - self.periods: list[FloatPeriod] = [ # type: ignore[assignment] - FloatPeriod( - float_spread=self.float_spread, - start=period[defaults.headers["a_acc_start"]], - end=period[defaults.headers["a_acc_end"]], - payment=self.schedule.pschedule[-1], # set payment to Leg payment - notional=self.notional, - currency=self.currency, - convention=self.convention, - termination=self.schedule.termination, - frequency=self.schedule.frequency, - stub=period[defaults.headers["stub_type"]] == "Stub", - roll=self.schedule.roll, - calendar=self.schedule.calendar, - fixing_method=self.fixing_method, - fixings=self.fixings[i], - method_param=self.method_param, - spread_compound_method=self.spread_compound_method, - ) - for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) - ] + return super(_FloatLegMixin, self)._set_periods() @property def dcf(self) -> float: - _ = [period.dcf for period in self.periods] + _ = [period.dcf for period in self.periods if isinstance(period, FloatPeriod)] return sum(_) def rate(self, curve: Curve | NoInput) -> DualTypes: @@ -1390,9 +1402,10 @@ def rate(self, curve: Curve | NoInput) -> DualTypes: ------- float, Dual, Dual2 """ - compounded_rate: DualTypes = 1.0 - for period in self.periods: - compounded_rate *= 1.0 + period.dcf * period.rate(curve) / 100 + rates = ( + (1.0 + p.dcf * p.rate(curve) / 100) for p in self.periods if isinstance(p, FloatPeriod) + ) + compounded_rate: DualTypes = prod(rates) # type: ignore[arg-type] return 100 * (compounded_rate - 1.0) / self.dcf def npv( @@ -1476,7 +1489,7 @@ def fixings_table( else: dfs = [] prod = 1 + self.dcf * self.rate(curve) / 100.0 - for period in self.periods: + for period in [_ for _ in self.periods if isinstance(_, FloatPeriod)]: df = period.fixings_table(curve, approximate, disc_curve) scalar = prod / (1 + period.dcf * period.rate(curve) / 100.0) df[(curve.id, "risk")] *= scalar # type: ignore[operator] @@ -1503,12 +1516,13 @@ def analytic_delta( """ disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve) fx_, base = _get_fx_and_base(self.currency, fx, base) - compounded_rate: DualTypes = 1.0 - for period in self.periods: - compounded_rate *= 1 + period.dcf * period.rate(curve) / 100 + + float_periods: list[FloatPeriod] = [_ for _ in self.periods if isinstance(_, FloatPeriod)] + rates = ((1 + p.dcf * p.rate(curve) / 100) for p in float_periods) + compounded_rate: DualTypes = prod(rates) # type: ignore[arg-type] a_sum: DualTypes = 0.0 - for period in self.periods: + for period in float_periods: _ = period.analytic_delta(curve, disc_curve_, fx_, base) / disc_curve_[period.payment] _ *= compounded_rate / (1 + period.dcf * period.rate(curve) / 100) a_sum += _ @@ -1570,7 +1584,7 @@ def cashflows( return DataFrame.from_records(seq) -class ZeroFixedLeg(_FixedLegMixin, BaseLeg): +class ZeroFixedLeg(_FixedLegMixin, BaseLeg): # type: ignore[misc] """ Create a zero coupon fixed leg composed of a single :class:`~rateslib.periods.FixedPeriod` . @@ -1622,6 +1636,7 @@ class ZeroFixedLeg(_FixedLegMixin, BaseLeg): zfl.cashflows(curve) """ + periods: list[FixedPeriod] # type: ignore[assignment] def __init__( @@ -1773,7 +1788,7 @@ def _spread( target_npv: DualTypes, fore_curve: Curve, disc_curve: Curve, - fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0) + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), ) -> DualTypes: """ Overload the _spread calc to use analytic delta based on period rate @@ -1860,22 +1875,35 @@ class ZeroIndexLeg(_IndexLegMixin, BaseLeg): """ + periods: list[IndexFixedPeriod | Cashflow] # type: ignore[assignment] + def __init__( self, *args: Any, - index_base: DualTypes | Series[DualTypes] | NoInput = NoInput(0), - index_fixings: float | Series | NoInput = NoInput(0), + index_base: DualTypes | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] + index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), **kwargs: Any, ) -> None: - self.index_method = ( - defaults.index_method if index_method is NoInput.blank else index_method.lower() - ) - self.index_lag = defaults.index_lag if index_lag is NoInput.blank else index_lag + self.index_method = _drb(defaults.index_method, index_method).lower() + self.index_lag = _drb(defaults.index_lag, index_lag) super().__init__(*args, **kwargs) self.index_fixings = index_fixings # set index fixings after periods init - self.index_base = index_base # set after periods initialised + # set after periods initialised + self.index_base = index_base # type: ignore[assignment] + + def _regular_period( # type: ignore[empty-body] + self, + start: datetime, + end: datetime, + payment: datetime, + stub: bool, + notional: DualTypes, + iterator: int, + ) -> IndexFixedPeriod: + # set_periods has override + pass def _set_periods(self) -> None: self.periods = [ @@ -1893,7 +1921,7 @@ def _set_periods(self) -> None: roll=self.schedule.roll, calendar=self.schedule.calendar, index_base=self.index_base, - index_fixings=self.index_fixings, + index_fixings=NoInput(0), # set during init index_lag=self.index_lag, index_method=self.index_method, ), @@ -1909,7 +1937,7 @@ def _set_periods(self) -> None: def cashflow(self, curve: Curve | NoInput = NoInput(0)) -> DualTypes: """Aggregate the cashflows on the *IndexFixedPeriod* and *Cashflow* period using a *Curve*.""" - _: DualTypes = self.periods[0].cashflow(curve) + self.periods[1].cashflow + _: DualTypes = self.periods[0].cashflow(curve) + self.periods[1].cashflow # type: ignore[operator] return _ def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: @@ -1920,7 +1948,7 @@ def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: :meth:`BasePeriod.cashflows()`. """ cfs = super().cashflows(*args, **kwargs) - _ = cfs.iloc[[0]].copy() + _: DataFrame = cfs.iloc[[0]].copy() for attr in ["Cashflow", "NPV", "NPV Ccy"]: _[attr] += cfs.iloc[1][attr] _["Type"] = "ZeroIndexLeg" @@ -2000,19 +2028,25 @@ class CreditPremiumLeg(_FixedLegMixin, BaseLeg): premium_leg.npv(hazard_curve, disc_curve) """ # noqa: E501 + periods: list[CreditPremiumPeriod] # type: ignore[assignment] + def __init__( self, - *args, - fixed_rate: float | NoInput = NoInput(0), + *args: Any, + fixed_rate: DualTypes | NoInput = NoInput(0), premium_accrued: bool | NoInput = NoInput(0), - **kwargs, + **kwargs: Any, ): self._fixed_rate = fixed_rate self.premium_accrued = _drb(defaults.cds_premium_accrued, premium_accrued) super().__init__(*args, **kwargs) + if self.initial_exchange or self.final_exchange: + raise ValueError( + "`initial_exchange` and `final_exchange` cannot be True on CreditPremiumLeg." + ) self._set_periods() - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *CreditPremiumLeg* via summing all periods. @@ -2021,7 +2055,7 @@ def analytic_delta(self, *args, **kwargs): """ return super().analytic_delta(*args, **kwargs) - def cashflows(self, *args, **kwargs) -> DataFrame: + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *CreditPremiumLeg* used in calculating cashflows. @@ -2030,7 +2064,7 @@ def cashflows(self, *args, **kwargs) -> DataFrame: """ return super().cashflows(*args, **kwargs) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *CreditPremiumLeg* via summing all periods. @@ -2039,7 +2073,7 @@ def npv(self, *args, **kwargs): """ return super().npv(*args, **kwargs) - def accrued(self, settlement): + def accrued(self, settlement: datetime) -> DualTypes | None: """ Calculate the amount of premium accrued until a specific date within the relevant *Period*. @@ -2063,15 +2097,15 @@ def accrued(self, settlement): def _set_periods(self) -> None: return super()._set_periods() - def _regular_period( + def _regular_period( # type: ignore[override] self, start: datetime, end: datetime, payment: datetime, - notional: float, + notional: DualTypes, stub: bool, iterator: int, - ): + ) -> CreditPremiumPeriod: return CreditPremiumPeriod( fixed_rate=self.fixed_rate, premium_accrued=self.premium_accrued, @@ -2142,19 +2176,25 @@ class CreditProtectionLeg(BaseLeg): protection_leg.npv(hazard_curve, disc_curve) """ # noqa: E501 + periods: list[CreditProtectionPeriod] # type: ignore[assignment] + def __init__( self, - *args, + *args: Any, recovery_rate: DualTypes | NoInput = NoInput(0), discretization: int | NoInput = NoInput(0), - **kwargs, - ): - self._recovery_rate = _drb(defaults.cds_recovery_rate, recovery_rate) - self.discretization = _drb(defaults.cds_protection_discretization, discretization) + **kwargs: Any, + ) -> None: + self._recovery_rate: DualTypes = _drb(defaults.cds_recovery_rate, recovery_rate) + self.discretization: int = _drb(defaults.cds_protection_discretization, discretization) super().__init__(*args, **kwargs) + if self.initial_exchange or self.final_exchange: + raise ValueError( + "`initial_exchange` and `final_exchange` cannot be True on CreditProtectionLeg." + ) self._set_periods() - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic delta of the *CreditProtectionLeg* via summing all periods. @@ -2163,7 +2203,7 @@ def analytic_delta(self, *args, **kwargs): """ return super().analytic_delta(*args, **kwargs) - def analytic_rec_risk(self, *args, **kwargs): + def analytic_rec_risk(self, *args: Any, **kwargs: Any) -> DualTypes: """ Return the analytic recovery risk of the *CreditProtectionLeg* via summing all periods. @@ -2173,7 +2213,7 @@ def analytic_rec_risk(self, *args, **kwargs): _ = (period.analytic_rec_risk(*args, **kwargs) for period in self.periods) return sum(_) - def cashflows(self, *args, **kwargs) -> DataFrame: + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: """ Return the properties of the *CreditProtectionLeg* used in calculating cashflows. @@ -2182,7 +2222,7 @@ def cashflows(self, *args, **kwargs) -> DataFrame: """ return super().cashflows(*args, **kwargs) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *CreditProtectionLeg* via summing all periods. @@ -2199,10 +2239,10 @@ def _regular_period( start: datetime, end: datetime, payment: datetime, - notional: float, stub: bool, + notional: DualTypes, iterator: int, - ): + ) -> CreditProtectionPeriod: return CreditProtectionPeriod( recovery_rate=self.recovery_rate, discretization=self.discretization, @@ -2220,11 +2260,11 @@ def _regular_period( ) @property - def recovery_rate(self): + def recovery_rate(self) -> DualTypes: return self._recovery_rate @recovery_rate.setter - def recovery_rate(self, value): + def recovery_rate(self, value: DualTypes) -> None: self._recovery_rate = value for _ in self.periods: if isinstance(_, CreditProtectionPeriod): @@ -2236,7 +2276,7 @@ def recovery_rate(self, value): # Contact rateslib at gmail.com if this code is observed outside its intended sphere. -class IndexFixedLeg(_IndexLegMixin, _FixedLegMixin, BaseLeg): +class IndexFixedLeg(_IndexLegMixin, _FixedLegMixin, BaseLeg): # type: ignore[misc] """ Create a leg of :class:`~rateslib.periods.IndexFixedPeriod` s and initial and final :class:`~rateslib.periods.IndexCashflow` s. @@ -2316,19 +2356,17 @@ class IndexFixedLeg(_IndexLegMixin, _FixedLegMixin, BaseLeg): # TODO: spread calculations to determine the fixed rate on this leg do not work. def __init__( self, - *args, - index_base: float, - index_fixings: float | Series | NoInput = NoInput(0), + *args: Any, + index_base: DualTypes, + index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), - fixed_rate: float | NoInput = NoInput(0), - **kwargs, + fixed_rate: DualTypes | NoInput = NoInput(0), + **kwargs: Any, ) -> None: self._fixed_rate = fixed_rate - self.index_lag = defaults.index_lag if index_lag is NoInput.blank else index_lag - self.index_method = ( - defaults.index_method if index_method is NoInput.blank else index_method.lower() - ) + self.index_lag: int = _drb(defaults.index_lag, index_lag) + self.index_method = _drb(defaults.index_method, index_method).lower() if self.index_method not in ["daily", "monthly"]: raise ValueError("`index_method` must be in {'daily', 'monthly'}.") super().__init__(*args, **kwargs) @@ -2382,7 +2420,7 @@ def _set_periods(self) -> None: index_method=self.index_method, index_fixings=self.index_fixings, ) - for i, period in self.schedule.table.to_dict(orient="index").items() + for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) ] if self.final_exchange and self.amortization != 0: amortization = [ @@ -2425,13 +2463,13 @@ def _set_periods(self) -> None: ), ) - def npv(self, *args, **kwargs): + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: return super().npv(*args, **kwargs) - def cashflows(self, *args, **kwargs): + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: return super().cashflows(*args, **kwargs) - def analytic_delta(self, *args, **kwargs): + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: return super().analytic_delta(*args, **kwargs) @@ -2466,52 +2504,52 @@ class BaseLegMtm(BaseLeg, metaclass=ABCMeta): _do_not_repeat_set_periods: bool = False _is_mtm: bool = True + _delay_set_periods: bool = True def __init__( self, *args: Any, - fx_fixings: list | float | Dual | Dual2 | NoInput = NoInput(0), + fx_fixings: NoInput # type: ignore[type-var] + | DualTypes + | list[DualTypes] + | Series[DualTypes] + | tuple[DualTypes, Series[DualTypes]] = NoInput(0), alt_currency: str | NoInput = NoInput(0), - alt_notional: float | NoInput = NoInput(0), + alt_notional: DualTypes | NoInput = NoInput(0), **kwargs: Any, ) -> None: - if alt_currency is NoInput.blank: + if isinstance(alt_currency, NoInput): raise ValueError("`alt_currency` and `currency` must be supplied for MtmLeg.") - self.alt_currency = alt_currency.lower() - self.alt_notional = defaults.notional if alt_notional is NoInput.blank else alt_notional - self._delay_set_periods = True + self.alt_currency: str = alt_currency.lower() + self.alt_notional: DualTypes = _drb(defaults.notional, alt_notional) if "initial_exchange" not in kwargs: kwargs["initial_exchange"] = True kwargs["final_exchange"] = True super().__init__(*args, **kwargs) if self.amortization != 0: raise ValueError("`amortization` cannot be supplied to a `FixedLegExchangeMtm` type.") - self.fx_fixings = fx_fixings # calls the setter + + # calls the fixings setter, will convert the input types to list + self.fx_fixings = fx_fixings # type: ignore[assignment] @property def notional(self) -> DualTypes: return self._notional @notional.setter - def notional(self, value: DualTypes): + def notional(self, value: DualTypes) -> None: self._notional = value - @property - def fx_fixings(self): - """ - list : FX fixing values used for consecutive periods. - """ - return self._fx_fixings - - def _get_fx_fixings_from_series(self, ser: Series, ini_period: int = 0): + def _get_fx_fixings_from_series( + self, + ser: Series[DualTypes], # type: ignore[type-var] + ini_period: int = 0, + ) -> list[DualTypes]: last_fixing_date = ser.index[-1] - fixings_list = [] + fixings_list: list[DualTypes] = [] for i in range(ini_period, self.schedule.n_periods): - required_date = add_tenor( - self.schedule.aschedule[i], - f"{self.payment_lag_exchange}B", - NoInput(0), - self.schedule.calendar, + required_date = self.schedule.calendar.lag( + self.schedule.aschedule[i], self.payment_lag_exchange, True ) if required_date > last_fixing_date: break @@ -2526,13 +2564,27 @@ def _get_fx_fixings_from_series(self, ser: Series, ini_period: int = 0): ) return fixings_list + @property + def fx_fixings(self) -> list[DualTypes]: + """ + list : FX fixing values used for consecutive periods. + """ + return self._fx_fixings + @fx_fixings.setter - def fx_fixings(self, value): - if value is NoInput.blank: - self._fx_fixings = [] + def fx_fixings( + self, + value: NoInput # type: ignore[type-var] + | DualTypes + | list[DualTypes] + | Series[DualTypes] + | tuple[DualTypes, Series[DualTypes]], + ) -> None: + if isinstance(value, NoInput): + self._fx_fixings: list[DualTypes] = [] elif isinstance(value, list): self._fx_fixings = value - elif isinstance(value, float | Dual | Dual2): + elif isinstance(value, float | Dual | Dual2 | Variable): self._fx_fixings = [value] elif isinstance(value, Series): self._fx_fixings = self._get_fx_fixings_from_series(value) @@ -2549,7 +2601,7 @@ def fx_fixings(self, value): # Commercial use of this code, and/or copying and redistribution is prohibited. # Contact rateslib at gmail.com if this code is observed outside its intended sphere. - def _get_fx_fixings(self, fx): + def _get_fx_fixings(self, fx: DualTypes | FXRates | FXForwards | NoInput) -> list[DualTypes]: """ Return the calculated FX fixings. @@ -2565,11 +2617,12 @@ def _get_fx_fixings(self, fx): ``fx_fixings``. """ n_given, n_req = len(self.fx_fixings), self.schedule.n_periods - fx_fixings = self.fx_fixings.copy() + fx_fixings_: list[DualTypes] = self.fx_fixings.copy() + # Only FXForwards can correctly forecast rates. Other inputs may raise Errros or Warnings. if isinstance(fx, FXForwards): for i in range(n_given, n_req): - fx_fixings.append( + fx_fixings_.append( fx.rate( self.alt_currency + self.currency, self.schedule.calendar.lag( @@ -2594,7 +2647,7 @@ def _get_fx_fixings(self, fx): "'warn'.", UserWarning, ) - fx_fixings = [1.0] * n_req + fx_fixings_ = [1.0] * n_req else: if defaults.no_fx_fixings_for_xcs.lower() == "warn": warnings.warn( @@ -2603,13 +2656,13 @@ def _get_fx_fixings(self, fx): "'warn'.", UserWarning, ) - fx_fixings.extend([fx_fixings[-1]] * (n_req - n_given)) - return fx_fixings + fx_fixings_.extend([fx_fixings_[-1]] * (n_req - n_given)) + return fx_fixings_ - def _set_periods(self, fx): - fx_fixings = self._get_fx_fixings(fx) - self.notional = fx_fixings[0] * self.alt_notional - notionals = [self.alt_notional * fx_fixings[i] for i in range(len(fx_fixings))] + def _set_periods(self, fx: DualTypes | FXRates | FXForwards | NoInput) -> None: # type: ignore[override] + fx_fixings_: list[DualTypes] = self._get_fx_fixings(fx) + self.notional = fx_fixings_[0] * self.alt_notional + notionals = [self.alt_notional * fx_fixings_[i] for i in range(len(fx_fixings_))] # initial exchange self.periods = ( @@ -2623,7 +2676,7 @@ def _set_periods(self, fx): ), self.currency, "Exchange", - fx_fixings[0], + fx_fixings_[0], ), ] if self.initial_exchange @@ -2651,9 +2704,9 @@ def _set_periods(self, fx): ), self.currency, "Mtm", - fx_fixings[i + 1], + fx_fixings_[i + 1], ) - for i in range(len(fx_fixings) - 1) + for i in range(len(fx_fixings_) - 1) ] interleaved_periods = [ val for pair in zip(regular_periods, mtm_flows, strict=False) for val in pair @@ -2672,7 +2725,7 @@ def _set_periods(self, fx): ), self.currency, "Exchange", - fx_fixings[-1], + fx_fixings_[-1], ), ) @@ -2717,7 +2770,7 @@ def analytic_delta( return ret -class FixedLegMtm(_FixedLegMixin, BaseLegMtm): +class FixedLegMtm(_FixedLegMixin, BaseLegMtm): # type: ignore[misc] """ Create a leg of :class:`~rateslib.periods.FixedPeriod` s and initial, mtm and final :class:`~rateslib.periods.Cashflow` s. @@ -2840,7 +2893,11 @@ def __init__( self, *args: Any, float_spread: DualTypes | NoInput = NoInput(0), - fixings: DualTypes | list[DualTypes] | NoInput = NoInput(0), + fixings: Series[DualTypes] # type: ignore[type-var] + | list[DualTypes | list[DualTypes] | Series[DualTypes] | NoInput] + | tuple[DualTypes, Series[DualTypes]] + | DualTypes + | NoInput = NoInput(0), fixing_method: str | NoInput = NoInput(0), method_param: int | NoInput = NoInput(0), spread_compound_method: str | NoInput = NoInput(0), @@ -2916,14 +2973,14 @@ class CustomLeg(BaseLeg): """ # noqa: E501 - def __init__(self, periods): + def __init__(self, periods: list[Period]) -> None: if not all(isinstance(p, Period) for p in periods): raise ValueError( "Each object in `periods` must be of type {FixedPeriod, FloatPeriod, " "Cashflow}.", ) self._set_periods(periods) - def _set_periods(self, periods: list[Period]) -> None: + def _set_periods(self, periods: list[Period]) -> None: # type: ignore[override] self.periods: list[Any] = periods def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index 3fc1cc38..5c622fb0 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -2678,21 +2678,21 @@ class Cashflow: def __init__( self, - notional: float, + notional: DualTypes, payment: datetime, currency: str | NoInput = NoInput(0), stub_type: str | NoInput = NoInput(0), - rate: float | NoInput = NoInput(0), + rate: DualTypes | NoInput = NoInput(0), ): self.notional, self.payment = notional, payment self.currency = _drb(defaults.base_currency, currency).lower() self.stub_type = stub_type - self._rate: float | NoInput = rate if isinstance(rate, NoInput) else _dual_float(rate) + self._rate: DualTypes | NoInput = rate if isinstance(rate, NoInput) else _dual_float(rate) def __repr__(self) -> str: return f"" - def rate(self) -> float | None: + def rate(self) -> DualTypes | None: """ Return the associated rate initialised with the *Cashflow*. Not used for calculations. """ @@ -2766,15 +2766,19 @@ def cashflows( } @property - def cashflow(self) -> float: + def cashflow(self) -> DualTypes: return -self.notional + # @property + # def dcf(self) -> float: + # return 0.0 + def analytic_delta( self, - curve: Curve | None = None, - disc_curve: Curve | None = None, - fx: float | FXRates | FXForwards | None = None, - base: str | None = None, + curve: Curve | NoInput = NoInput(0), + disc_curve: Curve | NoInput = NoInput(0), + fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), + base: str | NoInput = NoInput(0), ) -> DualTypes: """ Return the analytic delta of the *Cashflow*. @@ -3230,7 +3234,7 @@ class IndexCashflow(IndexMixin, Cashflow): # type: ignore[misc] def __init__( self, *args: Any, - index_base: float, + index_base: DualTypes, index_fixings: DualTypes | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), diff --git a/python/tests/test_legs.py b/python/tests/test_legs.py index a5032201..3c2c7623 100644 --- a/python/tests/test_legs.py +++ b/python/tests/test_legs.py @@ -982,8 +982,8 @@ def test_analytic_delta_no_fixed_rate(self, curve) -> None: frequency="A", fixed_rate=NoInput(0), ) - result = zfl.analytic_delta(curve) - assert result is None + with pytest.raises(ValueError, match="Must have `fixed_rate` on ZeroFixedLeg for analy"): + zfl.analytic_delta(curve) class TestZeroIndexLeg: @@ -1346,6 +1346,21 @@ def test_premium_leg_accrued(self, date, exp): result = leg.accrued(date) assert abs(result - exp) < 1e-6 + @pytest.mark.parametrize("final", [True, False]) + def test_exchanges_raises(self, final): + with pytest.raises(ValueError, match="`initial_exchange` and `final_exchange` cannot be"): + CreditPremiumLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 6, 1), + payment_lag=2, + notional=-1e9, + convention="ActActICMA", + frequency="Q", + fixed_rate=2.0, + initial_exchange=final, + final_exchange=not final, + ) + class TestCreditProtectionLeg: def test_leg_analytic_delta(self, hazard_curve, curve) -> None: From 8dd97e2de9756c0f8859151690bf9d264c5302c6 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Mon, 6 Jan 2025 17:11:30 +0100 Subject: [PATCH 6/6] TST/TYP: legs.py `index_fixings` for amortised IndexFixedLeg with `final_exchange` (#600) --- pyproject.toml | 1 - python/rateslib/legs.py | 461 +++++++++++++++++++------------------ python/rateslib/periods.py | 6 +- python/tests/test_legs.py | 36 +++ 4 files changed, 278 insertions(+), 226 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 14d65f80..4a2bdb44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -144,7 +144,6 @@ packages = [ exclude = [ "/instruments", "fx_volatility.py", - "legs.py", "solver.py", ] strict = true diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index 8d127819..4301a035 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -1161,7 +1161,7 @@ class _IndexLegMixin: index_method: str _index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0) # type: ignore[type-var] _index_base: DualTypes | NoInput = NoInput(0) - periods: list[Period] + periods: list[IndexFixedPeriod | IndexCashflow | Cashflow] index_lag: int # def _set_index_fixings_on_periods(self): @@ -1202,30 +1202,39 @@ def index_fixings( self, value: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput, # type: ignore[type-var] ) -> None: - self._index_fixings = value - for i, period in enumerate(self.periods): - if isinstance(period, IndexFixedPeriod | IndexCashflow): - if isinstance(value, Series): - val: DualTypes | None = IndexMixin._index_value( - i_fixings=value, - i_method=self.index_method, - i_lag=self.index_lag, - i_date=period.end, - i_curve=NoInput(0), # ! NoInput returned for periods beyond Series end. - ) - if val is None: - _: DualTypes | NoInput = NoInput(0) - else: - _ = val - elif isinstance(value, list): - if i >= len(value): - _ = NoInput(0) # some fixings are unknown, list size is limited - else: - _ = value[i] - else: - # value is float or NoInput - _ = value if i == 0 else NoInput(0) - period.index_fixings = _ + self._index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = value # type: ignore[type-var] + + def _index_from_series(ser: Series[DualTypes], end: datetime) -> DualTypes | NoInput: # type: ignore[type-var] + val: DualTypes | None = IndexMixin._index_value( + i_fixings=ser, + i_method=self.index_method, + i_lag=self.index_lag, + i_date=end, + i_curve=NoInput(0), # ! NoInput returned for periods beyond Series end. + ) + if val is None: + _: DualTypes | NoInput = NoInput(0) + else: + _ = val + return _ + + def _index_from_list(ls: list[DualTypes], i: int) -> DualTypes | NoInput: + return NoInput(0) if i >= len(ls) else ls[i] + + if isinstance(value, NoInput): + for p in [_ for _ in self.periods if type(_) is not Cashflow]: + p.index_fixings = NoInput(0) + elif isinstance(value, Series): + for p in [_ for _ in self.periods if type(_) is not Cashflow]: + date_: datetime = p.end if type(p) is IndexFixedPeriod else p.payment + p.index_fixings = _index_from_series(value, date_) + elif isinstance(value, list): + for i, p in enumerate([_ for _ in self.periods if type(_) is not Cashflow]): + p.index_fixings = _index_from_list(value, i) + else: + self.periods[0].index_fixings = value # type: ignore[union-attr] + for p in [_ for _ in self.periods[1:] if type(_) is not Cashflow]: + p.index_fixings = NoInput(0) @property def index_base(self) -> DualTypes | NoInput: @@ -1974,6 +1983,211 @@ def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: return super().npv(*args, **kwargs) +class IndexFixedLeg(_IndexLegMixin, _FixedLegMixin, BaseLeg): # type: ignore[misc] + """ + Create a leg of :class:`~rateslib.periods.IndexFixedPeriod` s and initial and + final :class:`~rateslib.periods.IndexCashflow` s. + + Parameters + ---------- + args : dict + Required positional args to :class:`BaseLeg`. + index_base : float or None, optional + The base index to determine the cashflow. + index_fixings : float, list or Series, optional + If a float scalar, will be applied as the index fixing for the first period. + If a datetime indexed ``Series``, will use the fixings that are available + in that object for relevant periods, and derive the rest from the ``curve``. + If a list, will apply those values as the fixings for the first set of periods + and derive the rest from the ``curve``. + index_method : str + Whether the indexing uses a daily measure for settlement or the most recently + monthly data taken from the first day of month. + index_lag : int, optional + The number of months by which the index value is lagged. Used to ensure + consistency between curves and forecast values. Defined by default. + fixed_rate : float or None + The fixed rate applied to determine cashflows. Can be set to `None` and + designated later, perhaps after a mid-market rate for all periods has been + calculated. + kwargs : dict + Required keyword arguments to :class:`BaseLeg`. + + Notes + ----- + + .. warning:: + + An initial exchange is not currently implemented for this leg. + + The final cashflow notional is set as the notional. The payment date is set equal + to the final accrual date adjusted by ``payment_lag_exchange``. + + If ``amortization`` is specified an exchanged notional equivalent to the + amortization amount is added to the list of periods. For similar examples see + :class:`~rateslib.legs.FloatLeg`. + + The NPV of a *IndexFixedLeg* is the sum of the period NPVs. + + .. math:: + + P = - R \\sum_{i=1}^n N_i d_i v(m_i) I(m_i) - \\sum_{i=1}^{n-1}(N_{i}-N_{i+1})v(m_i)I(m_i) - N_n v(m_n)I(m_n) + + The analytic delta is defined as that of a *FixedLeg*. + + .. math:: + + A = \\sum_{i=1}^n N_i d_i v(m_i) I(m_i) + + Examples + -------- + + .. ipython:: python + + curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.98}) + index_curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.99}, index_base=100.0) + index_leg_exch = IndexFixedLeg( + dt(2022, 1, 1), "9M", "Q", + notional=1000000, + amortization=200000, + index_base=100.0, + initial_exchange=False, + final_exchange=True, + fixed_rate=1.0, + ) + index_leg_exch.cashflows(index_curve, curve) + index_leg_exch.npv(index_curve, curve) + + """ # noqa: E501 + + periods: list[IndexCashflow | IndexFixedPeriod] # type: ignore[assignment] + + # TODO: spread calculations to determine the fixed rate on this leg do not work. + def __init__( + self, + *args: Any, + index_base: DualTypes, + index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] + index_method: str | NoInput = NoInput(0), + index_lag: int | NoInput = NoInput(0), + fixed_rate: DualTypes | NoInput = NoInput(0), + **kwargs: Any, + ) -> None: + self._fixed_rate = fixed_rate + self.index_lag: int = _drb(defaults.index_lag, index_lag) + self.index_method = _drb(defaults.index_method, index_method).lower() + if self.index_method not in ["daily", "monthly"]: + raise ValueError("`index_method` must be in {'daily', 'monthly'}.") + super().__init__(*args, **kwargs) + self.index_fixings = index_fixings # set index fixings after periods init + self.index_base = index_base # set after periods initialised + + def _set_periods(self) -> None: + self.periods = [] + + # initial exchange + if self.initial_exchange: + raise NotImplementedError( + "Cannot construct `IndexFixedLeg` with `initial_exchange` " + "due to not implemented `index_fixings` input argument applicable to " + "the indexing-up the initial exchange.", + ) + # self.periods.append( + # IndexCashflow( + # notional=-self.notional, + # payment=add_tenor( + # self.schedule.aschedule[0], + # f"{self.payment_lag_exchange}B", + # None, + # self.schedule.calendar, + # ), + # currency=self.currency, + # stub_type="Exchange", + # rate=None, + # index_base=self.index_base, + # index_fixings=self.index_fixings, + # index_method=self.index_method, + # ) + # ) + + # regular periods + regular_periods = [ + IndexFixedPeriod( + fixed_rate=self.fixed_rate, + start=period[defaults.headers["a_acc_start"]], + end=period[defaults.headers["a_acc_end"]], + payment=period[defaults.headers["payment"]], + notional=self.notional - self.amortization * i, + convention=self.convention, + currency=self.currency, + termination=self.schedule.termination, + frequency=self.schedule.frequency, + stub=period[defaults.headers["stub_type"]] == "Stub", + roll=self.schedule.roll, + calendar=self.schedule.calendar, + index_base=self.index_base, + index_method=self.index_method, + index_fixings=self.index_fixings[i] + if isinstance(self.index_fixings, list) + else self.index_fixings, + ) + for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) + ] + if self.final_exchange and self.amortization != 0: + amortization = [ + IndexCashflow( + notional=self.amortization, + payment=self.schedule.pschedule[1 + i], + currency=self.currency, + stub_type="Amortization", + rate=NoInput(0), + index_base=self.index_base, + index_fixings=self.index_fixings[i] + if isinstance(self.index_fixings, list) + else self.index_fixings, + index_method=self.index_method, + ) + for i in range(self.schedule.n_periods - 1) + ] + interleaved_periods: list[IndexCashflow | IndexFixedPeriod] = [ + val for pair in zip(regular_periods, amortization, strict=False) for val in pair + ] + interleaved_periods.append(regular_periods[-1]) # add last regular period + else: + interleaved_periods = regular_periods # type: ignore[assignment] + self.periods.extend(interleaved_periods) + + # final cashflow + if self.final_exchange: + self.periods.append( + IndexCashflow( + notional=self.notional - self.amortization * (self.schedule.n_periods - 1), + payment=self.schedule.calendar.lag( + self.schedule.aschedule[-1], + self.payment_lag_exchange, + True, + ), + currency=self.currency, + stub_type="Exchange", + rate=NoInput(0), + index_base=self.index_base, + index_fixings=self.index_fixings[-1] + if isinstance(self.index_fixings, list) + else self.index_fixings, + index_method=self.index_method, + ), + ) + + def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: + return super().npv(*args, **kwargs) + + def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: + return super().cashflows(*args, **kwargs) + + def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: + return super().analytic_delta(*args, **kwargs) + + class CreditPremiumLeg(_FixedLegMixin, BaseLeg): """ Create a credit premium leg composed of :class:`~rateslib.periods.CreditPremiumPeriod` s. @@ -2276,203 +2490,6 @@ def recovery_rate(self, value: DualTypes) -> None: # Contact rateslib at gmail.com if this code is observed outside its intended sphere. -class IndexFixedLeg(_IndexLegMixin, _FixedLegMixin, BaseLeg): # type: ignore[misc] - """ - Create a leg of :class:`~rateslib.periods.IndexFixedPeriod` s and initial and - final :class:`~rateslib.periods.IndexCashflow` s. - - Parameters - ---------- - args : dict - Required positional args to :class:`BaseLeg`. - index_base : float or None, optional - The base index to determine the cashflow. - index_fixings : float, list or Series, optional - If a float scalar, will be applied as the index fixing for the first period. - If a datetime indexed ``Series``, will use the fixings that are available - in that object for relevant periods, and derive the rest from the ``curve``. - If a list, will apply those values as the fixings for the first set of periods - and derive the rest from the ``curve``. - index_method : str - Whether the indexing uses a daily measure for settlement or the most recently - monthly data taken from the first day of month. - index_lag : int, optional - The number of months by which the index value is lagged. Used to ensure - consistency between curves and forecast values. Defined by default. - fixed_rate : float or None - The fixed rate applied to determine cashflows. Can be set to `None` and - designated later, perhaps after a mid-market rate for all periods has been - calculated. - kwargs : dict - Required keyword arguments to :class:`BaseLeg`. - - Notes - ----- - - .. warning:: - - An initial exchange is not currently implemented for this leg. - - The final cashflow notional is set as the notional. The payment date is set equal - to the final accrual date adjusted by ``payment_lag_exchange``. - - If ``amortization`` is specified an exchanged notional equivalent to the - amortization amount is added to the list of periods. For similar examples see - :class:`~rateslib.legs.FloatLeg`. - - The NPV of a *IndexFixedLeg* is the sum of the period NPVs. - - .. math:: - - P = - R \\sum_{i=1}^n N_i d_i v(m_i) I(m_i) - \\sum_{i=1}^{n-1}(N_{i}-N_{i+1})v(m_i)I(m_i) - N_n v(m_n)I(m_n) - - The analytic delta is defined as that of a *FixedLeg*. - - .. math:: - - A = \\sum_{i=1}^n N_i d_i v(m_i) I(m_i) - - Examples - -------- - - .. ipython:: python - - curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.98}) - index_curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.99}, index_base=100.0) - index_leg_exch = IndexFixedLeg( - dt(2022, 1, 1), "9M", "Q", - notional=1000000, - amortization=200000, - index_base=100.0, - initial_exchange=False, - final_exchange=True, - fixed_rate=1.0, - ) - index_leg_exch.cashflows(index_curve, curve) - index_leg_exch.npv(index_curve, curve) - - """ # noqa: E501 - - # TODO: spread calculations to determine the fixed rate on this leg do not work. - def __init__( - self, - *args: Any, - index_base: DualTypes, - index_fixings: DualTypes | list[DualTypes] | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] - index_method: str | NoInput = NoInput(0), - index_lag: int | NoInput = NoInput(0), - fixed_rate: DualTypes | NoInput = NoInput(0), - **kwargs: Any, - ) -> None: - self._fixed_rate = fixed_rate - self.index_lag: int = _drb(defaults.index_lag, index_lag) - self.index_method = _drb(defaults.index_method, index_method).lower() - if self.index_method not in ["daily", "monthly"]: - raise ValueError("`index_method` must be in {'daily', 'monthly'}.") - super().__init__(*args, **kwargs) - self.index_fixings = index_fixings # set index fixings after periods init - self.index_base = index_base # set after periods initialised - - def _set_periods(self) -> None: - self.periods = [] - - # initial exchange - if self.initial_exchange: - raise NotImplementedError( - "Cannot construct `IndexFixedLeg` with `initial_exchange` " - "due to not implemented `index_fixings` input argument applicable to " - "the indexing-up the initial exchange.", - ) - # self.periods.append( - # IndexCashflow( - # notional=-self.notional, - # payment=add_tenor( - # self.schedule.aschedule[0], - # f"{self.payment_lag_exchange}B", - # None, - # self.schedule.calendar, - # ), - # currency=self.currency, - # stub_type="Exchange", - # rate=None, - # index_base=self.index_base, - # index_fixings=self.index_fixings, - # index_method=self.index_method, - # ) - # ) - - # regular periods - regular_periods = [ - IndexFixedPeriod( - fixed_rate=self.fixed_rate, - start=period[defaults.headers["a_acc_start"]], - end=period[defaults.headers["a_acc_end"]], - payment=period[defaults.headers["payment"]], - notional=self.notional - self.amortization * i, - convention=self.convention, - currency=self.currency, - termination=self.schedule.termination, - frequency=self.schedule.frequency, - stub=period[defaults.headers["stub_type"]] == "Stub", - roll=self.schedule.roll, - calendar=self.schedule.calendar, - index_base=self.index_base, - index_method=self.index_method, - index_fixings=self.index_fixings, - ) - for i, period in enumerate(self.schedule.table.to_dict(orient="index").values()) - ] - if self.final_exchange and self.amortization != 0: - amortization = [ - IndexCashflow( - notional=self.amortization, - payment=self.schedule.pschedule[1 + i], - currency=self.currency, - stub_type="Amortization", - rate=NoInput(0), - index_base=self.index_base, - index_fixings=self.index_fixings, - index_method=self.index_method, - ) - for i in range(self.schedule.n_periods - 1) - ] - interleaved_periods = [ - val for pair in zip(regular_periods, amortization, strict=False) for val in pair - ] - interleaved_periods.append(regular_periods[-1]) # add last regular period - else: - interleaved_periods = regular_periods - self.periods.extend(interleaved_periods) - - # final cashflow - if self.final_exchange: - self.periods.append( - IndexCashflow( - notional=self.notional - self.amortization * (self.schedule.n_periods - 1), - payment=self.schedule.calendar.lag( - self.schedule.aschedule[-1], - self.payment_lag_exchange, - True, - ), - currency=self.currency, - stub_type="Exchange", - rate=NoInput(0), - index_base=self.index_base, - index_fixings=self.index_fixings, - index_method=self.index_method, - ), - ) - - def npv(self, *args: Any, **kwargs: Any) -> DualTypes | dict[str, DualTypes]: - return super().npv(*args, **kwargs) - - def cashflows(self, *args: Any, **kwargs: Any) -> DataFrame: - return super().cashflows(*args, **kwargs) - - def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes: - return super().analytic_delta(*args, **kwargs) - - class BaseLegMtm(BaseLeg, metaclass=ABCMeta): """ Abstract base class with common parameters for all ``LegMtm`` subclasses. diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index 5c622fb0..b8a3efa3 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -3165,8 +3165,8 @@ class IndexCashflow(IndexMixin, Cashflow): # type: ignore[misc] ---------- args : dict Required positional args to :class:`Cashflow`. - index_base : float or None, optional - The base index to determine the cashflow. + index_base : float, optional + The base index to determine the cashflow. Required but may be set after initialisation. index_fixings : float, or Series, optional If a float scalar, will be applied as the index fixing for the whole period. If a datetime indexed ``Series`` will use the @@ -3234,7 +3234,7 @@ class IndexCashflow(IndexMixin, Cashflow): # type: ignore[misc] def __init__( self, *args: Any, - index_base: DualTypes, + index_base: DualTypes | NoInput, index_fixings: DualTypes | Series[DualTypes] | NoInput = NoInput(0), # type: ignore[type-var] index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), diff --git a/python/tests/test_legs.py b/python/tests/test_legs.py index 3c2c7623..4b429bf3 100644 --- a/python/tests/test_legs.py +++ b/python/tests/test_legs.py @@ -1727,6 +1727,42 @@ def test_initial_exchange_raises(self) -> None: initial_exchange=True, ) + def test_index_fixings_as_list(self) -> None: + leg = IndexFixedLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 10, 1), + payment_lag=2, + convention="Act360", + frequency="Q", + notional=1e6, + amortization=250e3, + index_base=NoInput(0), + index_fixings=[100.0, 200.0], + ) + assert leg.periods[0].index_fixings == 100.0 + assert leg.periods[1].index_fixings == 200.0 + assert leg.periods[2].index_fixings == NoInput(0) + + def test_index_fixings_as_list_final_exchange(self) -> None: + leg = IndexFixedLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 10, 1), + payment_lag=2, + convention="Act360", + frequency="Q", + notional=1e6, + amortization=250e3, + index_base=NoInput(0), + index_fixings=[100.0, 100.0, 200.0, 199.0], + final_exchange=True, + ) + assert leg.periods[0].index_fixings == 100.0 + assert leg.periods[1].index_fixings == 100.0 + assert leg.periods[2].index_fixings == 200.0 + assert leg.periods[3].index_fixings == 199.0 + assert leg.periods[4].index_fixings == NoInput(0) + assert leg.periods[5].index_fixings == NoInput(0) + class TestFloatLegExchangeMtm: @pytest.mark.parametrize(