run_cell — the unit of work

worker

The cell runner — the unit of work that runs identically local / Spark / Ray.

run_cell fits, optionally backtests, and predicts ONE (ts_id, model) cell and returns a CellResult. Engines differ only in how they call it and collect its results — that symmetry is what makes "same code everywhere" real.

CellResult is defined here because it is the worker's output type; the registry writers (registry/) consume it. It carries plain data (frames + scalars), no behavior, so it is the clean seam between compute and lineage.

CellResult dataclass

CellResult(
    run_id,
    ts_id,
    model_type,
    compute_engine,
    model_hash,
    status,
    error,
    predictions,
    oof,
    metrics,
    best_params=dict(),
    fit_seconds=0.0,
    worker_id=None,
    cell_started_at=None,
    cell_ended_at=None,
    artifact_bytes=None,
    model_artifact_uri=None,
    cpu_seconds=None,
    process_rss_bytes=None,
    peak_gpu_bytes=None,
    intraop_threads=None,
    n_obs=None,
    device_requested=None,
    device_available=None,
    device_used=None,
    device_name=None,
    backtest_status=None,
    n_folds_achieved=None,
    backtest_note=None,
    backtest_refit=None,
    staleness_gap=None,
    achieved_step=None,
    achieved_min_train=None,
    first_val_date=None,
    last_val_date=None,
    error_class=None,
    interval_source=None,
    point_forecast_source=None,
    point_forecast_decision=None,
    interval_calibration=None,
    point_forecast_margin=None,
    hpo_scoring=None,
    diagnostics=dict(),
    demand_profile=None,
    explanations=None,
    n_fits=None,
    train_rows_total=None,
    n_hpo_fits=None,
)

The result of one (ts_id, model) cell.

A failing cell sets status="error" with error populated and empty predictions — it never raises out of run_cell, so one bad cell can't sink a 100k-series batch.

classify_error

classify_error(exc)

Which ERROR_CLASSES token describes exc — "UNKNOWN" if none does (pure).

UNKNOWN is a real answer, not a failure of the table. A token invented per stack trace would make the column unqueryable, so the vocabulary stays fixed and the honest response to an unrecognised failure is to say so and leave the full text in error_detail. A rising UNKNOWN share is the signal that this table needs a row, and it is visible in one GROUP BY.

Source code in src/scale_forecasting/worker.py

def classify_error(exc: BaseException) -> str:
    """Which `ERROR_CLASSES` token describes ``exc`` — ``"UNKNOWN"`` if none does (pure).

    ``UNKNOWN`` is a real answer, not a failure of the table. A token invented per stack trace
    would make the column unqueryable, so the vocabulary stays fixed and the honest response to an
    unrecognised failure is to say so and leave the full text in ``error_detail``. A rising
    ``UNKNOWN`` share is the signal that this table needs a row, and it is visible in one GROUP BY.
    """
    names = {t.__name__ for t in type(exc).__mro__}
    text = f"{type(exc).__name__}: {exc}".lower()
    for token, types, needles in ERROR_CLASSES:
        if names.intersection(types) or any(needle in text for needle in needles):
            return token
    return "UNKNOWN"

hpo_scoring_basis

hpo_scoring_basis(n_obs, cfg, params)

Whether the search behind this cell's params reserved the newest fold (pure).

None when no search ran, which is most cells. Otherwise "holdout" when an inner fold existed for the search to score on, or "in_sample" when it did not and the search fell back to every fold — a distinction that decides whether the cell's own metrics are honest, since the search and the leaderboard would otherwise be reading the same window.

The two granularities are asked different questions. Per-series tuning searched this series, so the split exists only if this series is long enough for two folds. Fleetwide tuning searched a sample the cell never saw, so the only thing the cell can honestly report is the run's fold geometry; a short series inside that sample fell back on its own, and it is the run header's scoring block, not this column, that would show it.

Source code in src/scale_forecasting/worker.py

def hpo_scoring_basis(n_obs: int, cfg: RunConfig, params: dict[str, Any] | None) -> str | None:
    """Whether the search behind this cell's params reserved the newest fold (pure).

    ``None`` when no search ran, which is most cells. Otherwise ``"holdout"`` when an inner fold
    existed for the search to score on, or ``"in_sample"`` when it did not and the search fell
    back to every fold — a distinction that decides whether the cell's own metrics are honest,
    since the search and the leaderboard would otherwise be reading the same window.

    The two granularities are asked different questions. Per-series tuning searched *this* series,
    so the split exists only if this series is long enough for two folds. Fleetwide tuning searched
    a sample the cell never saw, so the only thing the cell can honestly report is the run's fold
    geometry; a short series inside that sample fell back on its own, and it is the run header's
    ``scoring`` block, not this column, that would show it.
    """
    if not cfg.hpo.enabled:
        return None
    if params is not None:
        return "holdout" if cfg.backtest.n_folds >= 2 else "in_sample"
    if cfg.hpo.granularity != "per_series":
        return None
    return "holdout" if achievable_folds(n_obs, cfg) >= 2 else "in_sample"

