CIR Curve¶

quantflow.rates.cir.CIRCurve `pydantic-model` ¶

Bases: YieldCurve

Yield curve derived from the Cox-Ingersoll-Ross short-rate model.

The CIR model describes the short rate as a mean-reverting square-root diffusion:

\[\begin{equation} dr_t = \kappa(\theta - r_t)\, dt + \sigma\sqrt{r_t}\, dW_t \end{equation}\]

The model admits a closed-form discount factor; see discount_factor.

Throughout, the auxiliary quantities are:

\[\begin{equation} \begin{aligned} \gamma &= \sqrt{\kappa^2 + 2\sigma^2} \\ d_e(\tau) &= (\gamma + \kappa) + (\gamma - \kappa)e^{-\gamma\tau} \end{aligned} \end{equation}\]

Fields:

ref_date (datetime)
curve_type (Literal['cir_curve'])
rate (DecimalNumber)
kappa (DecimalNumber)
theta (DecimalNumber)
sigma (DecimalNumber)

curve_type `pydantic-field` ¶

curve_type = 'cir_curve'

rate `pydantic-field` ¶

rate = Decimal('0.05')

Initial short rate \(r_0\)

kappa `pydantic-field` ¶

kappa = Decimal('1.0')

Mean reversion speed \(\kappa\)

theta `pydantic-field` ¶

theta = Decimal('0.05')

Long-run mean \(\theta\)

sigma `pydantic-field` ¶

sigma = Decimal('0.1')

Volatility \(\sigma\)

ref_date `pydantic-field` ¶

ref_date

Reference date for the yield curve

calibrator ¶

calibrator()

Return a CIRCurveCalibration wrapping this curve.

Source code in quantflow/rates/cir.py

def calibrator(self) -> CIRCurveCalibration:
    """Return a [CIRCurveCalibration][...CIRCurveCalibration] wrapping
    this curve."""
    return CIRCurveCalibration(yield_curve=self)

process ¶

process()

Return the underlying CIR stochastic process.

Source code in quantflow/rates/cir.py

def process(self) -> CIR:
    """Return the underlying [CIR][quantflow.sp.cir.CIR] stochastic process."""
    return CIR(
        rate=float(self.rate),
        kappa=float(self.kappa),
        theta=float(self.theta),
        sigma=float(self.sigma),
    )

instantaneous_forward_rate ¶

instantaneous_forward_rate(ttm)

Calculate the instantaneous forward rate for the CIR model.

The forward rate is:

\[\begin{equation} f(\tau) = -\frac{\partial}{\partial \tau}\ln D(\tau) = r_0 B'(\tau) - A'(\tau) \end{equation}\]

where:

\[\begin{equation} \begin{aligned} B'(\tau) &= \frac{4\gamma^2 e^{-\gamma\tau}}{d_e(\tau)^2} \\ A'(\tau) &= \frac{2\kappa\theta}{\sigma^2} \left[-\frac{\gamma - \kappa}{2} + \frac{\gamma(\gamma - \kappa)e^{-\gamma\tau}}{d_e(\tau)}\right] \end{aligned} \end{equation}\]

Source code in quantflow/rates/cir.py

