Workflows

The workflows package provides high-level classes that orchestrate multi-step processes like daily model calibration, historical backtesting and per-ticker tuning of the calibration settings.

BacktestWorkflow

BacktestWorkflow(ticker: str, model_config: dict, *, settings: CalibrationSettings | None = None, allow_mc: bool = False, dates: list[str] | None = None, warm_start: bool = True)

Walk-forward backtest of one model on one ticker.

Snapshot dates are processed in ascending order; the model is calibrated on day t and evaluated, with day t+1's own market parameters, on the liquid options of day t+1. Everything is offline (no network calls). Two baselines accompany every period:

• BSM-ATM - a flat Black-Scholes volatility equal to day t's ATM implied volatility;
• Stale - the model parameters calibrated one snapshot earlier (t-1) applied on t+1, i.e. what skipping the daily re-calibration costs.

Initializes the backtest workflow.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker to run the backtest for.	required
model_config	dict	A dictionary "recipe" defining how to calibrate the model.	required
settings	CalibrationSettings | None	Calibration settings (filters, objective, optimiser).	None
allow_mc	bool	Permit Monte Carlo pricing for simulation-only models.	False
dates	list[str] | None	Snapshot dates to use (any order); defaults to every stored date.	None
warm_start	bool	Seed each calibration with the previous period's parameters.	True

Source code in src/optpricing/workflows/backtest_workflow.py

def __init__(
    self,
    ticker: str,
    model_config: dict,
    *,
    settings: CalibrationSettings | None = None,
    allow_mc: bool = False,
    dates: list[str] | None = None,
    warm_start: bool = True,
):
    """
    Initializes the backtest workflow.

    Parameters
    ----------
    ticker : str
        The stock ticker to run the backtest for.
    model_config : dict
        A dictionary "recipe" defining how to calibrate the model.
    settings : CalibrationSettings | None, optional
        Calibration settings (filters, objective, optimiser).
    allow_mc : bool, optional
        Permit Monte Carlo pricing for simulation-only models.
    dates : list[str] | None, optional
        Snapshot dates to use (any order); defaults to every stored date.
    warm_start : bool, optional
        Seed each calibration with the previous period's parameters.
    """
    self.ticker = ticker.upper()
    self.model_config = dict(model_config)
    self.model_config.setdefault("ticker", self.ticker)
    self.settings = settings or CalibrationSettings()
    self.allow_mc = allow_mc
    self.dates = sorted(dates) if dates is not None else None
    self.warm_start = warm_start
    self.results: list[dict[str, Any]] = []
    self.bucket_results: list[dict[str, Any]] = []

run

run() -> None

Executes the full backtesting loop.

For each consecutive pair of snapshot dates, calibrates on the earlier day and records out-of-sample metrics on the later day in self.results (one row per period) and self.bucket_results.

Source code in src/optpricing/workflows/backtest_workflow.py