is_panel_model

is_panel_model(model_name, cfg, params=None)

Whether model_name is configured for panel training (global or hybrid).

Source code in src/scale_forecasting/worker.py

def is_panel_model(model_name: str, cfg: RunConfig, params: dict[str, Any] | None = None) -> bool:
    """Whether ``model_name`` is configured for panel training (`global` or `hybrid`)."""
    base_name = base_model_name(model_name)
    authored: dict[str, Any] = {}
    if base_name in cfg.model_params:
        authored.update(cfg.model_params[base_name])
    if model_name in cfg.model_params:
        authored.update(cfg.model_params[model_name])
    if params is not None:
        authored.update(params)
    if "training_mode" in authored:
        return str(authored["training_mode"]) in ("global", "hybrid")
    if "@" in model_name:
        return resolve_model_variant(cfg, model_name).training_mode in ("global", "hybrid")
    return False

rollup_metrics

rollup_metrics(fold_metrics)

Average the per-fold metric panels into one panel.

NaNs are ignored per metric (a metric undefined on one fold shouldn't sink the mean); a metric NaN on every fold stays NaN. Always returns the full panel.

Source code in src/scale_forecasting/worker.py

def rollup_metrics(fold_metrics: list[dict[str, float]]) -> dict[str, float]:
    """Average the per-fold metric panels into one panel.

    NaNs are ignored per metric (a metric undefined on one fold shouldn't sink the mean);
    a metric NaN on every fold stays NaN. Always returns the full panel.
    """
    out: dict[str, float] = {}
    for name in METRIC_NAMES:
        vals = np.array([fm.get(name, np.nan) for fm in fold_metrics], dtype=float)
        finite = vals[~np.isnan(vals)]
        out[name] = float(finite.mean()) if finite.size else float("nan")
    return out

run_cell

run_cell(series, model_name, cfg, params=None)

Fit + (optional backtest) + predict ONE (ts_id, model) cell.

Pure-ish and deterministic: reads nothing global, writes nothing (no BQ), returns a CellResult carrying plain data. A failing cell returns status="error" and never raises, so one bad series can't sink a batch.

params are the hyperparameters this cell's model is built with. Resolution order:

• params given (not None) → use them. This is the fleetwide path: the driver tuned the model once on a sample (resolve_fleetwide) and threads the winning params here — never through cfg (the config is the run_id identity key).
• else if cfg.hpo.enabled and granularity == "per_series" → tune on this series now.
• else → {} (the default: today's untuned behavior).

The resolved params drive both the backtest folds and the final fit, so best_params = model.get_params() reflects what actually ran.

Source code in src/scale_forecasting/worker.py

def run_cell(
    series: pd.DataFrame,
    model_name: str,
    cfg: RunConfig,
    params: dict[str, Any] | None = None,
) -> CellResult:
    """Fit + (optional backtest) + predict ONE ``(ts_id, model)`` cell.

    Pure-ish and deterministic: reads nothing global, writes nothing (no BQ), returns a
    `CellResult` carrying plain data. A failing cell returns ``status="error"`` and
    never raises, so one bad series can't sink a batch.

    ``params`` are the hyperparameters this cell's model is built with. Resolution order:

    * ``params`` given (not None) → use them. This is the **fleetwide** path: the driver tuned the
      model once on a sample (`resolve_fleetwide`) and threads the
      winning params here — never through ``cfg`` (the config is the run_id identity key).
    * else if ``cfg.hpo.enabled`` and ``granularity == "per_series"`` → tune on *this* series now.
    * else → ``{}`` (the default: today's untuned behavior).

    The resolved params drive **both** the backtest folds and the final fit, so
    ``best_params = model.get_params()`` reflects what actually ran.
    """
    ts_id = _ts_id(series, cfg)
    run_id = make_run_id(cfg)
    model_hash = make_model_hash(run_id, ts_id, model_name, cfg)
    # Wall-clock lane + worker identity for the trace, captured for every return path (ok or error).
    cell_started_at = datetime.now(UTC)
    worker_id = _worker_id()
    # What this worker can see, memoized per process. Read up here so an error cell carries it too:
    # a cell that failed *because* the accelerator was missing is the row most worth the evidence.
    available, device_name = visible_device()
    # Two tallies, one question each: what the published forecast cost, and what a per-series
    # search cost on top of it. Created up here so the error path below can report them — a cell
    # that died after forty fits burned forty fits, and that is the row worth finding.
    fits, hpo_fits = FitTally(), FitTally()

    def _error(exc: BaseException, engine: str) -> CellResult:
        return CellResult(
            run_id=run_id,
            ts_id=ts_id,
            model_type=model_name,
            compute_engine=engine,
            model_hash=model_hash,
            status="error",
            error=repr(exc),
            # Classified here rather than by a reader after the fact, because this is the only
            # place the exception object still exists: `error` is a string by the time it leaves
            # the worker, and the type is the strongest signal the table has.
            error_class=classify_error(exc),
            predictions=_empty_predictions(),
            oof=None,
            metrics={name: float("nan") for name in METRIC_NAMES},
            worker_id=worker_id,
            cell_started_at=cell_started_at,
            cell_ended_at=datetime.now(UTC),
            device_available=available,
            device_name=device_name,
            n_fits=fits.n_fits,
            train_rows_total=fits.train_rows,
            n_hpo_fits=hpo_fits.n_fits,
        )

    try:
        model_cls = get_model(model_name)
    except Exception as e:  # noqa: BLE001 - unknown model name → error cell, engine unknown
        return _error(e, cfg.python_runtime)

    engine = _compute_engine(model_cls, cfg)
    # Harvest: record what this fit costs so a later run can be sized from it. Three cheap probes
    # around work the run was doing anyway — no sample, no pre-pass. Neither RSS nor the CUDA
    # high-water is reset first: the absolute peak is the number that sizes a slot (see
    # `CellResult.process_rss_bytes`), and resetting would also perturb whatever else shares this
    # worker. `intraop_threads` is captured before the fit because that is when it is in force.
    measuring = cfg.compute.profile.records_measurements
    intraop_threads = _intraop_threads() if measuring else None
    cpu_started = time.process_time()
    started = time.perf_counter()
    try:
        series = regularize_series_grid(series, cfg)
        fit_model_name = model_name
        if cfg.features.regime_routing:
            compat = select_models_for_series(series, cfg)
            if compat and model_name not in compat:
                fit_model_name = compat[0]
                model_cls = get_model(fit_model_name)
        model_cfg = effective_config_for_model(cfg, model_cls, fit_model_name)
        raw_target_series = _target(series, model_cfg)
        demand_prof = compute_demand_profile(raw_target_series.to_numpy(dtype=float))
        # Fit the transform's stateful λ once per cell (None for none/log1p), on the cleaned target.
        # It lives on ctx so the backtest folds and the final fit share one λ — never refit at
        # predict (the whole point of carrying it on the cell).
        lam = fit_transform_lambda(raw_target_series, model_cfg.features.transform)
        static_covs = extract_static_covariates(series, model_cfg) or None
        ctx = _model_context(
            model_cfg,
            transform_lambda=lam,
            family=model_cls.family,
            static_covariates=static_covs,
        )
        _require_device(ctx.device, model_cls.family, engine)
        resolved = _resolve_params(series, fit_model_name, model_cfg, ctx, params, hpo_fits)

        # Optional backtest first (fresh model per fold) → OOF frame + rolled-up metrics.
        #
        # Its own try/except, and this is the point of the block. Backtesting *scores* a model; it
        # does not produce the forecast. A scoring failure that propagated turned the whole cell
        # into an error and threw away a forecast that had not even been attempted yet — which is
        # how short history became the largest error class in the registry. Whatever happens here,
        # the final fit below still runs, and the outcome is recorded rather than inferred from a
        # NULL metric (which is also what "backtesting was off" looks like).
        oof: pd.DataFrame | None = None
        metrics = {name: float("nan") for name in METRIC_NAMES}
        backtest_status: str | None = None
        n_folds_achieved: int | None = None
        backtest_note: str | None = None
        # How the model was carried between fold origins, and what the blind control arm cost.
        # NULL when backtesting is off or failed, because "refit per fold" is a claim about a loop
        # that never ran.
        backtest_refit: str | None = None
        staleness_gap: float | None = None
        # The geometry actually used and the window it covered — NULL on the same terms, since both
        # describe a fold loop rather than the config that asked for one.
        achieved_step: int | None = None
        achieved_min_train: int | None = None
        first_val_date: date | None = None
        last_val_date: date | None = None
        if model_cfg.backtest.enabled:
            try:
                oof, fold_metrics, bt = backtest_cell(
                    series,
                    lambda: model_cls(resolved, ctx),
                    model_cfg,
                    lam,
                    fits,
                    model_cls.lags_covariates_internally,
                )
                metrics = rollup_metrics(fold_metrics)
                n_folds_achieved = len(fold_metrics)
                backtest_status, backtest_note = _backtest_outcome(
                    n_folds_achieved, model_cfg, series
                )
                backtest_refit, staleness_gap = bt.refit_mode, bt.staleness_gap
                achieved_step, achieved_min_train = bt.achieved_step, bt.achieved_min_train
                first_val_date, last_val_date = bt.first_val_date, bt.last_val_date
                if n_folds_achieved == 0:
                    oof = None  # an empty frame would write zero rows and read as "not asked"
            except Exception as e:  # noqa: BLE001 - scoring is not the forecast; degrade, don't fail
                _log.warning(
                    "backtest failed for %s/%s (forecast unaffected): %r", ts_id, model_name, e
                )
                oof = None
                metrics = {name: float("nan") for name in METRIC_NAMES}
                backtest_status, n_folds_achieved, backtest_note = "failed", 0, repr(e)
                backtest_refit, staleness_gap = None, None
                achieved_step, achieved_min_train = None, None
                first_val_date, last_val_date = None, None

        # Final fit on the full history, then forecast the horizon.
        y, X = build_features(series, model_cfg, lam, model_cls.lags_covariates_internally)
        model = model_cls(resolved, ctx)
        model.fit(y, X)
        fits.record(len(y))
        # The design frame for the horizon, indexed by the *future* dates: holiday flags and
        # Fourier phase are recomputed there (exact — they are functions of the date), the
        # level-shift step is carried forward, lagged covariates are read off their own source
        # column, and only user-supplied exog falls back to a recency stand-in because it is
        # genuinely unknown.
        future_exog = build_future_features(y, X, model_cfg)
        eff_qs = cfg.output.effective_quantiles
        predictions = model.predict(model_cfg.data.horizon, future_exog, quantiles=eff_qs)
        explanations = _collect_explanations(
            model, model_cfg.data.horizon, future_exog, ts_id, model_name
        )

        # Recalibrate the forward forecast against held-out error.
        #
        # The model's own band comes from *in-sample* residuals, which are one-step-ahead by
        # construction and are systematically too small for any model that fits its training data
        # tightly. Measured consequence: fleet coverage of 0.601 against a nominal 0.8. The OOF
        # frame above is a genuine held-out sample that this run has already paid for, and it
        # carries `horizon_step`, so it can also say how the error *grows* — which in-sample
        # residuals structurally cannot. Both the band and the bias correction are re-derived from
        # it when it is available, per horizon step where there are enough residuals to support
        # one, and the provenance is recorded either way rather than left to be inferred from the
        # fold count. `cal is None` leaves the model's own band untouched.
        #
        # Which arm ships is a separate question from how the band was built, and under
        # `point_forecast="auto"` it is answered per series from this cell's own folds rather than
        # once for the whole fleet. The margin is always reported corrected-minus-raw whatever the
        # cell chose, so a fleet-wide average stays comparable across cells that chose differently.
        metric = cfg.backtest.decision_metric
        corrected = corrected_arm_for(metric)
        arm, arm_decision = resolve_arm(cfg.output.point_forecast, oof, metric, corrected)
        cal = calibrate_from_oof(oof, eff_qs) if oof is not None else None
        predictions, interval_calibration = apply_calibration(predictions, cal, arm)
        predictions, arm_override = apply_quantile_and_cost_rules(predictions, cfg.output)
        if arm_override is not None:
            arm = arm_override
            arm_decision = "configured"
        if has_output_constraints(cfg.output):
            bounds_src = (
                future_exog if future_exog is not None and not future_exog.empty else series
            )
            predictions = apply_output_constraints(predictions, cfg.output, bounds_df=bounds_src)
        # `corrected`, not `arm`: the margin has to mean the same thing on every row for a
        # fleet-wide GROUP BY to be worth reading, so the comparison is always
        # corrected-vs-raw under the metric's own functional. Passing the selected arm would
        # have made a `raw`-configured run with `decision_metric="rmse"` report a *median*
        # comparison under a column the rest of the run reads as the mean one.
        arm_comparison = compare_arms(oof, metric, corrected) if oof is not None else {}

        # Persist the fitted model as an artifact only when the run opts in (model-artifact
        # lineage). A serialize failure must not sink an otherwise-good forecast, so it degrades to
        # no artifact rather than turning the cell into an error.
        artifact_bytes: bytes | None = None
        if cfg.compute.persist_models:
            try:
                artifact_bytes = model.serialize()
            except Exception as e:  # noqa: BLE001 - persistence is best-effort, never fatal
                _log.warning("serialize failed for %s/%s: %r", ts_id, model_name, e)

        diag = _collect_diagnostics(model, ts_id, model_name)
        if should_stamp_variant_diagnostics(model_name, cfg):
            diag.update(variant_diagnostics_for(model_name, cfg, model_cls=model_cls))
        if cfg.features.regime_routing:
            diag["demand_profile"] = demand_prof
            if fit_model_name != model_name:
                diag["regime_routing"] = {
                    "demand_class": demand_prof["demand_class"],
                    "routed_to": fit_model_name,
                }

        fit_seconds = time.perf_counter() - started
        cpu_seconds = time.process_time() - cpu_started
        return CellResult(
            run_id=run_id,
            ts_id=ts_id,
            model_type=model_name,
            compute_engine=engine,
            model_hash=model_hash,
            status="ok",
            error=None,
            predictions=predictions,
            oof=oof,
            metrics=metrics,
            best_params=model.get_params(),
            fit_seconds=fit_seconds,
            worker_id=worker_id,
            cell_started_at=cell_started_at,
            cell_ended_at=datetime.now(UTC),
            artifact_bytes=artifact_bytes,
            cpu_seconds=cpu_seconds if measuring else None,
            process_rss_bytes=_process_rss_bytes() if measuring else None,
            # Unconditional, unlike its neighbours: this is the only record of whether a device
            # the run paid for was ever visible to a cell, and gating it on profiling left 8 of
            # 21 historical GPU jobs with no evidence either way. It is also the cheapest probe
            # here — `_peak_gpu_bytes` memoizes "no accelerator" per process, so a CPU-only
            # worker pays one failed import for its whole life and every later cell short-circuits.
            peak_gpu_bytes=_peak_gpu_bytes(),
            intraop_threads=intraop_threads,
            n_obs=len(series) if measuring else None,
            # Unconditional for the same reason: this is the only per-cell record of whether the
            # device a run paid for was ever visible, and of where the fit actually landed. Both
            # probes are memoized or trivial, and `device_used` is asked of the fitted model.
            device_requested=ctx.device,
            device_available=available,
            device_used=model.device_used(),
            device_name=device_name,
            backtest_status=backtest_status,
            n_folds_achieved=n_folds_achieved,
            backtest_note=backtest_note,
            backtest_refit=backtest_refit,
            staleness_gap=staleness_gap,
            achieved_step=achieved_step,
            achieved_min_train=achieved_min_train,
            first_val_date=first_val_date,
            last_val_date=last_val_date,
            # A class attribute, so this is the model's own declaration rather than an inference
            # from the frame — a residual band on a model with no recorded residuals collapses to
            # bounds equal to `yhat`, which is indistinguishable from a native zero-width interval
            # by inspection and very distinguishable by what it means.
            interval_source="native" if model_cls.supports_native_intervals else "residual",
            # Which functional `yhat` carries, and where the band came from. Two separate facts:
            # a run can select the raw arm and still ship an OOF-calibrated interval.
            point_forecast_source=arm,
            point_forecast_decision=arm_decision,
            interval_calibration=interval_calibration,
            # The arm comparison, scored on the folds. Held-out by construction — each fold's
            # correction came from that fold's own training window and never saw the slice it is
            # measured against. This is the number that says whether the correction earned its
            # keep *for this series*, rather than asking anyone to trust a fleetwide average.
            point_forecast_margin=arm_comparison.get("margin"),
            hpo_scoring=hpo_scoring_basis(len(series), cfg, params),
            diagnostics=diag,
            demand_profile=demand_prof,
            explanations=explanations,
            n_fits=fits.n_fits,
            train_rows_total=fits.train_rows,
            n_hpo_fits=hpo_fits.n_fits,
        )
    except Exception as e:  # noqa: BLE001 - any failure → error cell, batch survives
        return _error(e, engine)

run_panel

run_panel(panel, model_name, cfg, params=None)

Fit + (optional panel backtest) + predict ONE global/hybrid model across a panel.

Returns one CellResult per distinct ts_id in panel, preserving the exact per-series schema and lineage contract (forecasts, backtest_oof, forecast_metadata) as run_cell. Never raises out of run_panel — an unhandled failure produces status="error" cells.

Source code in src/scale_forecasting/worker.py

def run_panel(
    panel: pd.DataFrame,
    model_name: str,
    cfg: RunConfig,
    params: dict[str, Any] | None = None,
) -> list[CellResult]:
    """Fit + (optional panel backtest) + predict ONE global/hybrid model across a panel.

    Returns one `CellResult` per distinct ``ts_id`` in ``panel``, preserving the exact per-series
    schema and lineage contract (`forecasts`, `backtest_oof`, `forecast_metadata`) as `run_cell`.
    Never raises out of `run_panel` — an unhandled failure produces `status="error"` cells.
    """
    from .backtest import OOF_COLUMNS, make_folds
    from .metrics import compute_metrics
    from .seasonality import seasonal_period

    id_col = cfg.data.ts_id_col
    date_col = cfg.data.date_col
    target_col = cfg.data.target_col
    run_id = make_run_id(cfg)
    cell_started_at = datetime.now(UTC)
    worker_id = _worker_id()
    available, device_name = visible_device()
    series_frames: dict[str, pd.DataFrame] = {}
    for raw_uid, sub in panel.groupby(id_col, sort=True):
        reg_sub = regularize_series_grid(sub, cfg)
        series_frames[str(raw_uid)] = reg_sub.sort_values(date_col).reset_index(drop=True)
    uids = sorted(series_frames.keys())
    if not uids:
        return []

    def _panel_errors(exc: BaseException, engine: str, n_fits: int = 0) -> list[CellResult]:
        ended = datetime.now(UTC)
        return [
            CellResult(
                run_id=run_id,
                ts_id=uid,
                model_type=model_name,
                compute_engine=engine,
                model_hash=make_model_hash(run_id, uid, model_name, cfg),
                status="error",
                error=repr(exc),
                error_class=classify_error(exc),
                predictions=_empty_predictions(),
                oof=None,
                metrics={name: float("nan") for name in METRIC_NAMES},
                worker_id=worker_id,
                cell_started_at=cell_started_at,
                cell_ended_at=ended,
                device_available=available,
                device_name=device_name,
                n_fits=n_fits,
                train_rows_total=0,
                n_hpo_fits=0,
            )
            for uid in uids
        ]

    try:
        model_cls = get_model(model_name)
    except Exception as e:  # noqa: BLE001
        return _panel_errors(e, cfg.python_runtime)

    engine = _compute_engine(model_cls, cfg)
    measuring = cfg.compute.profile.records_measurements
    intraop_threads = _intraop_threads() if measuring else None
    cpu_started = time.process_time()
    started = time.perf_counter()
    n_fits = 0
    train_rows_total = 0

    try:
        model_cfg = effective_config_for_model(cfg, model_cls, model_name)
        ctx = _model_context(model_cfg, family=model_cls.family)
        _require_device(ctx.device, model_cls.family, engine)
        variant_spec = resolve_model_variant(cfg, model_name, model_cls=model_cls)
        authored: dict[str, Any] = dict(variant_spec.clean_params)
        clean_override = clean_model_params(params) if params is not None else {}
        resolved = {**authored, **clean_override} if params is not None else authored

        lams: dict[str, float | None] = {
            uid: fit_transform_lambda(
                _target(series_frames[uid], model_cfg), model_cfg.features.transform
            )
            for uid in uids
        }
        static_map: dict[str, dict[str, Any]] = {}
        for uid in uids:
            sc = extract_static_covariates(series_frames[uid], model_cfg)
            if sc:
                static_map[uid] = sc

        # Optional panel-wide backtest across folds.
        oof_by_uid: dict[str, pd.DataFrame | None] = dict.fromkeys(uids, None)
        metrics_by_uid: dict[str, dict[str, float]] = {
            uid: {name: float("nan") for name in METRIC_NAMES} for uid in uids
        }
        bt_status_by_uid: dict[str, str | None] = dict.fromkeys(uids, None)
        bt_note_by_uid: dict[str, str | None] = dict.fromkeys(uids, None)
        n_folds_by_uid: dict[str, int | None] = dict.fromkeys(uids, None)
        bt_refit_by_uid: dict[str, str | None] = dict.fromkeys(uids, None)
        ach_step_by_uid: dict[str, int | None] = dict.fromkeys(uids, None)
        ach_min_by_uid: dict[str, int | None] = dict.fromkeys(uids, None)
        first_val_by_uid: dict[str, date | None] = dict.fromkeys(uids, None)
        last_val_by_uid: dict[str, date | None] = dict.fromkeys(uids, None)

        if model_cfg.backtest.enabled:
            try:
                folds_by_uid = {uid: make_folds(len(series_frames[uid]), model_cfg) for uid in uids}
                max_folds = max((len(fl) for fl in folds_by_uid.values()), default=0)
                oof_chunks_by_uid: dict[str, list[pd.DataFrame]] = {u: [] for u in uids}
                fmetrics_by_uid: dict[str, list[dict[str, float]]] = {u: [] for u in uids}
                m_period = seasonal_period(model_cfg.data.freq)
                gap = model_cfg.backtest.gap
                bt_h = model_cfg.backtest.horizon

                for fold_idx in range(max_folds):
                    active_uids = [u for u in uids if fold_idx < len(folds_by_uid[u])]
                    if not active_uids:
                        continue
                    fold_series_map: dict[str, tuple[pd.Series, pd.DataFrame | None]] = {}
                    fold_futr_map: dict[str, pd.DataFrame | None] = {}
                    fold_lams: dict[str, float | None] = {}
                    for uid in active_uids:
                        fold = folds_by_uid[uid][fold_idx]
                        sub = series_frames[uid]
                        train_slice = sub.iloc[: fold.train_end]
                        val_future = sub.iloc[fold.train_end : fold.val_end]
                        if model_cfg.features.past_covariates:
                            val_future = val_future.drop(
                                columns=[
                                    c
                                    for c in model_cfg.features.past_covariates
                                    if c in val_future.columns
                                ]
                            )
                        y_tr, X_tr = build_features(
                            train_slice, model_cfg, lams[uid], model_cls.lags_covariates_internally
                        )
                        fold_series_map[uid] = (y_tr, X_tr)
                        fold_futr_map[uid] = build_future_features(
                            y_tr,
                            X_tr,
                            model_cfg,
                            horizon=gap + bt_h,
                            future_covariates_df=val_future,
                        )
                        fold_lams[uid] = lams[uid]
                        train_rows_total += len(y_tr)

                    fold_model = model_cls(resolved, ctx)
                    fold_model.fit_panel(
                        fold_series_map, static_map=static_map if static_map else None
                    )
                    n_fits += 1
                    preds_map = fold_model.predict_panel(
                        gap + bt_h,
                        fold_futr_map,
                        transform_lambdas=fold_lams,
                    )
                    for uid in active_uids:
                        fold = folds_by_uid[uid][fold_idx]
                        sub = series_frames[uid]
                        train_slice = sub.iloc[: fold.train_end]
                        val_slice = sub.iloc[fold.val_start : fold.val_end]
                        pred = preds_map[uid].iloc[gap:].reset_index(drop=True)
                        y_train_raw = train_slice[target_col].to_numpy(dtype=float)
                        y_val = val_slice[target_col].to_numpy(dtype=float)
                        yhat_raw = pred["yhat_raw"].to_numpy(dtype=float)
                        yhat_adjusted = pred["yhat"].to_numpy(dtype=float)
                        yhat = (
                            yhat_raw if model_cfg.output.point_forecast == "raw" else yhat_adjusted
                        )
                        lower = pred["yhat_lower"].to_numpy(dtype=float)
                        upper = pred["yhat_upper"].to_numpy(dtype=float)
                        cutoff = pd.Timestamp(train_slice[date_col].iloc[-1]).normalize()
                        fold_oof = pd.DataFrame(
                            {
                                "ds": pd.to_datetime(val_slice[date_col]).to_numpy(),
                                "fold_id": fold.fold_id,
                                "y_true": y_val,
                                "yhat": yhat,
                                "yhat_raw": yhat_raw,
                                "yhat_adjusted": yhat_adjusted,
                                "yhat_lower": lower,
                                "yhat_upper": upper,
                                "cutoff_date": cutoff,
                                "horizon_step": np.arange(1, len(y_val) + 1),
                                "yhat_stale": np.full(len(y_val), np.nan),
                            },
                            columns=list(OOF_COLUMNS),
                        )
                        if has_output_constraints(model_cfg.output):
                            fold_oof = apply_output_constraints(
                                fold_oof, model_cfg.output, bounds_df=val_slice
                            )
                            yhat = fold_oof["yhat"].to_numpy(dtype=float)
                            lower = fold_oof["yhat_lower"].to_numpy(dtype=float)
                            upper = fold_oof["yhat_upper"].to_numpy(dtype=float)
                        fmetrics_by_uid[uid].append(
                            compute_metrics(
                                y_val,
                                yhat,
                                y_train=y_train_raw,
                                lower=lower,
                                upper=upper,
                                seasonal_period=m_period,
                            )
                        )
                        oof_chunks_by_uid[uid].append(fold_oof)

                for uid in uids:
                    sub = series_frames[uid]
                    flist = folds_by_uid[uid]
                    n_ach = len(fmetrics_by_uid[uid])
                    n_folds_by_uid[uid] = n_ach
                    metrics_by_uid[uid] = rollup_metrics(fmetrics_by_uid[uid])
                    bt_status_by_uid[uid], bt_note_by_uid[uid] = _backtest_outcome(
                        n_ach, model_cfg, sub
                    )
                    if n_ach > 0:
                        oof_by_uid[uid] = pd.concat(oof_chunks_by_uid[uid], ignore_index=True)
                        bt_refit_by_uid[uid] = "per_fold"
                        geom = resolve_geometry(len(sub), model_cfg)
                        ach_step_by_uid[uid] = geom.step
                        ach_min_by_uid[uid] = geom.min_train
                        first_val_by_uid[uid] = pd.Timestamp(
                            sub[date_col].iloc[flist[0].val_start]
                        ).date()
                        last_val_by_uid[uid] = pd.Timestamp(
                            sub[date_col].iloc[flist[-1].val_end - 1]
                        ).date()
            except Exception as e:  # noqa: BLE001
                _log.warning("panel backtest failed for %s: %r", model_name, e)
                for uid in uids:
                    bt_status_by_uid[uid] = "failed"
                    n_folds_by_uid[uid] = 0
                    bt_note_by_uid[uid] = repr(e)

        # Final fit across the full panel.
        full_series_map: dict[str, tuple[pd.Series, pd.DataFrame | None]] = {}
        full_futr_map: dict[str, pd.DataFrame | None] = {}
        for uid in uids:
            sub = series_frames[uid]
            y_full, X_full = build_features(
                sub, model_cfg, lams[uid], model_cls.lags_covariates_internally
            )
            full_series_map[uid] = (y_full, X_full)
            full_futr_map[uid] = build_future_features(y_full, X_full, model_cfg)
            train_rows_total += len(y_full)

        model = model_cls(resolved, ctx)
        model.fit_panel(full_series_map, static_map=static_map if static_map else None)
        n_fits += 1
        eff_qs = cfg.output.effective_quantiles
        raw_preds_map = model.predict_panel(
            model_cfg.data.horizon,
            full_futr_map,
            quantiles=eff_qs,
            transform_lambdas=lams,
        )

        total_fit_seconds = time.perf_counter() - started
        total_cpu_seconds = time.process_time() - cpu_started
        per_series_fit_s = total_fit_seconds / max(1, len(uids))
        per_series_cpu_s = total_cpu_seconds / max(1, len(uids))
        rss_bytes = _process_rss_bytes() if measuring else None
        peak_gpu = _peak_gpu_bytes()
        dev_used = model.device_used()
        best_params = model.get_params()

        metric = cfg.backtest.decision_metric
        corrected = corrected_arm_for(metric)
        ended_at = datetime.now(UTC)

        results: list[CellResult] = []
        for idx, uid in enumerate(uids):
            oof = oof_by_uid[uid]
            arm, arm_decision = resolve_arm(cfg.output.point_forecast, oof, metric, corrected)
            cal = calibrate_from_oof(oof, eff_qs) if oof is not None else None
            predictions, interval_calibration = apply_calibration(raw_preds_map[uid], cal, arm)
            predictions, arm_override = apply_quantile_and_cost_rules(predictions, cfg.output)
            if arm_override is not None:
                arm = arm_override
                arm_decision = "configured"
            if has_output_constraints(cfg.output):
                fut_df = full_futr_map.get(uid)
                bounds_src = (
                    fut_df if fut_df is not None and not fut_df.empty else series_frames[uid]
                )
                predictions = apply_output_constraints(
                    predictions, cfg.output, bounds_df=bounds_src
                )
            arm_comparison = compare_arms(oof, metric, corrected) if oof is not None else {}

            artifact_bytes: bytes | None = None
            if cfg.compute.persist_models and idx == 0:
                try:
                    artifact_bytes = model.serialize()
                except Exception as e:  # noqa: BLE001
                    _log.warning("serialize failed for %s/%s: %r", uid, model_name, e)

            uid_demand_prof = compute_demand_profile(
                _target(series_frames[uid], model_cfg).to_numpy(dtype=float)
            )
            uid_diag = _collect_diagnostics(model, uid, model_name)
            if should_stamp_variant_diagnostics(model_name, cfg):
                uid_diag.update(variant_diagnostics_for(model_name, cfg, model_cls=model_cls))
            if cfg.features.regime_routing:
                uid_diag["demand_profile"] = uid_demand_prof

            results.append(
                CellResult(
                    run_id=run_id,
                    ts_id=uid,
                    model_type=model_name,
                    compute_engine=engine,
                    model_hash=make_model_hash(run_id, uid, model_name, cfg),
                    status="ok",
                    error=None,
                    predictions=predictions,
                    oof=oof,
                    metrics=metrics_by_uid[uid],
                    best_params=best_params,
                    fit_seconds=per_series_fit_s,
                    worker_id=worker_id,
                    cell_started_at=cell_started_at,
                    cell_ended_at=ended_at,
                    artifact_bytes=artifact_bytes,
                    cpu_seconds=per_series_cpu_s if measuring else None,
                    process_rss_bytes=rss_bytes,
                    peak_gpu_bytes=peak_gpu,
                    intraop_threads=intraop_threads,
                    n_obs=len(series_frames[uid]) if measuring else None,
                    device_requested=ctx.device,
                    device_available=available,
                    device_used=dev_used,
                    device_name=device_name,
                    backtest_status=bt_status_by_uid[uid],
                    n_folds_achieved=n_folds_by_uid[uid],
                    backtest_note=bt_note_by_uid[uid],
                    backtest_refit=bt_refit_by_uid[uid],
                    staleness_gap=None,
                    achieved_step=ach_step_by_uid[uid],
                    achieved_min_train=ach_min_by_uid[uid],
                    first_val_date=first_val_by_uid[uid],
                    last_val_date=last_val_by_uid[uid],
                    interval_source="native" if model_cls.supports_native_intervals else "residual",
                    point_forecast_source=arm,
                    point_forecast_decision=arm_decision,
                    interval_calibration=interval_calibration,
                    point_forecast_margin=arm_comparison.get("margin"),
                    hpo_scoring=hpo_scoring_basis(len(series_frames[uid]), cfg, params),
                    diagnostics=uid_diag,
                    demand_profile=uid_demand_prof,
                    n_fits=n_fits if idx == 0 else 0,
                    train_rows_total=train_rows_total if idx == 0 else 0,
                    n_hpo_fits=0,
                )
            )
        return results
    except Exception as e:  # noqa: BLE001
        return _panel_errors(e, engine, n_fits=n_fits)