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"""

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