def run(self) -> None:
    """
    Executes the full backtesting loop.

    For each consecutive pair of snapshot dates, calibrates on the earlier
    day and records out-of-sample metrics on the later day in
    ``self.results`` (one row per period) and ``self.bucket_results``.
    """
    dates = self.dates or get_available_snapshot_dates(self.ticker, ascending=True)
    if len(dates) < 2:
        logger.warning(
            "Backtest for %s requires at least 2 days of data. Skipping.",
            self.model_config["name"],
        )
        return

    model_name = self.model_config["name"]
    previous_params: dict[str, float] | None = None
    for calib_date, eval_date in zip(dates[:-1], dates[1:]):
        assert calib_date < eval_date, "dates must be ascending (no look-ahead)"
        logger.info(
            "--- %s: calibrate on %s, evaluate on %s ---",
            model_name,
            calib_date,
            eval_date,
        )
        calib_data = load_market_snapshot(self.ticker, calib_date)
        eval_data = load_market_snapshot(self.ticker, eval_date)
        if calib_data is None or eval_data is None:
            continue

        t0 = time.perf_counter()
        calib_workflow = DailyWorkflow(
            market_data=calib_data,
            model_config=self.model_config,
            settings=self.settings,
            allow_mc=self.allow_mc,
            warm_start=previous_params if self.warm_start else None,
        )
        calib_workflow.run()
        row: dict[str, Any] = {
            "Calibration Date": calib_date,
            "Eval Date": eval_date,
            "Model": model_name,
            "Status": calib_workflow.results["Status"],
        }
        if calib_workflow.results["Status"] != "Success":
            row["Error"] = calib_workflow.results.get("Error")
            row["Out-of-Sample RMSE"] = np.nan
            row["Out-of-Sample IV RMSE"] = np.nan
            self.results.append(row)
            logger.warning("Calibration failed; period skipped (%s).", row["Error"])
            continue

        params = calib_workflow.results["Calibrated Params"]
        model = calib_workflow.model
        eval_workflow = DailyWorkflow(
            market_data=eval_data,
            model_config=self.model_config,
            settings=self.settings,
            allow_mc=self.allow_mc,
        )
        eval_workflow._prepare_for_evaluation()
        summary, buckets, _ = eval_workflow.evaluate(model)

        atm_vol = self._atm_vol(calib_workflow.calibration_data)
        bsm_atm_iv_rmse = np.nan
        if atm_vol is not None:
            bsm_summary, _, _ = eval_workflow.evaluate(BSMModel({"sigma": atm_vol}))
            bsm_atm_iv_rmse = bsm_summary["rmse_iv"]
        # Stale baseline: the parameters fitted one snapshot earlier (t-1)
        # applied on t+1, i.e. what skipping today's re-calibration costs.
        stale_iv_rmse = np.nan
        if previous_params is not None:
            try:
                stale_model = model.with_params(**previous_params)
                stale_summary, _, _ = eval_workflow.evaluate(stale_model)
                stale_iv_rmse = stale_summary["rmse_iv"]
            except ValueError:
                pass

        row.update(
            {
                "In-Sample IV RMSE": calib_workflow.results.get(
                    "Calibration IV RMSE"
                ),
                "Out-of-Sample IV RMSE": summary["rmse_iv"],
                "Out-of-Sample RMSE": summary["rmse_price"],
                "Out-of-Sample Relative RMSE": summary["rmse_rel_price"],
                "BSM-ATM IV RMSE": bsm_atm_iv_rmse,
                "Stale-Params IV RMSE": stale_iv_rmse,
                "N Options": summary["n_options"],
                "N Calibration Options": calib_workflow.results.get(
                    "N Calibration Options"
                ),
                "Runtime (s)": time.perf_counter() - t0,
                "Params": {k: float(v) for k, v in params.items()},
            }
        )
        self.results.append(row)
        for _, b in buckets.iterrows():
            self.bucket_results.append(
                {"Eval Date": eval_date, "Model": model_name, **b.to_dict()}
            )
        logger.info(
            "Out-of-sample %s on %s: IV RMSE %.4f (BSM-ATM %.4f), price RMSE %.4f",
            model_name,
            eval_date,
            summary["rmse_iv"],
            bsm_atm_iv_rmse,
            summary["rmse_price"],
        )
        previous_params = params

save_results

save_results() -> tuple[Path, Path] | None

Saves the per-period table (CSV), the bucket table (CSV) and the summary with parameters (JSON) in the backtest-logs directory.

Returns:

Type	Description
tuple[Path, Path] | None	(csv_path, json_path) or None when there is nothing to save.

Source code in src/optpricing/workflows/backtest_workflow.py

def save_results(self) -> tuple[Path, Path] | None:
    """
    Saves the per-period table (CSV), the bucket table (CSV) and the summary
    with parameters (JSON) in the backtest-logs directory.

    Returns
    -------
    tuple[Path, Path] | None
        ``(csv_path, json_path)`` or None when there is nothing to save.
    """
    if not self.results:
        logger.info("No backtest results to save.")
        return None

    model_name = self.model_config["name"].replace(" ", "_")
    today_str = pd.Timestamp.now().strftime("%Y-%m-%d")
    base = BACKTEST_LOGS_DIR / f"{self.ticker}_{model_name}_backtest_{today_str}"
    base.parent.mkdir(parents=True, exist_ok=True)

    df = pd.DataFrame(self.results)
    if "Params" in df.columns:
        df["Params"] = df["Params"].map(
            lambda p: json.dumps(p) if isinstance(p, dict) else p
        )
    csv_path = base.with_suffix(".csv")
    df.to_csv(csv_path, index=False)
    if self.bucket_results:
        pd.DataFrame(self.bucket_results).to_csv(
            base.with_name(base.name + "_buckets.csv"), index=False
        )
    json_path = base.with_suffix(".json")
    payload = {
        "summary": self.summary(),
        "settings": self.settings.to_dict(),
        "periods": self.results,
    }
    with open(json_path, "w") as f:
        json.dump(payload, f, indent=2, default=_json_default)
    logger.info("Backtest log saved to %s", csv_path)
    return csv_path, json_path

