REF: Leg spreads (#141) (#1061)

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

Author: attack68

python/rateslib/legs/components/fixed.py

@@ -29,10 +29,13 @@
         LegFixings,
         Schedule,
         Series,
+        _BaseCurve_,
         _SettlementParams,
         datetime,
         int_,
         str_,
+        _BaseCurve,
+        FXForwards_,
     )
 
 
@@ -650,6 +653,56 @@ def fx_delivery(i: int) -> datetime:
         else:
             self._mtm_exchange_periods = None
 
+    def _spread(
+        self,
+        target_npv: DualTypes,
+        rate_curve: CurveOption_,
+        disc_curve: CurveOption_,
+        index_curve: _BaseCurve_,
+        fx: FXForwards_ = 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, index_curve=index_curve, fx=fx
+        )
+        return -target_npv / a_delta
+
 
 class ZeroFixedLeg(_BaseLeg):
     """
@@ -808,18 +861,28 @@ def fixed_rate(self, value: DualTypes_) -> None:
     def _spread(
         self,
         target_npv: DualTypes,
-        fore_curve: CurveOption_,
-        disc_curve: CurveOption_,
-        fx: FX_ = NoInput(0),
+        rate_curve: CurveOption_,
+        disc_curve: _BaseCurve,
+        index_curve: _BaseCurve_ = NoInput(0),
+        fx: FXForwards_ = NoInput(0),
     ) -> DualTypes:
         """
         Overload the _spread calc to use analytic delta based on period rate
         """
-        a_delta = self._analytic_delta(fore_curve, disc_curve, fx, self.currency)
-        period_rate = -target_npv / (a_delta * 100)
+
+        unindexed_target_npv = (
+            target_npv / self._regular_periods[0].index_up(1.0, index_curve=index_curve)
+        )
+        unindexed_reference_target_npv = (
+            unindexed_target_npv / self._regular_periods[0].convert_deliverable(1.0, fx=fx)
+        )
+
         f = self.schedule.periods_per_annum
-        _: DualTypes = f * ((1 + period_rate * self.dcf / 100) ** (1 / (self.dcf * f)) - 1)
-        return _ * 10000
+        d = self._regular_periods[0].dcf
+        N = self.settlement_params.notional
+        w = disc_curve[self.settlement_params.payment]
+        R = ((-unindexed_reference_target_npv / (N * w) + 1) ** (1 / (d * f)) - 1) * f * 10000.0
+        return R
 
 
 class ZeroIndexLeg(_BaseLeg):
@@ -938,8 +1001,4 @@ def _spread(
         """
         Overload the _spread calc to use analytic delta based on period rate
         """
-        a_delta = self._analytic_delta(fore_curve, disc_curve, fx, self.currency)
-        period_rate = -target_npv / (a_delta * 100)
-        f = self.schedule.periods_per_annum
-        _: DualTypes = f * ((1 + period_rate * self.dcf / 100) ** (1 / (self.dcf * f)) - 1)
-        return _ * 10000
+        raise NotImplementedError()

python/rateslib/legs/components/float.py

@@ -34,6 +34,7 @@
         datetime,
         int_,
         str_,
+        FXForwards_
     )
 
 
@@ -371,8 +372,8 @@ def _spread(
         target_npv: DualTypes,
         rate_curve: CurveOption_,
         disc_curve: _BaseCurve_,
-        fx: FX_ = NoInput(0),
         index_curve: _BaseCurve_ = NoInput(0),
+        fx: FXForwards_ = NoInput(0),
     ) -> DualTypes:
         """
         Calculates an adjustment to the ``fixed_rate`` or ``float_spread`` to match
