Nelson Siegel Curve¶

quantflow.rates.nelson_siegel.NelsonSiegelCurve `pydantic-model` ¶

Bases: YieldCurve

Class representing a Nelson-Siegel yield curve

The Nelson-Siegel model is a popular parametric model for fitting the term structure of interest rates. It is defined by the following formula for the instantaneous forward rate:

\[\begin{equation} f(\tau) = \beta_1 + \beta_2 e^{-\lambda \tau} + \beta_3 \lambda \tau e^{-\lambda \tau} \end{equation}\]

where \(\tau\) is the time to maturity, \(\beta_1\) is the level parameter, \(\beta_2\) is the slope parameter, \(\beta_3\) is the curvature parameter and \(\lambda\) is the decay factor.

Fields:

ref_date (datetime)
curve_type (Literal['nelson_siegel_curve'])
beta1 (Decimal)
beta2 (Decimal)
beta3 (Decimal)
lambda_ (Decimal)

curve_type `pydantic-field` ¶

curve_type = 'nelson_siegel_curve'

beta1 `pydantic-field` ¶

beta1 = Decimal(0)

Level parameter

beta2 `pydantic-field` ¶

beta2 = Decimal(0)

Slope parameter

beta3 `pydantic-field` ¶

beta3 = Decimal(0)

Curvature parameter

lambda_ `pydantic-field` ¶

lambda_ = Decimal(1)

Decay factor

ref_date `pydantic-field` ¶

ref_date

Reference date for the yield curve

calibrator ¶

calibrator()

Return a NelsonSiegelCalibration wrapping this curve.

Source code in quantflow/rates/nelson_siegel.py

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

instantaneous_forward_rate ¶

instantaneous_forward_rate(ttm)

Source code in quantflow/rates/nelson_siegel.py

def instantaneous_forward_rate(self, ttm: FloatArrayLike) -> FloatArrayLike:
    b1, b2, b3, lam = (
        float(self.beta1),
        float(self.beta2),
        float(self.beta3),
        float(self.lambda_),
    )
    ttm_ = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    lt = lam * ttm_
    et = np.exp(-lt)
    return maybe_float(b1 + b2 * et + b3 * lt * et)

discount_factor ¶

discount_factor(ttm)

Calculate the discount factor for a given time to maturity.

The discount factor is calculated using the formula:

\[\begin{align*} D(\tau) &= e^{-r(\tau) \tau} \\ r(\tau) &= \beta_1 + \beta_2 \frac{1 - e^{-\lambda \tau}} {\lambda \tau} + \beta_3 \left(\frac{1 - e^{-\lambda \tau}}{\lambda \tau} - e^{-\lambda \tau}\right) \end{align*}\]

Source code in quantflow/rates/nelson_siegel.py

def discount_factor(self, ttm: FloatArrayLike) -> FloatArrayLike:
    r"""Calculate the discount factor for a given time to maturity.

    The discount factor is calculated using the formula:

    \begin{align*}
        D(\tau) &= e^{-r(\tau) \tau} \\
        r(\tau) &= \beta_1 + \beta_2 \frac{1 - e^{-\lambda \tau}}
        {\lambda \tau} + \beta_3
        \left(\frac{1 - e^{-\lambda \tau}}{\lambda \tau}
          - e^{-\lambda \tau}\right)
    \end{align*}
    """
    ttma = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    tt = ttma * float(self.lambda_)
    et = np.exp(-tt)
    with np.errstate(divide="ignore", invalid="ignore"):
        ett = np.where(tt > 1e-10, (1 - et) / tt, 1.0)
    zero_coupon_rate = (
        float(self.beta1) + float(self.beta2) * ett + float(self.beta3) * (ett - et)
    )
    df = np.exp(-zero_coupon_rate * ttma)
    return maybe_float(df)

jacobian ¶

jacobian(ttm)

Analytical Jacobian of discount factors w.r.t. params.

Params order: \([\beta_1, \beta_2, \beta_3, \lambda]\). Shape: (len(ttm), 4).

Source code in quantflow/rates/nelson_siegel.py

def jacobian(self, ttm: FloatArrayLike) -> FloatArray:
    r"""Analytical Jacobian of discount factors w.r.t. params.

    Params order: $[\beta_1, \beta_2, \beta_3, \lambda]$. Shape: (len(ttm), 4).
    """
    b2, b3, lam = float(self.beta2), float(self.beta3), float(self.lambda_)
    ttma = np.maximum(np.asarray(ttm, dtype=float), 0.0)
    lt = lam * ttma
    et = np.exp(-lt)
    with np.errstate(divide="ignore", invalid="ignore"):
        ett = np.where(lt > 1e-10, (1.0 - et) / lt, 1.0 - lt / 2.0)
    a_term = ttma * ett
    b_term = a_term - ttma * et
    zero_rate = float(self.beta1) + b2 * ett + b3 * (ett - et)
    df = np.exp(-zero_rate * ttma)
    with np.errstate(divide="ignore", invalid="ignore"):
        da_dlam = np.where(lam > 1e-10, ttma * et / lam - a_term / lam, 0.0)
    db_dlam = da_dlam + ttma**2 * et
    return np.column_stack(
        [
            -ttma * df,
            -a_term * df,
            -b_term * df,
            -df * (b2 * da_dlam + b3 * db_dlam),
        ]
    )

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.nelson_siegel.NelsonSiegelCalibration `pydantic-model` ¶