summary

summary() -> dict[str, Any]

Aggregates the per-period rows.

Returns:

Type	Description
dict[str, Any]	model, ticker, n_periods, n_failed, means and medians of the out-of-sample IV / price RMSE, the BSM-ATM baseline mean and the share of periods beating it.

Source code in src/optpricing/workflows/backtest_workflow.py

def summary(self) -> dict[str, Any]:
    """
    Aggregates the per-period rows.

    Returns
    -------
    dict[str, Any]
        ``model``, ``ticker``, ``n_periods``, ``n_failed``, means and medians
        of the out-of-sample IV / price RMSE, the BSM-ATM baseline mean and
        the share of periods beating it.
    """
    frame = pd.DataFrame(self.results)
    ok = frame[frame["Status"] == "Success"] if not frame.empty else frame
    out: dict[str, Any] = {
        "model": self.model_config["name"],
        "ticker": self.ticker,
        "n_periods": int(len(frame)),
        "n_failed": int(len(frame) - len(ok)),
    }
    if ok.empty:
        return out
    oos_iv = ok["Out-of-Sample IV RMSE"].astype(float)
    bsm = ok["BSM-ATM IV RMSE"].astype(float)
    out.update(
        {
            "mean_oos_iv_rmse": float(oos_iv.mean()),
            "median_oos_iv_rmse": float(oos_iv.median()),
            "mean_oos_price_rmse": float(
                ok["Out-of-Sample RMSE"].astype(float).mean()
            ),
            "mean_in_sample_iv_rmse": float(
                ok["In-Sample IV RMSE"].astype(float).mean()
            ),
            "mean_bsm_atm_iv_rmse": float(bsm.mean()),
            "share_beating_bsm_atm": float(
                np.mean(oos_iv.to_numpy() < bsm.to_numpy())
            ),
            "mean_runtime_s": float(ok["Runtime (s)"].astype(float).mean()),
        }
    )
    return out

DailyWorkflow

DailyWorkflow(market_data: DataFrame, model_config: dict[str, Any], *, settings: CalibrationSettings | None = None, allow_mc: bool = False, warm_start: dict[str, float] | None = None)

Orchestrates the calibration of a single model for a single snapshot of market data.

The workflow is fully offline:

1. Reads the market parameters (spot, dividend yield, rate) stored in the snapshot (implied from the chain for pre-v2 files).
2. Selects the calibration set with the settings' liquidity filter and the (wider) evaluation set.
3. Builds the initial guess from the recipe, the ATM implied volatility and, for jump models with a local returns history, a historical prior.
4. Calibrates with :class:~optpricing.calibration.Calibrator (IV-space weighted objective, multi-start) and evaluates the fit.

Initializes the daily workflow.

Parameters:

Name	Type	Description	Default
market_data	DataFrame	A DataFrame containing the option chain for a single snapshot date.	required
model_config	dict[str, Any]	A recipe: name, model_class and optionally initial_guess, bounds, frozen, historical_prior, requires_mc, ticker.	required
settings	CalibrationSettings | None	Filter / objective / optimiser settings; defaults apply if omitted.	None
allow_mc	bool	Permit Monte Carlo pricing for simulation-only models.	False
warm_start	dict[str, float] | None	A previous calibration used as an extra starting point.	None

Source code in src/optpricing/workflows/daily_workflow.py

