diff --git a/financepy/market/curves/cds_curve.py b/financepy/market/curves/cds_curve.py index c35a6be6..5c846ce4 100644 --- a/financepy/market/curves/cds_curve.py +++ b/financepy/market/curves/cds_curve.py @@ -11,7 +11,7 @@ from ...utils.global_vars import G_DAYS_IN_YEAR from ...market.curves.discount_curve import DiscountCurve from ...market.curves.interpolator import _uinterpolate, InterpTypes -from ...utils.helpers import input_time, table_to_string +from ...utils.helpers import input_time, table_to_string, times_from_dates from ...utils.frequency import annual_frequency, FrequencyTypes from ...utils.helpers import check_argument_types, _func_name from ...utils.helpers import label_to_string @@ -139,11 +139,18 @@ def survival_prob(self, dt): supports vectorisation.""" if isinstance(dt, Date): - t = (dt - self.anchor_dt) / G_DAYS_IN_YEAR + t = times_from_dates(self.anchor_dt, dt, self.dc_type) elif isinstance(dt, list): - t = np.array(dt) - else: + if len(dt) > 0 and isinstance(dt[0], Date): + t = times_from_dates(self.anchor_dt, dt, self.dc_type) + else: + t = np.array(dt, dtype=float) + elif isinstance(dt, (float, int)): + t = float(dt) + elif isinstance(dt, np.ndarray): t = dt + else: + raise FinError("Unknown time type") if np.any(t < 0.0): raise FinError("Survival Date before curve anchor date") @@ -155,7 +162,7 @@ def survival_prob(self, dt): qs[i] = _uinterpolate(t[i], self._times, self._qs, self.interp_method.value) return qs elif np.isscalar(t): - q = _uinterpolate(t, self._times, self._qs, self.interp_method.value) + q = _uinterpolate(float(t), self._times, self._qs, self.interp_method.value) return q raise FinError("Unknown time type") @@ -167,15 +174,16 @@ def df(self, dt): function supports vectorisation.""" if isinstance(dt, Date): - t = (dt - self.anchor_dt) / G_DAYS_IN_YEAR + return self.libor_curve.df(dt) elif isinstance(dt, list): - t = np.array(dt) + if len(dt) > 0 and isinstance(dt[0], Date): + return self.libor_curve.df(dt) + else: + return self.libor_curve.df_t(np.array(dt, dtype=float)) + elif isinstance(dt, (float, int, np.ndarray)): + return self.libor_curve.df_t(dt) else: - t = dt - - df = self.libor_curve.df_t(t) - - return df + raise FinError(f"Unknown date or time type: {type(dt)}") ########################################################################### @@ -200,7 +208,7 @@ def build_curve(self): self.recovery_rate, ) - t_mat = (maturity_dt - self.anchor_dt) / G_DAYS_IN_YEAR + t_mat = times_from_dates(self.anchor_dt, maturity_dt, self.dc_type) q = self._qs[i] self._times = np.append(self._times, t_mat) diff --git a/tests/unit/test_FinCDSCurve.py b/tests/unit/test_FinCDSCurve.py index d149ab21..c50d25fa 100644 --- a/tests/unit/test_FinCDSCurve.py +++ b/tests/unit/test_FinCDSCurve.py @@ -17,70 +17,112 @@ ######################################################################################## -# def test_fin_cds_curve(): - -# curve_dt = Date(20, 12, 2018) - -# swaps = [] -# depos = [] -# fras = [] - -# fixed_dcc = DayCountTypes.ACT_365F -# fixed_freq = FrequencyTypes.SEMI_ANNUAL -# fixed_cpn = 0.05 - -# for i in range(1, 11): - -# maturity_dt = curve_dt.add_months(12 * i) -# swap = IborSwap( -# curve_dt, -# maturity_dt, -# SwapTypes.PAY, -# fixed_cpn, -# fixed_freq, -# fixed_dcc, -# ) -# swaps.append(swap) - -# libor_curve = IborSingleCurve(curve_dt, depos, fras, swaps) - -# cds_contracts = [] - -# for i in range(1, 11): -# maturity_dt = curve_dt.add_months(12 * i) -# cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) -# cds_contracts.append(cds) - -# recovery_rate = 0.40 - -# issuer_curve = CDSCurve( -# curve_dt, cds_contracts, libor_curve, recovery_rate, use_cache=False -# ) - -# assert round(issuer_curve._times[0], 4) == 0.0 -# assert round(issuer_curve._times[5], 4) == 5.0027 -# assert round(issuer_curve._times[9], 4) == 9.0055 -# assert round(issuer_curve._qs[0], 4) == 1.0 -# assert round(issuer_curve._qs[5], 4) == 0.9249 -# assert round(issuer_curve._qs[9], 4) == 0.8071 - -# i = 1 -# maturity_dt = curve_dt.add_months(12 * i) -# cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) -# v = cds.value(curve_dt, issuer_curve, recovery_rate) -# assert round(v["dirty_pv"] * 1000, 4) == -0.0086 -# assert round(v["clean_pv"] * 1000, 4) == -0.0086 - -# i = 5 -# maturity_dt = curve_dt.add_months(12 * i) -# cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) -# v = cds.value(curve_dt, issuer_curve, recovery_rate) -# assert round(v["dirty_pv"] * 1000, 4) == -0.1637 -# assert round(v["clean_pv"] * 1000, 4) == -0.1637 - -# i = 10 -# maturity_dt = curve_dt.add_months(12 * i) -# cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) -# v = cds.value(curve_dt, issuer_curve, recovery_rate) -# assert round(v["dirty_pv"] * 1000, 4) == -1.1457 -# assert round(v["clean_pv"] * 1000, 4) == -1.1457 +import numpy as np + + +def test_fin_cds_curve(): + + curve_dt = Date(20, 12, 2018) + + swaps = [] + depos = [] + fras = [] + + fixed_dcc = DayCountTypes.ACT_365F + fixed_freq = FrequencyTypes.SEMI_ANNUAL + fixed_cpn = 0.05 + + for i in range(1, 11): + + maturity_dt = curve_dt.add_months(12 * i) + swap = IborSwap( + curve_dt, + maturity_dt, + SwapTypes.PAY, + fixed_cpn, + fixed_freq, + fixed_dcc, + ) + swaps.append(swap) + + libor_curve = IborSingleCurve(curve_dt, depos, fras, swaps) + + cds_contracts = [] + + for i in range(1, 11): + maturity_dt = curve_dt.add_months(12 * i) + cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) + cds_contracts.append(cds) + + recovery_rate = 0.40 + + issuer_curve = CDSCurve( + curve_dt, cds_contracts, libor_curve, recovery_rate + ) + + assert round(issuer_curve._times[0], 4) == 0.0 + assert round(issuer_curve._times[5], 4) == 5.0027 + assert round(issuer_curve._times[9], 4) == 9.0055 + assert round(issuer_curve._qs[0], 4) == 1.0 + assert round(issuer_curve._qs[5], 4) == 0.9249 + assert round(issuer_curve._qs[9], 4) == 0.8071 + + for i in [1, 5, 10]: + maturity_dt = curve_dt.add_months(12 * i) + cds = CDS(curve_dt, maturity_dt, 0.005 + 0.001 * (i - 1)) + dirty_pv, clean_pv = cds.value(curve_dt, issuer_curve, recovery_rate) + assert abs(dirty_pv) < 1e-5 + assert abs(clean_pv) < 1e-5 + + # Test scalar Date evaluation and day-count consistency + d1 = curve_dt.add_months(6) + d2 = curve_dt.add_months(12) + df_single = issuer_curve.df(d1) + sp_single = issuer_curve.survival_prob(d1) + + assert isinstance(df_single, float) + assert isinstance(sp_single, float) + assert df_single == libor_curve.df(d1) + + # Test vectorization with list of Dates + dts = [d1, d2] + df_vector = issuer_curve.df(dts) + sp_vector = issuer_curve.survival_prob(dts) + + assert isinstance(df_vector, np.ndarray) + assert isinstance(sp_vector, np.ndarray) + assert len(df_vector) == 2 + assert len(sp_vector) == 2 + assert np.isclose(df_vector[0], df_single) + assert np.isclose(df_vector[1], issuer_curve.df(d2)) + assert np.isclose(sp_vector[0], sp_single) + assert np.isclose(sp_vector[1], issuer_curve.survival_prob(d2)) + + # Test numeric float and numpy array inputs + t_single = 0.5 + df_t = issuer_curve.df(t_single) + sp_t = issuer_curve.survival_prob(t_single) + assert isinstance(df_t, float) + assert isinstance(sp_t, float) + assert np.isclose(df_t, libor_curve.df_t(t_single)) + + t_arr = np.array([0.5, 1.0]) + df_arr = issuer_curve.df(t_arr) + sp_arr = issuer_curve.survival_prob(t_arr) + assert isinstance(df_arr, np.ndarray) + assert isinstance(sp_arr, np.ndarray) + assert len(df_arr) == 2 + assert len(sp_arr) == 2 + assert np.isclose(df_arr[0], df_t) + assert np.isclose(sp_arr[0], sp_t) + + # Test day count consistency when libor_curve uses a different day count + libor_curve_360 = IborSingleCurve( + curve_dt, depos, fras, swaps, curve_dc_type=DayCountTypes.ACT_360 + ) + issuer_curve_360 = CDSCurve( + curve_dt, cds_contracts, libor_curve_360, recovery_rate + ) + assert issuer_curve_360.df(d1) == libor_curve_360.df(d1) + assert issuer_curve_360.df(dts)[0] == libor_curve_360.df(dts)[0] + assert issuer_curve_360.df(dts)[1] == libor_curve_360.df(dts)[1]