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

prepare

prepare(ttm)

Hook called before optimisation with the observation times to maturity.

By default it does nothing. Curves whose parameters depend on the observation grid, such as interpolated curves, use it to seed their nodes.

PARAMETER	DESCRIPTION
ttm	Observation times to maturity in years TYPE: FloatArray

Source code in quantflow/rates/calibration.py

def prepare(
    self, ttm: Annotated[FloatArray, Doc("Observation times to maturity in years")]
) -> None:
    """Hook called before optimisation with the observation times to maturity.

    By default it does nothing. Curves whose parameters depend on the
    observation grid, such as interpolated curves, use it to seed their
    nodes."""

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)