def __init__(
    self,
    market_data: pd.DataFrame,
    model_config: dict[str, Any],
    *,
    settings: CalibrationSettings | None = None,
    allow_mc: bool = False,
    warm_start: dict[str, float] | None = None,
):
    """
    Initializes the daily workflow.

    Parameters
    ----------
    market_data : pd.DataFrame
        A DataFrame containing the option chain for a single snapshot date.
    model_config : dict[str, Any]
        A recipe: ``name``, ``model_class`` and optionally ``initial_guess``,
        ``bounds``, ``frozen``, ``historical_prior``, ``requires_mc``,
        ``ticker``.
    settings : CalibrationSettings | None, optional
        Filter / objective / optimiser settings; defaults apply if omitted.
    allow_mc : bool, optional
        Permit Monte Carlo pricing for simulation-only models.
    warm_start : dict[str, float] | None, optional
        A previous calibration used as an extra starting point.
    """
    self.market_data = market_data
    self.model_config = model_config
    self.settings = settings or CalibrationSettings()
    self.allow_mc = allow_mc
    self.warm_start = warm_start
    self.results: dict[str, Any] = {"Model": self.model_config["name"]}
    ticker = self.model_config.get("ticker")
    if ticker is None and "ticker" in getattr(market_data, "columns", []):
        ticker = str(market_data["ticker"].iloc[0])
    if ticker is not None:
        self.results["Ticker"] = ticker
    self.stock: Stock | None = None
    self.rate: Rate | None = None
    self.calibration_data: pd.DataFrame | None = None
    self.evaluation_data: pd.DataFrame | None = None
    self.calibrator: Calibrator | None = None
    self.model: BaseModel | None = None

evaluate

evaluate(model: BaseModel, stock: Stock | None = None, rate: Rate | None = None, data: DataFrame | None = None) -> tuple[dict[str, float], pd.DataFrame, pd.DataFrame]

Fit metrics of model on the evaluation set (or data).

Returns:

Type	Description
tuple[dict[str, float], DataFrame, DataFrame]	Summary metrics, bucket table and per-option residuals (see :func:optpricing.calibration.evaluate_model).

Source code in src/optpricing/workflows/daily_workflow.py

def evaluate(
    self,
    model: BaseModel,
    stock: Stock | None = None,
    rate: Rate | None = None,
    data: pd.DataFrame | None = None,
) -> tuple[dict[str, float], pd.DataFrame, pd.DataFrame]:
    """
    Fit metrics of ``model`` on the evaluation set (or ``data``).

    Returns
    -------
    tuple[dict[str, float], pd.DataFrame, pd.DataFrame]
        Summary metrics, bucket table and per-option residuals (see
        :func:`optpricing.calibration.evaluate_model`).
    """
    stock = stock or self.stock
    rate = rate or self.rate
    if stock is None or rate is None:
        self._prepare_for_evaluation()
        stock, rate = self.stock, self.rate
    if data is None:
        if self.evaluation_data is None:
            self._prepare_for_evaluation()
        data = self.evaluation_data
    return evaluate_model(model, data, stock, rate, allow_mc=self.allow_mc)

run

run() -> None

Executes the full calibration and evaluation workflow.

Populates self.results with the status, the calibrated parameters and the fit metrics; never raises (failures are recorded with Status = "Failed" and an Error message).

Source code in src/optpricing/workflows/daily_workflow.py

