ENH: add CDS protection pricing (#425)

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

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:

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

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

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

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()<rateslib.periods.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()<rateslib.periods.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()<rateslib.periods.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",
 ]

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()<rateslib.periods.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()<rateslib.periods.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()<rateslib.periods.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).