TYP: instruments (#622)

Co-authored-by: JHM Darbyshire (M1) <[email protected]>

pyproject.toml

@@ -155,7 +155,6 @@ exclude = [
     "/instruments/generics.py",
     "/instruments/rates/inflation.py",
     "/instruments/rates/multi_currency.py",
-    "/instruments/rates/single_currency.py",
     "solver.py",
 ]
 strict = true

python/rateslib/curves/_parsers.py

@@ -155,7 +155,9 @@ def _get_curves_maybe_from_solver(
     # refactor curves into a list
     if not isinstance(curves, list | tuple):
         # convert isolated value input to list
-        curves_as_list: list[Curve | dict[str, str | Curve] | NoInput | str] = [curves]
+        curves_as_list: list[
+            Curve | dict[str, str | Curve] | dict[str, str] | dict[str, Curve] | NoInput | str
+        ] = [curves]
     else:
         curves_as_list = curves
 

python/rateslib/instruments/base.py

@@ -266,8 +266,8 @@ def cashflows(
             self.leg1.currency,
         )
 
-        df1 = self.leg1.cashflows(curves_[0], curves_[1], fx_, base_)  # type: ignore[arg-type]
-        df2 = self.leg2.cashflows(curves_[2], curves_[3], fx_, base_)  # type: ignore[arg-type]
+        df1 = self.leg1.cashflows(curves_[0], curves_[1], fx_, base_)
+        df2 = self.leg2.cashflows(curves_[2], curves_[3], fx_, base_)
         # filter empty or all NaN
         dfs_filtered = [_ for _ in [df1, df2] if not (_.empty or isna(_).all(axis=None))]
 

python/rateslib/instruments/rates/single_currency.py

@@ -193,7 +193,7 @@ def __init__(
             leg2_fixing_method=leg2_fixing_method,
             leg2_method_param=leg2_method_param,
         )
-        self.kwargs = _update_not_noinput(self.kwargs, user_kwargs)
+        self.kwargs: dict[str, Any] = _update_not_noinput(self.kwargs, user_kwargs)
 
         self._fixed_rate = fixed_rate
         self._leg2_float_spread = leg2_float_spread
@@ -896,7 +896,7 @@ def __init__(
             leg2_fixing_method=leg2_fixing_method,
             leg2_method_param=leg2_method_param,
         )
-        self.kwargs = _update_not_noinput(self.kwargs, user_kwargs)
+        self.kwargs: dict[str, Any] = _update_not_noinput(self.kwargs, user_kwargs)
         self._fixed_rate = fixed_rate
         self._leg2_float_spread = leg2_float_spread
         self.leg1 = ZeroFixedLeg(**_get(self.kwargs, leg=1))
@@ -1251,14 +1251,14 @@ def __init__(
             leg2_fixing_method=leg2_fixing_method,
             leg2_method_param=leg2_method_param,
         )
-        self.kwargs = _update_not_noinput(self.kwargs, user_kwargs)
+        self.kwargs: dict[str, Any] = _update_not_noinput(self.kwargs, user_kwargs)
         self._float_spread = float_spread
         self._leg2_float_spread = leg2_float_spread
         self.leg1 = FloatLeg(**_get(self.kwargs, leg=1))
         self.leg2 = FloatLeg(**_get(self.kwargs, leg=2))
 
     def _set_pricing_mid(self, curves: Curves_, solver: Solver_) -> None:
-        if self.float_spread is NoInput.blank and self.leg2_float_spread is NoInput.blank:
+        if isinstance(self.float_spread, NoInput) and isinstance(self.leg2_float_spread, NoInput):
             # set a pricing parameter for the purpose of pricing NPV at zero.
             rate = self.rate(curves, solver)
             self.leg1.float_spread = _dual_float(rate)
@@ -1553,7 +1553,7 @@ def __init__(
             "leg2_fixing_method": "ibor",
             "leg2_float_spread": 0.0,
         }
-        self.kwargs = _update_not_noinput(self.kwargs, user_kwargs)
+        self.kwargs: dict[str, Any] = _update_not_noinput(self.kwargs, user_kwargs)
 
         # Build
         self._fixed_rate = self.kwargs["fixed_rate"]
@@ -1593,7 +1593,7 @@ def analytic_delta(  # type: ignore[override]
         """
         disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve)
         fx, base = _get_fx_and_base(self.leg1.currency, fx, base)
-        rate = self.rate([curve])  # type: ignore[list-item]
+        rate = self.rate([curve])
         dcf = self._fixed_period.dcf
         _: DualTypes = self.leg1.notional * dcf * disc_curve_[self._payment_date] / 10000
         return fx * _ / (1 + dcf * rate / 100)
@@ -1735,18 +1735,26 @@ def cashflows(
 
         if isinstance(self.fixed_rate, NoInput):
             _fix = None
-            _spd = None
             _cf = None
-            _df = None
-            _npv_local = None
-            _npv_fx = None
         else:
+            _fix = -_dual_float(self.fixed_rate)
             _cf = _dual_float(self.cashflow(curves_[0]))  # type: ignore[arg-type]
+
+        if isinstance(curves_[1], NoInput):
+            _df = None
+        else:
             _df = _dual_float(curves_[1][self._payment_date])
-            _fix = -_dual_float(self.fixed_rate)
-            _spd = -_dual_float(self.rate(curves_[1])) * 100
+
+        _spd = self.rate(curves_[0])
+        if _spd is not None:
+            _spd = -_dual_float(_spd) * 100.0
+
+        if _cf is not None and _df is not None:
             _npv_local = _cf * _df
             _npv_fx = _npv_local * _dual_float(fx__)
+        else:
+            _npv_local = None
+            _npv_fx = None
 
         cfs = self._fixed_period.cashflows(curves_[0], curves_[1], fx__, base_)
         cfs[defaults.headers["type"]] = "FRA"
@@ -1810,14 +1818,17 @@ def fixings_table(
             NoInput(0),
             self.leg2.currency,
         )
+        if isinstance(curves_[2], NoInput) or isinstance(curves_[3], NoInput):
+            raise ValueError("`curves` are not supplied correctly.")
+
         df = self.leg2.fixings_table(
             curve=curves_[2], approximate=approximate, disc_curve=curves_[3]
         )
         rate = self._float_period.rate(curve=curves_[2])
-        scalar = curves_[3][self._payment_date] / curves_[3][self._float_period.payment]
+        scalar: DualTypes = curves_[3][self._payment_date] / curves_[3][self._float_period.payment]
         scalar *= 1.0 / (1.0 + self._float_period.dcf * rate / 100.0)
-        df[(curves_[2].id, "risk")] *= scalar
-        df[(curves_[2].id, "notional")] *= scalar
+        df[(curves_[2].id, "risk")] *= scalar  # type: ignore[operator, union-attr]
+        df[(curves_[2].id, "notional")] *= scalar  # type: ignore[operator, union-attr]
         return _trim_df_by_index(df, NoInput(0), right)
 
     def delta(self, *args: Any, **kwargs: Any) -> DataFrame:
@@ -1848,8 +1859,8 @@ def _payment_date(self) -> datetime:
 
     @property
     def _fixed_period(self) -> FixedPeriod:
-        return self.leg1.periods[0]
+        return self.leg1.periods[0]  # type: ignore[return-value]
 
     @property
     def _float_period(self) -> FloatPeriod:
-        return self.leg2.periods[0]
+        return self.leg2.periods[0]  # type: ignore[return-value]

python/rateslib/instruments/utils.py

@@ -16,7 +16,7 @@
     from rateslib.typing import (
         FX,
         Curves_,
-        Curves_Tuple,
+        Curves_DiscTuple,
         Vol,
         Vol_,
         VolOption,
@@ -71,7 +71,7 @@ def _get_curves_fx_and_base_maybe_from_solver(
     fx: FX,
     base: str | NoInput,
     local_ccy: str | NoInput,
-) -> tuple[Curves_Tuple, FX, str | NoInput]:
+) -> tuple[Curves_DiscTuple, FX, str | NoInput]:
     """
     Parses the ``solver``, ``curves``, ``fx`` and ``base`` arguments in combination.
 

python/rateslib/periods.py

@@ -61,7 +61,7 @@
 from rateslib.splines import evaluate
 
 if TYPE_CHECKING:
-    from rateslib.typing import FX, CalInput, CalTypes, CurveOption_, DualTypes, Number
+    from rateslib.typing import FX, CalInput, CalTypes, Curve_, CurveOption_, DualTypes, Number
 
 # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International
 # Commercial use of this code, and/or copying and redistribution is prohibited.
@@ -567,9 +567,9 @@ def npv(
 
     def cashflows(
         self,
-        curve: Curve | dict[str, Curve] | NoInput = NoInput(0),
-        disc_curve: Curve | NoInput = NoInput(0),
-        fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0),
+        curve: CurveOption_ = NoInput(0),
+        disc_curve: Curve_ = NoInput(0),
+        fx: FX = NoInput(0),
         base: str | NoInput = NoInput(0),
     ) -> dict[str, Any]:
         """
@@ -962,9 +962,9 @@ def analytic_delta(
 
     def cashflows(
         self,
-        curve: Curve | dict[str, Curve] | NoInput = NoInput(0),
-        disc_curve: Curve | NoInput = NoInput(0),
-        fx: DualTypes | FXRates | FXForwards | NoInput = NoInput(0),
+        curve: CurveOption_ = NoInput(0),
+        disc_curve: Curve_ = NoInput(0),
+        fx: FX = NoInput(0),
         base: str | NoInput = NoInput(0),
     ) -> dict[str, Any]:
         """
@@ -1025,7 +1025,7 @@ def npv(
 
         return _maybe_local(value, local, self.currency, fx, base)
 
-    def cashflow(self, curve: Curve | dict[str, Curve] | NoInput = NoInput(0)) -> DualTypes | None:
+    def cashflow(self, curve: CurveOption_ = NoInput(0)) -> DualTypes | None:
         """
         Forecast the *Period* cashflow based on a *Curve* providing index rates.
 
@@ -1049,9 +1049,7 @@ def cashflow(self, curve: Curve | dict[str, Curve] | NoInput = NoInput(0)) -> Du
             # probably "needs a `curve` to forecast rate
             return None
 
-    def _maybe_get_cal_and_conv_from_curve(
-        self, curve: Curve | dict[str, Curve] | NoInput
-    ) -> tuple[CalTypes, str]:
+    def _maybe_get_cal_and_conv_from_curve(self, curve: CurveOption_) -> tuple[CalTypes, str]:
         if isinstance(curve, NoInput):
             cal_: CalTypes = get_calendar(self.calendar)
             conv_: str = self.convention
@@ -1073,7 +1071,7 @@ def _maybe_get_cal_and_conv_from_curve(
                 conv_ = curve.convention
         return cal_, conv_
 
-    def rate(self, curve: Curve | dict[str, Curve] | NoInput = NoInput(0)) -> DualTypes:
+    def rate(self, curve: CurveOption_ = NoInput(0)) -> DualTypes:
         """
         Calculating the floating rate for the period.
 

python/rateslib/typing.py

@@ -86,18 +86,19 @@
 from rateslib.curves import Curve as Curve  # noqa: E402
 
 Curve_: TypeAlias = "Curve | NoInput"
+CurveDict: TypeAlias = "dict[str, Curve | str] | dict[str, Curve] | dict[str, str]"
 
 CurveOrId: TypeAlias = "Curve | str"
 CurveOrId_: TypeAlias = "CurveOrId | NoInput"
 
-CurveInput: TypeAlias = "CurveOrId | dict[str, CurveOrId]"
+CurveInput: TypeAlias = "CurveOrId | CurveDict"
 CurveInput_: TypeAlias = "CurveInput | NoInput"
 
 CurveOption: TypeAlias = "Curve | dict[str, Curve]"
 CurveOption_: TypeAlias = "CurveOption | NoInput"
 
-Curves: TypeAlias = "CurveOrId | dict[str, CurveOrId] | list[CurveOrId | dict[str, CurveOrId]]"
-Curves_: TypeAlias = "CurveOrId_ | dict[str, CurveOrId] | list[CurveOrId_ | dict[str, CurveOrId]]"
+Curves: TypeAlias = "CurveOrId | CurveDict | list[CurveOrId | CurveDict]"
+Curves_: TypeAlias = "CurveOrId_ | CurveDict | list[CurveOrId_ | CurveDict]"
 
 Curves_Tuple: TypeAlias = "tuple[CurveOption_, CurveOption_, CurveOption_, CurveOption_]"
 Curves_DiscTuple: TypeAlias = "tuple[CurveOption_, Curve_, CurveOption_, Curve_]"