Ensembler (Blending Math)

ensembler

Consensus across base models — calculated + learned — pure logic, all in pandas.

The outer loop combines every base model's forecast per ts_id into a consensus. Two families, both pure here (no GCP calls):

• Calculated (mean, median, inverse_error, champion, horizon_specialist) — heuristics and per-series selectors that need no cross-series meta-learner training. combine_calculated blends the base forecast_predictions in pandas and returns ensemble_<s> prediction rows the caller appends via the Storage Write API — the same append path the learned strategies use (every append-only cell-table write goes through the Write API, no INSERT…SELECT DML).
• Learned (nnls, ridge, xgb) — meta-learners that train on the backtest OOF to learn per-model trust weights. fit_learned fits them; because it only ever sees backtest_oof (never in-sample fits) leakage is structurally impossible, and it refuses to run when backtesting is off.

A run may request several strategies at once (ensemble.strategies is a list); each yields its own weights/rows so calculated and learned consensuses sit side-by-side in one run.

Public surface: combine_calculated, combine_oof, fit_learned, calibrate_ensemble_predictions, ensemble_oof_bounds (plus the pure combine helpers mean_combine, median_combine, inverse_error_weights).

calibrate_ensemble_predictions

calibrate_ensemble_predictions(pred_rows, ens_oof, cfg)

Calibrate ensemble future prediction intervals from ens_oof & apply output constraints.

When cfg.output.calibrate_intervals is True and ens_oof has >= 5 residual rows for a (ts_id, model_type) cell, replaces the linearly blended bounds with empirical 10th/90th residual quantiles from calibration.calibrate_from_oof. Always enforces physical output constraints (calibration.apply_output_constraints) when configured.

Source code in src/scale_forecasting/ensembler.py

