From 99c24437151dd0e2d771ff78aea87b9be47dec3f Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Wed, 1 Jan 2025 19:05:23 +0100 Subject: [PATCH 1/3] TYP: periods.py --- python/rateslib/curves/curves.py | 2 +- python/rateslib/periods.py | 118 ++++++++++++++++++------------- 2 files changed, 68 insertions(+), 52 deletions(-) diff --git a/python/rateslib/curves/curves.py b/python/rateslib/curves/curves.py index 9a43dc05..8a446498 100644 --- a/python/rateslib/curves/curves.py +++ b/python/rateslib/curves/curves.py @@ -3058,7 +3058,7 @@ def _get_node_vector(self) -> Arr1dF64 | Arr1dObj: # pragma: no cover def average_rate( effective: datetime, termination: datetime, convention: str, rate: DualTypes -) -> tuple[DualTypes, float, int]: +) -> tuple[Number, float, int]: """ Return the geometric, 1 calendar day, average rate for the rate in a period. diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index a8263521..ebc89a9d 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -37,6 +37,7 @@ Dual, Dual2, DualTypes, + Number, Variable, _dual_float, dual_exp, @@ -147,8 +148,17 @@ def _maybe_local( if local: return {currency: value} else: - fx_, _ = _get_fx_and_base(currency, fx, base) - return value * fx_ + return _maybe_fx_converted(value, currency, fx, base) + + +def _maybe_fx_converted( + value: DualTypes, + currency: str, + fx: float | FXRates | FXForwards | NoInput, + base: str | NoInput, +) -> DualTypes: + fx_, _ = _get_fx_and_base(currency, fx, base) + return value * fx_ def _disc_maybe_from_curve( @@ -1715,7 +1725,7 @@ def _fixings_table_fast( v_vals = v_vals.interpolate(method="time") v_vals = Series(np.exp(v_vals.to_numpy()), index=obs_vals.index) - scalar = dcf_vals.values / obs_vals.values + scalar = dcf_vals.to_numpy() / obs_vals.to_numpy() if self.fixing_method in ["rfr_lockout", "rfr_lockout_avg"]: scalar[-self.method_param :] = 0.0 scalar[-(self.method_param + 1)] = ( @@ -1723,7 +1733,7 @@ def _fixings_table_fast( / obs_vals.iloc[-(self.method_param + 1)] ) # perform an efficient rate approximation - rate = curve_.rate( + rate = curve_._rate_with_raise( effective=obs_dates.iloc[0], termination=obs_dates.iloc[-1], ) @@ -1736,7 +1746,7 @@ def _fixings_table_fast( # approximate sensitivity to each fixing z = self.float_spread / 10000 if "avg" in self.fixing_method: - drdri = 1 / n + drdri: DualTypes = 1 / n elif self.spread_compound_method == "none_simple": drdri = (1 / n) * (1 + (r_bar / 100) * d) ** (n - 1) elif self.spread_compound_method == "isda_compounding": @@ -1762,7 +1772,7 @@ def _fixings_table_fast( "dcf": dcf_vals, "notional": notional_exposure, "risk": notional_exposure * v_vals * obs_vals * 0.0001, - "rates": Series(rate, index=obs_dates.index).astype( + "rates": Series(rate, index=obs_dates.index).astype( # type: ignore[arg-type] float, ), # .apply(float, convert_dtype=float), }, @@ -1804,21 +1814,25 @@ def _ibor_fixings_table( ------- DataFrame """ - if isinstance(curve, dict) and self.stub: - # then must perform an interpolated calculation - return self._ibor_stub_fixings_table(curve, disc_curve, right, risk) - elif isinstance(curve, dict) and not self.stub: - # then extract the one relevant curve from dict - curve = _get_ibor_curve_from_dict(self.freq_months, curve) + if isinstance(curve, dict): + if self.stub: + # then must perform an interpolated calculation + return self._ibor_stub_fixings_table(curve, disc_curve, right, risk) + else: # not self.stub: + # then extract the one relevant curve from dict + curve_: Curve = _get_ibor_curve_from_dict(self.freq_months, curve) + else: + curve_ = curve + return self._ibor_single_tenor_fixings_table( - curve, disc_curve, f"{self.freq_months}m", right, risk + curve_, disc_curve, f"{self.freq_months}m", right, risk ) def _ibor_single_tenor_fixings_table( self, - curve, - disc_curve, - tenor, + curve: Curve, + disc_curve: Curve, + tenor: str, right: datetime | NoInput, risk: DualTypes | NoInput = NoInput(0), ) -> DataFrame: @@ -1874,11 +1888,11 @@ def _ibor_single_tenor_fixings_table( def _ibor_stub_fixings_table( self, - curve: dict, + curve: dict[str, Curve], disc_curve: Curve, right: datetime | NoInput, risk: DualTypes | NoInput = NoInput(0), - ): + ) -> DataFrame: calendar = next(iter(curve.values())).calendar # note: ASSUMES all curve calendars are same values = {add_tenor(self.start, k, "MF", calendar): k for k, v in curve.items()} values = dict(sorted(values.items())) @@ -1926,9 +1940,9 @@ def _ibor_stub_fixings_table( def _rfr_fixings_array( self, - d: dict, + d: dict[str, Any], disc_curve: Curve, - ): + ) -> tuple[DualTypes, DataFrame]: """ Calculate the rate of a period via extraction and combination of every fixing. @@ -1976,9 +1990,9 @@ def _rfr_fixings_array( if self.fixing_method in ["rfr_lockout", "rfr_lockout_avg"]: rates_dual.iloc[-self.method_param :] = rates_dual.iloc[-self.method_param - 1] if "avg" in self.fixing_method: - rate = self._rate_rfr_avg_with_spread(rates_dual.to_numpy(), d["dcf_vals"].to_numpy()) + rate: Dual = self._rate_rfr_avg_with_spread(rates_dual.to_numpy(), d["dcf_vals"].to_numpy()) # type: ignore[assignment] else: - rate = self._rate_rfr_isda_compounded_with_spread( + rate = self._rate_rfr_isda_compounded_with_spread( # type: ignore[assignment] rates_dual.to_numpy(), d["dcf_vals"].to_numpy() ) @@ -2087,7 +2101,7 @@ def _get_method_dcf_endpoints( return start_obs, end_obs, start_dcf, end_dcf def _get_method_dcf_markers( - self, calendar: CalTypes, convention: str, exposure=False + self, calendar: CalTypes, convention: str, exposure: bool =False ) -> tuple[ Series[datetime], Series[datetime], @@ -2159,14 +2173,18 @@ def _get_method_dcf_markers( ], ) - return obs_dates, dcf_dates, dcf_vals, obs_vals + return obs_dates, dcf_dates, dcf_vals, obs_vals # type: ignore[return-value] - def _get_analytic_delta_quadratic_coeffs(self, fore_curve, disc_curve): + def _get_analytic_delta_quadratic_coeffs( + self, + fore_curve: Curve, + disc_curve: Curve + ) -> tuple[Number, Number]: """ For use in the Leg._spread calculation get the 'a' and 'b' coefficients """ - os, oe, _, _ = self._get_method_dcf_endpoints(fore_curve) - rate = fore_curve.rate( + os, oe, _, _ = self._get_method_dcf_endpoints(fore_curve.calendar) + rate = fore_curve._rate_with_raise( effective=os, termination=oe, float_spread=0.0, @@ -2176,8 +2194,8 @@ def _get_analytic_delta_quadratic_coeffs(self, fore_curve, disc_curve): # approximate sensitivity to each fixing z = 0.0 if self.float_spread is None else self.float_spread if self.spread_compound_method == "isda_compounding": - d2rdz2 = d * (n - 1) * (1 + (r / 100 + z / 10000) * d) ** (n - 2) / 1e8 - drdz = (1 + (r / 100 + z / 10000) * d) ** (n - 1) / 1e4 + d2rdz2: Number = d * (n - 1) * (1 + (r / 100 + z / 10000) * d) ** (n - 2) / 1e8 + drdz: Number = (1 + (r / 100 + z / 10000) * d) ** (n - 1) / 1e4 Nvd = -self.notional * disc_curve[self.payment] * self.dcf a, b = 0.5 * Nvd * d2rdz2, Nvd * drdz elif self.spread_compound_method == "isda_flat_compounding": @@ -2247,11 +2265,11 @@ class CreditPremiumPeriod(BasePeriod): def __init__( self, - *args, + *args: Any, fixed_rate: float | NoInput = NoInput(0), premium_accrued: bool | NoInput = NoInput(0), - **kwargs, - ): + **kwargs: Any, + ) -> None: self.premium_accrued = _drb(defaults.cds_premium_accrued, premium_accrued) self.fixed_rate = fixed_rate super().__init__(*args, **kwargs) @@ -2264,9 +2282,10 @@ def cashflow(self) -> float | None: if isinstance(self.fixed_rate, NoInput): return None else: - return -self.notional * self.dcf * self.fixed_rate * 0.01 + _: float = -self.notional * self.dcf * self.fixed_rate * 0.01 + return _ - def accrued(self, settlement: datetime): + def accrued(self, settlement: datetime) -> float | None: """ Calculate the amount of premium accrued until a specific date within the *Period*. @@ -2279,7 +2298,7 @@ def accrued(self, settlement: datetime): ------- float """ - if isinstance(self.fixed_rate, NoInput): + if self.cashflow is None: # self.fixed_rate is NoInput return None else: if settlement <= self.start or settlement >= self.end: @@ -2313,11 +2332,9 @@ def npv( if self.start < curve.node_dates[0]: # then mid-period valuation - r, q_start, _v_start = ( - _dual_float((curve.node_dates[0] - self.start).days), - 1.0, - 1.0, - ) + r: float = _dual_float((curve.node_dates[0] - self.start).days) + q_start: DualTypes = 1.0 + _v_start: DualTypes = 1.0 else: r, q_start, _v_start = 0.0, curve[self.start], disc_curve[self.start] @@ -2362,22 +2379,21 @@ def analytic_delta( if self.start < curve.node_dates[0]: # then mid-period valuation - r, q_start, _v_start = ( - _dual_float((curve.node_dates[0] - self.start).days), - 1.0, - 1.0, - ) + r: float = _dual_float((curve.node_dates[0] - self.start).days) + q_start: DualTypes = 1.0 + _v_start: DualTypes = 1.0 else: - r, q_start, _v_start = 0.0, curve[self.start], disc_curve[self.start] + r = 0.0 + q_start = curve[self.start] + _v_start = disc_curve[self.start] # method 1: _ = 0.5 * (1 + r / n) _ *= q_start - q_end _ *= v_end - return _maybe_local( + return _maybe_fx_converted( 0.0001 * self.notional * self.dcf * (q_end * v_payment + _), - False, self.currency, fx, base, @@ -2385,11 +2401,11 @@ def analytic_delta( def cashflows( self, - curve: Curve | dict | NoInput = NoInput(0), + curve: Curve | dict[str, 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), - ): + ) -> dict[str, Any]: """ Return the cashflows of the *CreditPremiumPeriod*. See From afb27008cd3fe11e6121b82f51f97e95763f0320 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Wed, 1 Jan 2025 21:16:20 +0100 Subject: [PATCH 2/3] TYP: periods.py --- python/rateslib/curves/curves.py | 2 +- python/rateslib/dual/__init__.py | 2 +- python/rateslib/periods.py | 117 ++++++++++++++++--------------- 3 files changed, 63 insertions(+), 58 deletions(-) diff --git a/python/rateslib/curves/curves.py b/python/rateslib/curves/curves.py index 8a446498..03e6628f 100644 --- a/python/rateslib/curves/curves.py +++ b/python/rateslib/curves/curves.py @@ -3081,7 +3081,7 @@ def average_rate( """ d: float = _DCF1d[convention.upper()] n: int = (termination - effective).days - _: DualTypes = ((1 + n * d * rate / 100) ** (1 / n) - 1) / d + _: Number = ((1 + n * d * rate / 100) ** (1 / n) - 1) / d return _ * 100, d, n diff --git a/python/rateslib/dual/__init__.py b/python/rateslib/dual/__init__.py index 5077f4f8..0903c452 100644 --- a/python/rateslib/dual/__init__.py +++ b/python/rateslib/dual/__init__.py @@ -22,7 +22,7 @@ def _dual_float(val: DualTypes) -> float: - """Overload for the float() builtin to handle Pyo3 issues with Variabe""" + """Overload for the float() builtin to handle Pyo3 issues with Variable""" try: return float(val) # type: ignore[arg-type] except TypeError as e: # val is not Number but a Variable diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index ebc89a9d..596420fc 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -248,15 +248,15 @@ def __init__( if end < start: raise ValueError("`end` cannot be before `start`.") self.start, self.end, self.payment = start, end, payment - self.frequency = frequency.upper() - self.notional = _drb(defaults.notional, notional) - self.currency = _drb(defaults.base_currency, currency).lower() - self.convention = _drb(defaults.convention, convention) + self.frequency: str = frequency.upper() + self.notional: float = _drb(defaults.notional, notional) + self.currency: str = _drb(defaults.base_currency, currency).lower() + self.convention: str = _drb(defaults.convention, convention) self.termination = termination self.freq_months = defaults.frequency_months[self.frequency] - self.stub = stub - self.roll = roll - self.calendar = calendar + self.stub: bool = stub + self.roll: int | str | NoInput = roll + self.calendar: CalInput = calendar def __repr__(self) -> str: return f"" @@ -2179,7 +2179,7 @@ def _get_analytic_delta_quadratic_coeffs( self, fore_curve: Curve, disc_curve: Curve - ) -> tuple[Number, Number]: + ) -> tuple[DualTypes, DualTypes]: """ For use in the Leg._spread calculation get the 'a' and 'b' coefficients """ @@ -2197,7 +2197,8 @@ def _get_analytic_delta_quadratic_coeffs( d2rdz2: Number = d * (n - 1) * (1 + (r / 100 + z / 10000) * d) ** (n - 2) / 1e8 drdz: Number = (1 + (r / 100 + z / 10000) * d) ** (n - 1) / 1e4 Nvd = -self.notional * disc_curve[self.payment] * self.dcf - a, b = 0.5 * Nvd * d2rdz2, Nvd * drdz + a: DualTypes = 0.5 * Nvd * d2rdz2 + b: DualTypes = Nvd * drdz elif self.spread_compound_method == "isda_flat_compounding": # d2rdz2 = 0.0 drdz = (1 + comb(n, 2) / n * r / 100 * d + comb(n, 3) / n * (r / 100 * d) ** 2) / 1e4 @@ -2401,7 +2402,7 @@ def analytic_delta( def cashflows( self, - curve: Curve | dict[str, Curve] | NoInput = NoInput(0), + curve: Curve | NoInput = NoInput(0), # type: ignore[override] disc_curve: Curve | NoInput = NoInput(0), fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), @@ -2414,7 +2415,8 @@ def cashflows( fx, base = _get_fx_and_base(self.currency, fx, base) if not isinstance(curve, NoInput) and not isinstance(disc_curve, NoInput): - npv = _dual_float(self.npv(curve, disc_curve)) + npv_: DualTypes = self.npv(curve, disc_curve) # type: ignore[assignment] + npv = _dual_float(npv_) npv_fx = npv * _dual_float(fx) survival = _dual_float(curve[self.end]) else: @@ -2422,9 +2424,9 @@ def cashflows( return { **super().cashflows(curve, disc_curve, fx, base), - defaults.headers["rate"]: _dual_float(self.fixed_rate), + defaults.headers["rate"]: None if isinstance(self.fixed_rate, NoInput) else _dual_float(self.fixed_rate), defaults.headers["survival"]: survival, - defaults.headers["cashflow"]: _dual_float(self.cashflow), + defaults.headers["cashflow"]: None if self.cashflow is None else _dual_float(self.cashflow), defaults.headers["npv"]: npv, defaults.headers["fx"]: _dual_float(fx), defaults.headers["npv_fx"]: npv_fx, @@ -2477,15 +2479,15 @@ class CreditProtectionPeriod(BasePeriod): 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) + **kwargs: Any, + ) -> None: + self.recovery_rate: DualTypes = _drb(defaults.cds_recovery_rate, recovery_rate) if self.recovery_rate < 0.0 and self.recovery_rate > 1.0: raise ValueError("`recovery_rate` value must be in [0.0, 1.0]") - self.discretization = _drb(defaults.cds_protection_discretization, discretization) + self.discretization: int = _drb(defaults.cds_protection_discretization, discretization) super().__init__(*args, **kwargs) @property @@ -2518,7 +2520,9 @@ def npv( else: s2 = self.start - value, q2, v2 = 0.0, curve[s2], disc_curve[s2] + value: DualTypes = 0.0 + q2: DualTypes = curve[s2] + v2: DualTypes = disc_curve[s2] while s2 < self.end: q1, v1 = q2, v2 s2 = s2 + timedelta(days=self.discretization) @@ -2547,11 +2551,11 @@ def analytic_delta( def cashflows( self, - curve: Curve | dict | NoInput = NoInput(0), + curve: Curve | NoInput = NoInput(0), # type: ignore[override] 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), - ): + ) -> dict[str, Any]: """ Return the cashflows of the *CreditProtectionPeriod*. See @@ -2560,7 +2564,8 @@ def cashflows( fx, base = _get_fx_and_base(self.currency, fx, base) if not isinstance(curve, NoInput) and not isinstance(disc_curve, NoInput): - npv = _dual_float(self.npv(curve, disc_curve)) + npv_: DualTypes = self.npv(curve, disc_curve) # type: ignore[assignment] + npv = _dual_float(npv_) npv_fx = npv * _dual_float(fx) survival = _dual_float(curve[self.end]) else: @@ -2598,9 +2603,9 @@ def analytic_rec_risk( self.recovery_rate = Variable(rr.real, ["__recovery_rate__"]) else: self.recovery_rate = Variable(_dual_float(rr), ["__recovery_rate__"]) - pv = self.npv(curve, disc_curve, fx, base, False) + pv: Dual | Dual2 | Variable = self.npv(curve, disc_curve, fx, base, False) # type: ignore[assignment] self.recovery_rate = rr - _ = _dual_float(gradient(pv, ["__recovery_rate__"], order=1)[0]) + _: float = _dual_float(gradient(pv, ["__recovery_rate__"], order=1)[0]) return _ * 0.01 @@ -2674,16 +2679,16 @@ def __init__( self.notional, self.payment = notional, payment self.currency = _drb(defaults.base_currency, currency).lower() self.stub_type = stub_type - self._rate = rate if isinstance(rate, NoInput) else _dual_float(rate) + self._rate: float | NoInput = rate if isinstance(rate, NoInput) else _dual_float(rate) def __repr__(self) -> str: return f"" - def rate(self) -> DualTypes | None: + def rate(self) -> float | None: """ Return the associated rate initialised with the *Cashflow*. Not used for calculations. """ - return self._rate + return None if isinstance(self._rate, NoInput) else self._rate def npv( self, @@ -2692,16 +2697,14 @@ def npv( fx: float | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), local: bool = False, - ) -> DualTypes: + ) -> DualTypes | dict[str, DualTypes]: """ Return the NPV of the *Cashflow*. See :meth:`BasePeriod.npv()` """ - disc_curve_: Curve | NoInput = _disc_maybe_from_curve(curve, disc_curve) - if not isinstance(disc_curve, Curve) and isinstance(curve, NoInput): - raise TypeError("`curves` have not been supplied correctly.") - value = self.cashflow * disc_curve_[self.payment] + disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve) + value: DualTypes = self.cashflow * disc_curve_[self.payment] return _maybe_local(value, local, self.currency, fx, base) def cashflows( @@ -2710,20 +2713,21 @@ def cashflows( disc_curve: Curve | NoInput = NoInput(0), fx: float | FXRates | FXForwards | NoInput = NoInput(0), base: str | NoInput = NoInput(0), - ) -> dict: + ) -> dict[str, Any]: """ Return the cashflows of the *Cashflow*. See :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_, _ = _get_fx_and_base(self.currency, fx, base) if isinstance(disc_curve_, NoInput): npv, npv_fx, df, collateral = None, None, None, None else: - npv = _dual_float(self.npv(curve, disc_curve_)) - npv_fx = npv * _dual_float(fx) + npv_: DualTypes = self.npv(curve, disc_curve_) # type: ignore[assignment] + npv = _dual_float(npv_) + npv_fx = npv * _dual_float(fx_) df, collateral = _dual_float(disc_curve_[self.payment]), disc_curve_.collateral try: @@ -2748,13 +2752,13 @@ def cashflows( # defaults.headers["spread"]: None, defaults.headers["cashflow"]: cashflow_, defaults.headers["npv"]: npv, - defaults.headers["fx"]: _dual_float(fx), + defaults.headers["fx"]: _dual_float(fx_), defaults.headers["npv_fx"]: npv_fx, defaults.headers["collateral"]: collateral, } @property - def cashflow(self): + def cashflow(self) -> float: return -self.notional def analytic_delta( @@ -2782,13 +2786,19 @@ class IndexMixin(metaclass=ABCMeta): Abstract base class to include methods and properties related to indexed *Periods*. """ - index_base: float | Series | NoInput = NoInput(0) + index_base: DualTypes | NoInput = NoInput(0) index_method: str = "" - index_fixings: float | Series | NoInput = NoInput(0) + index_fixings: float | Series[float] | NoInput = NoInput(0) index_lag: int | NoInput = NoInput(0) + index_only: bool = False payment: datetime = datetime(1990, 1, 1) currency: str = "" + @property + @abstractmethod + def real_cashflow(self) -> DualTypes | None: + pass # pragma: no cover + def cashflow(self, curve: Curve | NoInput = NoInput(0)) -> DualTypes | None: """ float, Dual or Dual2 : The calculated value from rate, dcf and notional, @@ -2796,16 +2806,17 @@ def cashflow(self, curve: Curve | NoInput = NoInput(0)) -> DualTypes | None: """ if self.real_cashflow is None: return None - index_ratio, _, _ = self.index_ratio(curve) - if index_ratio is None: - return None else: - if self.index_only: - _ = -1.0 + index_ratio, _, _ = self.index_ratio(curve) + if index_ratio is None: + return None else: - _ = 0.0 - _ = self.real_cashflow * (index_ratio + _) - return _ + if self.index_only: + notional_ = -1.0 + else: + notional_ = 0.0 + ret: DualTypes = self.real_cashflow * (index_ratio + notional_) + return ret def index_ratio(self, curve: Curve | NoInput = NoInput(0)) -> tuple: """ @@ -2933,11 +2944,6 @@ def npv( value = self.cashflow(curve) * disc_curve_[self.payment] return _maybe_local(value, local, self.currency, fx, base) - @property - @abstractmethod - def real_cashflow(self): - pass # pragma: no cover - class IndexFixedPeriod(IndexMixin, FixedPeriod): # type: ignore[misc] """ @@ -3028,7 +3034,6 @@ def __init__( # raise ValueError("`index_base` cannot be None.") self.index_base = index_base self.index_fixings = index_fixings - self.index_only = False self.index_method = ( defaults.index_method if isinstance(index_method, NoInput) else index_method.lower() ) @@ -3193,7 +3198,7 @@ def __init__( self, *args, index_base: float, - index_fixings: float | Series | NoInput = NoInput(0), + index_fixings: float | Series[float] | NoInput = NoInput(0), index_method: str | NoInput = NoInput(0), index_lag: int | NoInput = NoInput(0), index_only: bool = False, From 436143bc8f36180da4d08c856b5b072d6861bbd2 Mon Sep 17 00:00:00 2001 From: JHM Darbyshire <24256554+attack68@users.noreply.github.com> Date: Wed, 1 Jan 2025 21:18:30 +0100 Subject: [PATCH 3/3] REF: divisions to multiplications --- python/rateslib/periods.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index 596420fc..0b140b0d 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -1818,7 +1818,7 @@ def _ibor_fixings_table( if self.stub: # then must perform an interpolated calculation return self._ibor_stub_fixings_table(curve, disc_curve, right, risk) - else: # not self.stub: + else: # not self.stub: # then extract the one relevant curve from dict curve_: Curve = _get_ibor_curve_from_dict(self.freq_months, curve) else: @@ -1990,7 +1990,9 @@ def _rfr_fixings_array( if self.fixing_method in ["rfr_lockout", "rfr_lockout_avg"]: rates_dual.iloc[-self.method_param :] = rates_dual.iloc[-self.method_param - 1] if "avg" in self.fixing_method: - rate: Dual = self._rate_rfr_avg_with_spread(rates_dual.to_numpy(), d["dcf_vals"].to_numpy()) # type: ignore[assignment] + rate: Dual = self._rate_rfr_avg_with_spread( + rates_dual.to_numpy(), d["dcf_vals"].to_numpy() + ) # type: ignore[assignment] else: rate = self._rate_rfr_isda_compounded_with_spread( # type: ignore[assignment] rates_dual.to_numpy(), d["dcf_vals"].to_numpy() @@ -2101,7 +2103,7 @@ def _get_method_dcf_endpoints( return start_obs, end_obs, start_dcf, end_dcf def _get_method_dcf_markers( - self, calendar: CalTypes, convention: str, exposure: bool =False + self, calendar: CalTypes, convention: str, exposure: bool = False ) -> tuple[ Series[datetime], Series[datetime], @@ -2176,9 +2178,7 @@ def _get_method_dcf_markers( return obs_dates, dcf_dates, dcf_vals, obs_vals # type: ignore[return-value] def _get_analytic_delta_quadratic_coeffs( - self, - fore_curve: Curve, - disc_curve: Curve + self, fore_curve: Curve, disc_curve: Curve ) -> tuple[DualTypes, DualTypes]: """ For use in the Leg._spread calculation get the 'a' and 'b' coefficients @@ -2424,9 +2424,13 @@ def cashflows( return { **super().cashflows(curve, disc_curve, fx, base), - defaults.headers["rate"]: None if isinstance(self.fixed_rate, NoInput) else _dual_float(self.fixed_rate), + defaults.headers["rate"]: None + if isinstance(self.fixed_rate, NoInput) + else _dual_float(self.fixed_rate), defaults.headers["survival"]: survival, - defaults.headers["cashflow"]: None if self.cashflow is None else _dual_float(self.cashflow), + defaults.headers["cashflow"]: None + if self.cashflow is None + else _dual_float(self.cashflow), defaults.headers["npv"]: npv, defaults.headers["fx"]: _dual_float(fx), defaults.headers["npv_fx"]: npv_fx,