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.

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

where \(A(\tau)\) and \(B(\tau)\) are the affine_coefficients.

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.

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

    where $A(\tau)$ and $B(\tau)$ are the
    [affine_coefficients][..affine_coefficients].
    """
    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)

affine_coefficients

affine_coefficients(ttm)

Return the affine coefficients \(A(\tau)\) and \(B(\tau)\) of the log discount factor.

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

where

\[\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 affine_coefficients(
    self, ttm: FloatArrayLike
) -> tuple[FloatArrayLike, FloatArrayLike]:
    r"""Return the affine coefficients $A(\tau)$ and $B(\tau)$
    of the log discount factor.

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

    where

    \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)
    sigma2 = sigma * sigma
    gamma = np.sqrt(kappa * kappa + 2.0 * sigma2)
    gamma_m_kappa = gamma - kappa
    emgt = np.exp(-gamma * ttma)
    de = (gamma + kappa) + gamma_m_kappa * emgt
    b = 2.0 * (1.0 - emgt) / de
    log_a = (2.0 * kappa * theta / sigma2) * (
        np.log(2.0 * gamma) - 0.5 * gamma_m_kappa * ttma - np.log(de)
    )
    return maybe_float(log_a), maybe_float(b)

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)

rates

rates(ttm, frequency=2)

Calculate zero rates compounded at the given frequency.

The continuously compounded rate \(r_c(\tau)\) is converted to a rate compounded \(m\) times per year via:

\[\begin{equation} r_m(\tau) = m\,(e^{r_c(\tau)/m} - 1) \end{equation}\]

When frequency=0 the result is continuously compounded (same as continuously_compounded_rate).

PARAMETER	DESCRIPTION
ttm	Time to maturity in years TYPE: ArrayLike
frequency	Compounding periods per year (e.g. 2 for semi-annual). Pass 0 for continuously compounded. TYPE: int DEFAULT: 2

Source code in quantflow/rates/yield_curve.py

def rates(
    self,
    ttm: Annotated[ArrayLike, Doc("Time to maturity in years")],
    frequency: Annotated[
        int,
        Doc(
            "Compounding periods per year (e.g. 2 for semi-annual). "
            "Pass 0 for continuously compounded."
        ),
    ] = 2,
) -> FloatArrayLike:
    r"""Calculate zero rates compounded at the given frequency.

    The continuously compounded rate $r_c(\tau)$ is converted to a
    rate compounded $m$ times per year via:

    \begin{equation}
        r_m(\tau) = m\,(e^{r_c(\tau)/m} - 1)
    \end{equation}

    When ``frequency=0`` the result is continuously compounded (same as
    [continuously_compounded_rate][..continuously_compounded_rate]).
    """
    rc = self.continuously_compounded_rate(ttm)
    if frequency <= 0:
        return rc
    return frequency * np.expm1(rc / frequency)

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.