def calibrate_ensemble_predictions(
    pred_rows: list[dict[str, Any]],
    ens_oof: pd.DataFrame,
    cfg: RunConfig,
) -> list[dict[str, Any]]:
    """Calibrate ensemble future prediction intervals from ``ens_oof`` & apply output constraints.

    When ``cfg.output.calibrate_intervals`` is True and ``ens_oof`` has >= 5 residual rows for a
    ``(ts_id, model_type)`` cell, replaces the linearly blended bounds with empirical 10th/90th
    residual quantiles from `calibration.calibrate_from_oof`. Always enforces physical output
    constraints (`calibration.apply_output_constraints`) when configured.
    """

    if not pred_rows:
        return pred_rows
    need_constraints = has_output_constraints(cfg.output)
    need_q_or_cost = (
        cfg.output.quantiles is not None or cfg.output.effective_target_quantile is not None
    )
    need_cal = not ens_oof.empty
    if not need_constraints and not need_cal and not need_q_or_cost:
        return pred_rows

    df = pd.DataFrame(pred_rows)
    had_quantiles_col = "quantiles" in df.columns
    target_qs = cfg.output.effective_quantiles
    lo_q, hi_q = target_qs[0], target_qs[-1]
    oof_by_cell: dict[tuple[str, str], pd.DataFrame] = {}
    if need_cal:
        for (tid, mtype), g in ens_oof.groupby(["ts_id", "model_type"]):
            oof_by_cell[(str(tid), str(mtype))] = g

    emit_quantiles = cfg.output.quantiles is not None or had_quantiles_col
    out_parts: list[pd.DataFrame] = []
    for (tid, mtype), group in df.groupby(["ts_id", "model_type"], sort=False):
        cell_df = group.copy()
        if (need_constraints or need_q_or_cost) and "yhat_raw" not in cell_df.columns:
            cell_df["yhat_raw"] = cell_df["yhat"].to_numpy(dtype=float)
        calibrated_this_cell = False
        if need_cal:
            oof_slice = oof_by_cell.get((str(tid), str(mtype)))
            if oof_slice is not None and not oof_slice.empty:
                cal = calibrate_from_oof(oof_slice, target_qs, residual_column="yhat")
                if cal is not None:
                    calibrated_this_cell = True
                    yhat_vals = cell_df["yhat"].to_numpy(dtype=float)
                    qlo_arr = np.array(
                        [cal.for_step(s).quantiles[lo_q] for s in range(1, len(cell_df) + 1)],
                        dtype=float,
                    )
                    qhi_arr = np.array(
                        [cal.for_step(s).quantiles[hi_q] for s in range(1, len(cell_df) + 1)],
                        dtype=float,
                    )
                    cell_df["yhat_lower"] = yhat_vals + qlo_arr
                    cell_df["yhat_upper"] = yhat_vals + qhi_arr
                    if emit_quantiles:
                        qjson_rows: list[str] = []
                        for idx in range(len(cell_df)):
                            step_q = cal.for_step(idx + 1).quantiles
                            raw_qmap = {
                                q: float(yhat_vals[idx] + step_q[q])
                                for q in target_qs
                                if q in step_q
                            }
                            mono_qmap = rearrange_monotone_quantiles(raw_qmap)
                            qjson_rows.append(
                                json.dumps(
                                    {str(q): v for q, v in mono_qmap.items()}, sort_keys=True
                                )
                            )
                        cell_df["quantiles"] = qjson_rows
                    if cfg.output.effective_target_quantile is not None:
                        q_star = cfg.output.effective_target_quantile
                        qstar_arr = np.array(
                            [cal.for_step(s).quantiles[q_star] for s in range(1, len(cell_df) + 1)],
                            dtype=float,
                        )
                        cell_df["yhat"] = yhat_vals + qstar_arr
                else:
                    resid = (oof_slice["y_true"] - oof_slice["yhat"]).dropna().to_numpy(dtype=float)
                    resid = resid[np.isfinite(resid)]
                    if len(resid) >= 2:
                        calibrated_this_cell = True
                        qlo, qhi = np.percentile(resid, [lo_q * 100.0, hi_q * 100.0])
                        yhat_vals = cell_df["yhat"].to_numpy(dtype=float)
                        cell_df["yhat_lower"] = yhat_vals + float(qlo)
                        cell_df["yhat_upper"] = yhat_vals + float(qhi)
                        if emit_quantiles:
                            q_pcts = np.percentile(resid, [q * 100.0 for q in target_qs])
                            qjson_rows = []
                            for idx in range(len(cell_df)):
                                raw_qmap = {
                                    q: float(yhat_vals[idx] + q_pcts[j])
                                    for j, q in enumerate(target_qs)
                                }
                                mono_qmap = rearrange_monotone_quantiles(raw_qmap)
                                qjson_rows.append(
                                    json.dumps(
                                        {str(q): v for q, v in mono_qmap.items()}, sort_keys=True
                                    )
                                )
                            cell_df["quantiles"] = qjson_rows
                        if cfg.output.effective_target_quantile is not None:
                            qstar = float(
                                np.percentile(resid, cfg.output.effective_target_quantile * 100.0)
                            )
                            cell_df["yhat"] = yhat_vals + qstar
        if not calibrated_this_cell and need_q_or_cost:
            cell_df, _ = apply_quantile_and_cost_rules(cell_df, cfg.output)
            if not emit_quantiles and "quantiles" in cell_df.columns:
                cell_df = cell_df.drop(columns=["quantiles"])
        if need_constraints:
            cell_df = apply_output_constraints(cell_df, cfg.output)
        out_parts.append(cell_df)

    merged = pd.concat(out_parts, ignore_index=True)
    # Restore None for any NaN bounds so BigQuery Write API receives NULL rather than NaN.
    result: list[dict[str, Any]] = []
    for raw_rec in merged.to_dict(orient="records"):
        rec: dict[str, Any] = {str(k): v for k, v in raw_rec.items()}
        for bcol in ("yhat_lower", "yhat_upper"):
            val = rec.get(bcol)
            if val is not None and isinstance(val, float) and np.isnan(val):
                rec[bcol] = None
        result.append(rec)
    return result

combine_calculated

combine_calculated(
    base_df, cfg, metric_df=None, oof_df=None
)

Blend base predictions into ensemble_<s> rows for every calculated strategy (pure).

Builds the ensemble_<s> rows in pandas rather than via SQL INSERT…SELECT: reads the base forecast_predictions (long-format ts_id, model_type, forecast_date, yhat, yhat_lower, yhat_upper) and returns prediction-row dicts the caller stamps with run_id/ensemble_id and appends via the Storage Write API. Supports mean, median, inverse_error, champion (per-series single winner), and horizon_specialist (per-series, per-horizon-step inverse-error blend when oof_df is supplied, falling back to series-level inverse_error otherwise), as well as top_k and parsimony_margin pruning.

Source code in src/scale_forecasting/ensembler.py