def instantaneous_forward_rate(self, ttm: FloatArrayLike) -> FloatArrayLike:
    r"""Calculate the instantaneous forward rate for the CIR model.

    The forward rate is:

    \begin{equation}
        f(\tau) = -\frac{\partial}{\partial \tau}\ln D(\tau)
        = r_0 B'(\tau) - A'(\tau)
    \end{equation}

    where:

    \begin{equation}
    \begin{aligned}
        B'(\tau) &= \frac{4\gamma^2 e^{-\gamma\tau}}{d_e(\tau)^2} \\
        A'(\tau) &= \frac{2\kappa\theta}{\sigma^2}
            \left[-\frac{\gamma - \kappa}{2}
            + \frac{\gamma(\gamma - \kappa)e^{-\gamma\tau}}{d_e(\tau)}\right]
    \end{aligned}
    \end{equation}
    """
    arr = np.asarray(ttm, dtype=float)
    ttma = np.maximum(arr, 0.0)
    kappa = float(self.kappa)
    theta = float(self.theta)
    sigma = float(self.sigma)
    rate = float(self.rate)
    sigma2 = sigma * sigma
    gamma = np.sqrt(kappa * kappa + 2.0 * sigma2)
    gamma_kappa = gamma + kappa
    gamma_m_kappa = gamma - kappa
    emgt = np.exp(-gamma * ttma)
    de = gamma_kappa + gamma_m_kappa * emgt
    # dB/dτ = 4γ²e^{-γτ} / de²
    db = 4.0 * gamma * gamma * emgt / (de * de)
    # d(log_a)/dτ = (2κθ/σ²) * [-(γ-κ)/2 + γ(γ-κ)e^{-γτ}/de]
    kts = 2.0 * kappa * theta / sigma2
    d_log_a = kts * (-gamma_m_kappa / 2.0 + gamma * gamma_m_kappa * emgt / de)
    fwd = rate * db - d_log_a
    return maybe_float(fwd)

discount_factor ¶

discount_factor(ttm)

Calculate the discount factor using the CIR closed-form solution.

The discount factor is:

\[\begin{equation} D(\tau) = e^{A(\tau) - B(\tau)\, r_0} \end{equation}\]

The coefficients are:

\[\begin{equation} \begin{aligned} B(\tau) &= \frac{2(1 - e^{-\gamma\tau})}{d_e(\tau)} \\ A(\tau) &= \frac{2\kappa\theta}{\sigma^2} \ln\!\left( \frac{2\gamma\, e^{-(\gamma - \kappa)\tau/2}}{d_e(\tau)} \right) \end{aligned} \end{equation}\]

Source code in quantflow/rates/cir.py

def discount_factor(self, ttm: FloatArrayLike) -> FloatArrayLike:
    r"""Calculate the discount factor using the CIR closed-form solution.

    The discount factor is:

    \begin{equation}
        D(\tau) = e^{A(\tau) - B(\tau)\, r_0}
    \end{equation}

    The coefficients are:

    \begin{equation}
    \begin{aligned}
        B(\tau) &= \frac{2(1 - e^{-\gamma\tau})}{d_e(\tau)} \\
        A(\tau) &= \frac{2\kappa\theta}{\sigma^2}
            \ln\!\left(
            \frac{2\gamma\, e^{-(\gamma - \kappa)\tau/2}}{d_e(\tau)}
            \right)
    \end{aligned}
    \end{equation}
    """
    arr = np.asarray(ttm, dtype=float)
    ttma = np.maximum(arr, 0.0)
    kappa = float(self.kappa)
    theta = float(self.theta)
    sigma = float(self.sigma)
    rate = float(self.rate)
    sigma2 = sigma * sigma
    gamma = np.sqrt(kappa * kappa + 2.0 * sigma2)
    gamma_kappa = gamma + kappa
    gamma_m_kappa = gamma - kappa
    emgt = np.exp(-gamma * ttma)
    # d/e^{γτ} = (γ+κ) + (γ-κ)e^{-γτ}
    de = gamma_kappa + gamma_m_kappa * emgt
    b = 2.0 * (1.0 - emgt) / de
    # log(d) = γτ + log(de)
    log_a = (2.0 * kappa * theta / sigma2) * (
        np.log(2.0 * gamma) - 0.5 * gamma_m_kappa * ttma - np.log(de)
    )
    df = np.exp(log_a - b * rate)
    return maybe_float(df)

jacobian ¶

jacobian(ttm)

Analytical Jacobian of discount factors w.r.t. \([r_0, \kappa, \theta, \sigma]\). Returns shape (len(ttm), 4).

Source code in quantflow/rates/cir.py

