mypy fixes

pyproject.toml

@@ -146,7 +146,7 @@ exclude = [
     "/instruments",
     "fx_volatility.py",
     "legs.py",
-    "periods.py",
+    # "periods.py",
     "solver.py",
 ]
 strict = true

python/rateslib/curves/curves.py

@@ -494,31 +494,46 @@ def rate(
             )
             curve_act360.rate(dt(2022, 2, 1), dt(2022, 3, 1))
         """
+        try:
+            _: DualTypes = self._rate_with_raise(
+                effective, termination, modifier, float_spread, spread_compound_method
+            )
+        except ZeroDivisionError as e:
+            if not "effective:" in str(e):
+                return None
+            raise e
+        except ValueError as e:
+            if "`effective` date for rate period is before" in str(e):
+                return None
+            raise e
+        return _
+
+    def _rate_with_raise(
+        self,
+        effective: datetime,
+        termination: datetime | str | NoInput,
+        modifier: str | NoInput = NoInput(1),
+        float_spread: float | NoInput = NoInput(0),
+        spread_compound_method: str | NoInput = NoInput(0),
+    ) -> DualTypes:
         modifier_ = _drb(self.modifier, modifier)
-        # calendar = self.calendar if calendar is False else calendar
-        # convention = self.convention if convention is None else convention
 
         if effective < self.node_dates[0]:  # Alternative solution to PR 172.
-            return None
-            # raise ValueError(
-            #     "`effective` date for rate period is before the initial node date of the Curve.\n"
-            #     "If you are trying to calculate a rate for an historical FloatPeriod have you "
-            #     "neglected to supply appropriate `fixings`?\n"
-            #     "See Documentation > Cookbook > Working with Fixings."
-            # )
+            raise ValueError(
+                "`effective` date for rate period is before the initial node date of the Curve.\n"
+                "If you are trying to calculate a rate for an historical FloatPeriod have you "
+                "neglected to supply appropriate `fixings`?\n"
+                "See Documentation > Cookbook > Working with Fixings."
+            )
         if isinstance(termination, str):
             termination = add_tenor(effective, termination, modifier_, self.calendar)
         elif isinstance(termination, NoInput):
             raise ValueError("`termination` must be supplied for rate of DF based Curve.")
 
-        try:
-            n_, d_ = self[effective], self[termination]
-            df_ratio = n_ / d_
-        except ZeroDivisionError:
-            return None
-
         if termination == effective:
             raise ZeroDivisionError(f"effective: {effective}, termination: {termination}")
+
+        df_ratio = self[effective] / self[termination]
         n_, d_ = (df_ratio - 1), dcf(effective, termination, self.convention)
         _: DualTypes = n_ / d_ * 100
 
@@ -535,9 +550,9 @@ def rate(
                 r_bar, d, n = average_rate(effective, termination, self.convention, _)
                 rd = r_bar / 100 * d
                 _ = (
-                    (r_bar + float_spread / 100)
-                    / n
-                    * (comb(n, 1) + comb(n, 2) * rd + comb(n, 3) * rd**2)
+                        (r_bar + float_spread / 100)
+                        / n
+                        * (comb(n, 1) + comb(n, 2) * rd + comb(n, 3) * rd**2)
                 )
                 return _
             else:
@@ -548,6 +563,7 @@ def rate(
 
         return _
 
+
     def clear_cache(self) -> None:
         """
         Clear the cache of values on a *Curve* type.
@@ -1478,21 +1494,22 @@ class LineCurve(Curve):
     _ini_solve = 0  # No constraint placed on initial node in Solver
     _base_type = "values"
 
