path: root/python/analytics/credit_default_swap.py

import array
import datetime
import math
import numpy as np
import pandas as pd
import warnings

from dateutil.relativedelta import relativedelta
from itertools import chain
from pandas.tseries.offsets import BDay
from pyisda.curve import SpreadCurve
from pyisda.date import previous_twentieth
from pyisda.legs import ContingentLeg, FeeLeg
from termcolor import colored
from .utils import build_table
from weakref import WeakSet
from yieldcurve import get_curve, rate_helpers, YC, ql_to_jp


class CreditDefaultSwap:
    """ minimal class to represent a credit default swap """

    __slots__ = (
        "_observed",
        "fixed_rate",
        "notional",
        "_start_date",
        "_end_date",
        "recovery",
        "_version",
        "_fee_leg",
        "_default_leg",
        "_value_date",
        "_yc",
        "_sc",
        "_risky_annuity",
        "_spread",
        "_price",
        "name",
        "issue_date",
        "currency",
        "_step_in_date",
        "_accrued",
        "_cash_settle_date",
        "_dl_pv",
        "_pv",
        "_clean_pv",
        "_original_clean_pv",
        "_trade_date",
        "_factor",
    )

    def __init__(
        self, start_date, end_date, recovery, fixed_rate, notional=10e6, issue_date=None
    ):
        """
        start_date : :class:`datetime.date`
            index start_date (Could be issue date, or last imm date)
        end_date : :class:`datetime.date`
            index last date
        recovery :
            recovery rate (between 0 and 1)
        fixed_rate :
            fixed coupon (in bps)
        """
        self.fixed_rate = fixed_rate
        self.notional = notional
        self._start_date = start_date
        self._end_date = end_date
        self.recovery = recovery

        self._fee_leg = FeeLeg(self._start_date, end_date, True, 1.0, 1.0)
        self._default_leg = ContingentLeg(self._start_date, end_date, True)
        self._value_date = None
        self._yc, self._sc = None, None
        self._risky_annuity = None
        self._spread, self._price = None, None
        self.name = None
        self.issue_date = issue_date
        self._factor = 1
        for attr in [
            "currency",
            "_step_in_date",
            "_cash_settle_date",
            "_accrued",
            "_dl_pv",
            "_pv",
            "_clean_pv",
            "_original_clean_pv",
            "_trade_date",
        ]:
            setattr(self, attr, None)
        self._observed = WeakSet()

    def __hash__(self):
        return hash(tuple(getattr(self, k) for k in self._getslots()))

    def _getslots(self):
        classes = reversed(self.__class__.__mro__)
        next(classes)  # skip object
        slots = chain.from_iterable(cls.__slots__ for cls in classes)
        next(slots)  # skip _observed
        yield from slots

    def __getstate__(self):
        return {k: getattr(self, k) for k in self._getslots()}

    def __setstate__(self, state):
        for name, value in state.items():
            setattr(self, name, value)
        self._observed = WeakSet()

    @property
    def start_date(self):
        return self._start_date

    @property
    def end_date(self):
        return self._end_date

    @start_date.setter
    def start_date(self, d):
        self._fee_leg = FeeLeg(d, self.end_date, True, 1.0, 1.0)
        self._default_leg = ContingentLeg(d, self.end_date, True)
        self._start_date = d

    @end_date.setter
    def end_date(self, d):
        self._fee_leg = FeeLeg(self.start_date, d, True, 1.0, 1.0)
        self._default_leg = ContingentLeg(self.start_date, d, True)
        self._end_date = d

    @property
    def spread(self):
        if self._spread is not None:
            return self._spread * 1e4
        else:
            return None

    @property
    def direction(self):
        if self.notional > 0.0:
            return "Buyer"
        else:
            return "Seller"

    @direction.setter
    def direction(self, d):
        if d == "Buyer":
            self.notional = abs(self.notional)
        elif d == "Seller":
            self.notional = -abs(self.notional)
        else:
            raise ValueError("Direction needs to be either 'Buyer' or 'Seller'")

    def _update(self):
        self._sc = SpreadCurve(
            self._yc.base_date,
            self._yc,
            self.start_date,
            self._step_in_date,
            self._cash_settle_date,
            [self.end_date],
            np.array([self._spread]),
            np.zeros(1),
            np.array([self.recovery]),
        )

        self._risky_annuity = self._fee_leg.pv(
            self.value_date,
            self._step_in_date,
            self._cash_settle_date,
            self._yc,
            self._sc,
            False,
        )
        self._dl_pv = self._default_leg.pv(
            self.value_date,
            self._step_in_date,
            self._cash_settle_date,
            self._yc,
            self._sc,
            self.recovery,
        )
        self._pv = self._dl_pv - self._risky_annuity * self.fixed_rate * 1e-4
        self._clean_pv = self._pv + self._accrued * self.fixed_rate * 1e-4
        self._price = 100 * (1 - self._clean_pv)

    @spread.setter
    def spread(self, s):
        """ s: spread in bps """
        if self.spread is None or s != self.spread:
            self._spread = s * 1e-4
            self._update()
            self.notify()

    @property
    def flat_hazard(self):
        sc_data = self._sc.inspect()["data"]
        # conversion to continuous compounding
        return sc_data[0][1]

    @property
    def pv(self):
        return self.notional * self._factor * self._pv

    @pv.setter
    def pv(self, val):
        self._pv = -val / (self.notional * self._factor)
        self._clean_pv = self._pv + self._accrued * self.fixed_rate * 1e-4
        self.price = 100 * (1 - self._clean_pv)

    @property
    def accrued(self):
        return -self.notional * self._factor * self._accrued * self.fixed_rate * 1e-4

    @property
    def days_accrued(self):
        return int(self._accrued * 360)

    @property
    def clean_pv(self):
        return self.notional * self._factor * self._clean_pv

    @property
    def price(self):
        return self._price

    @price.setter
    def price(self, val):
        if self._price is None or math.fabs(val - self._price) > 1e-6:
            self._clean_pv = (100 - val) / 100
            self._sc = SpreadCurve(
                self.value_date,
                self._yc,
                self.start_date,
                self._step_in_date,
                self._cash_settle_date,
                [self.end_date],
                array.array("d", [self.fixed_rate * 1e-4]),
                array.array("d", [self._clean_pv]),
                array.array("d", [self.recovery]),
            )
            self._risky_annuity = self._fee_leg.pv(
                self.value_date,
                self._step_in_date,
                self._cash_settle_date,
                self._yc,
                self._sc,
                False,
            )
            self._dl_pv = self._default_leg.pv(
                self.value_date,
                self._step_in_date,
                self._cash_settle_date,
                self._yc,
                self._sc,
                self.recovery,
            )
            self._pv = self._clean_pv - self._accrued * self.fixed_rate * 1e-4
            self._spread = (
                self._clean_pv / (self._risky_annuity - self._accrued)
                + self.fixed_rate * 1e-4
            )
            self._price = val
            self.notify()

    @property
    def DV01(self):
        old_pv, old_spread = self.pv, self.spread
        self.spread += 1
        dv01 = self.pv - old_pv
        self.spread = old_spread
        return dv01

    @property
    def theta(self):
        old_pv, old_value_date = self.clean_pv, self.value_date
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            self.value_date = self.value_date + relativedelta(days=1)
        carry = self.notional * self.fixed_rate * 1e-4 / 360
        roll_down = self.clean_pv - old_pv
        self.value_date = old_value_date
        return carry + roll_down

    @property
    def IRDV01(self):
        old_pv, old_yc = self.pv, self._yc
        # for rh in self._helpers:
        #     rh.quote += 1e-4
        # self._yc = ql_to_jp(self._ql_yc)
        helpers = rate_helpers(self.currency, evaluation_date=self.value_date)
        for rh in helpers:
            rh.quote.value += 1e-4
        ql_yc = YC(helpers)
        self._yc = ql_to_jp(ql_yc)
        self._update()  # to force recomputation
        new_pv = self.pv
        # for r in self._helpers:
        #     r.quote -= 1e-4
        self._yc = old_yc
        self._update()
        return new_pv - old_pv

    @property
    def rec_risk(self):
        old_recovery = self.recovery
        self.recovery = old_recovery - 0.01
        self._update()
        pv_minus = self.pv
        self.recovery = old_recovery + 0.01
        self._update()
        pv_plus = self.pv
        self.recovery = old_recovery
        self._update()
        return (pv_plus - pv_minus) / 2

    @property
    def jump_to_default(self):
        return -self.notional * (self.recovery + self._clean_pv - 1)

    @property
    def risky_annuity(self):
        return self._risky_annuity - self._accrued

    @property
    def value_date(self):
        if self._value_date is None:
            raise AttributeError("Please set value_date first")
        else:
            return self._value_date

    @value_date.setter
    def value_date(self, d):
        if isinstance(d, datetime.datetime):
            d = d.date()
        self.start_date = previous_twentieth(d)
        self._yc = get_curve(d, self.currency)
        self._value_date = d
        self._step_in_date = d + datetime.timedelta(days=1)
        self._accrued = self._fee_leg.accrued(self._step_in_date)
        self._cash_settle_date = pd.Timestamp(self._value_date) + 3 * BDay()
        if self._spread is not None:
            self._update()
        self.notify()

    def reset_pv(self):
        self._original_clean_pv = self._clean_pv
        self._trade_date = self._value_date

    @property
    def pnl(self):
        if self._original_clean_pv is None:
            raise ValueError("original pv not set")
        else:
            days_accrued = (self.value_date - self._trade_date).days / 360
            return self.notional * (
                self._clean_pv
                - self._original_clean_pv
                - days_accrued * self.fixed_rate * 1e-4
            )

    def notify(self):
        for obj in self._observed:
            obj._update()

    def observe(self, obj):
        self._observed.add(obj)

    def shock(self, params, *, spread_shock, **kwargs):
        r = []
        actual_params = [p for p in params if hasattr(self, p)]
        orig_spread = self.spread
        for ss in spread_shock:
            self.spread = orig_spread * (1 + ss)
            r.append([getattr(self, p) for p in actual_params])
        self.spread = orig_spread
        ind = pd.Index(spread_shock, name="spread_shock", copy=False)
        return pd.DataFrame(r, index=ind, columns=actual_params)

    def __repr__(self):
        if not self.spread:
            raise ValueError("Market spread is missing!")
        if self.days_accrued > 1:
            accrued_str = "Accrued ({} Days)".format(self.days_accrued)
        else:
            accrued_str = "Accrued ({} Day)".format(self.days_accrued)

        s = [
            "{:<20}\tNotional {:>5}MM {}\tFactor {:>28}".format(
                "Buy Protection" if self.notional > 0.0 else "Sell Protection",
                abs(self.notional) / 1_000_000,
                self.currency,
                self._factor,
            ),
            "{:<20}\t{:>15}".format("CDS Index", colored(self.name, attrs=["bold"])),
            "",
        ]
        rows = [
            ["Trd Sprd (bp)", self.spread, "Coupon (bp)", self.fixed_rate],
            ["1st Accr Start", self.issue_date, "Payment Freq", "Quarterly"],
            ["Maturity Date", self.end_date, "Rec Rate", self.recovery],
            ["Bus Day Adj", "Following", "DayCount", "ACT/360"],
        ]
        format_strings = [
            [None, "{:.2f}", None, "{:.0f}"],
            [None, "{:%m/%d/%y}", None, None],
            [None, "{:%m/%d/%y}", None, None],
            [None, None, None, None],
        ]
        s += build_table(rows, format_strings, "{:<20}{:>19}\t\t{:<20}{:>15}")
        s += ["", colored("Calculator", attrs=["bold"])]
        rows = [
            ["Valuation Date", self.value_date],
            ["Cash Settled On", self._cash_settle_date],
        ]
        format_strings = [[None, "{:%m/%d/%y}"], [None, "{:%m/%d/%y}"]]
        s += build_table(rows, format_strings, "{:<20}\t{:>15}")
        s += [""]
        rows = [
            ["Price", self.price, "Spread DV01", self.DV01],
            ["Principal", self.clean_pv, "IR DV01", self.IRDV01],
            [accrued_str, self.accrued, "Rec Risk (1%)", self.rec_risk],
            ["Cash Amount", self.pv, "Def Exposure", self.jump_to_default],
        ]
        format_strings = [
            [None, "{:.8f}", None, "{:,.2f}"],
            [None, "{:,.0f}", None, "{:,.2f}"],
            [None, "{:,.0f}", None, "{:,.2f}"],
            [None, "{:,.0f}", None, "{:,.0f}"],
        ]
        s += build_table(rows, format_strings, "{:<20}{:>19}\t\t{:<20}{:>15}")
        return "\n".join(s)