def jacobian(self, ttm: FloatArrayLike) -> FloatArray | None:
    r"""Analytical Jacobian of discount factors w.r.t.
    $[r_0, \kappa, \theta, \sigma]$. Returns shape (len(ttm), 4).
    """
    arr = np.asarray(ttm, dtype=float)
    ttma = np.maximum(arr, 0.0)
    kappa = float(self.kappa)
    theta = float(self.theta)
    sigma = float(self.sigma)
    rate = float(self.rate)
    sigma2 = sigma * sigma
    gamma = np.sqrt(kappa * kappa + 2.0 * sigma2)
    gm_k = gamma - kappa
    emgt = np.exp(-gamma * ttma)
    de = (gamma + kappa) + gm_k * emgt
    b = 2.0 * (1.0 - emgt) / de
    c = 2.0 * kappa * theta / sigma2
    f = np.log(2.0 * gamma) - 0.5 * gm_k * ttma - np.log(de)
    log_a = c * f
    d = np.exp(log_a - b * rate)

    # ∂D/∂r0
    d_rate = -b * d

    # ∂D/∂κ: use dγ/dκ = κ/γ
    gk = kappa / gamma
    d_de_k = (gk + 1.0) + (gk - 1.0) * emgt - gm_k * ttma * gk * emgt
    d_b_k = 2.0 * (ttma * gk * emgt * de - (1.0 - emgt) * d_de_k) / (de * de)
    d_f_k = kappa / (gamma * gamma) - 0.5 * (gk - 1.0) * ttma - d_de_k / de
    d_loga_k = (2.0 * theta / sigma2) * f + c * d_f_k
    d_kappa = d * (d_loga_k - rate * d_b_k)

    # ∂D/∂θ: only c = 2κθ/σ² depends on θ, so d(log D)/dθ = log_a / θ
    d_theta = d * log_a / theta

    # ∂D/∂σ: use dγ/dσ = 2σ/γ
    gs = 2.0 * sigma / gamma
    d_de_s = gs * (1.0 + emgt * (1.0 - gm_k * ttma))
    d_b_s = 2.0 * (ttma * gs * emgt * de - (1.0 - emgt) * d_de_s) / (de * de)
    d_f_s = 2.0 * sigma / (gamma * gamma) - sigma * ttma / gamma - d_de_s / de
    d_loga_s = (-2.0 * c / sigma) * f + c * d_f_s
    d_sigma = d * (d_loga_s - rate * d_b_s)

    return np.column_stack([d_rate, d_kappa, d_theta, d_sigma])

continuously_compounded_rate ¶

continuously_compounded_rate(ttm)

Calculate the continuously compounded rate for a given time to maturity.

The continuously compounded rate is related to the discount factor by the following formula:

\[\begin{equation} r(\tau) = -\frac{\ln D(\tau)}{\tau} \end{equation}\]

where \(D(\tau)\) is the discount factor for a given time to maturity \(\tau\).

Accepts a scalar float or a float array. Returns a scalar float for scalar input and a numpy float array for array input.

PARAMETER	DESCRIPTION
`ttm`	Time to maturity in years TYPE: `ArrayLike`

Source code in quantflow/rates/yield_curve.py

def continuously_compounded_rate(
    self, ttm: Annotated[ArrayLike, Doc("Time to maturity in years")]
) -> FloatArrayLike:
    r"""Calculate the continuously compounded rate for a given time to maturity.

    The continuously compounded rate is related to the discount factor
    by the following formula:

    \begin{equation}
        r(\tau) = -\frac{\ln D(\tau)}{\tau}
    \end{equation}

    where $D(\tau)$ is the discount factor for a given time to maturity $\tau$.

    Accepts a scalar float or a float array. Returns a scalar float for scalar
    input and a numpy float array for array input.
    """
    ttm_ = np.asarray(ttm, dtype=float)
    df = np.asarray(self.discount_factor(ttm_), dtype=float)
    result = np.where(
        ttm_ <= 0, self.instantaneous_forward_rate(0.0), -np.log(df) / ttm_
    )
    return maybe_float(result)

plot ¶

plot(ttm_max=10.0, n=200, **kwargs)

Plot the continuously compounded rate vs time to maturity.

Requires plotly to be installed.

PARAMETER	DESCRIPTION
`ttm_max`	Maximum time to maturity in years TYPE: `float` DEFAULT: `10.0`
`n`	Number of points to evaluate TYPE: `int` DEFAULT: `200`

Source code in quantflow/rates/yield_curve.py

