REF: review all the *Leg* spreadfunctions adding tests (#143) (#1064)

Co-authored-by: Mike Lync <[email protected]>
Co-authored-by: JHM Darbyshire <[email protected]>

Author: mikelync

python/rateslib/legs/components/credit.py

@@ -12,13 +12,17 @@
 if TYPE_CHECKING:
     from rateslib.typing import (  # pragma: no cover
         FX_,
+        CurveOption_,
         DualTypes,
         DualTypes_,
+        FXForwards_,
+        FXVolOption_,
         Schedule,
         _BaseCurve_,
         _SettlementParams,
         bool_,
         datetime,
+        datetime_,
         str_,
     )
 
@@ -171,6 +175,28 @@ def __init__(
         # self._exchange_periods = (None, None)
         # self._mtm_exchange_periods = None
 
+    def spread(
+        self,
+        *,
+        target_npv: DualTypes,
+        rate_curve: CurveOption_ = NoInput(0),
+        index_curve: _BaseCurve_ = NoInput(0),
+        disc_curve: _BaseCurve_ = NoInput(0),
+        fx: FXForwards_ = NoInput(0),
+        fx_vol: FXVolOption_ = NoInput(0),
+        settlement: datetime_ = NoInput(0),
+        forward: datetime_ = NoInput(0),
+    ) -> DualTypes:
+        a_delta = self.local_analytic_delta(
+            rate_curve=rate_curve,
+            disc_curve=disc_curve,
+            index_curve=index_curve,
+            fx=fx,
+            forward=forward,
+            settlement=settlement,
+        )
+        return -target_npv / a_delta
+
 
 class CreditProtectionLeg(_BaseLeg):
     """

python/rateslib/legs/components/fixed.py

@@ -5,6 +5,9 @@
 
 import rateslib.errors as err
 from rateslib import defaults
+from rateslib.curves._parsers import (
+    _disc_required_maybe_from_curve,
+)
 from rateslib.data.fixings import _leg_fixings_to_list
 from rateslib.enums.generics import NoInput, _drb
 from rateslib.legs.components.amortization import Amortization, _AmortizationType, _get_amortization
@@ -21,16 +24,15 @@
 
 if TYPE_CHECKING:
     from rateslib.typing import (  # pragma: no cover
-        FX_,
         CurveOption_,
         DualTypes,
         DualTypes_,
         FXForwards_,
+        FXVolOption_,
         IndexMethod,
         LegFixings,
         Schedule,
         Series,
-        _BaseCurve,
         _BaseCurve_,
         _SettlementParams,
         datetime,
@@ -654,53 +656,18 @@ def fx_delivery(i: int) -> datetime:
         else:
             self._mtm_exchange_periods = None
 
-    def _spread(
+    def spread(
         self,
+        *,
         target_npv: DualTypes,
-        rate_curve: CurveOption_,
-        disc_curve: _BaseCurve_,
-        index_curve: _BaseCurve_,
+        rate_curve: CurveOption_ = NoInput(0),
+        index_curve: _BaseCurve_ = NoInput(0),
+        disc_curve: _BaseCurve_ = NoInput(0),
         fx: FXForwards_ = NoInput(0),
-        forward: datetime_ = NoInput(0),
+        fx_vol: FXVolOption_ = NoInput(0),
         settlement: datetime_ = NoInput(0),
+        forward: datetime_ = NoInput(0),
     ) -> DualTypes:
-        """
-        Calculates the ``fixed_rate`` to match a specific target NPV on the leg.
-
-        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.**
-        rate_curve : Curve or LineCurve
-            The forecast curve passed to analytic delta calculation.
-        disc_curve : Curve
-            The discounting curve passed to analytic delta calculation.
-        index_curve : _BaseCurve_
-            The curve used for forecasting index values.
-        fx : FXForwards, optional
-            Required for multi-currency legs which are MTM exchanged.
-
-        Returns
-        -------
-        float, Dual, Dual2
-
-        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
-        --------
-        """
         a_delta = self.local_analytic_delta(
             rate_curve=rate_curve,
             disc_curve=disc_curve,
@@ -866,19 +833,36 @@ def fixed_rate(self, value: DualTypes_) -> None:
         for period in self._regular_periods:
             period.rate_params.fixed_rate = value
 
-    def _spread(
+    def spread(
         self,
+        *,
         target_npv: DualTypes,
-        rate_curve: CurveOption_,
-        disc_curve: _BaseCurve,
+        rate_curve: CurveOption_ = NoInput(0),
         index_curve: _BaseCurve_ = NoInput(0),
+        disc_curve: _BaseCurve_ = NoInput(0),
         fx: FXForwards_ = NoInput(0),
+        fx_vol: FXVolOption_ = NoInput(0),
+        settlement: datetime_ = NoInput(0),
+        forward: datetime_ = NoInput(0),
     ) -> DualTypes:
-        """
-        Overload the _spread calc to use analytic delta based on period rate
-        """
+        disc_curve_ = _disc_required_maybe_from_curve(rate_curve, disc_curve)
 
-        unindexed_target_npv = target_npv / self._regular_periods[0].index_up(
+        if not isinstance(settlement, NoInput):
+            if settlement > self.settlement_params.ex_dividend:
+                raise ZeroDivisionError(
+                    "A `spread` cannot be determined when the *Leg* always has zero value.\n"
+                    "The given `settlement` is after the `ex_dividend` date."
+                )
+            else:
+                w_fwd = disc_curve_[_drb(settlement, forward)]
+        else:
+            if isinstance(forward, NoInput):
+                w_fwd = 1.0
+            else:
+                w_fwd = disc_curve_[forward]
+
+        immediate_target_npv = target_npv * w_fwd
+        unindexed_target_npv = immediate_target_npv / self._regular_periods[0].index_up(
             1.0, index_curve=index_curve
         )
         unindexed_reference_target_npv = unindexed_target_npv / self._regular_periods[
@@ -888,7 +872,7 @@ def _spread(
         f = self.schedule.periods_per_annum
         d = self._regular_periods[0].dcf
         N = self.settlement_params.notional
-        w = disc_curve[self.settlement_params.payment]
+        w = disc_curve_[self.settlement_params.payment]
         R = ((-unindexed_reference_target_npv / (N * w) + 1) ** (1 / (d * f)) - 1) * f * 10000.0
         return R
 
@@ -998,15 +982,3 @@ def __init__(
         )
         self._exchange_periods = (_ini_cf,)
         self._regular_periods = (_final_cf,)
-
-    def _spread(
-        self,
-        target_npv: DualTypes,
-        fore_curve: CurveOption_,
-        disc_curve: CurveOption_,
-        fx: FX_ = NoInput(0),
-    ) -> DualTypes:
-        """
-        Overload the _spread calc to use analytic delta based on period rate
-        """
-        raise NotImplementedError()

python/rateslib/legs/components/float.py

@@ -20,19 +20,20 @@
 
 if TYPE_CHECKING:
     from rateslib.typing import (  # pragma: no cover
-        FX_,
         CurveOption_,
         DualTypes,
         DualTypes_,
         FloatRateSeries,
         Frequency,
         FXForwards_,
+        FXVolOption_,
         IndexMethod,
         LegFixings,
         Schedule,
         _BaseCurve_,
         _BasePeriod,
         datetime,
+        datetime_,
         int_,
         str_,
     )
@@ -367,82 +368,51 @@ def _is_linear(self) -> bool:
             return False
         return True
 
-    def _spread(
+    def spread(
         self,
+        *,
         target_npv: DualTypes,
-        rate_curve: CurveOption_,
-        disc_curve: _BaseCurve_,
+        rate_curve: CurveOption_ = NoInput(0),
         index_curve: _BaseCurve_ = NoInput(0),
+        disc_curve: _BaseCurve_ = NoInput(0),
         fx: FXForwards_ = NoInput(0),
+        fx_vol: FXVolOption_ = NoInput(0),
+        settlement: datetime_ = NoInput(0),
+        forward: datetime_ = NoInput(0),
     ) -> DualTypes:
-        """
-        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.**
-        rate_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.
-        index_curve : _BaseCurve, optional
-            The index curve used for forecasting index values.
-
-        Returns
-        -------
-        float, Dual, Dual2
-
-        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: DualTypes = self.local_analytic_delta(
+            local_npv = self.local_npv(
                 rate_curve=rate_curve,
                 disc_curve=disc_curve,
                 index_curve=index_curve,
                 fx=fx,
+                forward=forward,
+                settlement=settlement,
             )
-            return -target_npv / a_delta
-        else:
-            original_z = self.float_spread
-            original_npv = self.npv(
+            a_delta = self.local_analytic_delta(
                 rate_curve=rate_curve,
                 disc_curve=disc_curve,
                 index_curve=index_curve,
                 fx=fx,
+                forward=forward,
+                settlement=settlement,
             )
+            return -(target_npv - local_npv) / a_delta
+        else:
+            original_z = self.float_spread
 
             def s(g: DualTypes) -> DualTypes:
                 """
                 This determines the NPV change subject to a given float spread change denoted, g.
                 """
-                self.float_spread = g + original_z
-                return (
-                    self.npv(  # type: ignore[operator]
-                        rate_curve=rate_curve,
-                        disc_curve=disc_curve,
-                        index_curve=index_curve,
-                        fx=fx,
-                    )
-                    - original_npv
+                self.float_spread = g
+                return self.local_npv(
+                    rate_curve=rate_curve,
+                    disc_curve=disc_curve,
+                    index_curve=index_curve,
+                    fx=fx,
+                    forward=forward,
+                    settlement=settlement,
                 )
 
             result = ift_1dim(
@@ -620,14 +590,46 @@ def __init__(
             ),
         )
 
-    def _spread(
+    def spread(
         self,
+        *,
         target_npv: DualTypes,
-        fore_curve: CurveOption_,
-        disc_curve: CurveOption_,
-        fx: FX_ = NoInput(0),
+        rate_curve: CurveOption_ = NoInput(0),
+        index_curve: _BaseCurve_ = NoInput(0),
+        disc_curve: _BaseCurve_ = NoInput(0),
+        fx: FXForwards_ = NoInput(0),
+        fx_vol: FXVolOption_ = NoInput(0),
+        settlement: datetime_ = NoInput(0),
+        forward: datetime_ = NoInput(0),
     ) -> DualTypes:
-        """
-        Overload the _spread calc to use analytic delta based on period rate
-        """
-        raise NotImplementedError()
+        original_z = self.float_spread
+
+        def s(g: DualTypes) -> DualTypes:
+            """
+            This determines the NPV of the *Leg* subject to a given float spread change denoted, g.
+            """
+            self.float_spread = g
+            iteration_local_npv = self.local_npv(
+                rate_curve=rate_curve,
+                disc_curve=disc_curve,
+                index_curve=index_curve,
+                fx=fx,
+                forward=forward,
+                settlement=settlement,
+            )
+            return iteration_local_npv
+
+        result = ift_1dim(
+            s=s,
+            s_tgt=target_npv,
+            h="ytm_quadratic",
+            ini_h_args=(-300, 300, 1200),
+            # h="modified_brent",
+            # ini_h_args=(-10000, 10000),
+            func_tol=1e-6,
+            conv_tol=1e-6,
+        )
+
+        self.float_spread = original_z
+        _: DualTypes = result["g"]
+        return _