def run(self) -> None:
    """
    Executes the full calibration and evaluation workflow.

    Populates ``self.results`` with the status, the calibrated parameters and
    the fit metrics; never raises (failures are recorded with ``Status`` =
    ``"Failed"`` and an ``Error`` message).
    """
    model_name = self.model_config["name"]
    ticker = self.results.get("Ticker", "N/A")
    logger.info("### Workflow for %s on %s", model_name, ticker)
    t0 = time.perf_counter()
    try:
        self._prepare_for_evaluation()
        self.results.update(
            {
                "Implied Rate": self.rate.get_rate(),
                "Known Dividend": self.stock.dividend,
            }
        )

        annotated = annotate_chain(self.market_data, self.stock, self.rate)
        self.calibration_data = filter_options(
            annotated, self.stock, self.rate, self.settings.liquidity
        )
        logger.info(
            "  -> %d of %d options selected for calibration.",
            len(self.calibration_data),
            len(self.market_data),
        )
        if len(self.calibration_data) < 5:
            raise ValueError(
                f"Only {len(self.calibration_data)} liquid options after "
                "filtering; cannot calibrate."
            )

        model_class = self.model_config["model_class"]
        if self.model_config.get("requires_mc") and not self.allow_mc:
            raise NotImplementedError(
                f"{model_name} needs Monte Carlo pricing; rerun with allow_mc."
            )
        model_instance = model_class(params=dict(model_class.default_params))

        initial_guess, bounds, frozen = self._initial_guess_and_bounds(
            self.calibration_data
        )
        logger.info(
            "  -> Fitting %s (frozen: %s)",
            [p for p in initial_guess if p not in frozen],
            list(frozen),
        )

        self.calibrator = Calibrator(
            model_instance,
            self.calibration_data,
            self.stock,
            self.rate,
            objective=self.settings.objective,
            weighting=self.settings.weighting,
            allow_mc=self.allow_mc,
            feller_penalty=self.settings.feller_penalty,
        )
        calibrated = self.calibrator.fit(
            initial_guess=initial_guess,
            bounds=bounds,
            frozen_params=frozen,
            n_starts=self.settings.n_starts,
            warm_start=self.warm_start if self.settings.warm_start else None,
            time_budget=self.settings.time_budget,
            log_scale=self.model_config.get("log_scale", ()),
        )
        self.results["Calibrated Params"] = calibrated
        fit = self.calibrator.result
        self.results["Calibration IV RMSE"] = fit.rmse_iv
        self.results["Calibration RMSE"] = fit.rmse_price
        self.results["N Calibration Options"] = fit.n_options
        self.results["Evaluations"] = fit.n_evaluations

        self.model = model_instance.with_params(**calibrated)
        summary, buckets, _ = self.evaluate(self.model)
        self.results["RMSE"] = summary["rmse_price"]
        self.results["IV RMSE"] = summary["rmse_iv"]
        self.results["Relative RMSE"] = summary["rmse_rel_price"]
        self.results["N Options"] = summary["n_options"]
        self.results["Buckets"] = buckets
        self.results["Status"] = "Success"
        logger.info(
            "  -> %s: calibration IV RMSE %.4f, evaluation IV RMSE %.4f, "
            "price RMSE %.4f",
            model_name,
            fit.rmse_iv,
            summary["rmse_iv"],
            summary["rmse_price"],
        )
    except Exception as e:
        logger.error("Workflow failed for %s: %s", model_name, e)
        logger.debug("Traceback:", exc_info=True)
        self.results.update({"RMSE": np.nan, "Status": "Failed", "Error": str(e)})
    finally:
        self.results["Runtime (s)"] = time.perf_counter() - t0

TuningResult dataclass

TuningResult(ticker: str, created: str, dates: list[str], leaderboard: list[dict[str, Any]] = list(), settings: dict[str, dict[str, Any]] = dict(), trials: list[dict[str, Any]] = list())

Cached outcome of :func:tune_ticker.

Attributes:

Name	Type	Description
ticker	str	Ticker symbol.
created	str	ISO timestamp of the run.
dates	list[str]	Snapshot dates used (ascending).
leaderboard	list[dict[str, Any]]	One row per model, best first: model, settings, oos_iv_rmse, in_sample_iv_rmse, oos_price_rmse, bsm_atm_iv_rmse, runtime_s, n_periods, n_failed.
settings	dict[str, dict[str, Any]]	Best settings (serialised) per model.
trials	list[dict[str, Any]]	Every (model, settings) score.

load_tuned_settings

load_tuned_settings(ticker: str, model: str) -> CalibrationSettings | None

Best cached settings for (ticker, model), or None when not tuned.

Source code in src/optpricing/workflows/tuning.py

def load_tuned_settings(ticker: str, model: str) -> CalibrationSettings | None:
    """Best cached settings for ``(ticker, model)``, or None when not tuned."""
    result = load_tuning(ticker)
    return result.settings_for(model) if result else None

load_tuning

load_tuning(ticker: str) -> TuningResult | None

The cached tuning result for a ticker, or None.

Source code in src/optpricing/workflows/tuning.py

def load_tuning(ticker: str) -> TuningResult | None:
    """The cached tuning result for a ticker, or None."""
    path = tuning_path(ticker)
    if not path.exists():
        return None
    try:
        return TuningResult.load(path)
    except (OSError, ValueError, KeyError) as e:
        logger.warning("Could not read %s: %s", path, e)
        return None