def plot(
    self,
    ttm_max: Annotated[float, Doc("Maximum time to maturity in years")] = 10.0,
    n: Annotated[int, Doc("Number of points to evaluate")] = 200,
    **kwargs: Any,
) -> Any:
    """Plot the continuously compounded rate vs time to maturity.

    Requires plotly to be installed.
    """
    return plot.plot_yield_curve(self, ttm_max=ttm_max, n=n, **kwargs)

register_curve_types `classmethod` ¶

register_curve_types(*curve_classes)

Register a yield curve subclass for deserialization.

Source code in quantflow/rates/yield_curve.py

@classmethod
def register_curve_types(cls, *curve_classes: type[YieldCurve]) -> None:
    """Register a yield curve subclass for deserialization."""
    for curve_cls in curve_classes:
        name = snake_case(curve_cls.__name__)
        if current_type := _CURVE_TYPES.pop(name, None):
            _TYPES_TO_NAMES.pop(current_type, None)
        _CURVE_TYPES[name] = curve_cls
        _TYPES_TO_NAMES[curve_cls] = name

curve_types `classmethod` ¶

curve_types()

Return the registered curve types.

Source code in quantflow/rates/yield_curve.py

@classmethod
def curve_types(cls) -> tuple[str, ...]:
    """Return the registered curve types."""
    return tuple(sorted(_CURVE_TYPES))

get_curve_class `classmethod` ¶

get_curve_class(curve_type)

Get the yield curve class for a given curve type.

Source code in quantflow/rates/yield_curve.py

@classmethod
def get_curve_class(cls, curve_type: str) -> type[YieldCurve] | None:
    """Get the yield curve class for a given curve type."""
    return _CURVE_TYPES.get(curve_type)

quantflow.rates.cir.CIRCurveCalibration `pydantic-model` ¶

Bases: YieldCurveCalibration[CIRCurve]

Calibration wrapper for a CIR yield curve.

Fields:

yield_curve (Y)

yield_curve `pydantic-field` ¶

yield_curve

Yield curve to be calibrated

get_params ¶

get_params()

Source code in quantflow/rates/cir.py

def get_params(self) -> FloatArray:
    c = self.yield_curve
    kappa = float(c.kappa)
    theta = float(c.theta)
    sigma_ratio = float(c.sigma) / np.sqrt(2.0 * kappa * theta)
    return np.array([float(c.rate), kappa, theta, sigma_ratio])

set_params ¶

set_params(params)

Source code in quantflow/rates/cir.py

def set_params(self, params: FloatArray) -> None:
    rate, kappa, theta, sigma_ratio = params
    sigma = sigma_ratio * np.sqrt(2.0 * kappa * theta)
    self.yield_curve.rate = Decimal(str(round(float(rate), 10)))
    self.yield_curve.kappa = Decimal(str(round(float(kappa), 10)))
    self.yield_curve.theta = Decimal(str(round(float(theta), 10)))
    self.yield_curve.sigma = Decimal(str(round(float(sigma), 10)))

get_bounds ¶

get_bounds()

Source code in quantflow/rates/cir.py

def get_bounds(self) -> Bounds:
    return Bounds([0.0, 1e-4, 1e-6, 1e-6], [1.0, 1000.0, 1.0, 1.0])

calibrate ¶

calibrate(ttm, rates)

Fit the CIR curve to continuously compounded rates via least squares.

The Feller condition is enforced by reparametrising \(\sigma\) as

\[\begin{equation} \sigma = \rho \sqrt{2\kappa\theta}, \quad \rho \in [0, 1] \end{equation}\]

Since CIR requires non-negative rates, any negative input rates are floored to a small positive value before fitting.

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `ArrayLike`
`rates`	Continuously compounded rates, same length as ttm. TYPE: `ArrayLike`

Source code in quantflow/rates/cir.py