-    def __init__(  # type: ignore[no-untyped-def]
+    def __init__(
         self,
-        *args,
-        **kwargs,
+        *args: Any,
+        **kwargs: Any,
     ) -> None:
         super().__init__(*args, **kwargs)
 
     # def plot(self, *args, **kwargs):
     #     return super().plot(*args, **kwargs)
 
-    def rate(  # type: ignore[no-untyped-def]
+    def rate(
         self,
         effective: datetime,
-        *args,
-    ):
+        *args: Any,
+        **kwargs: Any,
+    ) -> DualTypes | None:
         """
         Return the curve value for a given date.
 
@@ -1509,8 +1526,22 @@ def rate(  # type: ignore[no-untyped-def]
         -------
         float, Dual, or Dual2
         """
+        try:
+            _: DualTypes = self._rate_with_raise(effective, *args, **kwargs)
+        except ValueError as e:
+            if "`effective` before initial LineCurve date" in str(e):
+                return None
+            raise e
+        return _
+
+    def _rate_with_raise(
+        self,
+        effective: datetime,
+        *args: Any,
+        **kwargs: Any,
+    ) -> DualTypes:
         if effective < self.node_dates[0]:  # Alternative solution to PR 172.
-            return None
+            raise ValueError("`effective` before initial LineCurve date.")
         return self[effective]
 
     def shift(

python/rateslib/dual/__init__.py

@@ -24,8 +24,8 @@
 def _dual_float(val: DualTypes) -> float:
     """Overload for the float() builtin to handle Pyo3 issues with Variabe"""
     try:
-        return float(val)
-    except TypeError:  # val is not Dual or Dual2
+        return float(val)  # type: ignore[arg-type]
+    except TypeError:  # val is not Number but a Variable
         #  This does not work well with rust.
         #  See: https://github.com/PyO3/pyo3/issues/3672
         #  and https://github.com/PyO3/pyo3/discussions/3911

python/rateslib/periods.py

@@ -1118,7 +1118,9 @@ def _rate_ibor(self, curve: Curve | dict[str, Curve] | NoInput) -> DualTypes:
                 "Must supply a valid `curve` for forecasting rates for the FloatPeriod."
             )
         elif not isinstance(curve, dict):
-            return method[curve._base_type](curve)
+            # TODO (low); this doesnt type well because of floating _base_type attribute.
+            # consider wrapping to some NamedTuple record
+            return method[curve._base_type](curve)  # type: ignore[arg-type]
         else:
             if not self.stub:
                 curve = _get_ibor_curve_from_dict(self.freq_months, curve)
@@ -1137,18 +1139,18 @@ def _rate_ibor_from_line_curve(self, curve: LineCurve) -> DualTypes:
         fixing_date = curve.calendar.lag(self.start, -self.method_param, False)
         return curve[fixing_date] + self.float_spread / 100
 
-    def _rate_ibor_interpolated_ibor_from_dict(self, curve: dict[str, Curve]) -> DualTypes:
+    def _rate_ibor_interpolated_ibor_from_dict(self, curve: dict[str, Curve]) -> DualTypes | None:
         """
         Get the rate on all available curves in dict and then determine the ones to interpolate.
         """
         calendar = next(iter(curve.values())).calendar  # note: ASSUMES all curve calendars are same
         fixing_date = add_tenor(self.start, f"-{self.method_param}B", "NONE", calendar)
 
-        def _rate(c: Curve, tenor):
+        def _rate(c: Curve, tenor: str) -> DualTypes | None:
             if c._base_type == "dfs":
                 return c.rate(self.start, tenor)
             else:  # values
-                return c.rate(fixing_date)
+                return c.rate(fixing_date, tenor)  # tenor is not used on LineCurve
 
         values = {add_tenor(self.start, k, "MF", calendar): _rate(v, k) for k, v in curve.items()}
         values = dict(sorted(values.items()))
@@ -1169,12 +1171,12 @@ def _rate(c: Curve, tenor):
             return rates[0]
         else:
             i = index_left(dates, len(dates), self.end)
-            _ = rates[i] + (rates[i + 1] - rates[i]) * (
+            _: DualTypes = rates[i] + (rates[i + 1] - rates[i]) * (
                 (self.end - dates[i]) / (dates[i + 1] - dates[i])
             )
             return _
 
-    def _rate_rfr(self, curve: Curve | dict | NoInput) -> Number:
+    def _rate_rfr(self, curve: Curve | dict[str, Curve] | NoInput) -> DualTypes | None:
         if isinstance(self.fixings, float | Dual | Dual2):
             # if fixings is a single value then return that value (curve unused can be NoInput)
             if (
@@ -1209,7 +1211,7 @@ def _rate_rfr(self, curve: Curve | dict | NoInput) -> Number:
                 "Do not supply a dict of curves for RFR based methods.",
             )
 
-    def _rate_rfr_from_df_curve(self, curve: Curve):
+    def _rate_rfr_from_df_curve(self, curve: Curve) -> DualTypes:
         if isinstance(curve, NoInput):
             # then attempt to get rate from fixings
             return self._rfr_rate_from_individual_fixings(curve)
@@ -1299,7 +1301,7 @@ def _rate_rfr_isda_compounded_with_spread(self, rates, dcf_vals):
                 "'isda_compounding', 'isda_flat_compounding'}.",
             )
 
-    def _rfr_rate_from_individual_fixings(self, curve):
+    def _rfr_rate_from_individual_fixings(self, curve: Curve) -> DualTypes:
         cal_, conv_ = self._maybe_get_cal_and_conv_from_curve(curve)
 
         data = self._rfr_get_individual_fixings_data(cal_, conv_, curve)
@@ -1427,11 +1429,11 @@ def _rfr_get_individual_fixings_data(
 
     def fixings_table(
         self,
-        curve: Curve | dict,
+        curve: Curve | dict[str, Curve],
         approximate: bool = False,
-        disc_curve: Curve = NoInput(0),
+        disc_curve: Curve | NoInput = NoInput(0),
         right: datetime | NoInput = NoInput(0),
-    ):
+    ) -> DataFrame:
         """
         Return a DataFrame of fixing exposures.
 
@@ -1741,7 +1743,7 @@ def _fixings_table_fast(
         elif "ibor" in self.fixing_method:
             return self._ibor_fixings_table(curve, disc_curve, right=right)
 
-    def _ibor_fixings_table(self, curve, disc_curve, right, risk=None):
+    def _ibor_fixings_table(self, curve, disc_curve, right, risk: DualTypes | NoInput = NoInput(0)) -> DataFrame:
         """
         Calculate a fixings_table under an IBOR based methodology.
 
@@ -1767,7 +1769,14 @@ def _ibor_fixings_table(self, curve, disc_curve, right, risk=None):
             curve, disc_curve, f"{self.freq_months}m", right, risk
         )
 
