diff --git a/docs/source/d_legs.rst b/docs/source/d_legs.rst index 95f50410..cf8d6e23 100644 --- a/docs/source/d_legs.rst +++ b/docs/source/d_legs.rst @@ -47,6 +47,8 @@ The following *Legs* are provided, click on the links for a full description of rateslib.legs.ZeroIndexLeg rateslib.legs.FixedLegMtm rateslib.legs.FloatLegMtm + rateslib.legs.CreditProtectionLeg + rateslib.legs.CreditPremiumLeg rateslib.legs.CustomLeg *Legs*, similar to *Periods*, are defined as having the following the methods: diff --git a/docs/source/d_periods.rst b/docs/source/d_periods.rst index 8bd86eef..4b1a207a 100644 --- a/docs/source/d_periods.rst +++ b/docs/source/d_periods.rst @@ -94,3 +94,17 @@ Every volatility *Period* type is endowed with the following methods: rateslib.periods.FXOptionPeriod.analytic_greeks rateslib.periods.FXOptionPeriod.rate rateslib.periods.FXOptionPeriod.implied_vol + +Credit Period +************** + +*Credit* periods provide the calculations for default protection and premium calculations +using *recovery rates* and *survivial probabilities* defined by a *Curve* of hazard rates. + +.. inheritance-diagram:: rateslib.periods.CreditPremiumPeriod rateslib.periods.CreditProtectionPeriod + :private-bases: + :parts: 1 + +.. autosummary:: + rateslib.periods.CreditPremiumPeriod + rateslib.periods.CreditProtectionPeriod diff --git a/docs/source/e_multicurrency.rst b/docs/source/e_multicurrency.rst index 3b8230a5..26c2b8a1 100644 --- a/docs/source/e_multicurrency.rst +++ b/docs/source/e_multicurrency.rst @@ -22,4 +22,4 @@ structures. rateslib.instruments.XCS rateslib.instruments.FXSwap rateslib.instruments.FXExchange - rateslib.instruments.forward_fx + rateslib.fx.forward_fx diff --git a/python/rateslib/default.py b/python/rateslib/default.py index 7aff69c5..3379c4ac 100644 --- a/python/rateslib/default.py +++ b/python/rateslib/default.py @@ -173,6 +173,8 @@ def __init__(self): self.fx_option_metric = "pips" self.cds_premium_accrued = True + self.cds_recovery_rate = 0.40 + self.cds_protection_discretization = 31 # Curves @@ -246,6 +248,7 @@ def __init__(self): "strike": "Strike", # CDS headers "survival": "Survival", + "recovery": "Recovery", } # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International diff --git a/python/rateslib/legs.py b/python/rateslib/legs.py index b1c05668..50eebca9 100644 --- a/python/rateslib/legs.py +++ b/python/rateslib/legs.py @@ -40,6 +40,7 @@ from rateslib.periods import ( Cashflow, CreditPremiumPeriod, + CreditProtectionPeriod, FixedPeriod, FloatPeriod, IndexCashflow, @@ -1848,6 +1849,127 @@ def _regular_period( ) +class CreditProtectionLeg(BaseLeg): + """ + Create a credit protection leg composed of :class:`~rateslib.periods.CreditProtectionPeriod` s. + + Parameters + ---------- + args : tuple + Required positional args to :class:`BaseLeg`. + recovery_rate : float, Dual, Dual2, optional + The assumed recovery rate that defines payment on credit default. Set by ``defaults``. + discretization : int, optional + The number of days to discretize the numerical integration over possible credit defaults. + Set by ``defaults``. + kwargs : dict + Required keyword arguments to :class:`BaseLeg`. + + Notes + ----- + The NPV of a credit protection leg is the sum of the period NPVs. + + .. math:: + + P = \\sum_{i=1}^n P_i + + The analytic delta is the sum of the period analytic deltas. + + .. math:: + + A = -\\frac{\\partial P}{\\partial S} = \\sum_{i=1}^n -\\frac{\\partial P_i}{\\partial S} + + Examples + -------- + + .. ipython:: python + :suppress: + + from rateslib.curves import Curve + from rateslib.legs import CreditProtectionLeg + from datetime import datetime as dt + + .. ipython:: python + + disc_curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.98}) + hazard_curve = Curve({dt(2022, 1, 1): 1.0, dt(2023, 1, 1): 0.995}) + protection_leg = CreditProtectionLeg( + dt(2022, 1, 1), "9M", "Z", + recovery_rate=0.40, + notional=1000000, + ) + premium_leg.cashflows(hazard_curve, disc_curve) + premium_leg.npv(hazard_curve, disc_curve) + """ # noqa: E501 + + def __init__( + self, + *args, + recovery_rate: DualTypes | NoInput = NoInput(0), + discretization: int | NoInput = NoInput(0), + **kwargs, + ): + self.recovery_rate = _drb(defaults.cds_recovery_rate, recovery_rate) + self.discretization = _drb(defaults.cds_protection_discretization, discretization) + super().__init__(*args, **kwargs) + self._set_periods() + + def analytic_delta(self, *args, **kwargs): + """ + Return the analytic delta of the *CreditProtectionLeg* via summing all periods. + + For arguments see + :meth:`BasePeriod.analytic_delta()`. + """ + return super().analytic_delta(*args, **kwargs) + + def cashflows(self, *args, **kwargs) -> DataFrame: + """ + Return the properties of the *CreditProtectionLeg* used in calculating cashflows. + + For arguments see + :meth:`BasePeriod.cashflows()`. + """ + return super().cashflows(*args, **kwargs) + + def npv(self, *args, **kwargs): + """ + Return the NPV of the *CreditProtectionLeg* via summing all periods. + + For arguments see + :meth:`BasePeriod.npv()`. + """ + return super().npv(*args, **kwargs) + + def _set_periods(self) -> None: + return super()._set_periods() + + def _regular_period( + self, + start: datetime, + end: datetime, + payment: datetime, + notional: float, + stub: bool, + iterator: int, + ): + return CreditProtectionPeriod( + recovery_rate=self.recovery_rate, + discretization=self.discretization, + start=start, + end=end, + payment=payment, + frequency=self.schedule.frequency, + notional=notional, + currency=self.currency, + convention=self.convention, + termination=self.schedule.termination, + stub=stub, + roll=self.schedule.roll, + calendar=self.schedule.calendar, + ) + + # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International # Commercial use of this code, and/or copying and redistribution is prohibited. # Contact rateslib at gmail.com if this code is observed outside its intended sphere. @@ -2566,4 +2688,5 @@ def analytic_delta(self, *args, **kwargs): "ZeroFloatLeg", "ZeroIndexLeg", "CreditPremiumLeg", + "CreditProtectionLeg", ] diff --git a/python/rateslib/periods.py b/python/rateslib/periods.py index 3846227a..8d2882f7 100644 --- a/python/rateslib/periods.py +++ b/python/rateslib/periods.py @@ -22,7 +22,7 @@ import warnings from abc import ABCMeta, abstractmethod -from datetime import datetime +from datetime import datetime, timedelta from math import comb, log import numpy as np @@ -1914,6 +1914,151 @@ def cashflows( } +class CreditProtectionPeriod(BasePeriod): + """ + Create a credit protection period defined by a recovery rate. + + Parameters + ---------- + args : dict + Required positional args to :class:`BasePeriod`. + recovery_rate : float, Dual, Dual2, optional + The assumed recovery rate that defines payment on credit default. Set by ``defaults``. + discretization : int, optional + The number of days to discretize the numerical integration over possible credit defaults. + Set by ``defaults``. + kwargs : dict + Required keyword arguments to :class:`BasePeriod`. + + Notes + ----- + The ``cashflow``, paid on a credit event, is defined as follows; + + .. math:: + + C = -N(1-R) + + where *R* is the recovery rate. + + The :meth:`~rateslib.periods.BasePeriod.npv` is defined as a discretized sum of inter-period blocks whose + probability of default and protection payment sum to give an expected payment; + + .. math:: + + j &= [n/discretization] \\\\ + P &= C \\sum_{i=1}^{j} \\frac{1}{2} \\left ( v(m_{i-1}) + v_(m_{i}) \\right ) \\left ( Q(m_{i-1}) - Q(m_{i}) \\right ) \\\\ + + The *start* and *end* of the period are restricted by the *Curve* if the *Period* is current (i.e. *today* is + later than *start*) + + The :meth:`~rateslib.periods.BasePeriod.analytic_delta` is defined as; + + .. math:: + + A = 0 + """ # noqa: E501 + + def __init__( + self, + *args, + recovery_rate: DualTypes | NoInput = NoInput(0), + discretization: int | NoInput = NoInput(0), + **kwargs, + ): + self.recovery_rate = _drb(defaults.cds_recovery_rate, recovery_rate) + if float(self.recovery_rate) < 0.0 and float(self.recovery_rate) > 1.0: + raise ValueError("`recovery_rate` must be in [0.0, 1.0]") + self.discretization = _drb(defaults.cds_protection_discretization, discretization) + super().__init__(*args, **kwargs) + + @property + def cashflow(self) -> DualTypes: + """ + float, Dual or Dual2 : The calculated protection amount determined from notional + and recovery rate. + """ + return -self.notional * (1 - self.recovery_rate) + + def npv( + self, + curve: Curve | NoInput = NoInput(0), + disc_curve: Curve | NoInput = NoInput(0), + fx: float | FXRates | FXForwards | NoInput = NoInput(0), + base: str | NoInput = NoInput(0), + local: bool = False, + ) -> DualTypes | dict[str, DualTypes]: + """ + Return the NPV of the *CreditProtectionPeriod*. + See :meth:`BasePeriod.npv()` + """ + if not isinstance(disc_curve, Curve) and disc_curve is NoInput.blank: + raise TypeError("`curves` have not been supplied correctly.") + if not isinstance(curve, Curve) and curve is NoInput.blank: + raise TypeError("`curves` have not been supplied correctly.") + + if self.start < curve.node_dates[0]: + s2 = curve.node_dates[0] + else: + s2 = self.start + + value, q2, v2 = 0.0, curve[s2], disc_curve[s2] + while s2 < self.end: + q1, v1 = q2, v2 + s2 = s2 + timedelta(days=self.discretization) + if s2 > self.end: + s2 = self.end + q2, v2 = curve[s2], disc_curve[s2] + value += 0.5 * (v1 + v2) * (q1 - q2) + + value *= self.cashflow + return _maybe_local(value, local, self.currency, fx, base) + + def analytic_delta( + self, + curve: Curve | NoInput = NoInput(0), + disc_curve: Curve | NoInput = NoInput(0), + fx: float | FXRates | FXForwards | NoInput = NoInput(0), + base: str | NoInput = NoInput(0), + ) -> DualTypes: + """ + Return the analytic delta of the *CreditProtectionPeriod*. + See + :meth:`BasePeriod.analytic_delta()` + """ + return 0.0 + + def cashflows( + self, + curve: Curve | dict | NoInput = NoInput(0), + disc_curve: Curve | NoInput = NoInput(0), + fx: float | FXRates | FXForwards | NoInput = NoInput(0), + base: str | NoInput = NoInput(0), + ): + """ + Return the cashflows of the *CreditProtectionPeriod*. + See + :meth:`BasePeriod.cashflows()` + """ + fx, base = _get_fx_and_base(self.currency, fx, base) + + if curve is not NoInput.blank and disc_curve is not NoInput.blank: + npv = float(self.npv(curve, disc_curve)) + npv_fx = npv * float(fx) + survival = float(curve[self.end]) + else: + npv, npv_fx, survival = None, None, None + + return { + **super().cashflows(curve, disc_curve, fx, base), + defaults.headers["recovery"]: float(self.recovery_rate), + defaults.headers["survival"]: survival, + defaults.headers["cashflow"]: float(self.cashflow), + defaults.headers["npv"]: npv, + defaults.headers["fx"]: float(fx), + defaults.headers["npv_fx"]: npv_fx, + } + + class Cashflow: """ Create a single cashflow amount on a payment date (effectively a CustomPeriod). diff --git a/python/tests/test_legs.py b/python/tests/test_legs.py index 48314b03..11019242 100644 --- a/python/tests/test_legs.py +++ b/python/tests/test_legs.py @@ -11,6 +11,7 @@ from rateslib.legs import ( Cashflow, CreditPremiumLeg, + CreditProtectionLeg, CustomLeg, FixedLeg, FixedLegMtm, @@ -1159,6 +1160,57 @@ def test_premium_leg_set_credit_spread(self, curve) -> None: assert leg.periods[0].credit_spread == 2.0 +class TestCreditProtectionLeg: + def test_leg_analytic_delta(self, hazard_curve, curve) -> None: + leg = CreditProtectionLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 6, 1), + payment_lag=2, + notional=1e9, + frequency="Q", + ) + result = leg.analytic_delta(hazard_curve, curve) + assert abs(result) < 1e-7 + + @pytest.mark.parametrize(("premium_accrued"), [True, False]) + def test_leg_npv(self, hazard_curve, curve, premium_accrued) -> None: + leg = CreditProtectionLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 6, 1), + payment_lag=2, + notional=1e9, + frequency="Z", + ) + result = leg.npv(hazard_curve, curve) + expected = -1390937.7593346273 + assert abs(result - expected) < 1e-7 + + def test_leg_cashflows(self, hazard_curve, curve) -> None: + leg = CreditProtectionLeg( + effective=dt(2022, 1, 1), + termination=dt(2022, 6, 1), + notional=-1e9, + convention="Act360", + frequency="Q", + ) + result = leg.cashflows(hazard_curve, curve) + # test a couple of return elements + assert abs(result.loc[0, defaults.headers["cashflow"]] - 600e6) < 1e-4 + assert abs(result.loc[1, defaults.headers["df"]] - 0.98307) < 1e-4 + assert abs(result.loc[1, defaults.headers["notional"]] + 1e9) < 1e-7 + + def test_leg_zero_sched(self): + leg = CreditProtectionLeg( + effective=dt(2022, 1, 1), + termination=dt(2024, 6, 1), + notional=-1e9, + convention="Act360", + frequency="Z", + ) + assert len(leg.periods) == 1 + assert leg.periods[0].end == dt(2024, 6, 1) + + class TestIndexFixedLegExchange: @pytest.mark.parametrize( "i_fixings", diff --git a/python/tests/test_periods.py b/python/tests/test_periods.py index 6ab8236b..0d218279 100644 --- a/python/tests/test_periods.py +++ b/python/tests/test_periods.py @@ -16,6 +16,7 @@ from rateslib.periods import ( Cashflow, CreditPremiumPeriod, + CreditProtectionPeriod, FixedPeriod, FloatPeriod, FXCallPeriod, @@ -1786,6 +1787,184 @@ def test_null_cashflow(self): assert premium_period.cashflow is None +class TestCreditProtectionPeriod: + def test_period_npv(self, hazard_curve, curve, fxr) -> None: + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + exp = -596995.981290412 + result = period.npv(hazard_curve, curve) + assert abs(result - exp) < 1e-7 + + result = period.npv(hazard_curve, curve, fxr, "nok") + assert abs(result - exp * 10.0) < 1e-6 + + def test_period_npv_raises(self, curve, hazard_curve) -> None: + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + with pytest.raises( + TypeError, + match=re.escape("`curves` have not been supplied correctly."), + ): + period.npv(hazard_curve) + with pytest.raises( + TypeError, + match=re.escape("`curves` have not been supplied correctly."), + ): + period.npv(NoInput(0), curve) + + def test_period_analytic_delta(self, hazard_curve, curve, fxr) -> None: + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + result = period.analytic_delta(hazard_curve, curve) + assert abs(result - 0.0) < 1e-7 + + result = period.analytic_delta(hazard_curve, curve, fxr, "nok") + assert abs(result - 0.0 * 10.0) < 1e-7 + + def test_period_analytic_delta_fxr_base(self, hazard_curve, curve, fxr) -> None: + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + fxr = FXRates({"usdnok": 10.0}, base="NOK") + result = period.analytic_delta(hazard_curve, curve, fxr) + assert abs(result - 0.0) < 1e-7 + + def test_period_cashflows(self, hazard_curve, curve, fxr) -> None: + # also test the inputs to fx as float and as FXRates (10 is for + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + + cashflow = -period.notional * (1 - period.recovery_rate) + expected = { + defaults.headers["type"]: "CreditProtectionPeriod", + defaults.headers["stub_type"]: "Regular", + defaults.headers["a_acc_start"]: dt(2022, 1, 1), + defaults.headers["a_acc_end"]: dt(2022, 4, 1), + defaults.headers["payment"]: dt(2022, 4, 3), + defaults.headers["notional"]: 1e9, + defaults.headers["currency"]: "USD", + defaults.headers["convention"]: "Act360", + defaults.headers["dcf"]: period.dcf, + defaults.headers["df"]: 0.9897791268897856, + defaults.headers["recovery"]: 0.4, + defaults.headers["survival"]: 0.999, + defaults.headers["npv"]: -596995.9812904124, + defaults.headers["cashflow"]: cashflow, + defaults.headers["fx"]: 10.0, + defaults.headers["npv_fx"]: -596995.9812904124 * 10.0, + defaults.headers["collateral"]: None, + } + result = period.cashflows(hazard_curve, curve, fx=fxr, base="nok") + assert result == expected + + def test_period_cashflows_no_curves(self, fxr) -> None: + # also test the inputs to fx as float and as FXRates (10 is for + period = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 3), + notional=1e9, + convention="Act360", + termination=dt(2022, 4, 1), + frequency="Q", + currency="usd", + ) + cashflow = -period.notional * (1 - period.recovery_rate) + expected = { + defaults.headers["type"]: "CreditProtectionPeriod", + defaults.headers["stub_type"]: "Regular", + defaults.headers["a_acc_start"]: dt(2022, 1, 1), + defaults.headers["a_acc_end"]: dt(2022, 4, 1), + defaults.headers["payment"]: dt(2022, 4, 3), + defaults.headers["notional"]: 1e9, + defaults.headers["currency"]: "USD", + defaults.headers["convention"]: "Act360", + defaults.headers["dcf"]: period.dcf, + defaults.headers["df"]: None, + defaults.headers["recovery"]: 0.4, + defaults.headers["survival"]: None, + defaults.headers["npv"]: None, + defaults.headers["cashflow"]: cashflow, + defaults.headers["fx"]: 10.0, + defaults.headers["npv_fx"]: None, + defaults.headers["collateral"]: None, + } + result = period.cashflows(fx=fxr, base="nok") + assert result == expected + + def test_discretization_period(self, hazard_curve, curve): + p1 = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 1), + notional=1e9, + frequency="Q", + discretization=1, + ) + p2 = CreditProtectionPeriod( + start=dt(2022, 1, 1), + end=dt(2022, 4, 1), + payment=dt(2022, 4, 1), + notional=1e9, + frequency="Q", + discretization=31, + ) + r1 = p1.npv(hazard_curve, curve) + r2 = p2.npv(hazard_curve, curve) + assert 0.1 < abs(r1 - r2) < 1.0 # very similar result but not identical + + def test_mid_period(self, hazard_curve, curve): + period = CreditProtectionPeriod( + start=dt(2021, 10, 4), + end=dt(2022, 1, 4), + payment=dt(2022, 1, 4), + notional=1e9, + frequency="Q", + ) + r1 = period.npv(hazard_curve, curve) + exp = -20006.321837529074 + assert abs(r1 - exp) < 1e-7 + + class TestCashflow: def test_cashflow_analytic_delta(self, curve) -> None: cashflow = Cashflow(notional=1e6, payment=dt(2022, 1, 1))