@@ -389,6 +390,8 @@ def _spread(
             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
         -------
@@ -411,7 +414,7 @@ def _spread(
         --------
         """
         if self._is_linear:
-            a_delta: DualTypes = self.analytic_delta(  # type: ignore[assignment]
+            a_delta: DualTypes = self.local_analytic_delta(
                 rate_curve=rate_curve,
                 disc_curve=disc_curve,
                 index_curve=index_curve,
@@ -627,8 +630,4 @@ def _spread(
         """
         Overload the _spread calc to use analytic delta based on period rate
         """
-        a_delta = self._analytic_delta(fore_curve, disc_curve, fx, self.currency)
-        period_rate = -target_npv / (a_delta * 100)
-        f = self.schedule.periods_per_annum
-        _: DualTypes = f * ((1 + period_rate * self.dcf / 100) ** (1 / (self.dcf * f)) - 1)
-        return _ * 10000
+        raise NotImplementedError()

python/rateslib/legs/components/protocols/analytic_delta.py

@@ -24,6 +24,58 @@ class _WithAnalyticDelta(Protocol):
     @property
     def periods(self) -> list[_BasePeriod]: ...
 
+    def local_analytic_delta(
+        self,
+        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:
+        """
+        Calculate the analytic rate delta of a *Period* expressed in its local settlement currency.
+
+        Parameters
+        ----------
+        rate_curve: _BaseCurve or dict of such indexed by string tenor, optional
+            Used to forecast floating period rates, if necessary.
+        index_curve: _BaseCurve, optional
+            Used to forecast index values for indexation, if necessary.
+        disc_curve: _BaseCurve, optional
+            Used to discount cashflows.
+        fx: FXForwards, optional
+            The :class:`~rateslib.fx.FXForwards` object used for forecasting the
+            ``fx_fixing`` for deliverable cashflows, if necessary. Or, an
+            :class:`~rateslib.fx.FXRates` object purely for immediate currency conversion.
+        fx_vol: FXDeltaVolSmile, FXSabrSmile, FXDeltaVolSurface, FXSabrSurface, optional
+            The FX volatility *Smile* or *Surface* object used for determining Black calendar
+            day implied volatility values.
+        settlement: datetime, optional
+            The assumed settlement date of the *PV* determination. Used only to evaluate
+            *ex-dividend* status.
+        forward: datetime, optional
+            The future date to project the *PV* to using the ``disc_curve``.
+
+        Returns
+        -------
+        float, Dual, Dual2, Variable
+        """
+        local_analytic_delta: DualTypes = sum(
+            _.try_local_analytic_delta(
+                rate_curve=rate_curve,
+                index_curve=index_curve,
+                disc_curve=disc_curve,
+                fx=fx,
+                fx_vol=fx_vol,
+                settlement=settlement,
+                forward=forward,
+            ).unwrap()
+            for _ in self.periods
+        )
+        return local_analytic_delta
+
     def analytic_delta(
         self,
         *,

python/rateslib/periods/components/fixed_period.py

@@ -421,7 +421,7 @@ class ZeroFixedPeriod(_BasePeriodStatic):
            convention="1",
        )
        period.cashflows()
-    
+
     .. role:: red
 
     .. role:: green

python/rateslib/periods/components/float_period.py

@@ -526,7 +526,7 @@ class ZeroFloatPeriod(_BasePeriodStatic):
     For *analytic delta* purposes the :math:`\xi=-z`.
 
     .. rubric:: Examples
-    
+
     .. ipython:: python
       :suppress:
 
@@ -832,7 +832,7 @@ def unindexed_reference_cashflow(
         # determine each rate from individual Periods
         r_i = [period.rate(rate_curve=rate_curve) for period in self.float_periods]
         d_i = [period.period_params.dcf for period in self.float_periods]
-        r = np.prod(1.0 + np.array(r_i) * np.array(d_i) / 100.0) - 1.0
+        r: DualTypes = np.prod(1.0 + np.array(r_i) * np.array(d_i) / 100.0) - 1.0
         return -self.settlement_params.notional * r
 
     def try_unindexed_reference_cashflow_analytic_delta(

python/tests/legs/test_legs_legacy.py

@@ -1042,8 +1042,55 @@ def test_zero_fixed_spread(self, curve) -> None:
             convention="ActAct",
             fixed_rate=NoInput(0),
         )
-        result = zfl._spread(13140821.29 * curve[dt(2027, 1, 1)], NoInput(0), curve)
-        assert (result / 100 - 2.50) < 1e-3
+        result = zfl._spread(
+            target_npv=13140821.29 * curve[dt(2027, 1, 1)],
+            rate_curve=NoInput(0),
+            disc_curve=curve,
+        )
+        assert abs(result / 100 - 2.50) < 1e-3
+
+    def test_zero_fixed_spread_indexed(self, curve) -> None:
+        zfl = ZeroFixedLeg(
+            schedule=Schedule(
+                effective=dt(2022, 1, 1),
+                termination="5y",
+                payment_lag=0,
+                frequency="A",
+            ),
+            notional=-1e8,
+            convention="ActAct",
+            fixed_rate=NoInput(0),
+            index_base=100.0,
+            index_fixings=110.0,
+        )
+        result = zfl._spread(
+            target_npv=13140821.29 * curve[dt(2027, 1, 1)],
+            rate_curve=NoInput(0),
+            disc_curve=curve,
+        )
+        assert abs(result / 100 - 2.2826266057484057) < 1e-3
+
+    def test_zero_fixed_spread_non_deliverable(self, curve) -> None:
+        zfl = ZeroFixedLeg(
+            schedule=Schedule(
+                effective=dt(2022, 1, 1),
+                termination="5y",
+                payment_lag=0,
+                frequency="A",
+            ),
+            notional=-1e8,
+            convention="ActAct",
+            fixed_rate=NoInput(0),
+            currency="usd",
+            pair="eurusd",
+            fx_fixings=2.0,
+        )
+        result = zfl._spread(
+            target_npv=13140821.29 * curve[dt(2027, 1, 1)],
+            rate_curve=NoInput(0),
+            disc_curve=curve,
+        )
+        assert abs(result / 100 - 1.2808477472765924) < 1e-3
 
     def test_amortization_raises(self) -> None:
         with pytest.raises(TypeError, match="unexpected keyword argument 'amortization'"):
@@ -1504,6 +1551,25 @@ def test_non_deliverable(self, curve):
 
     # v2.5
 
+    def test_fixed_leg_spread(self, curve) -> None:
+        fixed_leg = FixedLeg(
+            schedule=Schedule(
+                effective=dt(2022, 1, 1),
+                termination=dt(2022, 7, 1),
+                payment_lag=2,
+                payment_lag_exchange=1,
+                frequency="Q",
+            ),
+            notional=-1e9,
+            convention="Act360",
+            fixed_rate=4.00,
+            currency="usd",
+        )
+        result = fixed_leg._spread(
+            target_npv=20000000, disc_curve=curve, rate_curve=curve, index_curve=curve
+        )
+        assert abs(result - 403.9491881327746) < 1e-6
+
     @pytest.mark.parametrize("initial", [True, False])
     @pytest.mark.parametrize("final", [True, False])
     @pytest.mark.parametrize("amortization", [True, False])
@@ -2587,7 +2653,12 @@ def test_mtm_leg_exchange_spread(self) -> None:
             rate_curve=fxf.curve("usd", "usd"), disc_curve=fxf.curve("usd", "usd"), fx=fxf
         )
         # a_delta = leg.analytic_delta(fxf.curve("usd", "usd"), fxf.curve("usd", "usd"), fxf)
-        result = leg._spread(100, fxf.curve("usd", "usd"), fxf.curve("usd", "usd"), fxf)
+        result = leg._spread(
+            target_npv=100,
+            rate_curve=fxf.curve("usd", "usd"),
+            disc_curve=fxf.curve("usd", "usd"),
+            fx=fxf,
+        )
         leg.float_spread = result
         npv2 = leg.npv(
             rate_curve=fxf.curve("usd", "usd"), disc_curve=fxf.curve("usd", "usd"), fx=fxf

python/tests/periods/test_periods_legacy.py

@@ -2875,8 +2875,8 @@ def test_fixed_period_npv_raises(self, curve) -> None:
             currency="usd",
         )
         with pytest.raises(
-            ValueError,
-            match=re.escape("`disc_curve` is required but it has not been provided, or c"),
+            TypeError,
+            match=re.escape("`curves` have not been supplied correctly"),
         ):
             fixed_period.npv()
 
@@ -3454,7 +3454,7 @@ def test_cashflow_cashflows(self, curve, fxr, crv, fx) -> None:
         assert result == expected
 
     def test_cashflow_npv_raises(self, curve) -> None:
-        with pytest.raises(ValueError, match="`disc_curve` is required but it has not been pr"):
+        with pytest.raises(TypeError, match="`curves` have not been supplied correctly."):
             Cashflow(notional=1e6, payment=dt(2022, 1, 1)).npv()
         cashflow = Cashflow(notional=1e6, payment=dt(2022, 1, 1))
         assert cashflow.analytic_delta(rate_curve=curve) == 0