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

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 ¶

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` ¶