def calibrate(
    self,
    ttm: Annotated[ArrayLike, Doc("Times to maturity in years.")],
    rates: Annotated[
        ArrayLike, Doc("Continuously compounded rates, same length as ttm.")
    ],
) -> CIRCurve:
    r"""Fit the CIR curve to continuously compounded rates via least squares.

    The Feller condition is enforced by reparametrising $\sigma$ as

    \begin{equation}
        \sigma = \rho \sqrt{2\kappa\theta}, \quad \rho \in [0, 1]
    \end{equation}

    Since CIR requires non-negative rates, any negative input rates are
    floored to a small positive value before fitting.
    """
    ttm_arr = np.asarray(ttm, dtype=float)
    rates_arr = np.maximum(np.asarray(rates, dtype=float), 1e-6)

    def residuals(params: np.ndarray) -> np.ndarray:
        self.set_params(params)
        df = np.asarray(self.yield_curve.discount_factor(ttm_arr), dtype=float)
        fitted = -np.log(df) / ttm_arr
        return fitted - rates_arr

    def jac(params: np.ndarray) -> FloatArray:
        self.set_params(params)
        _, kappa, theta, sigma_ratio = params
        sigma = sigma_ratio * np.sqrt(2.0 * kappa * theta)
        df = np.asarray(self.yield_curve.discount_factor(ttm_arr), dtype=float)
        jac_d = np.asarray(self.yield_curve.jacobian(ttm_arr), dtype=float)
        d_sigma = jac_d[:, 3]
        jac_d[:, 1] += d_sigma * sigma / (2.0 * kappa)
        jac_d[:, 2] += d_sigma * sigma / (2.0 * theta)
        jac_d[:, 3] = d_sigma * np.sqrt(2.0 * kappa * theta)
        return -jac_d / (df * ttm_arr)[:, None]

    x0 = np.array([rates_arr[0], 1.0, rates_arr[-1], 0.5])
    result = least_squares(
        residuals,
        jac=jac,
        x0=x0,
        bounds=([0.0, 1e-4, 1e-6, 1e-4], [1.0, 1000.0, 1.0, 1.0]),
    )
    self.set_params(result.x)
    return self.yield_curve

calibrate_df ¶

calibrate_df(ttm, target)

Fit the yield curve to target discount factors.

Converts discount factors to continuously compounded rates then calls [calibrate][quantflow.rates.cir.CIRCurveCalibration.calibrate_df.calibrate].

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `ArrayLike`
`target`	Target discount factors, same length as ttm. TYPE: `ArrayLike`

Source code in quantflow/rates/options.py

def calibrate_df(
    self,
    ttm: Annotated[ArrayLike, Doc("Times to maturity in years.")],
    target: Annotated[
        ArrayLike, Doc("Target discount factors, same length as ttm.")
    ],
) -> Y:
    """Fit the yield curve to target discount factors.

    Converts discount factors to continuously compounded rates then calls
    [calibrate][.calibrate].
    """
    ttm_ = np.asarray(ttm, dtype=float)
    rates = -np.log(np.asarray(target, dtype=float)) / ttm_
    return self.calibrate(ttm_, rates)

CIR Curve¶

quantflow.rates.cir.CIRCurve pydantic-model ¶

curve_type pydantic-field ¶

rate pydantic-field ¶

kappa pydantic-field ¶

theta pydantic-field ¶

sigma pydantic-field ¶

ref_date pydantic-field ¶

calibrator ¶

process ¶

instantaneous_forward_rate ¶

discount_factor ¶

jacobian ¶

continuously_compounded_rate ¶

plot ¶

register_curve_types classmethod ¶

curve_types classmethod ¶

get_curve_class classmethod ¶

quantflow.rates.cir.CIRCurveCalibration pydantic-model ¶

yield_curve pydantic-field ¶

get_params ¶

set_params ¶

get_bounds ¶

calibrate ¶

calibrate_df ¶

quantflow.rates.cir.CIRCurve `pydantic-model` ¶

curve_type `pydantic-field` ¶

rate `pydantic-field` ¶

kappa `pydantic-field` ¶

theta `pydantic-field` ¶

sigma `pydantic-field` ¶

ref_date `pydantic-field` ¶

register_curve_types `classmethod` ¶

curve_types `classmethod` ¶

get_curve_class `classmethod` ¶

quantflow.rates.cir.CIRCurveCalibration `pydantic-model` ¶

yield_curve `pydantic-field` ¶