-    def _ibor_single_tenor_fixings_table(self, curve, disc_curve, tenor, right, risk=None):
+    def _ibor_single_tenor_fixings_table(
+        self,
+        curve,
+        disc_curve,
+        tenor,
+        right: datetime | NoInput,
+        risk: DualTypes | NoInput = NoInput(0)
+    ) -> DataFrame:
         calendar = curve.calendar
         fixing_dt = calendar.lag(self.start, -self.method_param, False)
         if not isinstance(right, NoInput) and fixing_dt > right:
@@ -1785,7 +1794,7 @@ def _ibor_single_tenor_fixings_table(self, curve, disc_curve, tenor, right, risk
             reg_end_dt = add_tenor(self.start, tenor, curve.modifier, calendar)
             reg_dcf = dcf(self.start, reg_end_dt, curve.convention, reg_end_dt)
 
-            if self.fixings is not NoInput.blank or fixing_dt < curve.node_dates[0]:
+            if not isinstance(self.fixings, NoInput) or fixing_dt < curve.node_dates[0]:
                 # then fixing is set so return zero exposure.
                 _rate = NA if self.fixings is NoInput.blank else float(self.rate(curve))
                 df = DataFrame(
@@ -1799,7 +1808,7 @@ def _ibor_single_tenor_fixings_table(self, curve, disc_curve, tenor, right, risk
                 ).set_index("obs_dates")
             else:
                 risk = (
-                    -self.notional * self.dcf * disc_curve[self.payment] if risk is None else risk
+                    -self.notional * self.dcf * disc_curve[self.payment] if isinstance(risk, NoInput) else risk
                 )
                 df = DataFrame(
                     {
@@ -1817,7 +1826,7 @@ def _ibor_single_tenor_fixings_table(self, curve, disc_curve, tenor, right, risk
         return df
 
     def _ibor_stub_fixings_table(
-        self, curve: dict, disc_curve: Curve, right: datetime | NoInput, risk=None
+        self, curve: dict, disc_curve: Curve, right: datetime | NoInput, risk: DualTypes | NoInput = NoInput(0)
     ):
         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()}
@@ -1845,7 +1854,7 @@ def _ibor_stub_fixings_table(
             a2 = (self.end - reg_end_dts[i]) / (reg_end_dts[i + 1] - reg_end_dts[i])
             a1 = 1.0 - a2
 
-        risk = -self.notional * self.dcf * disc_curve[self.payment] if risk is None else risk
+        risk = -self.notional * self.dcf * disc_curve[self.payment] if isinstance(risk, NoInput) else risk
         tenor1, tenor2 = list(values.values())[i], list(values.values())[i + 1]
         df1 = self._ibor_single_tenor_fixings_table(
             curve[tenor1], disc_curve, tenor1, right, risk * a1