Bases: YieldCurveCalibration[NelsonSiegelCurve]

Calibration wrapper for a Nelson-Siegel yield curve.

Fields:

yield_curve (Y)
beta_bounds (tuple[float, float])
lambda_bounds (tuple[float, float])

beta_bounds `pydantic-field` ¶

beta_bounds = (-0.5, 0.5)

Lower and upper bounds for beta parameters

lambda_bounds `pydantic-field` ¶

lambda_bounds = (0.01, 50.0)

Lower and upper bounds for the decay parameter

yield_curve `pydantic-field` ¶

yield_curve

Yield curve to be calibrated

get_params ¶

get_params()

Source code in quantflow/rates/nelson_siegel.py

def get_params(self) -> FloatArray:
    ns = self.yield_curve
    return np.array(
        [float(ns.beta1), float(ns.beta2), float(ns.beta3), float(ns.lambda_)]
    )

set_params ¶

set_params(params)

Source code in quantflow/rates/nelson_siegel.py

def set_params(self, params: FloatArray) -> None:
    b1, b2, b3, lam = params
    self.yield_curve.beta1 = Decimal(str(round(float(b1), 10)))
    self.yield_curve.beta2 = Decimal(str(round(float(b2), 10)))
    self.yield_curve.beta3 = Decimal(str(round(float(b3), 10)))
    self.yield_curve.lambda_ = Decimal(str(round(float(lam), 10)))

get_bounds ¶

get_bounds()

Source code in quantflow/rates/nelson_siegel.py

def get_bounds(self) -> Bounds:
    lo, hi = self.beta_bounds
    lam_lo, lam_hi = self.lambda_bounds
    return Bounds([lo, lo, lo, lam_lo], [hi, hi, hi, lam_hi])

calibrate ¶

calibrate(ttm, rates)

Fit the curve using the fast profile-OLS solver.

Drops times to maturity below 1 day, which are often dominated by noise.

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

def calibrate(
    self,
    ttm: Annotated[ArrayLike, Doc("Times to maturity in years.")],
    rates: Annotated[
        ArrayLike, Doc("Continuously compounded rates, same length as ttm.")
    ],
) -> NelsonSiegelCurve:
    """Fit the curve using the fast profile-OLS solver.

    Drops times to maturity below 1 day, which are often dominated by noise.
    """
    ttm_ = np.asarray(ttm, dtype=float)
    rates_ = np.asarray(rates, dtype=float)
    mask = ttm_ >= 1 / 365
    ttm_arr = ttm_[mask]
    rates_arr = rates_[mask]
    lo, hi = self.lambda_bounds
    grid = np.linspace(lo, hi, 100)
    rss_values = [_rss(lam, ttm_arr, rates_arr) for lam in grid]
    best_idx = int(np.argmin(rss_values))
    refine_lo = grid[max(best_idx - 1, 0)]
    refine_hi = grid[min(best_idx + 1, len(grid) - 1)]
    result = minimize_scalar(
        _rss,
        bounds=(refine_lo, refine_hi),
        method="bounded",
        args=(ttm_arr, rates_arr),
    )
    lam: float = result.x
    b1, b2, b3 = _ols_betas(ttm_arr, rates_arr, lam)
    self.yield_curve.beta1 = Decimal(str(round(b1, 10)))
    self.yield_curve.beta2 = Decimal(str(round(b2, 10)))
    self.yield_curve.beta3 = Decimal(str(round(b3, 10)))
    self.yield_curve.lambda_ = Decimal(str(round(lam, 10)))
    return self.yield_curve

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.nelson_siegel.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"
    )

Nelson Siegel Curve¶

quantflow.rates.nelson_siegel.NelsonSiegelCurve pydantic-model ¶

curve_type pydantic-field ¶

beta1 pydantic-field ¶

beta2 pydantic-field ¶

beta3 pydantic-field ¶

lambda_ pydantic-field ¶

ref_date 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.nelson_siegel.NelsonSiegelCalibration pydantic-model ¶

beta_bounds pydantic-field ¶

lambda_bounds pydantic-field ¶

yield_curve pydantic-field ¶

get_params ¶

set_params ¶

get_bounds ¶

calibrate ¶

prepare ¶

calibrate_df ¶

calibrate_historical_rates_dataframe ¶

calibrate_historical_rates ¶

quantflow.rates.nelson_siegel.NelsonSiegelCurve `pydantic-model` ¶

curve_type `pydantic-field` ¶

beta1 `pydantic-field` ¶

beta2 `pydantic-field` ¶

beta3 `pydantic-field` ¶

lambda_ `pydantic-field` ¶

ref_date `pydantic-field` ¶

register_curve_types `classmethod` ¶

curve_types `classmethod` ¶

get_curve_class `classmethod` ¶

quantflow.rates.nelson_siegel.NelsonSiegelCalibration `pydantic-model` ¶

beta_bounds `pydantic-field` ¶

lambda_bounds `pydantic-field` ¶

yield_curve `pydantic-field` ¶