def combine_calculated(
    base_df: pd.DataFrame,
    cfg: RunConfig,
    metric_df: pd.DataFrame | None = None,
    oof_df: pd.DataFrame | None = None,
) -> list[dict[str, Any]]:
    """Blend base predictions into ``ensemble_<s>`` rows for every calculated strategy (pure).

    Builds the ``ensemble_<s>`` rows in pandas rather than via SQL ``INSERT…SELECT``: reads the base
    ``forecast_predictions`` (long-format ``ts_id, model_type, forecast_date, yhat, yhat_lower,
    yhat_upper``) and returns prediction-row dicts the caller stamps with ``run_id``/``ensemble_id``
    and appends via the Storage Write API. Supports ``mean``, ``median``, ``inverse_error``,
    ``champion`` (per-series single winner), and ``horizon_specialist`` (per-series,
    per-horizon-step inverse-error blend when ``oof_df`` is supplied, falling back to series-level
    ``inverse_error`` otherwise), as well as ``top_k`` and ``parsimony_margin`` pruning.
    """
    calculated = [s for s in cfg.ensemble.strategies if s in CALCULATED_STRATEGIES]
    if not calculated or base_df.empty:
        return []
    models = _pruned_models(cfg, metric_df, oof_df=oof_df)
    present = [m for m in models if m in set(base_df["model_type"].unique())]
    if not present:
        return []

    # One wide frame per value column, aligned to the present base models, reused across strategies.
    keys = ["ts_id", "forecast_date"]
    wide = {
        col: base_df.pivot_table(index=keys, columns="model_type", values=col).reindex(
            columns=present
        )
        for col in ("yhat", "yhat_lower", "yhat_upper")
    }
    index = wide["yhat"].index
    ts_ids_arr = np.asarray(index.get_level_values("ts_id"))
    has_selection_filter = cfg.ensemble.top_k is not None or cfg.ensemble.parsimony_margin > 0.0

    ie_weights: np.ndarray | None = None
    if "inverse_error" in calculated or (
        has_selection_filter and ("mean" in calculated or "median" in calculated)
    ):
        ie_weights = _inverse_error_weight_matrix(
            present, cfg, metric_df, ts_ids_arr, champion_only=False
        )

    champ_weights: np.ndarray | None = None
    if "champion" in calculated:
        champ_weights = _inverse_error_weight_matrix(
            present, cfg, metric_df, ts_ids_arr, champion_only=True
        )

    hs_weights: np.ndarray | None = None
    if "horizon_specialist" in calculated:
        hs_weights = _horizon_specialist_weight_matrix(present, cfg, metric_df, oof_df, index)

    rows: list[dict[str, Any]] = []
    for strategy in calculated:
        weights: np.ndarray | None = None
        if strategy == "inverse_error":
            weights = ie_weights
        elif strategy == "champion":
            weights = champ_weights
        elif strategy == "horizon_specialist":
            weights = hs_weights
        elif has_selection_filter and ie_weights is not None:
            weights = np.where(ie_weights > 0.0, 1.0, 0.0)
        blended = {
            col: _calc_blend(strategy, wide[col].to_numpy(dtype=float), weights)
            for col in ("yhat", "yhat_lower", "yhat_upper")
        }
        model_type = f"ensemble_{strategy}"
        for i, (ts_id, forecast_date) in enumerate(index):
            yh = blended["yhat"][i]
            if np.isnan(yh):
                continue
            lo, up = blended["yhat_lower"][i], blended["yhat_upper"][i]
            rows.append(
                {
                    "ts_id": ts_id,
                    "model_type": model_type,
                    "forecast_date": forecast_date,
                    "yhat": float(yh),
                    "yhat_lower": None if np.isnan(lo) else float(lo),
                    "yhat_upper": None if np.isnan(up) else float(up),
                }
            )
    return rows

combine_oof

combine_oof(oof_df, cfg, learned_weights=None)

Blend base-model OOF forecasts into ensemble OOF, per requested strategy (pure).

The scoring counterpart to the future-prediction consensus: because the base forecast_predictions are a true beyond-data forecast (no actuals to join), each ensemble is scored on the backtest OOF window — exactly the window the base models are scored on. This applies the same consensus rules in OOF space (where y_true lives) and returns long-format (ts_id, model_type='ensemble_<s>', fold_id, cutoff_date, forecast_date, y_true, yhat) for the caller to score with metrics.compute_metrics. The cutoff is carried out as well as joined on, because the caller cuts each series' history there to get the fold's MASE denominator.

