Yield Curve Calibration

quantflow.rates.calibration.YieldCurveCalibration pydantic-model

Bases: BaseModel, Generic[Y]

Fields:

• yield_curve (Y)

yield_curve pydantic-field

yield_curve

Yield curve to be calibrated

get_params abstractmethod

get_params()

Current model parameters as a flat array (starting point for fit)

Source code in quantflow/rates/calibration.py

@abstractmethod
def get_params(self) -> FloatArray:
    """Current model parameters as a flat array (starting point for fit)"""

set_params abstractmethod

set_params(params)

Update the yield curve from a flat parameter array

Source code in quantflow/rates/calibration.py

@abstractmethod
def set_params(self, params: FloatArray) -> None:
    """Update the yield curve from a flat parameter array"""

get_bounds abstractmethod

get_bounds()

Parameter bounds for the optimiser

Source code in quantflow/rates/calibration.py

@abstractmethod
def get_bounds(self) -> Bounds:
    """Parameter bounds for the optimiser"""

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 abstractmethod

calibrate(ttm, rates)

Fit the yield curve to continuously compounded rates.

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

@abstractmethod
def calibrate(
    self,
    ttm: Annotated[ArrayLike, Doc("Times to maturity in years.")],
    rates: Annotated[
        ArrayLike, Doc("Continuously compounded rates, same length as ttm.")
    ],
) -> Y:
    """Fit the yield curve to continuously compounded rates."""

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

quantflow.rates.calibration.OptionsDiscountingCalibration dataclass

OptionsDiscountingCalibration(
    asset_curve, quote_curve, cp, strikes, ttm
)

Calibrate yield curves from option price parity data.

The input data consists of arrays of call-put parity values, strikes, and times to maturity for a set of options on the same underlying. The calibration can be done jointly for both the asset and quote curves, or separately for one curve with the other fixed.

asset_curve instance-attribute

asset_curve

Yield curve for the underlying asset. An instance is treated as fixed; a YieldCurveCalibration will be calibrated from the parity data.

quote_curve instance-attribute

quote_curve

Yield curve for the quote asset. An instance is treated as fixed; a YieldCurveCalibration will be calibrated from the parity data.

cp instance-attribute

cp

(Call - Put) / Spot for each option pair

strikes instance-attribute

strikes

Strike / Spot for each option pair, same length as cp

ttm instance-attribute

ttm

Time to maturity in years for each option pair, same length as cp

calibrate

calibrate()

Source code in quantflow/rates/calibration.py

def calibrate(self) -> tuple[YieldCurve, YieldCurve]:
    if isinstance(self.asset_curve, YieldCurveCalibration):
        if isinstance(self.quote_curve, YieldCurveCalibration):
            return self.joint_calibration(self.asset_curve, self.quote_curve)
        else:
            return self.asset_calibration(self.asset_curve, self.quote_curve)
    elif isinstance(self.quote_curve, YieldCurveCalibration):
        return self.quote_calibration(self.asset_curve, self.quote_curve)
    else:
        return self.asset_curve, self.quote_curve

joint_calibration

joint_calibration(asset_cal, quote_cal)

Calibrate both curves jointly from all parity observations.

Source code in quantflow/rates/calibration.py

def joint_calibration(
    self,
    asset_cal: YieldCurveCalibration,
    quote_cal: YieldCurveCalibration,
) -> tuple[YieldCurve, YieldCurve]:
    """Calibrate both curves jointly from all parity observations."""
    asset_cal.prepare(self.ttm)
    quote_cal.prepare(self.ttm)
    pa = asset_cal.get_params()
    pq = quote_cal.get_params()
    has_jacobian = (
        asset_cal.yield_curve.jacobian(self.ttm) is not None
        and quote_cal.yield_curve.jacobian(self.ttm) is not None
    )
    n_a = len(pa)
    bounds = Bounds(
        np.concatenate([asset_cal.get_bounds().lb, quote_cal.get_bounds().lb]),
        np.concatenate([asset_cal.get_bounds().ub, quote_cal.get_bounds().ub]),
    )

    def residuals(params: np.ndarray) -> np.ndarray:
        asset_cal.set_params(params[:n_a])
        quote_cal.set_params(params[n_a:])
        da = asset_cal(self.ttm)
        dq = quote_cal(self.ttm)
        return self.cp - da + dq * self.strikes

    def jac(params: np.ndarray) -> FloatArray:
        asset_cal.set_params(params[:n_a])
        quote_cal.set_params(params[n_a:])
        ja = asset_cal.yield_curve.jacobian(self.ttm)
        jq = quote_cal.yield_curve.jacobian(self.ttm)
        if ja is None or jq is None:  # pragma: no cover
            raise TypeError("jacobian must not return None in joint calibration")
        return np.hstack([-ja, jq * self.strikes[:, None]])

    result = least_squares(
        residuals,
        np.concatenate([pa, pq]),
        jac=jac if has_jacobian else "2-point",
        bounds=bounds,
        method="trf",
    )
    asset_cal.set_params(result.x[:n_a])
    quote_cal.set_params(result.x[n_a:])
    return asset_cal.yield_curve, quote_cal.yield_curve

asset_calibration

asset_calibration(asset_cal, fixed_quote)

Calibrate only the asset curve; quote curve is fixed.

Source code in quantflow/rates/calibration.py

def asset_calibration(
    self,
    asset_cal: YieldCurveCalibration,
    fixed_quote: YieldCurve,
) -> tuple[YieldCurve, YieldCurve]:
    """Calibrate only the asset curve; quote curve is fixed."""
    dq = np.asarray(fixed_quote.discount_factor(self.ttm), dtype=float)
    target_da = self.cp + dq * self.strikes
    return asset_cal.calibrate_df(self.ttm, target_da), fixed_quote

quote_calibration

quote_calibration(fixed_asset, quote_cal)

Calibrate only the quote curve; asset curve is fixed.

Source code in quantflow/rates/calibration.py

def quote_calibration(
    self,
    fixed_asset: YieldCurve,
    quote_cal: YieldCurveCalibration,
) -> tuple[YieldCurve, YieldCurve]:
    """Calibrate only the quote curve; asset curve is fixed."""
    da = np.asarray(fixed_asset.discount_factor(self.ttm), dtype=float)
    target_dq = (da - self.cp) / self.strikes
    return fixed_asset, quote_cal.calibrate_df(self.ttm, target_dq)