RUFF: C90 Complexity and TYPs

pyproject.toml

@@ -101,6 +101,8 @@ docstring-code-format = false
 
 [tool.ruff.lint]
 select = [
+    # "ANN",  # flake8-annotations  -- Superceded by the use of mypy
+    # "COM",  # flake8-commas  -- conflicts with ruff format
     "E",  # pycodestyle
     "W",
     "F",  # Pyflakes
@@ -110,17 +112,17 @@ select = [
     "C4",  # flake8-comprehensions
     "S",  # flake8-bandit
     "PIE",  # flake8-pie
-    # "ANN",  # flake8-annotations  -- Superceded by the use of mypy
-    "A",  # flake8-builtins  -- Requires work
-    # "COM",  # flake8-commas  -- conflicts with ruff format
+    "A",  # flake8-builtins
     "Q",  # flake8-quotes
     "PT",  # flake8-pytest-style
-    # "C90",  # mccabe complexity  -- Requires work
+    "C90",  # mccabe complexity  -- Requires work
     "I",  # isort
     "N",  # pep8 naming
     # "RUF",  # -- Requires work
+    # "D", Pydocs -- requires work
 ]
 ignore = [
+    "A005",  # json and typing module name shadowing is allowed
     "PT011", # -- Requires work inputting match statements
     "PIE790",  # unnecessary pass
     "C408",  # unnecessary dict call
@@ -135,7 +137,12 @@ ignore = [
 
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["E402", "N801"]
-"python/tests/*" = ["F401", "B", "N", "S", "ANN"]
+"python/tests/*" = ["F401", "B", "N", "S", "ANN", "D"]
+"rust/*" = ["D"]
+
+[tool.ruff.lint.mccabe]
+# Flag errors (`C901`) whenever the complexity level exceeds 5.
+max-complexity = 14
 
 [tool.mypy]
 packages = [

python/rateslib/calendars/__init__.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import calendar as calendar_mod
+from collections.abc import Callable
 from datetime import datetime
 from typing import TYPE_CHECKING
 
@@ -586,6 +587,12 @@ def _get_fx_expiry_and_delivery(
     return expiry_, delivery_
 
 
+_IS_ROLL: dict[str, Callable[..., bool]] = {
+    "eom": _is_eom,
+    "som": _is_som,
+    "imm": _is_imm,
+}
+
 __all__ = (
     "add_tenor",
     "Cal",

python/rateslib/calendars/rs.py

@@ -140,50 +140,60 @@ def _parse_str_calendar(calendar: str, named: bool) -> CalTypes:
     """Parse the calendar string using Python and construct calendar objects."""
     vectors = calendar.split("|")
     if len(vectors) == 1:
-        calendars = vectors[0].lower().split(",")
-        if len(calendars) == 1:  # only one named calendar is found
-            return defaults.calendars[calendars[0]]  # lookup Hashmap
-        else:
-            # combined calendars are not yet predefined so this does not beenfit from hashmap speed
-            if named:
-                return NamedCal(calendar)
-            else:
-                cals = [defaults.calendars[_] for _ in calendars]
-                cals_: list[Cal] = []
-                for cal in cals:
-                    if isinstance(cal, Cal):
-                        cals_.append(cal)
-                    elif isinstance(cal, NamedCal):
-                        cals_.extend(cal.union_cal.calendars)
-                    else:
-                        cals_.extend(cal.calendars)
-                return UnionCal(cals_, None)
+        return _parse_str_calendar_no_associated(vectors[0], named)
     elif len(vectors) == 2:
+        return _parse_str_calendar_with_associated(vectors[0], vectors[1], named)
+    else:
+        raise ValueError("Cannot use more than one pipe ('|') operator in `calendar`.")
+
+
+def _parse_str_calendar_no_associated(calendar: str, named: bool) -> CalTypes:
+    calendars = calendar.lower().split(",")
+    if len(calendars) == 1:  # only one named calendar is found
+        return defaults.calendars[calendars[0]]  # lookup Hashmap
+    else:
+        # combined calendars are not yet predefined so this does not benefit from hashmap speed
         if named:
             return NamedCal(calendar)
         else:
-            calendars = vectors[0].lower().split(",")
             cals = [defaults.calendars[_] for _ in calendars]
-            cals_ = []
+            cals_: list[Cal] = []
             for cal in cals:
                 if isinstance(cal, Cal):
                     cals_.append(cal)
                 elif isinstance(cal, NamedCal):
                     cals_.extend(cal.union_cal.calendars)
                 else:
                     cals_.extend(cal.calendars)
+            return UnionCal(cals_, None)
 
-            settlement_calendars = vectors[1].lower().split(",")
-            sets = [defaults.calendars[_] for _ in settlement_calendars]
-            sets_: list[Cal] = []
-            for cal in sets:
-                if isinstance(cal, Cal):
-                    sets_.append(cal)
-                elif isinstance(cal, NamedCal):
-                    sets_.extend(cal.union_cal.calendars)
-                else:
-                    sets_.extend(cal.calendars)
 
-            return UnionCal(cals_, sets_)
+def _parse_str_calendar_with_associated(
+    calendar: str, associated_calendar: str, named: bool
+) -> CalTypes:
+    if named:
+        return NamedCal(calendar + "|" + associated_calendar)
     else:
-        raise ValueError("Cannot use more than one pipe ('|') operator in `calendar`.")
+        calendars = calendar.lower().split(",")
+        cals = [defaults.calendars[_] for _ in calendars]
+        cals_ = []
+        for cal in cals:
+            if isinstance(cal, Cal):
+                cals_.append(cal)
+            elif isinstance(cal, NamedCal):
+                cals_.extend(cal.union_cal.calendars)
+            else:
+                cals_.extend(cal.calendars)
+
+        settlement_calendars = associated_calendar.lower().split(",")
+        sets = [defaults.calendars[_] for _ in settlement_calendars]
+        sets_: list[Cal] = []
+        for cal in sets:
+            if isinstance(cal, Cal):
+                sets_.append(cal)
+            elif isinstance(cal, NamedCal):
+                sets_.extend(cal.union_cal.calendars)
+            else:
+                sets_.extend(cal.calendars)
+
+        return UnionCal(cals_, sets_)

python/rateslib/instruments/utils.py

@@ -22,7 +22,7 @@
         CurveInput,
         CurveOption,
         Curves,
-        CurvesList,
+        CurvesTuple,
         Vol,
         Vol_,
         VolOption,
@@ -125,7 +125,7 @@ def _get_curves_maybe_from_solver(
     curves_attr: Curves,
     solver: Solver | NoInput,
     curves: Curves,
-) -> CurvesList:
+) -> CurvesTuple:
     """
     Attempt to resolve curves as a variety of input types to a 4-tuple consisting of:
     (leg1 forecasting, leg1 discounting, leg2 forecasting, leg2 discounting)
@@ -146,18 +146,8 @@ def _get_curves_maybe_from_solver(
 
     # parse curves_as_list
     if isinstance(solver, NoInput):
-
-        def check_curve(curve: str | Curve | dict[str, Curve | str] | NoInput) -> CurveOption:
-            if isinstance(curve, str):
-                raise ValueError("`curves` must contain Curve, not str, if `solver` not given.")
-            elif curve is None or isinstance(curve, NoInput):
-                return NoInput(0)
-            elif isinstance(curve, dict):
-                return {k: check_curve(v) for k, v in curve.items()}  # type: ignore[misc]
-            return curve
-
         curves_parsed: tuple[CurveOption, ...] = tuple(
-            check_curve(curve) for curve in curves_as_list
+            _validate_curve_not_str(curve) for curve in curves_as_list
         )
     else:
         try:
@@ -170,15 +160,33 @@ def check_curve(curve: str | Curve | dict[str, Curve | str] | NoInput) -> CurveO
                 f"The available ids are {list(solver.pre_curves.keys())}.",
             )
 
-    if len(curves_parsed) == 1:
-        curves_parsed *= 4
-    elif len(curves_parsed) == 2:
-        curves_parsed *= 2
-    elif len(curves_parsed) == 3:
-        curves_parsed += (curves_parsed[1],)
-    elif len(curves_parsed) > 4:
+    return _make_4_tuple_of_curve(curves_parsed)
+
+
+def _validate_curve_not_str(curve: CurveInput) -> CurveOption:
+    """
+    Check a curve as a CurveInput type and convert to a more specific CurveOption
+    """
+    if isinstance(curve, str):
+        raise ValueError("`curves` must contain Curve, not str, if `solver` not given.")
+    elif curve is None or isinstance(curve, NoInput):
+        return NoInput(0)
+    elif isinstance(curve, dict):
+        return {k: _validate_curve_not_str(v) for k, v in curve.items()}  # type: ignore[misc]
+    return curve
+
+
+def _make_4_tuple_of_curve(curves: tuple[CurveOption, ...]) -> CurvesTuple:
+    n = len(curves)
+    if n == 1:
+        curves *= 4
+    elif n == 2:
+        curves *= 2
+    elif n == 3:
+        curves += (curves[1],)
+    elif n > 4:
         raise ValueError("Can only supply a maximum of 4 `curves`.")
-    return curves_parsed  # type: ignore[return-value]
+    return curves  # type: ignore[return-value]
 
 
 def _get_curves_fx_and_base_maybe_from_solver(
@@ -188,7 +196,7 @@ def _get_curves_fx_and_base_maybe_from_solver(
     fx: FX,
     base: str | NoInput,
     local_ccy: str | NoInput,
-) -> tuple[CurvesList, FX, str | NoInput]:
+) -> tuple[CurvesTuple, FX, str | NoInput]:
     """
     Parses the ``solver``, ``curves``, ``fx`` and ``base`` arguments in combination.
 

python/rateslib/rs.pyi

@@ -144,7 +144,11 @@ class Dual(_DualOps):
     dual: Arr1dF64 = ...
     @classmethod
     def vars_from(
-        cls, other: Dual, real: float, vars: Sequence[str], dual: Sequence[float] | Arr1dF64  # noqa: A002
+        cls,
+        other: Dual,
+        real: float,
+        vars: Sequence[str],  # noqa: A002
+        dual: Sequence[float] | Arr1dF64,
     ) -> Dual: ...
     def to_dual2(self) -> Dual2: ...
 

python/rateslib/scheduling.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import calendar as calendar_mod
-from collections.abc import Callable, Iterator
+from collections.abc import Iterator
 from datetime import datetime, timedelta
 from itertools import product
 from typing import TYPE_CHECKING, NamedTuple
@@ -10,15 +10,14 @@
 
 from rateslib import defaults
 from rateslib.calendars import (  # type: ignore[attr-defined]
+    _IS_ROLL,
     _adjust_date,
     _get_modifier,
     _get_roll,
     _get_rollday,
     _is_day_type_tenor,
     _is_eom,
     _is_eom_cal,
-    _is_imm,
-    _is_som,
     add_tenor,
     get_calendar,
 )
@@ -756,39 +755,38 @@ def _check_unadjusted_regular_swap(
     if not freq_check:
         return _InvalidSchedule("Months date separation not aligned with frequency.")
 
-    roll = "eom" if roll == 31 else roll
-    iter_: list[tuple[str, Callable[..., bool]]] = [
-        ("eom", _is_eom),
-        ("imm", _is_imm),
-        ("som", _is_som),
-    ]
-    for roll_, _is_roll in iter_:
-        if str(roll).lower() == roll_:
-            if not _is_roll(ueffective):
-                return _InvalidSchedule(f"Non-{roll_} effective date with {roll_} rolls.")
-            if not _is_roll(utermination):
-                return _InvalidSchedule(f"Non-{roll_} termination date with {roll_} rolls.")
-
-    if isinstance(roll, int):
-        if roll in [29, 30]:
-            if ueffective.day != roll and not (ueffective.month == 2 and _is_eom(ueffective)):
-                return _InvalidSchedule(f"Effective date not aligned with {roll} rolls.")
-            if utermination.day != roll and not (utermination.month == 2 and _is_eom(utermination)):
-                return _InvalidSchedule(f"Termination date not aligned with {roll} rolls.")
-        else:
-            if ueffective.day != roll:
-                return _InvalidSchedule(f"Termination date not aligned with {roll} rolls.")
-            if utermination.day != roll:
-                return _InvalidSchedule(f"Termination date not aligned with {roll} rolls.")
-
     if isinstance(roll, NoInput):
         roll = _get_unadjusted_roll(ueffective, utermination, eom)
         if roll == 0:
             return _InvalidSchedule("Roll day could not be inferred from given dates.")
+        else:
+            ueff_ret: _InvalidSchedule | None = None
+            uter_ret: _InvalidSchedule | None = None
+    else:
+        ueff_ret = _validate_date_and_roll(roll, ueffective)
+        uter_ret = _validate_date_and_roll(roll, utermination)
 
+    if isinstance(ueff_ret, _InvalidSchedule):
+        return ueff_ret
+    elif isinstance(uter_ret, _InvalidSchedule):
+        return uter_ret
     return _ValidSchedule(ueffective, utermination, NoInput(0), NoInput(0), frequency, roll, eom)
 
 
+def _validate_date_and_roll(roll: int | str, date: datetime) -> _InvalidSchedule | None:
+    roll = "eom" if roll == 31 else roll
+    if isinstance(roll, str) and not _IS_ROLL[roll.lower()](date):
+        return _InvalidSchedule(f"Non-{roll} effective date with {roll} rolls.")
+    elif isinstance(roll, int):
+        if roll in [29, 30]:
+            if date.day != roll and not (date.month == 2 and _is_eom(date)):
+                return _InvalidSchedule(f"Effective date not aligned with {roll} rolls.")
+        else:
+            if date.day != roll:
+                return _InvalidSchedule(f"Termination date not aligned with {roll} rolls.")
+    return None
+
+
 def _check_regular_swap(
     effective: datetime,
     termination: datetime,

python/rateslib/typing.py

@@ -64,7 +64,7 @@
 CurveOption_: TypeAlias = "Curve | dict[str, Curve]"
 CurveOption: TypeAlias = "CurveOption_ | NoInput"
 
-CurvesList: TypeAlias = "tuple[CurveOption, CurveOption, CurveOption, CurveOption]"
+CurvesTuple: TypeAlias = "tuple[CurveOption, CurveOption, CurveOption, CurveOption]"
 
 Vol_: TypeAlias = "DualTypes | FXDeltaVolSmile | FXDeltaVolSurface | str"
 Vol: TypeAlias = "Vol_ | NoInput"