diff --git a/frontend/src/yield-curve.md b/frontend/src/yield-curve.md index e849836..0f35823 100644 --- a/frontend/src/yield-curve.md +++ b/frontend/src/yield-curve.md @@ -13,7 +13,7 @@ import * as d3 from "npm:d3"; ``` ```js -const curveType = view(Inputs.select(["nelson_siegel", "vasicek_curve"], {label: "Curve type"})); +const curveType = view(Inputs.select(["nelson_siegel", "vasicek_curve", "cir_curve"], {label: "Curve type"})); ``` ```js diff --git a/quantflow/rates/cir.py b/quantflow/rates/cir.py index 6d8da9f..ec4fbf5 100644 --- a/quantflow/rates/cir.py +++ b/quantflow/rates/cir.py @@ -136,23 +136,28 @@ def jacobian(self, ttm: FloatArrayLike) -> FloatArray | None: @classmethod def calibrate(cls, ttm: ArrayLike, rates: ArrayLike) -> Self: - """Fit the CIR curve to continuously compounded rates via least squares.""" + """Fit the CIR curve to continuously compounded rates via least squares. + + The Feller condition is enforced by reparametrising sigma as + sigma = ratio * sqrt(2 * kappa * theta), with ratio in [0, 1]. + """ ttm_arr = np.asarray(ttm, dtype=float) rates_arr = np.asarray(rates, dtype=float) def residuals(params: np.ndarray) -> np.ndarray: - curve = cls( - rate=params[0], kappa=params[1], theta=params[2], sigma=params[3] - ) + rate, kappa, theta, sigma_ratio = params + sigma = sigma_ratio * np.sqrt(2.0 * kappa * theta) + curve = cls(rate=rate, kappa=kappa, theta=theta, sigma=sigma) df = np.asarray(curve.discount_factor(ttm_arr), dtype=float) fitted = -np.log(df) / ttm_arr return fitted - rates_arr - x0 = np.array([rates_arr[0], 1.0, rates_arr[-1], 0.1]) + x0 = np.array([rates_arr[0], 1.0, rates_arr[-1], 0.5]) result = least_squares( residuals, x0, - bounds=([0.0, 1e-4, 0.0, 1e-4], [1.0, 50.0, 1.0, 2.0]), + bounds=([0.0, 1e-4, 1e-6, 1e-4], [1.0, 50.0, 1.0, 1.0]), ) - r, k, th, s = result.x + r, k, th, sr = result.x + s = sr * np.sqrt(2.0 * k * th) return cls(rate=r, kappa=k, theta=th, sigma=s) diff --git a/quantflow_tests/test_cir_curve.py b/quantflow_tests/test_cir_curve.py new file mode 100644 index 0000000..6c83476 --- /dev/null +++ b/quantflow_tests/test_cir_curve.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from decimal import Decimal + +import numpy as np +import pytest + +from quantflow.rates.cir import CIRCurve + + +def test_cir_process_mapping() -> None: + curve = CIRCurve( + rate=Decimal("0.03"), + kappa=Decimal("1.2"), + theta=Decimal("0.04"), + sigma=Decimal("0.1"), + ) + process = curve.process() + assert process.rate == pytest.approx(0.03) + assert process.kappa == pytest.approx(1.2) + assert process.theta == pytest.approx(0.04) + assert process.sigma == pytest.approx(0.1) + + +def test_cir_forward_and_discount_shapes() -> None: + curve = CIRCurve( + rate=Decimal("0.03"), + kappa=Decimal("0.8"), + theta=Decimal("0.05"), + sigma=Decimal("0.1"), + ) + ttms = np.array([0.0, 0.5, 1.0, 2.0]) + fwd = np.asarray(curve.instantaneous_forward_rate(ttms)) + df = np.asarray(curve.discount_factor(ttms)) + assert fwd.shape == ttms.shape + assert df.shape == ttms.shape + assert df[0] == pytest.approx(1.0) + assert df[-1] < df[1] + + +def test_cir_forward_rate_at_zero() -> None: + curve = CIRCurve( + rate=Decimal("0.05"), + kappa=Decimal("1.0"), + theta=Decimal("0.04"), + sigma=Decimal("0.2"), + ) + fwd = curve.instantaneous_forward_rate(0.0) + assert fwd == pytest.approx(0.05, rel=1e-6) + + +def test_cir_calibrate_recovers_curve() -> None: + true_curve = CIRCurve( + rate=Decimal("0.03"), + kappa=Decimal("1.5"), + theta=Decimal("0.04"), + sigma=Decimal("0.06"), + ) + ttm = np.array([0.25, 0.5, 1.0, 2.0, 3.0, 5.0], dtype=float) + rates = -np.log(np.asarray(true_curve.discount_factor(ttm))) / ttm + fitted = CIRCurve.calibrate(ttm, rates) + fitted_df = np.asarray(fitted.discount_factor(ttm)) + true_df = np.asarray(true_curve.discount_factor(ttm)) + np.testing.assert_allclose(fitted_df, true_df, atol=3e-3) + + +def test_cir_calibrate_feller_condition() -> None: + """Calibration must always produce parameters satisfying the Feller condition.""" + ttm = np.array([0.25, 0.5, 1.0, 2.0, 5.0, 10.0], dtype=float) + rates = np.array([0.05, 0.048, 0.045, 0.042, 0.040, 0.039], dtype=float) + fitted = CIRCurve.calibrate(ttm, rates) + process = fitted.process() + assert process.feller_condition >= 0.0 + assert process.is_positive is True + + +def test_cir_continuously_compounded_rate() -> None: + curve = CIRCurve( + rate=Decimal("0.04"), + kappa=Decimal("1.0"), + theta=Decimal("0.05"), + sigma=Decimal("0.15"), + ) + ttm = np.array([1.0, 5.0, 10.0]) + df = np.asarray(curve.discount_factor(ttm)) + expected_rates = -np.log(df) / ttm + computed_rates = np.asarray(curve.continuously_compounded_rate(ttm)) + np.testing.assert_allclose(computed_rates, expected_rates, rtol=1e-10)