The registry key is the curve_type discriminator value rather than the class name, so the two can be named independently.

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.

    The registry key is the ``curve_type`` discriminator value rather than
    the class name, so the two can be named independently.
    """
    for curve_cls in curve_classes:
        name = curve_cls.model_fields["curve_type"].default
        if not isinstance(name, str):
            raise TypeError(
                f"{curve_cls.__name__} must define a string curve_type default"
            )
        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)

filtered_short_rate property

filtered_short_rate

Unscented-Kalman-filtered short rate at each observation date.

Populated by [calibrate_historical_rates][quantflow.rates.cir.CIRCurveCalibration.filtered_short_rate.calibrate_historical_rates]; accessing it before a historical fit raises an error.

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_historical_rates

calibrate_historical_rates(ttm, rates, dt)

Fit CIR by maximum likelihood with an unscented Kalman filter.

The short rate \(r_t\) is the latent state of a [CIRStateSpaceModel][quantflow.rates.cir.CIRCurveCalibration.CIRStateSpaceModel]. Its exact transition has a linear conditional mean but a state-dependent conditional variance, so the panel is filtered with the UnscentedKalmanFilter rather than the exact linear filter used for Vasicek.

The negative log-likelihood is minimised over \((\kappa, \theta, \rho, h)\), where \(h\) is the observation noise standard deviation and the Feller condition is enforced by reparametrising \(\sigma = \rho \sqrt{2\kappa\theta}\) with \(\rho \in (0, 1)\). The curve's rate is then set to the final filtered short rate.

PARAMETER	DESCRIPTION
ttm	Times to maturity in years, shape (n,). TYPE: FloatArray
rates	Continuously compounded rates, shape (T, n) (time by maturity). TYPE: FloatArray
dt	Per-step time increments in years, shape (T-1,); assumed uniform. TYPE: FloatArray

Source code in quantflow/rates/cir.py

def calibrate_historical_rates(
    self,
    ttm: Annotated[FloatArray, Doc("Times to maturity in years, shape (n,).")],
    rates: Annotated[
        FloatArray,
        Doc("Continuously compounded rates, shape (T, n) (time by maturity)."),
    ],
    dt: Annotated[
        FloatArray,
        Doc("Per-step time increments in years, shape (T-1,); assumed uniform."),
    ],
) -> CIRCurve:
    r"""Fit CIR by maximum likelihood with an unscented Kalman filter.

    The short rate $r_t$ is the latent state of a
    [CIRStateSpaceModel][..CIRStateSpaceModel]. Its exact transition has a
    linear conditional mean but a state-dependent conditional variance, so
    the panel is filtered with the
    [UnscentedKalmanFilter][quantflow.ta.kalman.UnscentedKalmanFilter]
    rather than the exact linear filter used for Vasicek.

    The negative log-likelihood is minimised over
    $(\kappa, \theta, \rho, h)$, where $h$ is the observation noise standard
    deviation and the Feller condition is enforced by reparametrising
    $\sigma = \rho \sqrt{2\kappa\theta}$ with $\rho \in (0, 1)$. The curve's
    rate is then set to the final filtered short rate.
    """
    rates = np.asarray(rates, dtype=float)
    ttm = np.asarray(ttm, dtype=float)
    dt = np.asarray(dt, dtype=float)
    if dt.size and not np.allclose(dt, dt[0], rtol=1e-2):
        raise ValueError(
            "calibrate_historical_rates assumes a uniform time step; "
            "the observation dates are not equally spaced"
        )
    step = float(dt[0]) if dt.size else 1.0

    def unpack(x: np.ndarray) -> tuple[float, float, float, float]:
        # x = (log kappa, log theta, logit rho, log h); enforce Feller via
        # sigma = rho * sqrt(2 kappa theta) with rho in (0, 1)
        kappa = float(np.exp(x[0]))
        theta = float(np.exp(x[1]))
        rho = 1.0 / (1.0 + np.exp(-x[2]))
        sigma = float(rho * np.sqrt(2.0 * kappa * theta))
        return kappa, theta, sigma, float(np.exp(x[3]))

    def filtered(
        kappa: float, theta: float, sigma: float, h: float
    ) -> UnscentedKalmanFilter:
        curve = CIRCurve(
            rate=Decimal(str(round(theta, 10))),
            kappa=Decimal(str(round(kappa, 10))),
            theta=Decimal(str(round(theta, 10))),
            sigma=Decimal(str(round(sigma, 10))),
        )
        model = CIRStateSpaceModel(curve=curve, ttm=ttm, dt=step, h=h)
        return model.unscented_filter(rates)

    theta0 = max(float(np.mean(rates)), 1e-4)
    short = rates[:, int(np.argmin(ttm))]
    short_mean = max(float(np.mean(short)), 1e-4)
    sigma0 = max(float(np.std(np.diff(short)) / np.sqrt(step * short_mean)), 1e-4)
    rho0 = min(max(sigma0 / np.sqrt(2.0 * 0.5 * theta0), 0.01), 0.99)
    h0 = max(float(np.std(rates - rates.mean(axis=0))) / 10.0, 1e-5)
    x0 = np.array(
        [np.log(0.5), np.log(theta0), np.log(rho0 / (1.0 - rho0)), np.log(h0)]
    )

    def neg_loglik(x: np.ndarray) -> float:
        return -filtered(*unpack(x)).filter()

    result = minimize(neg_loglik, x0, method="Nelder-Mead")
    kappa, theta, sigma, h = unpack(result.x)
    ukf = filtered(kappa, theta, sigma, h)
    ukf.filter()
    short_rate = np.array([float(s.mean.item()) for s in ukf.states])
    self._filtered_short_rate = short_rate
    c = self.yield_curve
    c.rate = Decimal(str(round(float(short_rate[-1]), 10)))
    c.kappa = Decimal(str(round(kappa, 10)))
    c.theta = Decimal(str(round(theta, 10)))
    c.sigma = Decimal(str(round(sigma, 10)))
    return c

calibrate_df

calibrate_df(ttm, target)

Fit the yield curve to target discount factors.

Converts discount factors to continuously compounded rates then calls 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/calibration.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)

calibrate_historical_rates_dataframe

calibrate_historical_rates_dataframe(rates, frequency=None)

Fit the yield curve from a historical panel of rates.

Tenor column labels are parsed into times to maturity, per-step time increments are inferred from the DatetimeIndex (irregular spacing supported), and rates are converted to continuously compounded if a finite frequency is supplied. The actual fit is delegated to [calibrate_historical_rates][quantflow.rates.cir.calibrate_historical_rates], which subclasses override.

PARAMETER	DESCRIPTION
rates	Historical zero rates with a DatetimeIndex and tenor column labels parsed by [ccy.Period][ccy.dates.period.Period] (e.g. '6m', '1y'). TYPE: DataFrame
frequency	Compounding periods per year of the input rates. None (default) means continuously compounded. TYPE: int | None DEFAULT: None

Source code in quantflow/rates/calibration.py

def calibrate_historical_rates_dataframe(
    self,
    rates: Annotated[
        pd.DataFrame,
        Doc(
            "Historical zero rates with a DatetimeIndex and tenor column "
            "labels parsed by [ccy.Period][ccy.dates.period.Period] "
            "(e.g. ``'6m'``, ``'1y'``)."
        ),
    ],
    frequency: Annotated[
        int | None,
        Doc(
            "Compounding periods per year of the input rates. ``None`` "
            "(default) means continuously compounded."
        ),
    ] = None,
) -> Y:
    """Fit the yield curve from a historical panel of rates.

    Tenor column labels are parsed into times to maturity, per-step
    time increments are inferred from the DatetimeIndex (irregular
    spacing supported), and rates are converted to continuously
    compounded if a finite ``frequency`` is supplied. The actual fit
    is delegated to [calibrate_historical_rates][...calibrate_historical_rates],
    which subclasses override.
    """
    ttm = np.array([tenor_to_years(str(c)) for c in rates.columns], dtype=float)
    rates_arr = _to_continuous(np.asarray(rates.values, dtype=float), frequency)
    dt = _dt_array(rates.index)
    return self.calibrate_historical_rates(ttm, rates_arr, dt)