tune_ticker

tune_ticker(ticker: str, *, models: list[str] | None = None, candidates: list[str] | str | None = None, dates: list[str] | None = None, max_pairs: int | None = None, allow_mc: bool = False, progress: Callable[[str], None] | None = None, save: bool = True) -> TuningResult

Walk-forward tuning of calibration settings and model ranking for a ticker.

Parameters:

Name	Type	Description	Default
ticker	str	Ticker with at least two stored snapshots.	required
models	list[str] | None	Recipe names; defaults to every calibratable model.	None
candidates	list[str] | str | None	Candidate setting names from :data:CANDIDATE_SETTINGS, "quick" (three candidates, the default) or "full".	None
dates	list[str] | None	Snapshot dates to use; defaults to all stored dates.	None
max_pairs	int | None	Use only the most recent max_pairs + 1 dates.	None
allow_mc	bool	Include simulation-only models (slow).	False
progress	Callable[[str], None] | None	Called with a short message before each trial.	None
save	bool	Write the result to artifacts/tuning, by default True.	True

Returns:

Type	Description
TuningResult	Best settings per model and the leaderboard.

Raises:

Type	Description
ValueError	If fewer than two snapshot dates are available.

Source code in src/optpricing/workflows/tuning.py

def tune_ticker(
    ticker: str,
    *,
    models: list[str] | None = None,
    candidates: list[str] | str | None = None,
    dates: list[str] | None = None,
    max_pairs: int | None = None,
    allow_mc: bool = False,
    progress: Callable[[str], None] | None = None,
    save: bool = True,
) -> TuningResult:
    """
    Walk-forward tuning of calibration settings and model ranking for a ticker.

    Parameters
    ----------
    ticker : str
        Ticker with at least two stored snapshots.
    models : list[str] | None, optional
        Recipe names; defaults to every calibratable model.
    candidates : list[str] | str | None, optional
        Candidate setting names from :data:`CANDIDATE_SETTINGS`, ``"quick"``
        (three candidates, the default) or ``"full"``.
    dates : list[str] | None, optional
        Snapshot dates to use; defaults to all stored dates.
    max_pairs : int | None, optional
        Use only the most recent ``max_pairs + 1`` dates.
    allow_mc : bool, optional
        Include simulation-only models (slow).
    progress : Callable[[str], None] | None, optional
        Called with a short message before each trial.
    save : bool, optional
        Write the result to ``artifacts/tuning``, by default True.

    Returns
    -------
    TuningResult
        Best settings per model and the leaderboard.

    Raises
    ------
    ValueError
        If fewer than two snapshot dates are available.
    """
    ticker = ticker.upper()
    dates = sorted(dates or get_available_snapshot_dates(ticker, ascending=True))
    if max_pairs is not None:
        dates = dates[-(max_pairs + 1) :]
    if len(dates) < 2:
        raise ValueError(
            f"Tuning {ticker} needs at least two snapshots; found {len(dates)}."
        )
    models = models or calibratable_model_names(allow_mc=allow_mc)
    if candidates is None or candidates == "quick":
        candidate_names = list(QUICK_CANDIDATES)
    elif candidates == "full":
        candidate_names = list(CANDIDATE_SETTINGS)
    else:
        candidate_names = list(candidates)
    unknown = [c for c in candidate_names if c not in CANDIDATE_SETTINGS]
    if unknown:
        raise ValueError(f"Unknown candidate settings: {unknown}")

    result = TuningResult(
        ticker=ticker,
        created=pd.Timestamp.now().isoformat(timespec="seconds"),
        dates=dates,
    )
    for model in models:
        cfg = get_model_config(model)
        best: dict[str, Any] | None = None
        for name in candidate_names:
            settings = CANDIDATE_SETTINGS[name]
            if progress:
                progress(f"{model} / {name}")
            t0 = time.perf_counter()
            bt = BacktestWorkflow(
                ticker, cfg, settings=settings, dates=dates, allow_mc=allow_mc
            )
            bt.run()
            summary = bt.summary()
            trial = {
                "model": cfg["name"],
                "settings": name,
                "oos_iv_rmse": _score(summary),
                "in_sample_iv_rmse": summary.get("mean_in_sample_iv_rmse", np.nan),
                "oos_price_rmse": summary.get("mean_oos_price_rmse", np.nan),
                "bsm_atm_iv_rmse": summary.get("mean_bsm_atm_iv_rmse", np.nan),
                "runtime_s": time.perf_counter() - t0,
                "n_periods": summary.get("n_periods", 0),
                "n_failed": summary.get("n_failed", 0),
            }
            result.trials.append(trial)
            logger.info(
                "%s / %s: OOS IV RMSE %.4f (%d periods, %.1fs)",
                cfg["name"],
                name,
                trial["oos_iv_rmse"],
                trial["n_periods"],
                trial["runtime_s"],
            )
            if np.isfinite(trial["oos_iv_rmse"]) and (
                best is None
                or trial["oos_iv_rmse"] < best["oos_iv_rmse"]
                or (
                    trial["oos_iv_rmse"] == best["oos_iv_rmse"]
                    and trial["runtime_s"] < best["runtime_s"]
                )
            ):
                best = trial
        if best is None:
            logger.warning(
                "No successful trial for %s; left out of the leaderboard.", model
            )
            continue
        result.leaderboard.append(dict(best))
        result.settings[cfg["name"]] = CANDIDATE_SETTINGS[best["settings"]].to_dict()

    result.leaderboard.sort(key=lambda row: row["oos_iv_rmse"])
    for rank, row in enumerate(result.leaderboard, start=1):
        row["rank"] = rank
    if save and result.leaderboard:
        path = result.save()
        logger.info("Tuning result saved to %s", path)
    return result

