Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion frontend/src/yield-curve.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
19 changes: 12 additions & 7 deletions quantflow/rates/cir.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
88 changes: 88 additions & 0 deletions quantflow_tests/test_cir_curve.py
Original file line number Diff line number Diff line change
@@ -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)
Loading