Blends over whichever base models are present per (ts_id, cutoff_date, forecast_date) key (see _fold_key for why the cutoff and not the fold ordinal): mean/median are unweighted; inverse_error weights each model per ts_id by 1/WAPE over that model's OOF (self-contained — the same signal forecast_metadata would carry); learned strategies apply learned_weights[strategy] (skipped if absent). Returns an empty frame when the OOF is empty or no strategy produces a blend.

Bounds are not blended, so an ensemble's coverage/pinball/interval_score are NaN. The base models' are not, any more — the OOF frame now carries yhat_lower and yhat_upper per fold, and the base cells score them. Blending them is a separate question with a real answer to choose (averaging two models' 80% intervals does not give an 80% interval), so the columns stay NaN rather than being filled with something plausible and wrong. An ensemble is compared to its base models on the point metrics.

Source code in src/scale_forecasting/ensembler.py

def combine_oof(
    oof_df: pd.DataFrame,
    cfg: RunConfig,
    learned_weights: dict[str, dict[str, float]] | None = None,
) -> pd.DataFrame:
    """Blend base-model OOF forecasts into ensemble OOF, per requested strategy (pure).

    The scoring counterpart to the future-prediction consensus: because the base
    ``forecast_predictions`` are a true beyond-data forecast (no actuals to join), each ensemble is
    scored on the **backtest OOF window** — exactly the window the base models are scored on. This
    applies the same consensus rules in OOF space (where ``y_true`` lives) and returns long-format
    ``(ts_id, model_type='ensemble_<s>', fold_id, cutoff_date, forecast_date, y_true, yhat)`` for
    the caller to score with `metrics.compute_metrics`. The cutoff is carried out as well as joined
    on, because the caller cuts each series' history there to get the fold's MASE denominator.

    Blends over whichever base models are present per ``(ts_id, cutoff_date, forecast_date)`` key
    (see `_fold_key` for why the cutoff and not the fold ordinal):
    ``mean``/``median`` are unweighted; ``inverse_error`` weights each model per ``ts_id`` by
    ``1/WAPE`` over that model's OOF (self-contained — the same signal ``forecast_metadata`` would
    carry); learned strategies apply ``learned_weights[strategy]`` (skipped if absent). Returns an
    empty frame when the OOF is empty or no strategy produces a blend.

    **Bounds are not blended, so an ensemble's ``coverage``/``pinball``/``interval_score`` are
    NaN.** The base models' are not, any more — the OOF frame now carries ``yhat_lower`` and
    ``yhat_upper`` per fold, and the base cells score them. Blending them is a separate question
    with a real answer to choose (averaging two models' 80% intervals does not give an 80%
    interval), so the columns stay NaN rather than being filled with something plausible and
    wrong. An ensemble is compared to its base models on the point metrics.
    """
    learned_weights = learned_weights or {}
    empty = pd.DataFrame(columns=list(_OOF_BLEND_COLS))
    if oof_df.empty:
        return empty
    models = filter_variants_for_ensemble(list(cfg.models), cfg, oof_df=oof_df)
    keys = _fold_key(oof_df)
    wide = oof_df.pivot_table(index=keys, columns="model_type", values="yhat", aggfunc="first")
    present_models = [m for m in models if m in wide.columns]
    if not present_models:
        return empty
    vals = wide.reindex(columns=present_models).to_numpy(dtype=float)
    aligned = oof_df.drop_duplicates(keys).set_index(keys).reindex(wide.index)
    truth = aligned["y_true"].to_numpy()
    # `fold_id` is still written out — it is the ordinal a reader recognises and the ensemble rows
    # have to line up with the base rows in the same table. It is carried, not joined on, so it is
    # read off whichever side of `aligned` the key put it on: the index when the fallback key is in
    # use, a column when the join went on the cutoff.
    fold_ids = _carried(aligned, keys, "fold_id", "float64")
    cutoffs = _carried(aligned, keys, "cutoff_date", "object")
    steps = _carried(aligned, keys, "horizon_step", "float64")
    ts_ids = wide.index.get_level_values("ts_id").to_numpy()
    # `inverse_error` is the one calculated strategy that *fits* something — a per-series weight
    # off these very rows — so it is held to the same rule as the meta-learners and only sees the
    # inner folds. Every strategy is then scored on all of them: the holdout is reserved from the
    # fit, not from the score. `mean` and `median` fit nothing, so the mask does not reach them.
    fit_rows = (fold_ids != holdout_fold_id(cfg)).to_numpy(dtype=bool)

    parts: list[pd.DataFrame] = []
    steps_arr = steps.to_numpy(dtype=float)
    for strategy in cfg.ensemble.strategies:
        yhat = _blend_oof_strategy(
            strategy,
            vals,
            present_models,
            ts_ids,
            steps_arr,
            truth,
            learned_weights,
            fit_rows,
            cfg,
        )
        if yhat is None:
            continue
        part = pd.DataFrame(
            {
                "ts_id": wide.index.get_level_values("ts_id"),
                "model_type": f"ensemble_{strategy}",
                "fold_id": fold_ids.to_numpy(),
                "cutoff_date": cutoffs.to_numpy(),
                "forecast_date": wide.index.get_level_values(keys[-1]),
                "horizon_step": steps.to_numpy(),
                "y_true": truth,
                "yhat": yhat,
            }
        )
        part = part[~part["yhat"].isna()].reset_index(drop=True)
        if has_output_constraints(cfg.output) and not part.empty:
            constrained_parts: list[pd.DataFrame] = []
            for (_tid, _fid), fg in part.groupby(["ts_id", "fold_id"], sort=False, dropna=False):
                fg_sorted = fg.sort_values("forecast_date").copy()
                fg_c = apply_output_constraints(fg_sorted[["yhat"]].copy(), cfg.output)
                fg_sorted["yhat"] = fg_c["yhat"].to_numpy(dtype=float)
                constrained_parts.append(fg_sorted)
            part = pd.concat(constrained_parts, ignore_index=True)
        parts.append(part)
    return pd.concat(parts, ignore_index=True)[list(_OOF_BLEND_COLS)] if parts else empty

ensemble_oof_bounds

ensemble_oof_bounds(ens_oof_group, fold_id, cfg)

Empirical 80% prediction intervals for one fold of an ensemble's OOF rows (pure).

Uses leave-one-fold-out OOF residuals (y_true - yhat from all other folds in ens_oof_group, falling back to all folds when only a single fold exists) so every ensemble_* pseudo-model earns non-NaN interval metrics (coverage_80, pinball_loss, winkler_score, interval_width, wis) on the same footing as base models. Applies calibration.apply_output_constraints when configured.

Source code in src/scale_forecasting/ensembler.py

def ensemble_oof_bounds(
    ens_oof_group: pd.DataFrame,
    fold_id: object,
    cfg: RunConfig,
) -> tuple[np.ndarray | None, np.ndarray | None]:
    """Empirical 80% prediction intervals for one fold of an ensemble's OOF rows (pure).

    Uses leave-one-fold-out OOF residuals (``y_true - yhat`` from all other folds in
    ``ens_oof_group``, falling back to all folds when only a single fold exists) so every
    ``ensemble_*`` pseudo-model earns non-NaN interval metrics (``coverage_80``, ``pinball_loss``,
    ``winkler_score``, ``interval_width``, ``wis``) on the same footing as base models. Applies
    `calibration.apply_output_constraints` when configured.
    """

    fg = ens_oof_group[ens_oof_group["fold_id"] == fold_id].sort_values("forecast_date")
    if fg.empty:
        return None, None
    other = ens_oof_group[ens_oof_group["fold_id"] != fold_id]
    ref = other if not other.empty else ens_oof_group
    residuals = (ref["y_true"] - ref["yhat"]).dropna().to_numpy(dtype=float)
    residuals = residuals[np.isfinite(residuals)]
    if len(residuals) == 0:
        return None, None
    if len(residuals) == 1:
        half = max(abs(float(residuals[0])), 1e-6)
        q10, q90 = -half, half
    else:
        q10, q90 = np.percentile(residuals, [10.0, 90.0])
        if q10 == q90:
            half = max(abs(float(q10)) * 0.1, 1e-6)
            q10, q90 = float(q10) - half, float(q90) + half

    yhat = fg["yhat"].to_numpy(dtype=float)
    frame = pd.DataFrame(
        {
            "yhat": yhat,
            "yhat_lower": yhat + float(q10),
            "yhat_upper": yhat + float(q90),
        }
    )
    if has_output_constraints(cfg.output):
        frame = apply_output_constraints(frame, cfg.output)
    return frame["yhat_lower"].to_numpy(dtype=float), frame["yhat_upper"].to_numpy(dtype=float)

filter_variants_for_ensemble

filter_variants_for_ensemble(
    models, cfg, *, oof_df=None, metric_df=None
)

Select the ensemble candidate pool across <base_model>@<tag> variants (pure).

Controlled by cfg.ensemble.variant_policy: - "all": keep all configured model variants. - "primary_only": keep at most one variant per base_model_name(m) — preferring the untagged base model if present, else the first variant of that base model in config order. - "best_per_base_model" (default): when multiple variants of the same base_model exist, keep the variant with the lowest inner-fold OOF loss (or mean decision_metric loss in metric_df when oof_df is absent), preventing collinear multi-variant dilution.

Source code in src/scale_forecasting/ensembler.py

def filter_variants_for_ensemble(
    models: list[str],
    cfg: RunConfig,
    *,
    oof_df: pd.DataFrame | None = None,
    metric_df: pd.DataFrame | None = None,
) -> list[str]:
    """Select the ensemble candidate pool across `<base_model>@<tag>` variants (pure).

    Controlled by ``cfg.ensemble.variant_policy``:
    - ``"all"``: keep all configured model variants.
    - ``"primary_only"``: keep at most one variant per ``base_model_name(m)`` — preferring the
      untagged base model if present, else the first variant of that base model in config order.
    - ``"best_per_base_model"`` (default): when multiple variants of the same ``base_model`` exist,
      keep the variant with the lowest inner-fold OOF loss (or mean ``decision_metric`` loss in
      ``metric_df`` when ``oof_df`` is absent), preventing collinear multi-variant dilution.
    """
    policy = cfg.ensemble.variant_policy
    if policy == "all" or len(models) <= 1:
        return list(models)

    groups: dict[str, list[str]] = {}
    for m in models:
        groups.setdefault(base_model_name(m), []).append(m)
    if all(len(g) == 1 for g in groups.values()):
        return list(models)

    if policy == "primary_only":
        chosen_primary = {(base if base in g else g[0]) for base, g in groups.items()}
        return [m for m in models if m in chosen_primary]

    # policy == "best_per_base_model"
    losses: dict[str, float] = {}
    metric = cfg.backtest.decision_metric
    if (
        oof_df is not None
        and not oof_df.empty
        and {"model_type", "y_true", "yhat"}.issubset(oof_df.columns)
    ):
        inner, _ = inner_fold_mask(oof_df, cfg)
        ref_oof = oof_df[inner] if inner.any() else oof_df
        for m in models:
            sub = ref_oof[ref_oof["model_type"] == m]
            if sub.empty:
                losses[m] = float("nan")
                continue
            yt = sub["y_true"].to_numpy(dtype=float)
            yh = sub["yhat"].to_numpy(dtype=float)
            valid = np.isfinite(yt) & np.isfinite(yh)
            if not valid.any():
                losses[m] = float("nan")
                continue
            denom = float(np.sum(np.abs(yt[valid])))
            err = float(np.sum(np.abs(yh[valid] - yt[valid])))
            losses[m] = err / denom if denom > 0.0 else err
    elif metric_df is not None and not metric_df.empty and metric in metric_df.columns:
        mean_by_model = metric_df.groupby("model_type")[metric].mean()
        for m in models:
            val = float(mean_by_model.get(m, float("nan")))
            losses[m] = loss_of(metric, val)

    order_map = {m: idx for idx, m in enumerate(models)}
    chosen_best: set[str] = set()
    for base, g in groups.items():
        if len(g) == 1:
            chosen_best.add(g[0])
            continue
        finite_candidates = [m for m in g if m in losses and np.isfinite(losses[m])]
        if not finite_candidates:
            chosen_best.add(base if base in g else g[0])
        else:
            winner = min(
                finite_candidates,
                key=lambda m: (
                    losses[m],
                    0 if "@" in m and "univariate" in m else (0 if "cov_none" in m else 1),
                    order_map[m],
                ),
            )
            chosen_best.add(winner)
    return [m for m in models if m in chosen_best]

fit_learned

fit_learned(oof_df, cfg)

Fit every learned strategy in cfg on the backtest OOF's inner folds.

Returns (weights, artifacts, scoring_basis): * weights[strategy] maps base-model name → learned weight, * artifacts[strategy] is the pickled, uploadable meta-learner payload, * scoring_basis is "holdout" when the newest fold was successfully reserved from the fit and "in_sample" when it could not be — the caller writes it to the registry beside the metrics it qualifies.

The newest fold is not in X. A meta-learner that trains and is scored on the same folds reports a fit statistic, and it lands in the same leaderboard column as the base models' genuinely out-of-fold numbers, where nothing distinguishes the two. Holding the newest fold out of the fit is what makes ensemble_nnls comparable to theta — see inner_fold_mask.

Refuses to run when backtesting is off — a learned ensemble has no legitimate, leakage-free data to train on (the guard that keeps stacking honest). Base models come from cfg.models in order; native vs python doesn't matter — only their OOF forecasts do.

Source code in src/scale_forecasting/ensembler.py

def fit_learned(
    oof_df: pd.DataFrame, cfg: RunConfig
) -> tuple[dict[str, dict[str, float]], dict[str, bytes], str]:
    """Fit every learned strategy in ``cfg`` on the backtest OOF's **inner** folds.

    Returns ``(weights, artifacts, scoring_basis)``:
      * ``weights[strategy]`` maps base-model name → learned weight,
      * ``artifacts[strategy]`` is the pickled, uploadable meta-learner payload,
      * ``scoring_basis`` is ``"holdout"`` when the newest fold was successfully reserved from the
        fit and ``"in_sample"`` when it could not be — the caller writes it to the registry beside
        the metrics it qualifies.

    **The newest fold is not in ``X``.** A meta-learner that trains and is scored on the same folds
    reports a fit statistic, and it lands in the same leaderboard column as the base models'
    genuinely out-of-fold numbers, where nothing distinguishes the two. Holding the newest fold out
    of the fit is what makes ``ensemble_nnls`` comparable to ``theta`` — see `inner_fold_mask`.

    Refuses to run when backtesting is off — a learned ensemble has no legitimate,
    leakage-free data to train on (the guard that keeps stacking honest). Base models come
    from ``cfg.models`` in order; native vs python doesn't matter — only their OOF forecasts do.
    """
    learned = [s for s in cfg.ensemble.strategies if s in LEARNED_STRATEGIES]
    if not learned:
        return {}, {}, "holdout"
    if not cfg.backtest.enabled:
        raise ConfigError(
            "learned ensemble strategies require backtest.enabled=true (they train on OOF)"
        )
    if oof_df.empty:
        raise ConfigError("cannot fit a learned ensemble: backtest_oof is empty")

    models = filter_variants_for_ensemble(list(cfg.models), cfg, oof_df=oof_df)
    inner, basis = inner_fold_mask(oof_df, cfg)
    X, y = _pivot_oof(oof_df[inner], models) if basis == "holdout" else (_EMPTY_X, _EMPTY_Y)
    if X.shape[0] == 0:
        # The inner folds exist but no *complete* row survives across every base model there --
        # one model only forecast the newest window, say. Refusing would cost the run its whole
        # learned ensemble over a fold-coverage gap; fitting on everything and saying so does not.
        basis = "in_sample"
        X, y = _pivot_oof(oof_df, models)
    if X.shape[0] == 0:
        raise ConfigError("no complete OOF rows across all base models to train on")

    weights: dict[str, dict[str, float]] = {}
    artifacts: dict[str, bytes] = {}
    for strategy in learned:
        if strategy == "nnls":
            w = _fit_nnls(X, y)
            payload: object = {"strategy": strategy, "models": models, "weights": w.tolist()}
        elif strategy == "ridge":
            w = _fit_ridge(X, y)
            payload = {"strategy": strategy, "models": models, "weights": w.tolist()}
        else:  # xgb
            w, payload = _fit_xgb(X, y, seed=0)
        weights[strategy] = {m: float(wi) for m, wi in zip(models, w, strict=True)}
        artifacts[strategy] = pickle.dumps(payload)
        _log.info("fit %s ensemble on %d %s OOF rows", strategy, X.shape[0], basis)

    return weights, artifacts, basis

inner_fold_mask

inner_fold_mask(oof_df, cfg)

Which OOF rows a fit is allowed to see, and what that says about the resulting numbers.

Returns (mask, basis). mask is True on the rows belonging to a fold other than backtest.holdout_fold_id — the folds a stacker, an inverse_error weight or a hyperparameter search may learn from. basis is "holdout" when a real split exists and "in_sample" when it does not, in which case the mask is all-True and the caller fits on everything it will later be scored on.

Two ways the split can fail to exist, and both are ordinary rather than exceptional: the run asked for n_folds: 1, or a short series in a ragged panel achieved only the newest fold. Falling back is the right behaviour — a stacker with no rows is worse than an optimistic one -- but it has to be said, because an in-sample number is indistinguishable from an honest one once it is sitting in a metric column. That is what basis is written to the registry for.

Source code in src/scale_forecasting/ensembler.py

def inner_fold_mask(oof_df: pd.DataFrame, cfg: RunConfig) -> tuple[np.ndarray, str]:
    """Which OOF rows a fit is allowed to see, and what that says about the resulting numbers.

    Returns ``(mask, basis)``. ``mask`` is True on the rows belonging to a fold other than
    `backtest.holdout_fold_id` — the folds a stacker, an ``inverse_error`` weight or a
    hyperparameter search may learn from. ``basis`` is ``"holdout"`` when a real split exists and
    ``"in_sample"`` when it does not, in which case the mask is all-True and the caller fits on
    everything it will later be scored on.

    Two ways the split can fail to exist, and both are ordinary rather than exceptional: the run
    asked for ``n_folds: 1``, or a short series in a ragged panel achieved only the newest fold.
    Falling back is the right behaviour — a stacker with no rows is worse than an optimistic one --
    but it has to be *said*, because an in-sample number is indistinguishable from an honest one
    once it is sitting in a metric column. That is what ``basis`` is written to the registry for.
    """
    holdout = holdout_fold_id(cfg)
    if "fold_id" not in oof_df.columns:
        return np.ones(len(oof_df), dtype=bool), "in_sample"
    mask = (oof_df["fold_id"] != holdout).to_numpy(dtype=bool)
    if not mask.any():
        return np.ones(len(oof_df), dtype=bool), "in_sample"
    return mask, "holdout"

inverse_error_weights

inverse_error_weights(errors)

Weights ∝ 1/error, normalized to sum to 1.

Lower-error models get more weight. A zero-error model would divide by zero, so zeros are treated as "perfectly trusted" and share the weight equally among themselves. Falls back to uniform weights when no error is finite and positive.

errors must be losses — non-negative, zero is perfect — not raw metric values. That is what metrics.loss_of produces, and every caller here routes through it. Handing this function a metric where larger is better inverts the weighting silently, which is exactly what used to happen with decision_metric="coverage".

Source code in src/scale_forecasting/ensembler.py

def inverse_error_weights(errors: np.ndarray) -> np.ndarray:
    """Weights ∝ 1/error, normalized to sum to 1.

    Lower-error models get more weight. A zero-error model would divide by zero, so zeros are
    treated as "perfectly trusted" and share the weight equally among themselves. Falls back to
    uniform weights when no error is finite and positive.

    ``errors`` must be **losses** — non-negative, zero is perfect — not raw metric values. That is
    what `metrics.loss_of` produces, and every caller here routes through it. Handing this function
    a metric where larger is better inverts the weighting silently, which is exactly what used to
    happen with ``decision_metric="coverage"``.
    """
    err = np.asarray(errors, dtype=float)
    if err.ndim != 1 or err.size == 0:
        raise ValueError("errors must be a non-empty 1-D array")

    zero = err == 0.0
    if zero.any():
        w = zero.astype(float)
        return w / w.sum()

    finite = np.isfinite(err) & (err > 0.0)
    if not finite.any():
        return np.full(err.size, 1.0 / err.size)

    inv = np.where(finite, 1.0 / err, 0.0)
    return inv / inv.sum()

mean_combine

mean_combine(yhats)

Row-wise mean across models. yhats is (n_models, horizon).

Source code in src/scale_forecasting/ensembler.py

def mean_combine(yhats: np.ndarray) -> np.ndarray:
    """Row-wise mean across models. ``yhats`` is ``(n_models, horizon)``."""
    return np.asarray(yhats, dtype=float).mean(axis=0)

median_combine

median_combine(yhats)

Row-wise median across models — robust to a single wild base forecast.

Source code in src/scale_forecasting/ensembler.py

def median_combine(yhats: np.ndarray) -> np.ndarray:
    """Row-wise median across models — robust to a single wild base forecast."""
    return np.median(np.asarray(yhats, dtype=float), axis=0)