Ensemble Driver

ensemble_run

Execute + score the run's ensembles into the registry (the orchestration seam).

ensembler is pure — it blends the base forecasts (calculated) and fits the learned meta-learners, but touches no GCP. This module is the thin I/O orchestrator that makes those consensuses real: it reads the base rows, computes both families in pandas, appends the ensemble_<s> prediction rows via the Storage Write API, and — the missing leaderboard link — scores every ensemble pseudo-model into forecast_metadata so it appears on v_model_leaderboard beside the base models. It is engine-agnostic (it only reads the shared registry) and runs either inline from main.run (after both engines join, under the one shared run_id) or standalone via python -m scale_forecasting.ensemble_run (re-ensembling an already-completed run — see _main).

Config-keyed ensembles. Every ensemble row carries an ensemble_id = make_ensemble_id(cfg.ensemble) — a digest of the ensemble configuration alone — so several ensemble configs can be scored under one run_id without their ensemble_<strategy> pseudo-models colliding. Re-running the same ensemble config lands the same ensemble_id (and so the same cells, deduped newest-first on read); a different config lands a different ensemble_id and sits beside the first on the leaderboard, distinctly keyed. The view groups by (run_id, model_type, ensemble_id) so the two never merge.

Three responsibilities, in order:

1. Calculated (mean / median / inverse_error) — read the base forecast_predictions (+ forecast_metadata for the inverse-error weights / pruning) and blend them in pandas (ensembler.combine_calculated), then append the ensemble_<s> rows via the Storage Write API — the same append path the learned strategies use (no INSERT…SELECT DML). compute_engine='ensemble'.
2. Learned (nnls / ridge / xgb) — read the base predictions + backtest_oof, fit_learned on the OOF, then apply the weights in pandas (yhat = Σ wₘ·yhatₘ, renormalized over whichever base models are present per (ts_id, forecast_date) — robust when the Spark future window and the native held-out window don't overlap), and append the resulting ensemble_<s> prediction rows via the Storage Write API. Each fitted meta-learner is uploaded as a GCS artifact and linked from its scored metadata row. These prediction rows are a true beyond-data forecast (the base predictions they blend are, too), so — like the base models — they carry no ground truth of their own.
3. Score — blend the base backtest_oof into an ensemble OOF with the same consensus rules (ensembler.combine_oof), append those blended rows back into backtest_oof (same table as the base rows, keyed apart by ensemble_id, so anything computed from that table — the pooled comparable leaderboard, a per-horizon coverage read — sees the consensuses and the base models on identical footing), and run the shared metrics.compute_metrics per (model, ts_id) → forecast_metadata rows with fold_id=NULL and compute_engine='ensemble'. Scoring lives on the OOF window because the base predictions (and therefore every ensemble prediction) are a true beyond-data forecast with no actuals to join — so an ensemble earns its leaderboard metric on exactly the window the base models are scored on (backtest_oof, where y_true lives). OOF carries no interval bounds, so ensemble coverage/pinball are unscored — computed as NaN and written as NULL, like every other metric this registry stores, consistent with the base models' OOF metrics. Learned consensuses are scored on the folds their meta-learner trained on (mildly optimistic, the price of stacking having no held-out-of-held-out window). Once these fold_id IS NULL rows land, the leaderboard shows the ensembles automatically — no view change beyond the ensemble_id group key.

Idempotency (append-only + dedupe-on-read). Every ensemble row is written through the Write API and is keyed in (run_id, ensemble_id, ts_id, model_type), so a re-run of the same ensemble config re-appends the same cells — correct-but-wasteful when the numbers repeat, and a genuine conflict when they don't (xgb is a stochastic meta-learner, and a repair re-fits). Both cases resolve the same way: every ensemble row carries a created_at, and each read dedupes to one row per cell by ORDER BY created_at DESC NULLS LAST — see base_read_sql, and registry.rows.cell_dedup_key for the same rule on the base tables. No pre-delete is issued because a DELETE matching rows still in the ~90-min Write API streaming buffer is rejected for the whole window (the constraint every cell writer lives under). A different ensemble config keys distinctly (different ensemble_id), so it never overwrites and never collides — both coexist.

Public surface: run_ensembles, python -m scale_forecasting.ensemble_run.

base_read_sql

base_read_sql(
    dataset,
    table,
    columns,
    model_list,
    ts_filter,
    extra="",
    dedupe_by="",
)

One run-, model- and series-scoped SELECT over a registry table (pure).

All three reads this module makes share a shape, and the model_type IN (...) clause is the part that has to be in every one of them. Two of the tables read here — forecast_predictions and now backtest_oof — hold the ensemble's own rows beside the base models', so a read that omits the filter blends consensuses of consensuses on any second pass over a run: a microbatch drain, a --force re-ensemble, a standalone re-score. Building the clause in one place is what stops that from being a thing to remember at each call site.

extra is for a clause only one read needs (forecast_metadata wants the full-fit rows only); ts_filter is the microbatch's AND ts_id IN UNNEST(@ts_ids), empty for a barrier.

dedupe_by names the cell grain of the table, and turns the read into "one row per cell, newest write wins" via QUALIFY ROW_NUMBER() … ORDER BY created_at DESC NULLS LAST. Three things downstream need this and none of them fail loudly without it: ensembler.combine_calculated pivots with pivot_table, whose default aggfunc is mean, so a duplicated base row is silently averaged with itself — harmless for two identical copies, wrong the moment a repair re-fit one of them. _inverse_error_run_weights takes an ungrouped mean of the decision metric, so a model with duplicate metadata rows is weighted by an average of its attempts. And the blended OOF is appended back into backtest_oof, so a duplicate on the way in becomes a duplicate on the way out.

NULLS LAST is the part that matters for old data: created_at has only had a writer on these two tables since P10, so every row from before it is NULL, and a NULL must lose to any real timestamp rather than sorting first and winning permanently.

Source code in src/scale_forecasting/ensemble_run.py

def base_read_sql(
    dataset: str,
    table: str,
    columns: str,
    model_list: str,
    ts_filter: str,
    extra: str = "",
    dedupe_by: str = "",
) -> str:
    """One run-, model- and series-scoped SELECT over a registry table (pure).

    All three reads this module makes share a shape, and the ``model_type IN (...)`` clause is the
    part that has to be in every one of them. Two of the tables read here — ``forecast_predictions``
    and now ``backtest_oof`` — hold *the ensemble's own rows* beside the base models', so a read
    that omits the filter blends consensuses of consensuses on any second pass over a run: a
    microbatch drain, a ``--force`` re-ensemble, a standalone re-score. Building the clause in one
    place is what stops that from being a thing to remember at each call site.

    ``extra`` is for a clause only one read needs (``forecast_metadata`` wants the full-fit rows
    only); ``ts_filter`` is the microbatch's ``AND ts_id IN UNNEST(@ts_ids)``, empty for a barrier.

    ``dedupe_by`` names the cell grain of the table, and turns the read into "one row per cell,
    newest write wins" via ``QUALIFY ROW_NUMBER() … ORDER BY created_at DESC NULLS LAST``. Three
    things downstream need this and none of them fail loudly without it:
    `ensembler.combine_calculated` pivots with ``pivot_table``, whose default ``aggfunc`` is
    ``mean``, so a duplicated base row is silently *averaged* with itself — harmless for two
    identical copies, wrong the moment a repair re-fit one of them. ``_inverse_error_run_weights``
    takes an ungrouped mean of the decision metric, so a model with duplicate metadata rows is
    weighted by an average of its attempts. And the blended OOF is appended back into
    ``backtest_oof``, so a duplicate on the way in becomes a duplicate on the way out.

    ``NULLS LAST`` is the part that matters for old data: ``created_at`` has only had a writer on
    these two tables since P10, so every row from before it is NULL, and a NULL must lose to any
    real timestamp rather than sorting first and winning permanently.
    """
    qualify = (
        ""
        if not dedupe_by
        else (
            f"\nQUALIFY ROW_NUMBER() OVER ("
            f"\n  PARTITION BY {dedupe_by}"
            f"\n  ORDER BY created_at DESC NULLS LAST"
            f"\n) = 1"
        )
    )
    return (
        f"SELECT {columns}\n"
        f"FROM `{dataset}.{table}`\n"
        f"WHERE run_id = @run_id AND model_type IN ({model_list}){extra}{ts_filter}{qualify}"
    )

cross_run_copy_sql

cross_run_copy_sql(dataset, table, model_list, dedupe_by)

Build an idempotent SQL INSERT INTO ... SELECT * REPLACE statement that consolidates deduplicated base-model rows from @source_run_ids into @target_run_id (pure).

Only base-model rows (ensemble_id IS NULL) matching model_list are copied, deduplicated across @source_run_ids with newest created_at winning per dedupe_by grain. Any (ts_id, model_type) cell already present under @target_run_id is skipped via NOT EXISTS so re-running cross-run ensembling with additional strategies never duplicates base rows.

Source code in src/scale_forecasting/ensemble_run.py

def cross_run_copy_sql(
    dataset: str,
    table: str,
    model_list: str,
    dedupe_by: str,
) -> str:
    """Build an idempotent SQL ``INSERT INTO ... SELECT * REPLACE`` statement that consolidates
    deduplicated base-model rows from ``@source_run_ids`` into ``@target_run_id`` (pure).

    Only base-model rows (``ensemble_id IS NULL``) matching ``model_list`` are copied, deduplicated
    across ``@source_run_ids`` with newest ``created_at`` winning per ``dedupe_by`` grain. Any
    ``(ts_id, model_type)`` cell already present under ``@target_run_id`` is skipped via ``NOT
    EXISTS`` so re-running cross-run ensembling with additional strategies never duplicates base
    rows.
    """
    return (
        f"INSERT INTO `{dataset}.{table}`\n"
        f"SELECT * REPLACE (@target_run_id AS run_id, @created_at AS created_at)\n"
        f"FROM `{dataset}.{table}` AS src\n"
        f"WHERE src.run_id IN UNNEST(@source_run_ids)\n"
        f"  AND src.model_type IN ({model_list})\n"
        f"  AND src.ensemble_id IS NULL\n"
        f"  AND NOT EXISTS (\n"
        f"    SELECT 1 FROM `{dataset}.{table}` AS existing\n"
        f"    WHERE existing.run_id = @target_run_id\n"
        f"      AND existing.ts_id = src.ts_id\n"
        f"      AND existing.model_type = src.model_type\n"
        f"      AND existing.ensemble_id IS NULL\n"
        f"  )\n"
        f"QUALIFY ROW_NUMBER() OVER (\n"
        f"  PARTITION BY {dedupe_by}\n"
        f"  ORDER BY src.created_at DESC NULLS LAST\n"
        f") = 1"
    )

ensemble_cross_runs

ensemble_cross_runs(
    source_run_ids,
    strategies=None,
    *,
    target_run_id=None,
    settings=None,
)

Blend base models across multiple completed runs into a consolidated cross-run ensemble.

Reads the stored RunConfig for each run in source_run_ids (registry.reads.read_run_config), merges them via merge_configs_for_ensemble, derives target_run_id (defaulting to make_run_id(merged_cfg)), copies deduplicated base rows (forecast_predictions, backtest_oof, forecast_metadata) into target_run_id via cross_run_copy_sql, records a COMPLETED header + ensemble job row in the registry, and executes run_ensembles so the cross-run ensembles and all contributing base models appear side-by-side on v_model_leaderboard and review.review_run. Returns target_run_id.

Source code in src/scale_forecasting/ensemble_run.py

def ensemble_cross_runs(
    source_run_ids: Iterable[str],
    strategies: list[str] | None = None,
    *,
    target_run_id: str | None = None,
    settings: Settings | None = None,
) -> str:  # pragma: no cover - GCP I/O, @gcp ensemble smoke
    """Blend base models across multiple completed runs into a consolidated cross-run ensemble.

    Reads the stored `RunConfig` for each run in ``source_run_ids``
    (`registry.reads.read_run_config`), merges them via `merge_configs_for_ensemble`, derives
    ``target_run_id`` (defaulting to ``make_run_id(merged_cfg)``), copies deduplicated base rows
    (``forecast_predictions``, ``backtest_oof``, ``forecast_metadata``) into ``target_run_id`` via
    `cross_run_copy_sql`, records a ``COMPLETED`` header + ``ensemble`` job row in the registry,
    and executes `run_ensembles` so the cross-run ensembles and all contributing base models
    appear side-by-side on ``v_model_leaderboard`` and `review.review_run`. Returns
    ``target_run_id``.
    """
    import time
    from datetime import UTC, datetime

    from google.cloud import bigquery

    from .config import RunConfig
    from .errors import ConfigError, get_logger
    from .registry.header import merge_header_telemetry, update_header, write_header
    from .registry.ids import make_job_key, make_run_id
    from .registry.jobs import update_job, write_job
    from .registry.reads import read_run_config, read_run_summary
    from .registry.rows import assemble_job_row
    from .registry.tables import resolve_settings

    run_ids = [str(r).strip() for r in source_run_ids if str(r).strip()]
    if len(run_ids) < 2:
        raise ConfigError(
            f"ensemble_cross_runs requires at least 2 source_run_ids; got {run_ids!r}."
        )
    resolved = resolve_settings(settings)
    configs: list[RunConfig] = []
    for rid in run_ids:
        raw = read_run_config(rid, settings=resolved)
        if raw is None:
            raise ConfigError(f"source run {rid!r} has no stored config in run_registry.")
        configs.append(RunConfig.model_validate(raw))

    merged_cfg = merge_configs_for_ensemble(configs, strategies=strategies)
    out_run_id = target_run_id or make_run_id(merged_cfg)
    log = get_logger(__name__)
    t0 = time.monotonic()
    created_at = datetime.now(UTC)

    existing_summary = read_run_summary(out_run_id, settings=resolved)
    if existing_summary is None:
        write_header(merged_cfg, out_run_id, settings=resolved, status="RUNNING")

    job_key = make_job_key(out_run_id, "ensemble", 1)
    if existing_summary is None:
        write_job(
            assemble_job_row(
                out_run_id,
                "ensemble",
                1,
                created_at,
                runtime="bigquery",
                started_at=created_at,
            ),
            settings=resolved,
        )

    dataset = resolved.registry_dataset_ref
    model_list = ", ".join(f"'{m}'" for m in merged_cfg.models)
    client = bigquery_client(resolved.project_id)
    copy_params = [
        bigquery.ArrayQueryParameter("source_run_ids", "STRING", run_ids),
        bigquery.ScalarQueryParameter("target_run_id", "STRING", out_run_id),
        bigquery.ScalarQueryParameter("created_at", "TIMESTAMP", created_at),
    ]
    for table, dedupe_by in (
        ("forecast_predictions", "src.ts_id, src.model_type, src.forecast_date"),
        ("backtest_oof", "src.ts_id, src.model_type, src.fold_id, src.forecast_date"),
        ("forecast_metadata", "src.ts_id, src.model_type, src.fold_id"),
    ):
        sql = cross_run_copy_sql(dataset, table, model_list, dedupe_by)
        client.query(sql, job_config=bigquery.QueryJobConfig(query_parameters=copy_params)).result()

    run_ensembles(merged_cfg, out_run_id, settings=resolved)
    elapsed = round(time.monotonic() - t0, 3)
    ended_at = datetime.now(UTC)
    if existing_summary is None:
        update_job(
            job_key,
            settings=resolved,
            status="COMPLETED",
            ended_at=ended_at,
            runtime_seconds=elapsed,
        )
        update_header(
            out_run_id,
            settings=resolved,
            status="COMPLETED",
            n_series=merged_cfg.data.series_limit,
            n_models=len(merged_cfg.models),
            runtime_seconds=elapsed,
        )
    try:
        merge_header_telemetry(
            out_run_id,
            {"cross_run_sources": run_ids},
            settings=resolved,
        )
    except Exception as exc:  # noqa: BLE001 - best-effort telemetry
        log.warning("cross_run_sources telemetry not recorded for %s: %s", out_run_id, exc)
    log.info(
        "cross-run ensemble complete: target_run_id=%s sources=%s models=%s strategies=%s",
        out_run_id,
        run_ids,
        merged_cfg.models,
        merged_cfg.ensemble.strategies,
    )
    return out_run_id

ensemble_refit_mode

ensemble_refit_mode(modes)

How this ensemble's members were scored, rolled up into one answer (pure).

Source code in src/scale_forecasting/ensemble_run.py

def ensemble_refit_mode(modes: Iterable[str | None]) -> str | None:
    """How this ensemble's *members* were scored, rolled up into one answer (pure)."""
    seen = {m for m in modes if m is not None}
    if not seen:
        return None
    return seen.pop() if len(seen) == 1 else "mixed"

ensemble_scoring_basis

ensemble_scoring_basis(
    strategy, oof_df, cfg, learned_basis
)

Whether this strategy's metrics were earned with the newest fold held out (pure).

Source code in src/scale_forecasting/ensemble_run.py

def ensemble_scoring_basis(
    strategy: str, oof_df: pd.DataFrame, cfg: RunConfig, learned_basis: str
) -> str | None:
    """Whether this strategy's metrics were earned with the newest fold held out (pure)."""
    from .config import LEARNED_STRATEGIES
    from .ensembler import inner_fold_mask

    if strategy in LEARNED_STRATEGIES:
        return learned_basis
    if strategy in ("inverse_error", "champion", "horizon_specialist"):
        return inner_fold_mask(oof_df, cfg)[1]
    if cfg.ensemble.top_k is not None or cfg.ensemble.parsimony_margin > 0.0:
        return inner_fold_mask(oof_df, cfg)[1]
    return None

merge_configs_for_ensemble

merge_configs_for_ensemble(configs, strategies=None)

Merge multiple compatible RunConfig objects into a single composite config for cross-run ensembling (pure).

Verifies that at least two configs are supplied and that all source configs share the same data.source_table, data.freq, data.horizon, and backtest geometry (enabled, n_folds, horizon, step), raising ConfigError on any mismatch. The returned config uses the first config as its base, sets models to the ordered deduplicated union of models across all source configs, merges model_params, and enables the requested ensemble strategies (defaulting to ["mean", "median", "inverse_error"] when neither strategies nor any source config enables strategies).

Source code in src/scale_forecasting/ensemble_run.py

def merge_configs_for_ensemble(
    configs: Iterable[RunConfig], strategies: list[str] | None = None
) -> RunConfig:
    """Merge multiple compatible `RunConfig` objects into a single composite config for cross-run
    ensembling (pure).

    Verifies that at least two configs are supplied and that all source configs share the same
    ``data.source_table``, ``data.freq``, ``data.horizon``, and ``backtest`` geometry (`enabled`,
    `n_folds`, `horizon`, `step`), raising `ConfigError` on any mismatch. The returned config uses
    the first config as its base, sets ``models`` to the ordered deduplicated union of models
    across all source configs, merges ``model_params``, and enables the requested ensemble
    ``strategies`` (defaulting to ``["mean", "median", "inverse_error"]`` when neither
    ``strategies`` nor any source config enables strategies).
    """
    from .config import EnsembleConfig, RunConfig
    from .errors import ConfigError

    cfg_list = list(configs)
    if len(cfg_list) < 2:
        raise ConfigError(
            f"merge_configs_for_ensemble requires at least 2 source configs; got {len(cfg_list)}."
        )
    first = cfg_list[0]
    merged_models: list[str] = []
    merged_params: dict[str, dict[str, Any]] = {}
    default_strategies: list[str] = []

    for idx, c in enumerate(cfg_list):
        if (c.data.source_table, c.data.freq) != (
            first.data.source_table,
            first.data.freq,
        ):
            raise ConfigError(
                f"cross-run ensemble source config[{idx}] has incompatible data source/freq "
                f"({c.data.source_table}, freq={c.data.freq!r}) vs "
                f"({first.data.source_table}, freq={first.data.freq!r})."
            )
        if c.data.horizon != first.data.horizon:
            raise ConfigError(
                f"cross-run ensemble source config[{idx}] has data.horizon={c.data.horizon}, "
                f"which does not match config[0].data.horizon={first.data.horizon}."
            )
        if (
            c.backtest.enabled != first.backtest.enabled
            or c.backtest.n_folds != first.backtest.n_folds
            or c.backtest.horizon != first.backtest.horizon
            or c.backtest.step != first.backtest.step
        ):
            raise ConfigError(
                f"cross-run ensemble source config[{idx}] has incompatible backtest geometry "
                f"(enabled={c.backtest.enabled}, n_folds={c.backtest.n_folds}, "
                f"horizon={c.backtest.horizon}, step={c.backtest.step}) vs "
                f"(enabled={first.backtest.enabled}, n_folds={first.backtest.n_folds}, "
                f"horizon={first.backtest.horizon}, step={first.backtest.step})."
            )
        for m in c.models:
            if m not in merged_models:
                merged_models.append(m)
        for m, p in c.model_params.items():
            merged_params[m] = dict(p)
        for s in c.ensemble.strategies:
            if s not in default_strategies:
                default_strategies.append(s)

    chosen_strategies = (
        list(strategies)
        if strategies is not None
        else (default_strategies or ["mean", "median", "inverse_error"])
    )
    ensemble = EnsembleConfig.model_validate(
        {
            "enabled": True,
            "strategies": chosen_strategies,
            "prune_threshold": first.ensemble.prune_threshold,
        }
    )
    payload = first.model_dump(mode="python")
    payload["models"] = merged_models
    payload["model_params"] = merged_params
    payload["ensemble"] = ensemble.model_dump(mode="python")
    return RunConfig.model_validate(payload)

run_ensemble_scoring

run_ensemble_scoring(bases)

Roll the per-strategy scoring bases up into one claim for the run (pure).

None when nothing in the run fit anything — a mean/median-only ensemble has no fold to reserve, so there is no claim to make and the header stays quiet rather than saying "holdout" about a fit that never happened. Otherwise the weakest answer wins: one strategy that fell back to in_sample makes the run's ensemble metrics partly in-sample, and a header that reported the best case would be exactly the reassurance this column exists to withhold.

Source code in src/scale_forecasting/ensemble_run.py

def run_ensemble_scoring(bases: Iterable[str | None]) -> str | None:
    """Roll the per-strategy scoring bases up into one claim for the run (pure).

    ``None`` when nothing in the run fit anything — a mean/median-only ensemble has no fold to
    reserve, so there is no claim to make and the header stays quiet rather than saying "holdout"
    about a fit that never happened. Otherwise the **weakest** answer wins: one strategy that fell
    back to ``in_sample`` makes the run's ensemble metrics partly in-sample, and a header that
    reported the best case would be exactly the reassurance this column exists to withhold.
    """
    seen = {b for b in bases if b is not None}
    if not seen:
        return None
    return "in_sample" if "in_sample" in seen else "holdout"

run_ensembles

run_ensembles(cfg, run_id, *, settings, job_id_prefix=None)

Execute + score every requested ensemble for run_id into the shared registry (barrier).

A no-op when cfg.ensemble.enabled is false. Otherwise blends the calculated + learned consensuses in pandas, appends their ensemble_<s> prediction rows via the Storage Write API, and scores every ensemble pseudo-model into forecast_metadata (see the module docstring), all stamped with ensemble_id = make_ensemble_id(cfg.ensemble) so multiple ensemble configs coexist under one run_id. Raises on any failure so main.run can finalize the shared header FAILED — mirroring how an engine error is surfaced. settings is the orchestrator's already-resolved infra (never re-resolved here, so one identity governs the whole run).

This is the barrier trigger: it blends every series in one pass, after all base jobs have joined. The microbatch counterpart (run_ensembles_microbatch) drains series incrementally as each one's base set completes; both call the shared per-batch core (_ensemble_batch). They land identical rows for the calculated strategies; the learned ones are re-fit per batch under microbatch and so can differ — see _ensemble_batch.

Source code in src/scale_forecasting/ensemble_run.py

def run_ensembles(
    cfg: RunConfig, run_id: str, *, settings: Settings, job_id_prefix: str | None = None
) -> None:  # pragma: no cover - GCP I/O, @gcp ensemble smoke
    """Execute + score every requested ensemble for ``run_id`` into the shared registry (barrier).

    A no-op when ``cfg.ensemble.enabled`` is false. Otherwise blends the calculated + learned
    consensuses in pandas, appends their ``ensemble_<s>`` prediction rows via the Storage Write API,
    and scores every ensemble pseudo-model into ``forecast_metadata`` (see the module docstring),
    all stamped with ``ensemble_id = make_ensemble_id(cfg.ensemble)`` so multiple ensemble configs
    coexist under one ``run_id``. Raises on any failure so `main.run` can finalize the shared
    header FAILED — mirroring how an engine error is surfaced. ``settings`` is the orchestrator's
    already-resolved infra (never re-resolved here, so one identity governs the whole run).

    This is the **barrier** trigger: it blends every series in one pass, after all base jobs have
    joined. The **microbatch** counterpart (`run_ensembles_microbatch`) drains series incrementally
    as each one's base set completes; both call the shared per-batch core (`_ensemble_batch`). They
    land identical rows for the *calculated* strategies; the learned ones are re-fit per batch under
    microbatch and so can differ — see `_ensemble_batch`.
    """
    from datetime import UTC, datetime

    from .errors import get_logger
    from .registry.ids import make_ensemble_id

    if not cfg.ensemble.enabled:
        return

    log = get_logger(__name__)
    ensemble_id = make_ensemble_id(cfg.ensemble)
    client = bigquery_client(settings.project_id)
    created_at = datetime.now(UTC)
    log.info(
        "ensemble run start (barrier): run_id=%s ensemble_id=%s strategies=%s",
        run_id,
        ensemble_id,
        cfg.ensemble.strategies,
    )
    _ensemble_batch(
        cfg,
        run_id,
        settings=settings,
        client=client,
        ensemble_id=ensemble_id,
        created_at=created_at,
        log=log,
        ts_ids=None,
        job_id_prefix=job_id_prefix,
    )

run_ensembles_microbatch

run_ensembles_microbatch(
    cfg,
    run_id,
    *,
    settings,
    poll_interval_s=15.0,
    max_polls=960,
    upstream_done=None,
    job_id_prefix=None,
)

Execute + score ensembles for run_id incrementally, per series as it completes.

The microbatch trigger: rather than waiting for every base model on every series (the run_ensembles barrier), this drains series in ready-batches — a series is ready once all cfg.models base models have landed a prediction row for it (_ready_series), and each ready-batch is blended + scored by the same core as the barrier (_ensemble_batch over that ts_id subset), so partial consensus output lands progressively and one slow family never blocks the series that are already complete. The loop (_drain_ready) polls every poll_interval_s seconds until upstream_done reports the base jobs finished and no ready series remain unprocessed (bounded by max_polls as a safety stop).

v1 default: the orchestrator calls this after the family join, so upstream_done defaults to "always done" and the loop drains every ready series in a single pass — the streaming semantics (per-series readiness, incremental writes) with a post-join trigger. Passing an upstream_done that tracks live base-job completion turns it into a consumer that overlaps base computation (the concurrent-trigger extension). Raises on any batch failure so main.run finalizes the header FAILED.

Source code in src/scale_forecasting/ensemble_run.py

def run_ensembles_microbatch(
    cfg: RunConfig,
    run_id: str,
    *,
    settings: Settings,
    poll_interval_s: float = 15.0,
    max_polls: int = 960,
    upstream_done: Callable[[], bool] | None = None,
    job_id_prefix: str | None = None,
) -> None:  # pragma: no cover - GCP I/O, @gcp ensemble smoke
    """Execute + score ensembles for ``run_id`` **incrementally, per series as it completes**.

    The microbatch trigger: rather than waiting for every base model on every series (the
    `run_ensembles` barrier), this drains series in ready-batches — a series is *ready* once all
    ``cfg.models`` base models have landed a prediction row for it (`_ready_series`), and each
    ready-batch is blended + scored by the same core as the barrier (`_ensemble_batch` over that
    ts_id subset), so partial consensus output lands progressively and one slow family never blocks
    the series that are already complete. The loop (`_drain_ready`) polls every ``poll_interval_s``
    seconds until ``upstream_done`` reports the base jobs finished *and* no ready series remain
    unprocessed (bounded by ``max_polls`` as a safety stop).

    v1 default: the orchestrator calls this **after** the family join, so ``upstream_done`` defaults
    to "always done" and the loop drains every ready series in a single pass — the streaming
    *semantics* (per-series readiness, incremental writes) with a post-join trigger. Passing an
    ``upstream_done`` that tracks live base-job completion turns it into a consumer that overlaps
    base computation (the concurrent-trigger extension). Raises on any batch failure so `main.run`
    finalizes the header FAILED.
    """
    import time
    from datetime import UTC, datetime

    from .errors import get_logger
    from .registry.ids import make_ensemble_id

    if not cfg.ensemble.enabled:
        return

    log = get_logger(__name__)
    ensemble_id = make_ensemble_id(cfg.ensemble)
    client = bigquery_client(settings.project_id)
    created_at = datetime.now(UTC)
    done = upstream_done or (lambda: True)
    log.info(
        "ensemble run start (microbatch): run_id=%s ensemble_id=%s strategies=%s",
        run_id,
        ensemble_id,
        cfg.ensemble.strategies,
    )

    def _process(ts_ids: list[str]) -> None:
        _ensemble_batch(
            cfg,
            run_id,
            settings=settings,
            client=client,
            ensemble_id=ensemble_id,
            created_at=created_at,
            log=log,
            ts_ids=ts_ids,
            job_id_prefix=job_id_prefix,
        )

    processed = _drain_ready(
        ready_fn=lambda: _ready_series(
            cfg, run_id, settings=settings, client=client, job_id_prefix=job_id_prefix
        ),
        process_fn=_process,
        done_fn=done,
        sleep_fn=lambda: time.sleep(poll_interval_s),
        max_polls=max_polls,
    )
    log.info(
        "ensemble run done (microbatch): run_id=%s ensemble_id=%s series_drained=%d",
        run_id,
        ensemble_id,
        len(processed),
    )