Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mikelync merged 1 commit into from
Oct 24, 2025
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions python/rateslib/legs/components/credit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_,
)

Expand Down Expand Up @@ -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):
"""
Expand Down
100 changes: 36 additions & 64 deletions python/rateslib/legs/components/fixed.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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[
Expand All @@ -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

Expand Down Expand Up @@ -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()
130 changes: 66 additions & 64 deletions python/rateslib/legs/components/float.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_,
)
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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 _
Loading
Loading