Interpolated Curves¶

Interpolated curves build the term structure directly from observed continuously compounded zero rates at a set of anchor dates, interpolating between them rather than fitting a parametric form.

The InterpolatedYieldCurve base class holds the interpolation nodes and derives the discount factor and instantaneous forward rate from the interpolated zero rate. Subclasses choose how the zero rate is interpolated between nodes.

InterpolatedLinearCurve interpolates the zero rate piecewise linearly, giving a forward rate that is linear on each segment.

InterpolatedMonotonicCubicCurve uses a shape-preserving PCHIP cubic spline, giving a smooth zero rate and forward rate that introduces no spurious local extrema between nodes.

quantflow.rates.interpolated.InterpolatedYieldCurve `pydantic-model` ¶

Bases: YieldCurve

Base class for yield curves built by interpolating the zero rate.

The curve is defined by continuously compounded zero rates \(r_i\) at a set of anchor dates with times to maturity \(\tau_i\) measured from ref_date on an ACT/365 basis. The zero rate is interpolated between the nodes and the discount factor follows directly:

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

The instantaneous forward rate is then \(f(\tau) = r(\tau) + \tau\,r'(\tau)\). Outside the node range the zero rate is held flat (the first node value below \(\tau_1\) and the last node value beyond \(\tau_N\)).

Subclasses choose how the zero rate is interpolated between nodes: InterpolatedLinearCurve (piecewise linear) or InterpolatedMonotonicCubicCurve (shape-preserving PCHIP spline).

The anchor lists default to empty, leaving the curve uncalibrated until its calibrator fills in the nodes.

Fields:

ref_date (datetime)
curve_type (str)
anchor_dates (list[datetime])
anchor_rates (list[DecimalNumber])

anchor_dates `pydantic-field` ¶

anchor_dates

Maturity dates of the interpolation nodes, strictly after the reference date and in increasing order

anchor_rates `pydantic-field` ¶

anchor_rates

Continuously compounded zero rates at each anchor date (0.05 means 5%), same length as anchor_dates

ref_date `pydantic-field` ¶

ref_date

Reference date for the yield curve

curve_type `pydantic-field` ¶

curve_type = 'unknown'

Type of the yield curve, used for serialization and discrimination

calibrator ¶

calibrator()

Return an [InterpolatedYieldCurveCalibration][ ...InterpolatedYieldCurveCalibration] wrapping this curve.

Source code in quantflow/rates/interpolated.py

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

instantaneous_forward_rate ¶

instantaneous_forward_rate(ttm)

Source code in quantflow/rates/interpolated.py

def instantaneous_forward_rate(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    dr = self._zero_rate_derivative(tau)
    return maybe_float(r + tau * dr)

discount_factor ¶

discount_factor(ttm)

Source code in quantflow/rates/interpolated.py

def discount_factor(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    return maybe_float(np.exp(-r * tau))

jacobian ¶

jacobian(ttm)

Analytical Jacobian of discount factors w.r.t. model parameters.

Returns None if no analytical Jacobian is available (default). Shape when not None: (len(ttm), n_params).

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `FloatArrayLike`

Source code in quantflow/rates/yield_curve.py

def jacobian(
    self, ttm: Annotated[FloatArrayLike, Doc("Times to maturity in years.")]
) -> FloatArray | None:
    """Analytical Jacobian of discount factors w.r.t. model parameters.

    Returns None if no analytical Jacobian is available (default).
    Shape when not None: (len(ttm), n_params).
    """
    return None

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.interpolated.InterpolatedLinearCurve `pydantic-model` ¶

Bases: InterpolatedYieldCurve

Yield curve interpolating the zero rate piecewise linearly.

The zero rate \(r(\tau)\) is linear between adjacent nodes, so the instantaneous forward rate is linear on each segment.

Fields:

ref_date (datetime)
anchor_dates (list[datetime])
anchor_rates (list[DecimalNumber])
curve_type (Literal['interpolated_linear_curve'])

curve_type `pydantic-field` ¶

curve_type = 'interpolated_linear_curve'

ref_date `pydantic-field` ¶

ref_date

Reference date for the yield curve

anchor_dates `pydantic-field` ¶

anchor_dates

Maturity dates of the interpolation nodes, strictly after the reference date and in increasing order

anchor_rates `pydantic-field` ¶

anchor_rates

Continuously compounded zero rates at each anchor date (0.05 means 5%), same length as anchor_dates

instantaneous_forward_rate ¶

instantaneous_forward_rate(ttm)

Source code in quantflow/rates/interpolated.py

def instantaneous_forward_rate(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    dr = self._zero_rate_derivative(tau)
    return maybe_float(r + tau * dr)

discount_factor ¶

discount_factor(ttm)

Source code in quantflow/rates/interpolated.py

def discount_factor(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    return maybe_float(np.exp(-r * tau))

calibrator ¶

calibrator()

Return an [InterpolatedYieldCurveCalibration][ ...InterpolatedYieldCurveCalibration] wrapping this curve.

Source code in quantflow/rates/interpolated.py

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

jacobian ¶

jacobian(ttm)

Analytical Jacobian of discount factors w.r.t. model parameters.

Returns None if no analytical Jacobian is available (default). Shape when not None: (len(ttm), n_params).

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `FloatArrayLike`

Source code in quantflow/rates/yield_curve.py

def jacobian(
    self, ttm: Annotated[FloatArrayLike, Doc("Times to maturity in years.")]
) -> FloatArray | None:
    """Analytical Jacobian of discount factors w.r.t. model parameters.

    Returns None if no analytical Jacobian is available (default).
    Shape when not None: (len(ttm), n_params).
    """
    return None

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.interpolated.InterpolatedMonotonicCubicCurve `pydantic-model` ¶

Bases: InterpolatedYieldCurve

Yield curve interpolating the zero rate with a monotone cubic spline.

The zero rate \(r(\tau)\) is interpolated with a shape-preserving cubic Hermite spline (PCHIP, Fritsch-Carlson) that never introduces a new local maximum or minimum between two nodes, giving a smooth zero rate and forward rate.

Fields:

ref_date (datetime)
anchor_dates (list[datetime])
anchor_rates (list[DecimalNumber])
curve_type (Literal['interpolated_monotonic_cubic_curve'])

curve_type `pydantic-field` ¶

curve_type = 'interpolated_monotonic_cubic_curve'

ref_date `pydantic-field` ¶

ref_date

Reference date for the yield curve

anchor_dates `pydantic-field` ¶

anchor_dates

Maturity dates of the interpolation nodes, strictly after the reference date and in increasing order

anchor_rates `pydantic-field` ¶

anchor_rates

Continuously compounded zero rates at each anchor date (0.05 means 5%), same length as anchor_dates

instantaneous_forward_rate ¶

instantaneous_forward_rate(ttm)

Source code in quantflow/rates/interpolated.py

def instantaneous_forward_rate(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    dr = self._zero_rate_derivative(tau)
    return maybe_float(r + tau * dr)

discount_factor ¶

discount_factor(ttm)

Source code in quantflow/rates/interpolated.py

def discount_factor(self, ttm: FloatArrayLike) -> FloatArrayLike:
    tau = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    r = self._zero_rate(tau)
    return maybe_float(np.exp(-r * tau))

calibrator ¶

calibrator()

Return an [InterpolatedYieldCurveCalibration][ ...InterpolatedYieldCurveCalibration] wrapping this curve.

Source code in quantflow/rates/interpolated.py

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

jacobian ¶

jacobian(ttm)

Analytical Jacobian of discount factors w.r.t. model parameters.

Returns None if no analytical Jacobian is available (default). Shape when not None: (len(ttm), n_params).

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `FloatArrayLike`

Source code in quantflow/rates/yield_curve.py

def jacobian(
    self, ttm: Annotated[FloatArrayLike, Doc("Times to maturity in years.")]
) -> FloatArray | None:
    """Analytical Jacobian of discount factors w.r.t. model parameters.

    Returns None if no analytical Jacobian is available (default).
    Shape when not None: (len(ttm), n_params).
    """
    return None

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.interpolated.InterpolatedYieldCurveCalibration `pydantic-model` ¶

Bases: YieldCurveCalibration[InterpolatedYieldCurve]

Calibration wrapper for an interpolated yield curve.

The interpolated curve passes exactly through its nodes, so calibration is direct: the anchor dates and rates are set from the input times to maturity and continuously compounded rates. The free parameters are the anchor rates.

Fields:

yield_curve (Y)

yield_curve `pydantic-field` ¶

yield_curve

Yield curve to be calibrated

get_params ¶

get_params()

Source code in quantflow/rates/interpolated.py

def get_params(self) -> FloatArray:
    return np.array([float(r) for r in self.yield_curve.anchor_rates], dtype=float)

set_params ¶

set_params(params)

Source code in quantflow/rates/interpolated.py

def set_params(self, params: FloatArray) -> None:
    curve = self.yield_curve
    rates = np.asarray(params, dtype=float)
    curve.anchor_rates = [Decimal(str(round(float(r), 10))) for r in rates]
    curve._rates = rates

get_bounds ¶

get_bounds()

Source code in quantflow/rates/interpolated.py

def get_bounds(self) -> Bounds:
    n = len(self.yield_curve.anchor_rates)
    return Bounds(np.full(n, -np.inf), np.full(n, np.inf))

calibrate ¶

calibrate(ttm, rates)

Set the curve nodes so it reprices the given rates exactly.

Maturity dates are reconstructed from the times to maturity relative to ref_date on an ACT/365 basis.

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/interpolated.py

def calibrate(
    self,
    ttm: Annotated[ArrayLike, Doc("Times to maturity in years.")],
    rates: Annotated[
        ArrayLike, Doc("Continuously compounded rates, same length as ttm.")
    ],
) -> InterpolatedYieldCurve:
    """Set the curve nodes so it reprices the given rates exactly.

    Maturity dates are reconstructed from the times to maturity relative to
    [ref_date][quantflow.rates.yield_curve.YieldCurve.ref_date] on an
    ACT/365 basis.
    """
    ttm_ = np.asarray(ttm, dtype=float)
    rates_ = np.asarray(rates, dtype=float)
    order = np.argsort(ttm_)
    curve = self.yield_curve
    ref = curve.ref_date
    curve.anchor_dates = [
        ref + timedelta(seconds=float(t) * _YEAR) for t in ttm_[order]
    ]
    curve.anchor_rates = [Decimal(str(round(float(r), 10))) for r in rates_[order]]
    curve._ttm = ttm_[order]
    curve._rates = rates_[order]
    return 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.

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.interpolated.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)

calibrate_historical_rates ¶

calibrate_historical_rates(ttm, rates, dt)

Model-specific hook for historical rate calibration.

Default implementation raises NotImplementedError. Subclasses with a stochastic short-rate dynamic override this method.

PARAMETER	DESCRIPTION
`ttm`	Times to maturity in years. TYPE: `FloatArray`
`rates`	Continuously compounded rates, same shape as ttm. TYPE: `FloatArray`
`dt`	Time increments between observations, same length as rates. TYPE: `FloatArray`

Source code in quantflow/rates/calibration.py

def calibrate_historical_rates(
    self,
    ttm: Annotated[FloatArray, Doc("Times to maturity in years.")],
    rates: Annotated[
        FloatArray, Doc("Continuously compounded rates, same shape as ttm.")
    ],
    dt: Annotated[
        FloatArray,
        Doc("Time increments between observations, same length as rates."),
    ],
) -> Y:
    """Model-specific hook for historical rate calibration.

    Default implementation raises NotImplementedError. Subclasses with a
    stochastic short-rate dynamic override this method.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support historical rate calibration"
    )

Interpolated Curves¶

quantflow.rates.interpolated.InterpolatedYieldCurve pydantic-model ¶

anchor_dates pydantic-field ¶

anchor_rates pydantic-field ¶

ref_date pydantic-field ¶

curve_type pydantic-field ¶

calibrator ¶

instantaneous_forward_rate ¶

discount_factor ¶

jacobian ¶

continuously_compounded_rate ¶

rates ¶

plot ¶

register_curve_types classmethod ¶

curve_types classmethod ¶

get_curve_class classmethod ¶

quantflow.rates.interpolated.InterpolatedLinearCurve pydantic-model ¶

curve_type pydantic-field ¶

ref_date pydantic-field ¶

anchor_dates pydantic-field ¶

anchor_rates pydantic-field ¶

instantaneous_forward_rate ¶

discount_factor ¶

calibrator ¶

jacobian ¶

continuously_compounded_rate ¶

rates ¶

plot ¶

register_curve_types classmethod ¶

curve_types classmethod ¶

get_curve_class classmethod ¶

quantflow.rates.interpolated.InterpolatedMonotonicCubicCurve pydantic-model ¶

curve_type pydantic-field ¶

ref_date pydantic-field ¶

anchor_dates pydantic-field ¶

anchor_rates pydantic-field ¶

instantaneous_forward_rate ¶

discount_factor ¶

calibrator ¶

jacobian ¶

continuously_compounded_rate ¶

rates ¶

plot ¶

register_curve_types classmethod ¶

curve_types classmethod ¶

get_curve_class classmethod ¶

quantflow.rates.interpolated.InterpolatedYieldCurveCalibration pydantic-model ¶

yield_curve pydantic-field ¶

get_params ¶

set_params ¶

get_bounds ¶

calibrate ¶

calibrate_df ¶

calibrate_historical_rates_dataframe ¶

calibrate_historical_rates ¶

quantflow.rates.interpolated.InterpolatedYieldCurve `pydantic-model` ¶

anchor_dates `pydantic-field` ¶

anchor_rates `pydantic-field` ¶

ref_date `pydantic-field` ¶

curve_type `pydantic-field` ¶

register_curve_types `classmethod` ¶

curve_types `classmethod` ¶

get_curve_class `classmethod` ¶

quantflow.rates.interpolated.InterpolatedLinearCurve `pydantic-model` ¶

curve_type `pydantic-field` ¶

ref_date `pydantic-field` ¶

anchor_dates `pydantic-field` ¶

anchor_rates `pydantic-field` ¶

register_curve_types `classmethod` ¶

curve_types `classmethod` ¶

get_curve_class `classmethod` ¶

quantflow.rates.interpolated.InterpolatedMonotonicCubicCurve `pydantic-model` ¶

curve_type `pydantic-field` ¶

ref_date `pydantic-field` ¶

anchor_dates `pydantic-field` ¶

anchor_rates `pydantic-field` ¶

register_curve_types `classmethod` ¶

curve_types `classmethod` ¶

get_curve_class `classmethod` ¶

quantflow.rates.interpolated.InterpolatedYieldCurveCalibration `pydantic-model` ¶

yield_curve `pydantic-field` ¶