tuning_path

tuning_path(ticker: str) -> Path

artifacts/tuning/{TICKER}.json.

Source code in src/optpricing/workflows/tuning.py

def tuning_path(ticker: str) -> Path:
    """``artifacts/tuning/{TICKER}.json``."""
    return TUNING_DIR / f"{ticker.upper()}.json"

Calibration recipes

calibratable_model_names

calibratable_model_names(*, allow_mc: bool = False) -> list[str]

Recipe keys whose model can be fitted to a chain: models with a vectorised closed form or a characteristic function, plus (with allow_mc) the simulation-only ones.

Source code in src/optpricing/workflows/configs/__init__.py

def calibratable_model_names(*, allow_mc: bool = False) -> list[str]:
    """
    Recipe keys whose model can be fitted to a chain: models with a vectorised
    closed form or a characteristic function, plus (with ``allow_mc``) the
    simulation-only ones.
    """
    names = []
    for key, cfg in ALL_MODEL_CONFIGS.items():
        model = cfg["model_class"](params=dict(cfg["model_class"].default_params))
        if supports_vectorized_pricing(model, allow_mc=allow_mc):
            names.append(key)
    return names

get_model_config

get_model_config(name: str) -> dict[str, Any]

A copy of the recipe for name (see :func:resolve_model_name).

Source code in src/optpricing/workflows/configs/__init__.py

def get_model_config(name: str) -> dict[str, Any]:
    """A copy of the recipe for ``name`` (see :func:`resolve_model_name`)."""
    return dict(ALL_MODEL_CONFIGS[resolve_model_name(name)])

resolve_model_name

resolve_model_name(name: str) -> str

Canonical recipe key for a user-supplied model name (case-insensitive, aliases such as "Variance Gamma" accepted).

Raises:

Type	Description
KeyError	If the name matches no recipe.

Source code in src/optpricing/workflows/configs/__init__.py

def resolve_model_name(name: str) -> str:
    """
    Canonical recipe key for a user-supplied model name (case-insensitive,
    aliases such as ``"Variance Gamma"`` accepted).

    Raises
    ------
    KeyError
        If the name matches no recipe.
    """
    key = name.strip()
    if key in ALL_MODEL_CONFIGS:
        return key
    lowered = key.lower()
    for canonical in ALL_MODEL_CONFIGS:
        if canonical.lower() == lowered:
            return canonical
    if lowered in MODEL_ALIASES:
        return MODEL_ALIASES[lowered]
    for canonical, cfg in ALL_MODEL_CONFIGS.items():
        if cfg["model_class"].name.lower() == lowered:
            return canonical
    raise KeyError(
        f"Unknown model '{name}'. Available: {', '.join(ALL_MODEL_CONFIGS)}."
    )