Forecaster (SDK)

sdk

The easy path: a thin, well-typed Forecaster facade over main.run.

This is the "simple SDK on top" — one class you point at a config and call. It adds no forecasting logic; every method delegates to the same code the CLI and Composer run (scale_forecasting.main.run, the run registry, the config layer), so a run driven from the SDK is byte-for-byte the run driven from python -m scale_forecasting.main. Users who need to drive Spark or Ray themselves skip this class entirely and call the direct surface (run_group, make_group_runner, make_chunk_runner, run_cell) — both paths reuse the identical model machinery. See docs/using_the_sdk.md.

Import cost: this module is cheap to import. The heavy model modules load only when a method that runs models is called (main is imported lazily inside run/dry_run), which preserves the near-instant import scale_forecasting contract enforced by test_sdk.py.

DryRunResult dataclass

DryRunResult(
    run_id, fanout, python_models, bq_models, workload=None
)

What a run would do, computed offline — no GCP calls, no compute launched.

run_id is the deterministic config hash (identical to what an actual run lands under); workload is the estimated work (cells always; fits and fold cohorts only once a caller has supplied per-series lengths, which this offline path has not); python_models/bq_models are the runtime split. fanout is the same estimate narrowed to its four count fields, kept because it is the older name and is in the public __init__.

Forecaster

Forecaster(config, *, settings=None)

The easy path: point it at a config, then dry_run, run, status, wait, or results.

A run driven here is identical to the CLI/Composer run — this class only wraps scale_forecasting.main.run. Construct from an in-memory RunConfig, or use from_file / from_dict, or from_run_id to recover the config a run already landed under. An optional settings injects the GCP infra identity; None resolves it from the SF_* environment at run time (the default deployments use).

The lifecycle closes the loop from one object: dry_run (offline plan), dag (the planned per-job DAG), run (execute), status/wait (track a submission), results (read the per-model leaderboard), jobs (the per-job cross-system trace), and trace (the per-job + per-cell execution timeline) — all keyed by the config's deterministic run_id, so status/results/jobs/trace work as a reattach path even in a fresh process.

Source code in src/scale_forecasting/sdk.py

def __init__(self, config: RunConfig, *, settings: Settings | None = None) -> None:
    self._config = config
    self._settings = settings

config property

config

The validated run configuration this forecaster will execute.

run_id property

run_id

The deterministic run_id for this config (pure hash; no GCP call).

attributions_df

attributions_df(
    run_id=None,
    *,
    ts_id=None,
    models=None,
    model_type=None,
    level="global",
    limit=5000,
)

Return a tidy DataFrame of Tier 1 (level="global") or Tier 2 (level="local") feature attributions for run_id.

• level="global" reads forecast_metadata.fit_diagnostics["feature_attributions"] and returns ["ts_id", "model_type", "compute_engine", "feature", "importance"].
• level="local" reads per-horizon-step forecast_predictions.explanations and returns ["ts_id", "model_type", "compute_engine", "forecast_date", "yhat", "baseline_score", "feature", "attribution"].

Source code in src/scale_forecasting/sdk.py

def attributions_df(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | list[str] | None = None,
    models: str | list[str] | None = None,
    model_type: str | None = None,
    level: str = "global",
    limit: int = 5000,
) -> pd.DataFrame:
    """Return a tidy ``DataFrame`` of Tier 1 (``level="global"``) or Tier 2 (``level="local"``)
    feature attributions for ``run_id``.

    - ``level="global"`` reads ``forecast_metadata.fit_diagnostics["feature_attributions"]``
      and returns ``["ts_id", "model_type", "compute_engine", "feature", "importance"]``.
    - ``level="local"`` reads per-horizon-step ``forecast_predictions.explanations`` and
      returns ``["ts_id", "model_type", "compute_engine", "forecast_date", "yhat",
      "baseline_score", "feature", "attribution"]``.
    """
    from .registry.reads import read_feature_attributions
    from .review import build_attributions_frame

    rid = run_id or self.run_id
    ts_list = [ts_id] if isinstance(ts_id, str) else (list(ts_id) if ts_id else None)
    eff_models = models if models is not None else model_type
    model_list = (
        [eff_models]
        if isinstance(eff_models, str)
        else (list(eff_models) if eff_models else None)
    )
    rows = read_feature_attributions(
        rid,
        ts_ids=ts_list,
        models=model_list,
        level=level,
        limit=limit,
        settings=self._settings,
    )
    return build_attributions_frame(rows, level=level)

best_params_df

best_params_df(run_id=None, *, limit=5000)

Return a DataFrame unpacking forecast_metadata.best_params for run_id.

Includes both Optuna HPO winning hyperparameters (for base models) and learned ensemble stacking weights (nnls / ridge / xgb for ensemble_* rows).

Source code in src/scale_forecasting/sdk.py

def best_params_df(self, run_id: str | None = None, *, limit: int = 5000) -> pd.DataFrame:
    """Return a ``DataFrame`` unpacking ``forecast_metadata.best_params`` for ``run_id``.

    Includes both Optuna HPO winning hyperparameters (for base models) and learned ensemble
    stacking weights (``nnls`` / ``ridge`` / ``xgb`` for ``ensemble_*`` rows).
    """
    from .registry.reads import read_best_params

    rid = run_id or self.run_id
    return build_best_params_frame(read_best_params(rid, limit=limit, settings=self._settings))

bias_df

bias_df(
    run_id=None,
    *,
    threshold_pct=0.05,
    tracking_signal_limit=4.0,
    limit=20000,
)

Alias for :meth:bias_report_df matching :meth:BakeOff.bias_df.

Source code in src/scale_forecasting/sdk.py

def bias_df(
    self,
    run_id: str | None = None,
    *,
    threshold_pct: float = 0.05,
    tracking_signal_limit: float = 4.0,
    limit: int = 20000,
) -> pd.DataFrame:
    """Alias for :meth:`bias_report_df` matching :meth:`BakeOff.bias_df`."""
    return self.bias_report_df(
        run_id,
        threshold_pct=threshold_pct,
        tracking_signal_limit=tracking_signal_limit,
        limit=limit,
    )

bias_report_df

bias_report_df(
    run_id=None,
    *,
    threshold_pct=0.05,
    tracking_signal_limit=4.0,
    limit=20000,
)

Audit every (ts_id, model_type) cell for directional bias and Trigg Tracking Signal on run_id.

Source code in src/scale_forecasting/sdk.py

def bias_report_df(
    self,
    run_id: str | None = None,
    *,
    threshold_pct: float = 0.05,
    tracking_signal_limit: float = 4.0,
    limit: int = 20000,
) -> pd.DataFrame:
    """Audit every ``(ts_id, model_type)`` cell for directional bias and Trigg Tracking Signal
    on ``run_id``.
    """
    import pandas as pd

    from .diagnostics import compute_bias_report
    from .registry.reads import read_oof

    rid = run_id or self.run_id
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else pd.DataFrame()
    return compute_bias_report(
        oof_df,
        threshold_pct=threshold_pct,
        tracking_signal_limit=tracking_signal_limit,
    )

calibration

calibration(run_id=None)

Return the review.CalibrationReport for run_id (point-forecast arm selection and per-step prediction-interval coverage).

Source code in src/scale_forecasting/sdk.py

def calibration(self, run_id: str | None = None) -> Any:
    """Return the `review.CalibrationReport` for ``run_id`` (point-forecast arm selection and
    per-step prediction-interval coverage).
    """
    from .review import calibration_report

    return calibration_report(run_id or self.run_id, settings=self._settings)

calibration_df

calibration_df(run_id=None, *, by=None)

Return (arms_df, coverage_df) DataFrames from calibration for run_id (or a single DataFrame when by='horizon' or by='arms').

Source code in src/scale_forecasting/sdk.py

def calibration_df(
    self,
    run_id: str | None = None,
    *,
    by: str | None = None,
) -> Any:
    """Return ``(arms_df, coverage_df)`` ``DataFrame``s from `calibration` for ``run_id`` (or a
    single ``DataFrame`` when ``by='horizon'`` or ``by='arms'``).
    """
    from .review import build_calibration_frames

    arms_df, cov_df = build_calibration_frames(self.calibration(run_id))
    if by == "horizon":
        return cov_df
    if by == "arms":
        return arms_df
    return arms_df, cov_df

cancel

cancel(run_id=None, job=None, *, confirm=False, reason='')

Cancel a run (or one family) — a preview by default; stop for real only on confirm.

Delegates to probes.cancel.cancel_run for run_id (default: this config's id): it probes the run to reconcile live state, then returns a CancelReport. Without confirm the report is a blast-radius preview — it touches no runtime and no registry, so you see exactly which jobs would stop and what partial data is retained. With confirm=True it stops each non-terminal family's runtime job, finalizes its registry row to CANCELLED (retaining — never deleting — landed partial results), records reason + the ADC actor for audit, and rolls the run header up. job narrows to one family (statistical/ml/deep_learning/native/ensemble); an already-terminal job is a no-op.

Source code in src/scale_forecasting/sdk.py

def cancel(
    self,
    run_id: str | None = None,
    job: str | None = None,
    *,
    confirm: bool = False,
    reason: str = "",
) -> Any:
    """Cancel a run (or one family) — a **preview by default**; stop for real only on confirm.

    Delegates to `probes.cancel.cancel_run` for ``run_id`` (default: this config's id): it
    probes the run to reconcile live state, then returns a `CancelReport`. Without ``confirm``
    the report is a blast-radius **preview** — it touches no runtime and no registry, so you see
    exactly which jobs would stop and what partial data is retained. With ``confirm=True`` it
    stops each non-terminal family's runtime job, finalizes its registry row to ``CANCELLED``
    (retaining — never deleting — landed partial results), records ``reason`` + the ADC actor
    for audit, and rolls the run header up. ``job`` narrows to one family
    (``statistical``/``ml``/``deep_learning``/``native``/``ensemble``); an already-terminal job
    is a no-op.
    """
    from .probes.cancel import cancel_run

    return cancel_run(
        run_id or self.run_id,
        job=job,
        confirm=confirm,
        reason=reason,
        settings=self._settings,
    )

cohorts_df

cohorts_df(run_id=None)

Return a DataFrame of backtest cohort coverage, achieved fold histograms, refit modes, staleness_gap, and holdout-fold pooled_wape for run_id.

Source code in src/scale_forecasting/sdk.py

def cohorts_df(self, run_id: str | None = None) -> pd.DataFrame:
    """Return a ``DataFrame`` of backtest cohort coverage, achieved fold histograms, refit
    modes, ``staleness_gap``, and holdout-fold ``pooled_wape`` for ``run_id``.
    """
    from .review import build_cohorts_frame

    return build_cohorts_frame(self.review_run(run_id))

dag

dag()

The planned execution DAG for this config — pure, offline, no GCP.

Resolves the config into its nodes (dag.dag_nodes): one per model family plus the ensemble (when enabled), each with the deterministic job_key it will run under, its resolved runtime/hardware, and its upstream dependencies. Because job_key\ s are derived from the config alone, this gives every job's identity before the run — the offline counterpart to jobs, so a caller can line up planned nodes against executed rows by job_key.

Planning never refuses — this stays total, so inspecting a config is always safe — but it does log dag.gpu_usefulness_report, because someone reading the planned hardware is exactly the person who should be told a device will be billed and barely used.

Source code in src/scale_forecasting/sdk.py

def dag(self) -> tuple[DagNode, ...]:
    """The planned execution DAG for this config — pure, offline, no GCP.

    Resolves the config into its nodes (`dag.dag_nodes`): one per model family plus the ensemble
    (when enabled), each with the deterministic ``job_key`` it will run under, its resolved
    runtime/hardware, and its upstream dependencies. Because ``job_key``\\ s are derived from
    the config alone, this gives every job's identity *before* the run — the offline counterpart
    to `jobs`, so a caller can line up planned nodes against executed rows by ``job_key``.

    Planning never refuses — this stays total, so inspecting a config is always safe — but it
    does log `dag.gpu_usefulness_report`, because someone reading the planned hardware is
    exactly the person who should be told a device will be billed and barely used.
    """
    from .dag import dag_nodes, gpu_usefulness_report, plan_dag

    run_dag = plan_dag(self._config)
    for line in gpu_usefulness_report(self._config, run_dag.jobs):
        _log.warning("%s", line)
    return dag_nodes(run_dag)

detect_anomalies_df

detect_anomalies_df(
    run_id=None,
    *,
    model_name=None,
    anomaly_prob_threshold=0.95,
    cutoff_back_steps=None,
    anomalies_only=False,
    actuals_df=None,
    engine="auto",
    limit=50000,
)

Detect time-series anomalies across any of the 37 models, 9 ensembles, or past runs.

Supports four complementary execution paths controlled by model_name and engine:

1. In-database BigQuery ML (engine="auto" or "bqml" with arima_plus / timesfm): Runs ML.DETECT_ANOMALIES(MODEL ...) for arima_plus / arima_plus@xreg or serverless AI.DETECT_ANOMALIES (TimesFM) directly in BigQuery. If inspecting an older past run_id whose BQML model object was already dropped and engine="auto", automatically falls back to the stored backtest_oof intervals.
2. Universal Model / Ensemble OOF (model_name="<any model>", "ensemble_*", "all", or engine="oof"): Scores stored backtest_oof out-of-fold predictions and calibrated prediction intervals for any statistical, ML, deep-learning, Vertex AI AutoML, BigQuery ML, or ensemble model on current or past runs via :func:diagnostics.compute_anomalies — zero re-fitting required.
3. Forward Forecast Realized Actuals (actuals_df=...): Joins published forecast_predictions intervals against newly arrived actuals_df to flag live post-cutoff anomalies.
4. Model-Free STL + Robust MAD (model_name="stl" or engine="stl"): Decomposes raw historical series when no backtest OOF intervals are available.

Returns a tidy DataFrame with columns ["ts_id", "model_type", "ds", "y", "yhat", "lower_bound", "upper_bound", "anomaly_probability", "is_anomaly"].

Source code in src/scale_forecasting/sdk.py

def detect_anomalies_df(
    self,
    run_id: str | None = None,
    *,
    model_name: str | None = None,
    anomaly_prob_threshold: float = 0.95,
    cutoff_back_steps: int | None = None,
    anomalies_only: bool = False,
    actuals_df: pd.DataFrame | None = None,
    engine: str = "auto",
    limit: int = 50000,
) -> pd.DataFrame:
    """Detect time-series anomalies across any of the 37 models, 9 ensembles, or past runs.

    Supports four complementary execution paths controlled by ``model_name`` and ``engine``:

    1. **In-database BigQuery ML (`engine="auto"` or `"bqml"` with `arima_plus` / `timesfm`):**
       Runs `ML.DETECT_ANOMALIES(MODEL ...)` for `arima_plus` / `arima_plus@xreg` or serverless
       `AI.DETECT_ANOMALIES` (`TimesFM`) directly in BigQuery. If inspecting an older past
       ``run_id`` whose BQML model object was already dropped and ``engine="auto"``,
       automatically falls back to the stored ``backtest_oof`` intervals.
    2. **Universal Model / Ensemble OOF (`model_name="<any model>"`, `"ensemble_*"`, `"all"`,
       or `engine="oof"`):** Scores stored ``backtest_oof`` out-of-fold predictions and
       calibrated prediction intervals for **any** statistical, ML, deep-learning, Vertex AI
       AutoML, BigQuery ML, or ensemble model on current or past runs via
       :func:`diagnostics.compute_anomalies` — zero re-fitting required.
    3. **Forward Forecast Realized Actuals (`actuals_df=...`):** Joins published
       ``forecast_predictions`` intervals against newly arrived ``actuals_df`` to flag live
       post-cutoff anomalies.
    4. **Model-Free STL + Robust MAD (`model_name="stl"` or `engine="stl"`):** Decomposes raw
       historical series when no backtest OOF intervals are available.

    Returns a tidy ``DataFrame`` with columns ``["ts_id", "model_type", "ds", "y", "yhat",
    "lower_bound", "upper_bound", "anomaly_probability", "is_anomaly"]``.
    """
    import pandas as pd

    from .diagnostics import compute_anomalies
    from .registry.reads import (
        read_oof,
        read_predictions,
        read_series_history,
        resolve_settings,
    )

    rid = run_id or self.run_id
    eff_cfg = self._effective_config_for_run(rid)
    conf_level = 0.80
    freq = str(eff_cfg.data.freq)

    # Mode 3: Forward predictions scored against newly arrived actuals_df.
    if actuals_df is not None:
        model_filter = [model_name] if model_name and model_name not in ("all", "stl") else None
        pred_rows = read_predictions(
            rid, models=model_filter, limit=limit, settings=self._settings
        )
        oof_rows = read_oof(rid, models=model_filter, limit=limit, settings=self._settings)
        pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
        oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else None
        return compute_anomalies(
            predictions_df=pred_df,
            actuals_df=actuals_df,
            oof_df=oof_df,
            model_name=model_name,
            anomaly_prob_threshold=anomaly_prob_threshold,
            confidence_level=conf_level,
            anomalies_only=anomalies_only,
            freq=freq,
        )

    # Mode 4: Explicit model-free STL decomposition over historical series.
    if engine == "stl" or model_name == "stl":
        hist_rows = read_series_history(
            rid, history_tail=cutoff_back_steps or 1000, settings=self._settings
        )
        hist_df = pd.DataFrame.from_records(hist_rows) if hist_rows else pd.DataFrame()
        return compute_anomalies(
            series_df=hist_df,
            model_name="stl",
            anomaly_prob_threshold=anomaly_prob_threshold,
            confidence_level=conf_level,
            anomalies_only=anomalies_only,
            freq=freq,
        )

    # Resolve default model_name when omitted.
    eff_model = model_name
    if eff_model is None:
        if engine == "bqml":
            native_candidates = [
                m
                for m in eff_cfg.models
                if m.split("@", 1)[0] in ("arima_plus", "arima_plus_xreg", "timesfm")
            ]
            eff_model = native_candidates[0] if native_candidates else "arima_plus"
        elif (
            engine == "auto"
            and len(eff_cfg.models) == 1
            and eff_cfg.models[0].split("@", 1)[0]
            in ("arima_plus", "arima_plus_xreg", "timesfm")
            and not eff_cfg.ensemble.enabled
        ):
            eff_model = eff_cfg.models[0]

    base_requested = eff_model.split("@", 1)[0] if eff_model else ""
    if base_requested == "arima_plus_xreg":
        eff_model = "arima_plus@xreg"
        base_requested = "arima_plus"

    # Mode 1: Native BigQuery ML.DETECT_ANOMALIES / AI.DETECT_ANOMALIES when requested.
    if (
        engine in ("auto", "bqml")
        and eff_model is not None
        and base_requested in ("arima_plus", "timesfm")
    ):
        from google.cloud import bigquery

        from .engines.bigquery_sql import build_detect_anomalies_query

        resolved = resolve_settings(self._settings)
        sql = build_detect_anomalies_query(
            eff_cfg,
            model_name=eff_model,
            dataset=resolved.dataset_ref,
            registry_dataset=resolved.registry_dataset_ref,
            anomaly_prob_threshold=anomaly_prob_threshold,
            cutoff_back_steps=cutoff_back_steps,
            run_id=rid,
        )
        client = bigquery_client(resolved.project_id)
        try:
            df = client.query(
                sql, job_config=bigquery.QueryJobConfig(use_query_cache=False)
            ).to_dataframe()
            if not df.empty:
                if "model_type" not in df.columns:
                    df.insert(1, "model_type", eff_model)
                if (
                    "yhat" not in df.columns
                    and "lower_bound" in df.columns
                    and "upper_bound" in df.columns
                ):
                    df.insert(
                        4,
                        "yhat",
                        0.5
                        * (
                            pd.to_numeric(df["lower_bound"], errors="coerce")
                            + pd.to_numeric(df["upper_bound"], errors="coerce")
                        ),
                    )
                ordered = [
                    c
                    for c in (
                        "ts_id",
                        "model_type",
                        "ds",
                        "y",
                        "yhat",
                        "lower_bound",
                        "upper_bound",
                        "anomaly_probability",
                        "is_anomaly",
                    )
                    if c in df.columns
                ]
                df = df[ordered + [c for c in df.columns if c not in ordered]]
            if anomalies_only and not df.empty and "is_anomaly" in df.columns:
                return df.loc[df["is_anomaly"] == True].reset_index(drop=True)  # noqa: E712 - nullable boolean series
            return df
        except Exception:
            if engine == "bqml":
                raise
            # Under engine="auto", if a past run's BQML model artifact was dropped, fall
            # through to stored backtest_oof below.

    # Mode 2: Universal backtest_oof residual & interval scoring across any model / ensemble.
    model_filter = [eff_model] if eff_model and eff_model != "all" else None
    oof_rows = read_oof(rid, models=model_filter, limit=limit, settings=self._settings)
    if oof_rows:
        oof_df = pd.DataFrame.from_records(oof_rows)
        if cutoff_back_steps is not None and cutoff_back_steps > 0:
            oof_df = (
                oof_df.sort_values(["ts_id", "model_type", "forecast_date"])
                .groupby(["ts_id", "model_type"], as_index=False)
                .tail(int(cutoff_back_steps))
            )
        return compute_anomalies(
            oof_df=oof_df,
            model_name=eff_model,
            anomaly_prob_threshold=anomaly_prob_threshold,
            confidence_level=conf_level,
            anomalies_only=anomalies_only,
            freq=freq,
        )

    # Fallback when a run had backtest.enabled=False: use STL over historical series.
    hist_rows = read_series_history(
        rid, history_tail=cutoff_back_steps or 1000, settings=self._settings
    )
    hist_df = pd.DataFrame.from_records(hist_rows) if hist_rows else pd.DataFrame()
    return compute_anomalies(
        series_df=hist_df,
        model_name=eff_model or "stl",
        anomaly_prob_threshold=anomaly_prob_threshold,
        confidence_level=conf_level,
        anomalies_only=anomalies_only,
        freq=freq,
    )

detect_change_points_df

detect_change_points_df(
    run_id=None,
    *,
    model_name=None,
    exclude_gap_edges=True,
    min_gap_steps=1,
    sigma_threshold=3.0,
    engine="auto",
    limit=50000,
)

Detect structural regime change points across raw series or any model's residuals on current or past runs.

• Raw series mode (model_name=None or "raw"): Runs BigQuery's native ML.DETECT_CHANGE_POINTS with gap-edge artifact filtering when engine in ("auto", "bqml"), or pure-Python binary segmentation (:func:diagnostics.compute_change_points) when engine="oof".
• Model-residual mode (model_name="<model>", "ensemble_*", or "all"): Reads stored backtest_oof trajectories for run_id and segments the model's residuals $e_t = y_t - \hat{y}_t$ via :func:diagnostics.compute_change_points to pinpoint structural regime shifts that the model failed to anticipate (distinguishing unmodeled structural breaks from seasonal or promotional jumps the model already explained).

Returns a tidy DataFrame with columns ["ts_id", "model_type", "begin_date", "end_date", "window_count", "window_avg", "window_min", "window_max", "window_stddev", "gap_edge_missing_steps", "is_gap_edge", "status"].

Source code in src/scale_forecasting/sdk.py

def detect_change_points_df(
    self,
    run_id: str | None = None,
    *,
    model_name: str | None = None,
    exclude_gap_edges: bool = True,
    min_gap_steps: int = 1,
    sigma_threshold: float = 3.0,
    engine: str = "auto",
    limit: int = 50000,
) -> pd.DataFrame:
    """Detect structural regime change points across raw series or any model's residuals on
    current or past runs.

    * **Raw series mode (`model_name=None` or `"raw"`):** Runs BigQuery's native
      `ML.DETECT_CHANGE_POINTS` with gap-edge artifact filtering when ``engine in ("auto",
      "bqml")``, or pure-Python binary segmentation (:func:`diagnostics.compute_change_points`)
      when ``engine="oof"``.
    * **Model-residual mode (`model_name="<model>"`, `"ensemble_*"`, or `"all"`):** Reads
      stored ``backtest_oof`` trajectories for ``run_id`` and segments the model's residuals
      $e_t = y_t - \\hat{y}_t$ via :func:`diagnostics.compute_change_points` to pinpoint
      structural regime shifts that the model failed to anticipate (distinguishing unmodeled
      structural breaks from seasonal or promotional jumps the model already explained).

    Returns a tidy ``DataFrame`` with columns ``["ts_id", "model_type", "begin_date",
    "end_date", "window_count", "window_avg", "window_min", "window_max", "window_stddev",
    "gap_edge_missing_steps", "is_gap_edge", "status"]``.
    """
    import pandas as pd

    from .diagnostics import compute_change_points
    from .registry.reads import read_oof, read_series_history, resolve_settings

    rid = run_id or self.run_id
    eff_cfg = self._effective_config_for_run(rid)
    freq = str(eff_cfg.data.freq)

    # Model-residual mode: segment backtest_oof residuals for any model or ensemble.
    if model_name is not None and model_name != "raw":
        eff_model = "arima_plus@xreg" if model_name == "arima_plus_xreg" else model_name
        model_filter = [eff_model] if eff_model != "all" else None
        oof_rows = read_oof(rid, models=model_filter, limit=limit, settings=self._settings)
        oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else pd.DataFrame()
        return compute_change_points(
            oof_df=oof_df,
            model_name=eff_model,
            exclude_gap_edges=exclude_gap_edges,
            min_gap_steps=min_gap_steps,
            freq=freq,
            sigma_threshold=sigma_threshold,
        )

    # Raw series mode: BigQuery ML.DETECT_CHANGE_POINTS when engine in ("auto", "bqml").
    if engine in ("auto", "bqml"):
        from google.cloud import bigquery

        from .engines.bigquery_sql import build_detect_change_points_query

        resolved = resolve_settings(self._settings)
        sql = build_detect_change_points_query(
            eff_cfg,
            dataset=resolved.dataset_ref,
            exclude_gap_edges=exclude_gap_edges,
            min_gap_steps=min_gap_steps,
        )
        client = bigquery_client(resolved.project_id)
        try:
            df = client.query(
                sql, job_config=bigquery.QueryJobConfig(use_query_cache=False)
            ).to_dataframe()
            if "model_type" not in df.columns:
                df.insert(1 if "ts_id" in df.columns else 0, "model_type", "raw")
            return df
        except Exception:
            if engine == "bqml":
                raise

    # Offline / fallback raw series change-point segmentation from stored history.
    hist_rows = read_series_history(rid, history_tail=2000, settings=self._settings)
    hist_df = pd.DataFrame.from_records(hist_rows) if hist_rows else pd.DataFrame()
    return compute_change_points(
        series_df=hist_df,
        model_name="raw",
        exclude_gap_edges=exclude_gap_edges,
        min_gap_steps=min_gap_steps,
        freq=freq,
        sigma_threshold=sigma_threshold,
    )

dry_run

dry_run()

Validate the config and report the planned fan-out + runtime split — touches no GCP.

Delegates the run_id to main.run (dry_run=True) so it is the single source of truth, then adds the estimated fanout and the python/BigQuery model split.

Source code in src/scale_forecasting/sdk.py

def dry_run(self) -> DryRunResult:
    """Validate the config and report the planned fan-out + runtime split — touches no GCP.

    Delegates the run_id to `main.run` (``dry_run=True``) so it is the single source of
    truth, then adds the estimated fanout and the python/BigQuery model split.
    """
    from . import main

    run_id = main.run(self._config, dry_run=True)
    python_models, bq_models = split_by_runtime(self._config)
    workload = estimate_workload(self._config)
    return DryRunResult(
        run_id=run_id,
        fanout=Fanout(workload.n_series, workload.n_models, workload.n_folds, workload.n_cells),
        python_models=python_models,
        bq_models=bq_models,
        workload=workload,
    )

emit_airflow

emit_airflow(
    config_uri=None, *, dag_id=None, with_retry=False
)

Render this config's Airflow DAG as a dag_<run_id>.py source string — pure, offline.

The Composer counterpart to dag: instead of the node list, it returns a standalone Airflow DAG file (airflow_emit.emit_airflow_dag) whose tasks reproduce this exact run — one job per family on its resolved runtime, the ensemble node, and (when the run has several ephemeral Ray/Dataproc-cluster families) the shared-cluster bracket — all under the same deterministic run_id. config_uri is embedded verbatim as the DAG's CONFIG_URI (what each task loads the config from); pass the gs:// URI of a staged config for a Composer-runnable DAG, or omit it to reference this config's own run_id file under the convention (runs/<run_id>.json). dag_id overrides the default scale_forecasting_<run_id>. Touches no GCP — write the returned string to the Airflow DAGs folder (or stage it with staging.stage_dag).

with_retry adds the repair node between the family join and the ensemble — the unattended form of retry, so a scheduled run recovers what it can before anything reads it. Off by default, and an argument rather than a config field: config fields feed the run_id digest, and repairing on a schedule must not be a different run from repairing by hand. A microbatch ensemble gets no node even when asked (it has already drained by then); airflow_emit.emit_airflow_dag logs when it declines.

Source code in src/scale_forecasting/sdk.py

def emit_airflow(
    self,
    config_uri: str | None = None,
    *,
    dag_id: str | None = None,
    with_retry: bool = False,
) -> str:
    """Render this config's Airflow DAG as a ``dag_<run_id>.py`` source string — pure, offline.

    The Composer counterpart to `dag`: instead of the node list, it returns a standalone Airflow
    DAG file (`airflow_emit.emit_airflow_dag`) whose tasks reproduce this exact run — one job
    per family on its resolved runtime, the ensemble node, and (when the run has several
    ephemeral Ray/Dataproc-cluster families) the shared-cluster bracket — all under the same
    deterministic ``run_id``. ``config_uri`` is embedded verbatim as the DAG's ``CONFIG_URI``
    (what each task loads the config from); pass the ``gs://`` URI of a staged config for a
    Composer-runnable DAG, or omit it to reference this config's own `run_id` file under the
    convention (``runs/<run_id>.json``). ``dag_id`` overrides the default
    ``scale_forecasting_<run_id>``. Touches no GCP — write the returned string to the Airflow
    DAGs folder (or stage it with `staging.stage_dag`).

    ``with_retry`` adds the repair node between the family join and the ensemble — the
    unattended form of `retry`, so a scheduled run recovers what it can before anything reads
    it. Off by default, and an argument rather than a config field: config fields feed the
    ``run_id`` digest, and repairing on a schedule must not be a different run from repairing by
    hand. A microbatch ensemble gets no node even when asked (it has already drained by then);
    `airflow_emit.emit_airflow_dag` logs when it declines.
    """
    from .airflow_emit import emit_airflow_dag

    uri = config_uri if config_uri is not None else f"runs/{self.run_id}.json"
    return emit_airflow_dag(self._config, uri, dag_id=dag_id, with_retry=with_retry)

ensemble_runs

ensemble_runs(
    source_run_ids, strategies=None, *, target_run_id=None
)

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

Convenience forwarder to Registry.ensemble_runs (ensemble_run.ensemble_cross_runs). If source_run_ids contains a single run id distinct from self.run_id, self.run_id is automatically prepended so forecaster_b.ensemble_runs([run_a_id]) combines run_b with run_a.

Source code in src/scale_forecasting/sdk.py

def ensemble_runs(
    self,
    source_run_ids: Sequence[str],
    strategies: list[str] | None = None,
    *,
    target_run_id: str | None = None,
) -> str:
    """Blend base models across multiple completed runs into a consolidated cross-run ensemble;
    return ``target_run_id``.

    Convenience forwarder to `Registry.ensemble_runs` (`ensemble_run.ensemble_cross_runs`). If
    ``source_run_ids`` contains a single run id distinct from ``self.run_id``, ``self.run_id``
    is automatically prepended so ``forecaster_b.ensemble_runs([run_a_id])`` combines ``run_b``
    with ``run_a``.
    """
    ids = list(source_run_ids)
    if len(ids) == 1 and ids[0] != self.run_id:
        ids = [self.run_id, ids[0]]
    return self.registry().ensemble_runs(
        ids, strategies=strategies, target_run_id=target_run_id
    )

ensemble_weights_df

ensemble_weights_df(run_id=None, *, limit=5000)

Return a long-form DataFrame of learned ensemble stacking weights (nnls, ridge, xgb) unpacked from forecast_metadata.best_params for run_id.

Source code in src/scale_forecasting/sdk.py

def ensemble_weights_df(self, run_id: str | None = None, *, limit: int = 5000) -> pd.DataFrame:
    """Return a long-form ``DataFrame`` of learned ensemble stacking weights (`nnls`, `ridge`,
    `xgb`) unpacked from ``forecast_metadata.best_params`` for ``run_id``.
    """
    from .registry.reads import read_best_params
    from .review import build_ensemble_weights_frame

    rid = run_id or self.run_id
    return build_ensemble_weights_frame(
        read_best_params(rid, limit=limit, settings=self._settings)
    )

exceptions_df

exceptions_df(
    run_id=None,
    *,
    actuals_df=None,
    prior_run_ids=None,
    top_n=50,
    limit=20000,
)

Build the Top-N operational exception triage queue for run_id.

Source code in src/scale_forecasting/sdk.py

def exceptions_df(
    self,
    run_id: str | None = None,
    *,
    actuals_df: pd.DataFrame | None = None,
    prior_run_ids: Sequence[str] | None = None,
    top_n: int = 50,
    limit: int = 20000,
) -> pd.DataFrame:
    """Build the Top-N operational exception triage queue for ``run_id``."""
    import pandas as pd

    from .diagnostics import compute_exceptions_queue
    from .registry.reads import read_oof, read_predictions

    rid = run_id or self.run_id
    pred_rows = read_predictions(rid, limit=limit, settings=self._settings)
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else pd.DataFrame()

    real_df = (
        self.score_realized(actuals_df, run_id=rid, limit=limit)
        if actuals_df is not None and not actuals_df.empty
        else None
    )
    stab_df = (
        self.revision_stability_df([*prior_run_ids, rid], limit=limit)
        if prior_run_ids
        else None
    )
    return compute_exceptions_queue(
        oof_df=oof_df,
        predictions_df=pred_df,
        realized_df=real_df,
        stability_df=stab_df,
        floor=self._config.output.floor,
        cap=self._config.output.cap,
        metric=self._config.backtest.decision_metric,
        top_n=top_n,
    )

explain

explain(*, validate_data=False)

Return a per-family execution plan DataFrame for this config — pure, offline.

Each row represents one node in the planned execution DAG (dag), detailing the resolved runtime, machine_type, workers, hardware, gpu_type, selected models, per-model training_modes (local / global / hybrid), expected cell count, active covariate tiers, and upstream depends_on keys. validate_data is accepted for parity with dry-run workflows (when False, stays 100% offline).

Source code in src/scale_forecasting/sdk.py

def explain(self, *, validate_data: bool = False) -> pd.DataFrame:
    """Return a per-family execution plan ``DataFrame`` for this config — pure, offline.

    Each row represents one node in the planned execution DAG (`dag`), detailing the resolved
    ``runtime``, ``machine_type``, ``workers``, ``hardware``, ``gpu_type``, selected ``models``,
    per-model ``training_modes`` (``local`` / ``global`` / ``hybrid``), expected cell count,
    active covariate tiers, and upstream ``depends_on`` keys. ``validate_data`` is accepted for
    parity with dry-run workflows (when ``False``, stays 100% offline).
    """
    _ = validate_data
    return build_explain_frame(self._config)

explain_forecast

explain_forecast(
    run_id=None,
    *,
    ts_id=None,
    model_type=None,
    seasonal_period=7,
    history_tail=120,
)

Decompose a series' historical trajectory, OOF backtest, and future forecast into structural trend, level shift, seasonal cycle, exogenous covariate attribution, and OOF residuals / interval width for run_id.

Source code in src/scale_forecasting/sdk.py

def explain_forecast(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | None = None,
    model_type: str | None = None,
    seasonal_period: int = 7,
    history_tail: int = 120,
) -> pd.DataFrame:
    """Decompose a series' historical trajectory, OOF backtest, and future forecast into
    structural trend, level shift, seasonal cycle, exogenous covariate attribution, and OOF
    residuals / interval width for ``run_id``.
    """
    import pandas as pd

    from .registry.reads import read_series_covariates
    from .review import explain_forecast_frame

    rid = run_id or self.run_id
    pred_df = self.predictions_df(
        rid,
        ts_id=ts_id,
        models=[model_type] if model_type is not None else None,
        include_history=True,
        include_oof=True,
        history_tail=history_tail,
        series_limit=1 if ts_id is None else None,
    )
    if pred_df.empty:
        return explain_forecast_frame(pred_df)
    chosen_ts = ts_id or str(pred_df["ts_id"].iloc[0])
    cov_rows = read_series_covariates(
        rid, ts_id=chosen_ts, history_tail=history_tail, settings=self._settings
    )
    cov_df = pd.DataFrame.from_records(cov_rows) if cov_rows else None
    return explain_forecast_frame(
        pred_df,
        ts_id=chosen_ts,
        model_type=model_type,
        seasonal_period=seasonal_period,
        covariate_df=cov_df,
    )

export_report

export_report(
    out_path, *, run_id=None, title=None, write_json=False
)

Export a self-contained .html report for run_id.

Source code in src/scale_forecasting/sdk.py

def export_report(
    self,
    out_path: str | Any,
    *,
    run_id: str | None = None,
    title: str | None = None,
    write_json: bool = False,
) -> Any:
    """Export a self-contained `.html` report for ``run_id``."""
    from .report import export_report

    rid = run_id or self.run_id
    payload_dict = {
        "run_id": rid,
        "leaderboard_df": self.leaderboard_df(rid),
        "fva_df": self.fva_df(rid),
        "horizon_decay_df": self.horizon_decay_df(rid),
        "bias_df": self.bias_report_df(rid),
        "exceptions_df": self.exceptions_df(rid),
        "inventory_policy_df": self.inventory_policy_df(rid),
        "predictions_df": self.predictions_df(rid),
    }
    return export_report(
        payload_dict,
        out_path,
        title=title,
        metric=self._config.backtest.decision_metric,
        write_json=write_json,
    )

feasibility

feasibility()

Read the source panel's series lengths and report what this run's fold geometry does.

The SDK face of --dry-run --feasibility. Reads one aggregation off the panel and returns the report as lines; launches nothing and writes nothing. See launch_plan.feasibility_report for why this one planning question cannot be answered offline.

Source code in src/scale_forecasting/sdk.py

def feasibility(self) -> list[str]:
    """Read the source panel's series lengths and report what this run's fold geometry does.

    The SDK face of ``--dry-run --feasibility``. Reads one aggregation off the panel and returns
    the report as lines; launches nothing and writes nothing. See
    `launch_plan.feasibility_report` for why this one planning question cannot be answered
    offline.
    """
    from .launch_plan import feasibility_report

    return feasibility_report(self._config, settings=self._settings)

from_dict classmethod

from_dict(data, *, settings=None)

Build from an already-parsed config dict (validates via RunConfig, raising ConfigError on a schema violation).

Source code in src/scale_forecasting/sdk.py

@classmethod
def from_dict(cls, data: dict[str, Any], *, settings: Settings | None = None) -> Forecaster:
    """Build from an already-parsed config dict (validates via `RunConfig`, raising
    `ConfigError` on a schema violation)."""
    try:
        cfg = RunConfig.model_validate(data)
    except Exception as exc:  # pydantic ValidationError → the package's ConfigError
        from .errors import ConfigError

        raise ConfigError(f"invalid config: {exc}") from exc
    return cls(cfg, settings=settings)

from_file classmethod

from_file(path, *, settings=None)

Build from a JSON config file (delegates to load_config, which raises ConfigError on a bad file).

Source code in src/scale_forecasting/sdk.py

@classmethod
def from_file(cls, path: str | Path, *, settings: Settings | None = None) -> Forecaster:
    """Build from a JSON config file (delegates to
    `load_config`, which raises
    `ConfigError` on a bad file)."""
    return cls(load_config(path), settings=settings)

from_run_id classmethod

from_run_id(run_id, *, settings=None)

Build from a run that already happened, by reading its config back out of the registry.

The third way in, for the operator who inherits a broken run: they have its id, from a pager or a dashboard or a ledger row, and not its config file. The run's header carries the config it landed under, so this recovers a Forecaster identical to the one that launched it — and every method here then works as usual, retry included.

Delegates to retry_run.config_for_run, which refuses loudly rather than returning something plausible: no stored config, a config that no longer validates, or a config whose own digest is not run_id each raise errors.ConfigError naming the id. That matters most for retry, where every one of those failures has a wrong answer available that looks like a small repair.

This is why retry still takes no run_id of its own. A repair is planned from a config; what an id buys you is the config, not a second way to plan.

Source code in src/scale_forecasting/sdk.py

@classmethod
def from_run_id(cls, run_id: str, *, settings: Settings | None = None) -> Forecaster:
    """Build from a run that already happened, by reading its config back out of the registry.

    The third way in, for the operator who inherits a broken run: they have its id, from a
    pager or a dashboard or a ledger row, and not its config file. The run's header carries the
    config it landed under, so this recovers a `Forecaster` identical to the one that launched
    it — and every method here then works as usual, `retry` included.

    Delegates to `retry_run.config_for_run`, which **refuses loudly rather than returning
    something plausible**: no stored config, a config that no longer validates, or a config
    whose own digest is not ``run_id`` each raise `errors.ConfigError` naming the id. That
    matters most for `retry`, where every one of those failures has a wrong answer available
    that looks like a small repair.

    This is why `retry` still takes no ``run_id`` of its own. A repair is planned from a config;
    what an id buys you is the config, not a second way to plan.
    """
    from .retry_run import config_for_run

    return cls(config_for_run(run_id, settings=settings), settings=settings)

fva_df

fva_df(
    run_id=None, *, baseline=None, metric=None, limit=20000
)

Compute Forecast Value Added (FVA) & Diebold–Mariano statistical significance for run_id (diagnostics.compute_fva).

Source code in src/scale_forecasting/sdk.py

def fva_df(
    self,
    run_id: str | None = None,
    *,
    baseline: str | None = None,
    metric: str | None = None,
    limit: int = 20000,
) -> pd.DataFrame:
    """Compute Forecast Value Added (FVA) & Diebold–Mariano statistical significance for
    ``run_id`` (`diagnostics.compute_fva`).
    """
    import pandas as pd

    from .diagnostics import compute_fva
    from .registry.reads import read_oof

    rid = run_id or self.run_id
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else pd.DataFrame()
    eff_metric = metric or self._config.backtest.decision_metric
    return compute_fva(oof_df, baseline=baseline, metric=eff_metric)

hierarchy_df

hierarchy_df(run_id=None, *, limit=20000)

Summarize hierarchical predictions by level (total, aggregate, bottom) and verify additive coherence ($\max_t |\hat{y}{\text{total},t} - \sum_b \hat{y}|$) for run_id.

Source code in src/scale_forecasting/sdk.py

def hierarchy_df(self, run_id: str | None = None, *, limit: int = 20000) -> pd.DataFrame:
    """Summarize hierarchical predictions by level (``total``, ``aggregate``, ``bottom``) and
    verify additive coherence ($\\max_t |\\hat{y}_{\\text{total},t} - \\sum_b \\hat{y}_{b,t}|$)
    for ``run_id``.
    """
    from .registry.reads import read_predictions
    from .review import build_hierarchy_frame

    rid = run_id or self.run_id
    rows = read_predictions(rid, limit=limit, settings=self._settings)
    return build_hierarchy_frame(rows)

horizon_decay_df

horizon_decay_df(
    run_id=None, *, metric=None, baseline=None, limit=20000
)

Compute lead-time accuracy decay across horizon steps ($h = 1 \dots H$).

Source code in src/scale_forecasting/sdk.py

def horizon_decay_df(
    self,
    run_id: str | None = None,
    *,
    metric: str | None = None,
    baseline: str | None = None,
    limit: int = 20000,
) -> pd.DataFrame:
    """Compute lead-time accuracy decay across horizon steps ($h = 1 \\dots H$)."""
    import pandas as pd

    from .diagnostics import compute_horizon_decay
    from .registry.reads import read_oof

    rid = run_id or self.run_id
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else pd.DataFrame()
    eff_metric = metric or self._config.backtest.decision_metric
    return compute_horizon_decay(oof_df, metric=eff_metric, baseline=baseline)

inventory_policy_df

inventory_policy_df(
    run_id=None,
    *,
    inventory_df=None,
    model_name=None,
    model_type=None,
    lead_time=7.0,
    lead_time_std=0.0,
    review_period=0.0,
    service_level=0.95,
    fill_rate=None,
    on_hand=0.0,
    on_order=0.0,
    backorders=0.0,
    holding_cost=1.0,
    stockout_cost=9.0,
    min_order_qty=0.0,
    order_multiple=0.0,
    limit=20000,
)

Translate run_id calibrated forecasts into operational inventory & replenishment policies (diagnostics.compute_inventory_policy).

Computes continuous-review $(s, S)$ and periodic-review $(R, S)$ Reorder Points (ROP), Safety Stock (SS), Order-Up-To Levels ($S$), Recommended Order Quantities ($Q$), and Expected Holding vs. Stockout Cost tables using AR(1) multi-step forecast error covariance ($\hat{\rho}_1$ from backtest_oof) and supplier lead-time variability ($\mu_L, \sigma_L$).

Source code in src/scale_forecasting/sdk.py

def inventory_policy_df(
    self,
    run_id: str | None = None,
    *,
    inventory_df: pd.DataFrame | Mapping[str, Mapping[str, Any]] | None = None,
    model_name: str | None = None,
    model_type: str | None = None,
    lead_time: float = 7.0,
    lead_time_std: float = 0.0,
    review_period: float = 0.0,
    service_level: float | None = 0.95,
    fill_rate: float | None = None,
    on_hand: float = 0.0,
    on_order: float = 0.0,
    backorders: float = 0.0,
    holding_cost: float = 1.0,
    stockout_cost: float = 9.0,
    min_order_qty: float = 0.0,
    order_multiple: float = 0.0,
    limit: int = 20000,
) -> pd.DataFrame:
    """Translate ``run_id`` calibrated forecasts into operational inventory & replenishment
    policies (`diagnostics.compute_inventory_policy`).

    Computes continuous-review $(s, S)$ and periodic-review $(R, S)$ **Reorder Points
    (`ROP`)**, **Safety Stock (`SS`)**, **Order-Up-To Levels ($S$)**, **Recommended Order
    Quantities ($Q$)**, and **Expected Holding vs. Stockout Cost** tables using AR(1)
    multi-step forecast error covariance ($\\hat{\\rho}_1$ from ``backtest_oof``) and supplier
    lead-time variability ($\\mu_L, \\sigma_L$).
    """
    import pandas as pd

    from .diagnostics import compute_inventory_policy
    from .registry.reads import read_oof, read_predictions

    rid = run_id or self.run_id
    eff_model = model_name or model_type
    model_filter = [eff_model] if eff_model and eff_model not in ("all", "*") else None
    pred_rows = read_predictions(rid, models=model_filter, limit=limit, settings=self._settings)
    oof_rows = read_oof(rid, models=model_filter, limit=limit, settings=self._settings)
    pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else None
    return compute_inventory_policy(
        pred_df,
        oof_df=oof_df,
        inventory_df=inventory_df,
        model_name=eff_model,
        lead_time=lead_time,
        lead_time_std=lead_time_std,
        review_period=review_period,
        service_level=service_level,
        fill_rate=fill_rate,
        on_hand=on_hand,
        on_order=on_order,
        backorders=backorders,
        holding_cost=holding_cost,
        stockout_cost=stockout_cost,
        min_order_qty=min_order_qty,
        order_multiple=order_multiple,
    )

jobs

jobs(run_id=None)

The per-job cross-system trace for a run — one JobTrace per family, plus the ensemble.

Reads v_run_jobs (registry.jobs.read_run_jobs) for run_id (default: this config's id): the authoritative map from each family's canonical job_key to the platform job that actually ran it (its system_job_id on Spark/Ray/BigQuery) and how it fared. Returns [] when the run has no jobs yet. This is the "where did each family run, and can I go look at that platform's job?" read — the executed counterpart to the offline dag.

Source code in src/scale_forecasting/sdk.py

def jobs(self, run_id: str | None = None) -> list[JobTrace]:
    """The per-job cross-system trace for a run — one `JobTrace` per family, plus the ensemble.

    Reads ``v_run_jobs`` (`registry.jobs.read_run_jobs`) for ``run_id`` (default: this config's
    id): the authoritative map from each family's canonical ``job_key`` to the platform job that
    actually ran it (its ``system_job_id`` on Spark/Ray/BigQuery) and how it fared. Returns
    ``[]`` when the run has no jobs yet. This is the "where did each family run, and can I go
    look at that platform's job?" read — the executed counterpart to the offline `dag`.
    """
    from .registry.jobs import read_run_jobs

    rows = read_run_jobs(run_id or self.run_id, settings=self._settings)
    return [
        JobTrace(
            family=r["family"],
            job_key=r["job_id"],
            system_job_id=r.get("system_job_id"),
            runtime=r["runtime"],
            status=r.get("status"),
            attempt=r.get("attempt"),
            hardware=r.get("hardware"),
            gpu_type=r.get("gpu_type"),
            spark_mode=r.get("spark_mode"),
            runtime_seconds=r.get("runtime_seconds"),
        )
        for r in rows
    ]

jobs_df

jobs_df(run_id=None)

Return a tidy DataFrame of per-family job execution traces from v_run_jobs for run_id (default: this config's id).

Source code in src/scale_forecasting/sdk.py

def jobs_df(self, run_id: str | None = None) -> pd.DataFrame:
    """Return a tidy ``DataFrame`` of per-family job execution traces from ``v_run_jobs`` for
    ``run_id`` (default: this config's id).
    """
    from dataclasses import asdict

    import pandas as pd

    cols = [
        "family",
        "job_key",
        "system_job_id",
        "runtime",
        "status",
        "attempt",
        "hardware",
        "gpu_type",
        "spark_mode",
        "runtime_seconds",
    ]
    traces = self.jobs(run_id)
    if not traces:
        return pd.DataFrame(columns=cols)
    return pd.DataFrame.from_records([asdict(t) for t in traces], columns=cols)

leaderboard

leaderboard(run_id=None, *, all_metrics=False)

Alias for :meth:leaderboard_df matching :meth:RunResult.leaderboard.

Source code in src/scale_forecasting/sdk.py

def leaderboard(self, run_id: str | None = None, *, all_metrics: bool = False) -> pd.DataFrame:
    """Alias for :meth:`leaderboard_df` matching :meth:`RunResult.leaderboard`."""
    return self.leaderboard_df(run_id, all_metrics=all_metrics)

leaderboard_df

leaderboard_df(run_id=None, *, all_metrics=False)

Return a ranked leaderboard DataFrame for run_id (default: this config's id).

Delegates to review_run and review.build_leaderboard_frame. Pass all_metrics=True to include mean_<metric> and p50_<metric> for all 21 evaluation metrics.

Source code in src/scale_forecasting/sdk.py

def leaderboard_df(
    self, run_id: str | None = None, *, all_metrics: bool = False
) -> pd.DataFrame:
    """Return a ranked leaderboard ``DataFrame`` for ``run_id`` (default: this config's id).

    Delegates to `review_run` and `review.build_leaderboard_frame`. Pass ``all_metrics=True``
    to include ``mean_<metric>`` and ``p50_<metric>`` for all 21 evaluation metrics.
    """
    from .review import build_leaderboard_frame

    return build_leaderboard_frame(self.review_run(run_id), all_metrics=all_metrics)

monitor

monitor(run_id=None, *, probe=False)

Live progress of a run: per-family job state + series done vs. expected (RunProgress).

Delegates to review.monitor_run for run_id (default: this config's id) — header status, each family's runner and status, how many cells have landed vs. the expected total, and how long each family has been quiet. Poll it while a run is in flight; pair with review.plot_progress for the bar.

probe=True also escalates the non-terminal jobs to their runtime and attaches the reconciled probes.reconcile.ProbeReport — use it when a bar has stopped moving and you need to know whether the job is still alive. It is off by default because a poll loop must stay registry-only; probe (the drill-down) is the deliberate version of the same read.

Source code in src/scale_forecasting/sdk.py

def monitor(self, run_id: str | None = None, *, probe: bool = False) -> Any:
    """Live progress of a run: per-family job state + series done vs. expected (`RunProgress`).

    Delegates to `review.monitor_run` for ``run_id`` (default: this config's id) — header
    status, each family's runner and status, how many cells have landed vs. the expected total,
    and how long each family has been quiet. Poll it while a run is in flight; pair with
    `review.plot_progress` for the bar.

    ``probe=True`` also escalates the non-terminal jobs to their runtime and attaches the
    reconciled `probes.reconcile.ProbeReport` — use it when a bar has stopped moving and you
    need to know whether the job is still alive. It is off by default because a poll loop must
    stay registry-only; `probe` (the drill-down) is the deliberate version of the same read.
    """
    from .review import monitor_run

    return monitor_run(run_id or self.run_id, probe=probe, settings=self._settings)

plot_attributions

plot_attributions(
    run_id=None,
    *,
    ts_id=None,
    model_type=None,
    level="global",
    top_k=12,
    ax=None,
    title=None,
)

Plot Tier 1 global driver importance or Tier 2 per-horizon-step local feature attributions for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_attributions(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | None = None,
    model_type: str | None = None,
    level: str = "global",
    top_k: int = 12,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot Tier 1 global driver importance or Tier 2 per-horizon-step local feature
    attributions for ``run_id``.
    """
    from .review import plot_attributions

    df = self.attributions_df(
        run_id,
        ts_id=ts_id,
        models=[model_type] if model_type else None,
        level=level,
    )
    return plot_attributions(
        df,
        ts_id=ts_id,
        model_type=model_type,
        top_k=top_k,
        ax=ax,
        title=title,
    )

plot_calibration

plot_calibration(run_id=None, *, ax=None, title=None)

Plot empirical interval coverage and mean interval width by horizon step ($h = 1 \dots H$) against the nominal coverage target for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_calibration(
    self,
    run_id: str | None = None,
    *,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot empirical interval coverage and mean interval width by horizon step ($h = 1 \\dots
    H$) against the nominal coverage target for ``run_id``.
    """
    from .review import plot_calibration

    return plot_calibration(self.calibration(run_id), ax=ax, title=title)

plot_ensemble_weights

plot_ensemble_weights(
    run_id=None, *, limit=5000, ax=None, title=None
)

Plot normalized base-model weights across learned ensemble strategies (nnls, ridge, xgb) as a horizontal stacked bar chart for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_ensemble_weights(
    self,
    run_id: str | None = None,
    *,
    limit: int = 5000,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot normalized base-model weights across learned ensemble strategies (`nnls`, `ridge`,
    `xgb`) as a horizontal stacked bar chart for ``run_id``.
    """
    from .review import plot_ensemble_weights

    rid = run_id or self.run_id
    return plot_ensemble_weights(
        self.ensemble_weights_df(rid, limit=limit),
        ax=ax,
        title=title or f"{rid} — Learned Ensemble Weights",
    )

plot_forecast_explanation

plot_forecast_explanation(
    run_id=None,
    *,
    ts_id=None,
    model_type=None,
    seasonal_period=7,
    history_tail=120,
    title=None,
)

Render a 4-panel forecast explainability & decomposition chart for one series and model on run_id.

Source code in src/scale_forecasting/sdk.py

def plot_forecast_explanation(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | None = None,
    model_type: str | None = None,
    seasonal_period: int = 7,
    history_tail: int = 120,
    title: str | None = None,
) -> Any:
    """Render a 4-panel forecast explainability & decomposition chart for one series and model
    on ``run_id``.
    """
    from .review import plot_forecast_explanation

    df = self.explain_forecast(
        run_id,
        ts_id=ts_id,
        model_type=model_type,
        seasonal_period=seasonal_period,
        history_tail=history_tail,
    )
    return plot_forecast_explanation(df, history_tail=history_tail, title=title)

plot_forecasts

plot_forecasts(
    run_id=None,
    *,
    ts_id=None,
    ts_ids=None,
    models=None,
    max_series=3,
    history_tail=120,
    ax=None,
    title=None,
)

Plot historical actuals, backtest OOF trajectories, and future forecasts with 80% prediction intervals for up to max_series series on run_id.

Source code in src/scale_forecasting/sdk.py

def plot_forecasts(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | None = None,
    ts_ids: Sequence[str] | None = None,
    models: Sequence[str] | None = None,
    max_series: int = 3,
    history_tail: int = 120,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot historical actuals, backtest OOF trajectories, and future forecasts with 80%
    prediction intervals for up to ``max_series`` series on ``run_id``.
    """
    from .registry.reads import read_predictions
    from .review import plot_forecasts_frame

    rid = run_id or self.run_id
    effective_ts_ids = [ts_id] if ts_id is not None else ts_ids
    chosen_ts = list(effective_ts_ids)[: max(1, max_series)] if effective_ts_ids else None
    if chosen_ts is None:
        sample_preds = read_predictions(
            rid,
            models=list(models) if models else None,
            limit=max(500, max_series * 100),
            settings=self._settings,
        )
        chosen_ts = list(dict.fromkeys(str(r["ts_id"]) for r in sample_preds))[
            : max(1, max_series)
        ]
    frame = self.predictions_df(
        rid,
        ts_ids=chosen_ts,
        models=models,
        include_history=True,
        include_oof=True,
        history_tail=history_tail,
    )
    return plot_forecasts_frame(
        frame,
        ts_ids=chosen_ts,
        models=models,
        max_series=max_series,
        history_tail=history_tail,
        ax=ax,
        title=title or rid,
    )

plot_fva

plot_fva(
    run_id=None,
    *,
    baseline=None,
    metric=None,
    limit=20000,
    ax=None,
    title=None,
)

Plot the Forecast Value Added (FVA) accuracy waterfall for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_fva(
    self,
    run_id: str | None = None,
    *,
    baseline: str | None = None,
    metric: str | None = None,
    limit: int = 20000,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot the Forecast Value Added (FVA) accuracy waterfall for ``run_id``."""
    from .diagnostics import plot_fva

    rid = run_id or self.run_id
    df = self.fva_df(rid, baseline=baseline, metric=metric, limit=limit)
    return plot_fva(df, ax=ax, title=title or f"{rid} — Forecast Value Added")

plot_hierarchy

plot_hierarchy(
    run_id=None,
    *,
    model_type=None,
    limit=20000,
    ax=None,
    title=None,
)

Plot the top-level (__total__) forecast against the sum of bottom-level leaf forecasts for run_id to visually verify hierarchical coherence.

Source code in src/scale_forecasting/sdk.py

def plot_hierarchy(
    self,
    run_id: str | None = None,
    *,
    model_type: str | None = None,
    limit: int = 20000,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot the top-level (``__total__``) forecast against the sum of bottom-level leaf
    forecasts for ``run_id`` to visually verify hierarchical coherence.
    """
    from .registry.reads import read_predictions
    from .review import plot_hierarchy_frame

    rid = run_id or self.run_id
    rows = read_predictions(rid, limit=limit, settings=self._settings)
    return plot_hierarchy_frame(rows, model_type=model_type, ax=ax, title=title or rid)

plot_horizon_decay

plot_horizon_decay(
    run_id=None,
    *,
    metric=None,
    baseline=None,
    limit=20000,
    ax=None,
    title=None,
)

Plot lead-time accuracy decay by horizon step ($h = 1 \dots H$) for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_horizon_decay(
    self,
    run_id: str | None = None,
    *,
    metric: str | None = None,
    baseline: str | None = None,
    limit: int = 20000,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot lead-time accuracy decay by horizon step ($h = 1 \\dots H$) for ``run_id``."""
    from .diagnostics import plot_horizon_decay

    rid = run_id or self.run_id
    df = self.horizon_decay_df(rid, metric=metric, baseline=baseline, limit=limit)
    return plot_horizon_decay(df, ax=ax, title=title or f"{rid} — Horizon Accuracy Decay")

plot_inventory_policy

plot_inventory_policy(
    run_id=None,
    *,
    inventory_df=None,
    model_name=None,
    model_type=None,
    lead_time=7.0,
    lead_time_std=0.0,
    review_period=0.0,
    service_level=0.95,
    fill_rate=None,
    on_hand=0.0,
    on_order=0.0,
    backorders=0.0,
    holding_cost=1.0,
    stockout_cost=9.0,
    min_order_qty=0.0,
    order_multiple=0.0,
    max_rows=15,
    limit=20000,
    ax=None,
    title=None,
)

Plot Reorder Points (ROP), Safety Stock (SS), and Recommended Order Quantities ($Q$) for run_id.

Source code in src/scale_forecasting/sdk.py

def plot_inventory_policy(
    self,
    run_id: str | None = None,
    *,
    inventory_df: pd.DataFrame | Mapping[str, Mapping[str, Any]] | None = None,
    model_name: str | None = None,
    model_type: str | None = None,
    lead_time: float = 7.0,
    lead_time_std: float = 0.0,
    review_period: float = 0.0,
    service_level: float | None = 0.95,
    fill_rate: float | None = None,
    on_hand: float = 0.0,
    on_order: float = 0.0,
    backorders: float = 0.0,
    holding_cost: float = 1.0,
    stockout_cost: float = 9.0,
    min_order_qty: float = 0.0,
    order_multiple: float = 0.0,
    max_rows: int = 15,
    limit: int = 20000,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot Reorder Points (`ROP`), Safety Stock (`SS`), and Recommended Order Quantities ($Q$)
    for ``run_id``.
    """
    from .diagnostics import plot_inventory_policy

    rid = run_id or self.run_id
    df = self.inventory_policy_df(
        rid,
        inventory_df=inventory_df,
        model_name=model_name or model_type,
        lead_time=lead_time,
        lead_time_std=lead_time_std,
        review_period=review_period,
        service_level=service_level,
        fill_rate=fill_rate,
        on_hand=on_hand,
        on_order=on_order,
        backorders=backorders,
        holding_cost=holding_cost,
        stockout_cost=stockout_cost,
        min_order_qty=min_order_qty,
        order_multiple=order_multiple,
        limit=limit,
    )
    return plot_inventory_policy(
        df,
        max_rows=max_rows,
        ax=ax,
        title=title or f"{rid} — Inventory & Replenishment Policy",
    )

plot_scenarios

plot_scenarios(
    sim_df,
    *,
    ts_id=None,
    model_type=None,
    show_revenue=False,
    ax=None,
    title=None,
)

Plot baseline vs. counterfactual scenario trajectories from simulate_scenarios.

Source code in src/scale_forecasting/sdk.py

def plot_scenarios(
    self,
    sim_df: pd.DataFrame,
    *,
    ts_id: str | None = None,
    model_type: str | None = None,
    show_revenue: bool = False,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Plot baseline vs. counterfactual scenario trajectories from `simulate_scenarios`."""
    from .scenarios import plot_scenarios

    return plot_scenarios(
        sim_df,
        ts_id=ts_id,
        model_type=model_type,
        show_revenue=show_revenue,
        ax=ax,
        title=title,
    )

plot_series

plot_series(
    ts_id=None,
    *,
    run_id=None,
    models=None,
    history_tail=120,
    ax=None,
    title=None,
)

Convenience wrapper around :meth:plot_forecasts for a single ts_id.

Source code in src/scale_forecasting/sdk.py

def plot_series(
    self,
    ts_id: str | None = None,
    *,
    run_id: str | None = None,
    models: Sequence[str] | None = None,
    history_tail: int = 120,
    ax: Any = None,
    title: str | None = None,
) -> Any:
    """Convenience wrapper around :meth:`plot_forecasts` for a single ``ts_id``."""
    return self.plot_forecasts(
        run_id,
        ts_id=ts_id,
        models=models,
        max_series=1,
        history_tail=history_tail,
        ax=ax,
        title=title,
    )

predictions_df

predictions_df(
    run_id=None,
    *,
    ts_id=None,
    ts_ids=None,
    models=None,
    include_history=True,
    include_oof=True,
    history_tail=120,
    series_limit=None,
    limit=10000,
)

Return a tidy DataFrame of forecasts (plus optional OOF and historical actuals) for run_id.

Columns: ts_id, segment ("history" / "oof" / "forecast"), model_type, ds, y_true, yhat, yhat_lower, yhat_upper, fold_id.

Source code in src/scale_forecasting/sdk.py

def predictions_df(
    self,
    run_id: str | None = None,
    *,
    ts_id: str | None = None,
    ts_ids: Sequence[str] | None = None,
    models: Sequence[str] | None = None,
    include_history: bool = True,
    include_oof: bool = True,
    history_tail: int = 120,
    series_limit: int | None = None,
    limit: int = 10000,
) -> pd.DataFrame:
    """Return a tidy ``DataFrame`` of forecasts (plus optional OOF and historical actuals) for
    ``run_id``.

    Columns: ``ts_id``, ``segment`` (``"history"`` / ``"oof"`` / ``"forecast"``),
    ``model_type``, ``ds``, ``y_true``, ``yhat``, ``yhat_lower``, ``yhat_upper``, ``fold_id``.
    """
    from .registry.reads import read_oof, read_predictions, read_series_history
    from .review import build_predictions_frame

    rid = run_id or self.run_id
    ts_list = [ts_id] if ts_id is not None else (list(ts_ids) if ts_ids is not None else None)
    model_list = list(models) if models is not None else None
    pred_rows = read_predictions(
        rid, ts_ids=ts_list, models=model_list, limit=limit, settings=self._settings
    )
    resolved_ts = ts_list or (
        list(dict.fromkeys(str(r["ts_id"]) for r in pred_rows)) if pred_rows else None
    )
    if series_limit is not None and resolved_ts is not None:
        resolved_ts = resolved_ts[: max(1, series_limit)]
        pred_rows = [r for r in pred_rows if str(r["ts_id"]) in set(resolved_ts)]
    oof_rows = (
        read_oof(
            rid, ts_ids=resolved_ts, models=model_list, limit=limit, settings=self._settings
        )
        if include_oof
        else None
    )
    hist_rows = (
        read_series_history(
            rid, ts_ids=resolved_ts, history_tail=history_tail, settings=self._settings
        )
        if include_history
        else None
    )
    return build_predictions_frame(pred_rows, oof_rows=oof_rows, history_rows=hist_rows)

probe

probe(run_id=None, job=None)

Reconcile a run against its runtime: per-family verdict + a disagreement flag.

Delegates to probes.reconcile.probe_run for run_id (default: this config's id): the registry-first drill-down that fuses the header + landed artifacts with, only for the incomplete or stale jobs, their live native state (Spark / Ray / BigQuery) — flagging each row where the runtime contradicts the registry. job narrows the escalation to one family (statistical/ml/deep_learning/native/ensemble). A run that is already terminal short-circuits and touches no runtime; for the fleet-wide view use monitor.

Source code in src/scale_forecasting/sdk.py

def probe(self, run_id: str | None = None, job: str | None = None) -> Any:
    """Reconcile a run against its runtime: per-family verdict + a disagreement flag.

    Delegates to `probes.reconcile.probe_run` for ``run_id`` (default: this config's id): the
    registry-first drill-down that fuses the header + landed artifacts with, **only for the
    incomplete or stale jobs**, their live native state (Spark / Ray / BigQuery) — flagging each
    row where the runtime contradicts the registry. ``job`` narrows the escalation to one family
    (``statistical``/``ml``/``deep_learning``/``native``/``ensemble``). A run that is already
    terminal short-circuits and touches no runtime; for the fleet-wide view use `monitor`.
    """
    from .probes.reconcile import probe_run

    return probe_run(run_id or self.run_id, job=job, settings=self._settings)

reensemble

reensemble(strategies=None, *, run_id=None)

Execute post-hoc re-ensembling on an already-completed run without re-fitting base models; return the resulting ensemble_id.

Delegates to ensemble_run.run_ensembles with strategies overriding the config's ensemble.strategies block. New strategies land under a distinct ensemble_id and appear alongside existing ensembles on v_model_leaderboard.

Source code in src/scale_forecasting/sdk.py

def reensemble(
    self,
    strategies: list[str] | None = None,
    *,
    run_id: str | None = None,
) -> str:
    """Execute post-hoc re-ensembling on an already-completed run without re-fitting base
    models; return the resulting ``ensemble_id``.

    Delegates to `ensemble_run.run_ensembles` with ``strategies`` overriding the config's
    ``ensemble.strategies`` block. New strategies land under a distinct ``ensemble_id`` and
    appear alongside existing ensembles on ``v_model_leaderboard``.
    """
    from .ensemble_run import _override_ensemble, run_ensembles
    from .registry.ids import make_ensemble_id
    from .registry.tables import resolve_settings

    cfg = _override_ensemble(self._config, strategies)
    if not cfg.ensemble.enabled:
        cfg = _override_ensemble(
            self._config, list(cfg.ensemble.strategies) or ["mean", "median", "inverse_error"]
        )
    rid = run_id or self.run_id
    resolved = resolve_settings(self._settings)
    run_ensembles(cfg, rid, settings=resolved)
    return make_ensemble_id(cfg.ensemble)

registry

registry()

A Registry handle for the registry this config's runs land in.

The lifecycle verbs (Registry.doctor, Registry.drop_run, …) are keyed on the registry, not on a config, so they live on their own object — but reaching them from the Forecaster you already have is the common case (f.registry().drop_run(f.run_id)).

Source code in src/scale_forecasting/sdk.py

def registry(self) -> Registry:
    """A `Registry` handle for the registry this config's runs land in.

    The lifecycle verbs (`Registry.doctor`, `Registry.drop_run`, …) are keyed on the *registry*,
    not on a config, so they live on their own object — but reaching them from the `Forecaster`
    you already have is the common case (`f.registry().drop_run(f.run_id)`).
    """
    return Registry(settings=self._settings)

results

results(run_id=None)

The per-model leaderboard for a run — one ModelResult each, best (lowest WAPE) first.

Reads v_model_leaderboard (registry.reads.read_leaderboard) for run_id (default: this config's id). Returns [] when the run has produced no scored rows yet. This is the first-class "which model won?" read that RunResult only pointed at before.

Source code in src/scale_forecasting/sdk.py

def results(self, run_id: str | None = None) -> list[ModelResult]:
    """The per-model leaderboard for a run — one `ModelResult` each, best (lowest WAPE) first.

    Reads ``v_model_leaderboard`` (`registry.reads.read_leaderboard`) for ``run_id`` (default:
    this config's id). Returns ``[]`` when the run has produced no scored rows yet. This is the
    first-class "which model won?" read that `RunResult` only pointed at before.
    """
    from .registry.reads import read_leaderboard

    rows = read_leaderboard(run_id or self.run_id, settings=self._settings)
    return [
        ModelResult(
            model_type=r["model_type"],
            ensemble_id=r.get("ensemble_id"),
            compute_engine=r.get("compute_engine"),
            n_cells=r["n_cells"],
            no_artifact_rate=r.get("no_artifact_rate"),
            median_fit_seconds=r.get("median_fit_seconds"),
            mean_wape=r.get("mean_wape"),
            mean_mae=r.get("mean_mae"),
        )
        for r in rows
    ]

retry

retry(*, confirm=False, reason='')

Repair what this run failed to produce — a preview by default.

Delegates to retry_run.retry_run. It reads the run's per-cell outcomes, derives the cells that were never attempted at all (expected series at the run's pinned snapshot, minus what landed), classifies every one of them with retry_policy, and returns the decision table inside a retry_run.RetryReport. Without confirm that is all it does — no launch, no write. With confirm=True it re-submits only the models that could still succeed as a fresh attempt under the same run_id, and records the table it acted on under job_telemetry.$.retry.

Two things it will refuse, both on purpose. A cell that already has predictions is never retried — a retry appends, so it can add a second forecast beside the first but never replace it, and quietly swapping a reviewed forecast for a fresh fit is a different run, not a repair. And because v1 submits per model over the whole series universe, a model with any landed predictions is refused wholesale; the report names those under blocked rather than dropping them silently. Changing an answer that already landed means changing the config, which changes the run_id — the honest way to say the answer changed.

There is no run_id parameter, unlike probe / cancel / settle. Those inspect a run; this one plans one, and the plan comes from the config — the DAG to narrow, the family each model belongs to, the source table and the series subset. A different run has a different config by definition, so build a Forecaster on that config instead. If all you have is the id, from_run_id reads the config back out of the run's header and gives you exactly that — which is why the id is a constructor argument here and not a method one.

Source code in src/scale_forecasting/sdk.py

def retry(self, *, confirm: bool = False, reason: str = "") -> Any:
    """Repair what this run failed to produce — a **preview by default**.

    Delegates to `retry_run.retry_run`. It reads the run's per-cell outcomes, derives the cells
    that were never attempted at all (expected series at the run's pinned snapshot, minus what
    landed), classifies every one of them with `retry_policy`, and returns the decision table
    inside a `retry_run.RetryReport`. Without ``confirm`` that is all it does — no launch, no
    write. With ``confirm=True`` it re-submits **only the models that could still succeed** as a
    fresh attempt under the same ``run_id``, and records the table it acted on under
    ``job_telemetry.$.retry``.

    Two things it will refuse, both on purpose. A cell that already has predictions is never
    retried — a retry appends, so it can add a second forecast beside the first but never
    replace it, and quietly swapping a reviewed forecast for a fresh fit is a different run, not
    a repair. And because v1 submits per *model* over the whole series universe, a model with
    *any* landed predictions is refused wholesale; the report names those under ``blocked``
    rather than dropping them silently. Changing an answer that already landed means changing
    the config, which changes the ``run_id`` — the honest way to say the answer changed.

    There is no ``run_id`` parameter, unlike `probe` / `cancel` / `settle`. Those inspect a run;
    this one *plans* one, and the plan comes from the config — the DAG to narrow, the family
    each model belongs to, the source table and the series subset. A different run has a
    different config by definition, so build a `Forecaster` on that config instead. If all you
    have is the id, `from_run_id` reads the config back out of the run's header and gives you
    exactly that — which is why the id is a *constructor* argument here and not a method one.
    """
    from .retry_run import retry_run

    return retry_run(self._config, confirm=confirm, reason=reason, settings=self._settings)

review

review()

Return a pointer to this config's results without running anything (offline).

The run_id is the deterministic config hash, so this resolves the same location a completed run landed under. dataset_ref is None when the GCP identity can't be resolved (no SF_* env and no injected settings) — the run_id + view names are still returned so a caller can construct the query once infra is known.

Source code in src/scale_forecasting/sdk.py

def review(self) -> RunResult:
    """Return a pointer to this config's results **without** running anything (offline).

    The run_id is the deterministic config hash, so this resolves the same location a completed
    run landed under. ``dataset_ref`` is ``None`` when the GCP identity can't be resolved (no
    ``SF_*`` env and no injected ``settings``) — the run_id + view names are still returned so a
    caller can construct the query once infra is known.
    """
    return RunResult(
        run_id=self.run_id, dataset_ref=self._resolved_dataset_ref(), views=VIEW_NAMES
    )

review_run

review_run(run_id=None)

Data-science review of a finished run: bests, metric panel, ensemble lift (RunReview).

Delegates to review.review_run for run_id (default: this config's id) — best model per family and overall, the full metric panel aggregated across series, and each ensemble's lift over the best base model. The reading counterpart to the offline review pointer; pair with review.plot_leaderboard / review.plot_metric_distribution.

Source code in src/scale_forecasting/sdk.py

def review_run(self, run_id: str | None = None) -> Any:
    """Data-science review of a finished run: bests, metric panel, ensemble lift (`RunReview`).

    Delegates to `review.review_run` for ``run_id`` (default: this config's id) — best model per
    family and overall, the full metric panel aggregated across series, and each ensemble's lift
    over the best base model. The reading counterpart to the offline `review` pointer; pair with
    `review.plot_leaderboard` / `review.plot_metric_distribution`.
    """
    from .review import review_run as _review_run

    return _review_run(run_id or self.run_id, settings=self._settings)

revision_stability_df

revision_stability_df(
    run_ids, *, jump_threshold_pct=0.3, limit=20000
)

Measure cross-run forecast revision stability ("whipsaw") across run_ids.

Source code in src/scale_forecasting/sdk.py

def revision_stability_df(
    self,
    run_ids: Sequence[str],
    *,
    jump_threshold_pct: float = 0.30,
    limit: int = 20000,
) -> pd.DataFrame:
    """Measure cross-run forecast revision stability ("whipsaw") across ``run_ids``."""
    return self.registry().revision_stability_df(
        run_ids, jump_threshold_pct=jump_threshold_pct, limit=limit
    )

rollup_time

rollup_time(
    freq="W", *, run_id=None, level=0.8, limit=20000
)

Aggregate run_id future predictions to Weekly ("W"), Monthly ("MS"/"M"), or Quarterly ("QS"/"Q") buckets with autocorrelation-aware prediction intervals.

Source code in src/scale_forecasting/sdk.py

def rollup_time(
    self,
    freq: str = "W",
    *,
    run_id: str | None = None,
    level: float = 0.80,
    limit: int = 20000,
) -> pd.DataFrame:
    """Aggregate ``run_id`` future predictions to Weekly (`"W"`), Monthly (`"MS"`/`"M"`), or
    Quarterly (`"QS"`/`"Q"`) buckets with autocorrelation-aware prediction intervals.
    """
    import pandas as pd

    from .diagnostics import rollup_time
    from .registry.reads import read_oof, read_predictions

    rid = run_id or self.run_id
    pred_rows = read_predictions(rid, limit=limit, settings=self._settings)
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else None
    return rollup_time(pred_df, freq=freq, oof_df=oof_df, level=level)

run

run(*, spark=None, n_series=None, max_executors=None)

Execute the run (Spark/Ray ∥ BigQuery under one run_id) and return where to query it.

Delegates to main.run, threading this forecaster's settings (so an injected identity is honored). spark optionally injects a SparkSession / DataprocSparkSession for the in-process Spark path (notebook / Connect demo). n_series overrides data.series_limit (the scale knob — it changes the run_id, so each scale is its own run); max_executors caps the remote Spark batch's executor ceiling. Returns a RunResult pointing at the registry views under the resolved dataset.

Note the returned run_id reflects any n_series override, so it may differ from this forecaster's base run_id; use it (or status/results with it) to track this run.

Source code in src/scale_forecasting/sdk.py

def run(
    self,
    *,
    spark: object | None = None,
    n_series: int | None = None,
    max_executors: int | None = None,
) -> RunResult:
    """Execute the run (Spark/Ray ∥ BigQuery under one run_id) and return where to query it.

    Delegates to `main.run`, threading this forecaster's ``settings`` (so an injected
    identity is honored). ``spark`` optionally injects a `SparkSession` /
    ``DataprocSparkSession`` for the in-process Spark path (notebook / Connect demo).
    ``n_series`` overrides ``data.series_limit`` (the scale knob — it changes the ``run_id``, so
    each scale is its own run); ``max_executors`` caps the remote Spark batch's executor
    ceiling. Returns a `RunResult` pointing at the registry views under the resolved dataset.

    Note the returned ``run_id`` reflects any ``n_series`` override, so it may differ from this
    forecaster's base `run_id`; use it (or `status`/`results` with it) to track this run.
    """
    from . import main

    run_id = main.run(
        self._config,
        spark=spark,
        settings=self._settings,
        n_series=n_series,
        max_executors=max_executors,
    )
    return RunResult(run_id=run_id, dataset_ref=self._resolved_dataset_ref(), views=VIEW_NAMES)

run_live

run_live(
    *,
    spark=None,
    n_series=None,
    max_executors=None,
    poll_seconds=10.0,
    poll_s=None,
    probe=False,
    plot=True,
)

Execute run while polling monitor for live progress updates, then return RunResult.

Launches run in a background thread and polls monitor every poll_seconds (or poll_s) seconds until the run finishes (or when poll_seconds <= 0, executes synchronously). When plot=True and the terminal progress snapshot contains family rows, renders review.plot_progress once at completion. Safe in both interactive notebooks and headless automation (notebook_acceptance).

Source code in src/scale_forecasting/sdk.py

def run_live(
    self,
    *,
    spark: object | None = None,
    n_series: int | None = None,
    max_executors: int | None = None,
    poll_seconds: float = 10.0,
    poll_s: float | None = None,
    probe: bool = False,
    plot: bool = True,
) -> RunResult:
    """Execute `run` while polling `monitor` for live progress updates, then return `RunResult`.

    Launches `run` in a background thread and polls `monitor` every ``poll_seconds`` (or
    ``poll_s``) seconds until the run finishes (or when ``poll_seconds <= 0``, executes
    synchronously). When ``plot=True`` and the terminal progress snapshot contains family rows,
    renders `review.plot_progress` once at completion. Safe in both interactive notebooks and
    headless automation (`notebook_acceptance`).
    """
    import threading

    effective_poll = poll_s if poll_s is not None else poll_seconds
    t0 = time.monotonic()
    if effective_poll <= 0:
        res = self.run(spark=spark, n_series=n_series, max_executors=max_executors)
        elapsed = time.monotonic() - t0
        return RunResult(
            run_id=res.run_id,
            dataset_ref=res.dataset_ref,
            views=res.views,
            status=res.status or "COMPLETED",
            runtime_seconds=res.runtime_seconds or elapsed,
        )

    effective_cfg = (
        self._config
        if n_series is None
        else self._config.model_copy(
            update={"data": self._config.data.model_copy(update={"series_limit": n_series})}
        )
    )
    target_rid = make_run_id(effective_cfg)
    outcome: dict[str, Any] = {}

    def _worker() -> None:
        try:
            outcome["result"] = self.run(
                spark=spark, n_series=n_series, max_executors=max_executors
            )
        except BaseException as exc:  # noqa: BLE001 - re-raised on main thread
            outcome["error"] = exc

    thread = threading.Thread(target=_worker, name=f"sf-run-{target_rid}", daemon=True)
    thread.start()
    last_snap: Any = None
    while thread.is_alive():
        thread.join(timeout=effective_poll)
        try:
            snap = self.monitor(target_rid, probe=probe)
            last_snap = snap
            if snap.status is not None:
                frac = f"{snap.fraction:.0%}" if snap.fraction is not None else "?"
                _log.info(
                    "live progress [%s]: status=%s cells=%s/%s (%s)",
                    target_rid,
                    snap.status,
                    snap.n_done,
                    snap.n_expected if snap.n_expected is not None else "?",
                    frac,
                )
        except Exception:  # noqa: BLE001 - transient read errors never abort the run
            pass
    thread.join()
    elapsed = time.monotonic() - t0
    if "error" in outcome:
        raise outcome["error"]
    res = outcome["result"]
    final_snap = last_snap
    with contextlib.suppress(Exception):  # the closing snapshot is a nicety, never a failure
        final_snap = self.monitor(res.run_id)
    if plot and final_snap is not None and getattr(final_snap, "families", None):
        from .review import plot_progress

        plot_progress(final_snap)
    final_status = (
        (getattr(final_snap, "status", None) if final_snap is not None else None)
        or res.status
        or "COMPLETED"
    )
    return RunResult(
        run_id=res.run_id,
        dataset_ref=res.dataset_ref,
        views=res.views,
        status=final_status,
        runtime_seconds=res.runtime_seconds or elapsed,
    )

score_realized

score_realized(
    actuals_df,
    *,
    run_id=None,
    metric=None,
    drift_threshold=1.5,
    limit=20000,
)

Score published forward predictions for run_id against newly arrived actuals_df and flag series whose live error exceeds drift_threshold times backtest expectation.

Source code in src/scale_forecasting/sdk.py

def score_realized(
    self,
    actuals_df: pd.DataFrame,
    *,
    run_id: str | None = None,
    metric: str | None = None,
    drift_threshold: float = 1.5,
    limit: int = 20000,
) -> pd.DataFrame:
    """Score published forward predictions for ``run_id`` against newly arrived ``actuals_df``
    and flag series whose live error exceeds ``drift_threshold`` times backtest expectation.
    """
    import pandas as pd

    from .diagnostics import compute_realized_accuracy
    from .registry.reads import read_oof, read_predictions

    rid = run_id or self.run_id
    pred_rows = read_predictions(rid, limit=limit, settings=self._settings)
    oof_rows = read_oof(rid, limit=limit, settings=self._settings)
    pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else None
    eff_metric = metric or self._config.backtest.decision_metric
    return compute_realized_accuracy(
        pred_df,
        actuals_df,
        oof_df=oof_df,
        metric=eff_metric,
        drift_threshold=drift_threshold,
    )

settle

settle(run_id=None, job=None, *, yes=False, reason='')

Repair stale job rows from the probe's own verdicts — a preview by default.

Delegates to probes.settle.settle_run for run_id (default: this config's id). probe already decides that a family whose row still says RUNNING finished forty minutes ago and then discards that answer; this is the verb that writes it down. Without yes it probes, returns the SettlePlan inside a SettleReport, and changes nothing. With yes=True it sets each settleable family's status (COMPLETED, or FAILED with a RUNTIME_FAILED / RUNTIME_LOST / CAPACITY_ABANDONED token), records reason + the ADC actor and the full evidence under job_telemetry.$.settle, and re-reads every row so the outcome is the status it actually ended up with. job narrows to one family.

Only five (verdict, runtime-state) readings settle; UNKNOWN and RUNNING_CONFIRMED are always refused, and the plan says why each untouched family was left alone. Settle never deletes and never stamps ended_at/runtime_seconds — a row settled days later would report days of runtime. It leaves the run header to registry.ops.close_runs, which settling the job rows is precisely what unblocks; the report carries the hint.

yes= rather than confirm= on purpose: cancel's confirm= gates stopping running work in the cloud, yes= gates repairing a status field in our own registry — the same gate registry.ops.close_runs and registry.ops.sweep_orphans use.

Source code in src/scale_forecasting/sdk.py

def settle(
    self,
    run_id: str | None = None,
    job: str | None = None,
    *,
    yes: bool = False,
    reason: str = "",
) -> Any:
    """Repair stale job rows from the probe's own verdicts — a **preview by default**.

    Delegates to `probes.settle.settle_run` for ``run_id`` (default: this config's id). `probe`
    already decides that a family whose row still says RUNNING finished forty minutes ago and
    then discards that answer; this is the verb that writes it down. Without ``yes`` it probes,
    returns the `SettlePlan` inside a ``SettleReport``, and changes nothing. With ``yes=True``
    it sets each settleable family's status (``COMPLETED``, or ``FAILED`` with a
    ``RUNTIME_FAILED`` / ``RUNTIME_LOST`` / ``CAPACITY_ABANDONED`` token), records ``reason`` +
    the ADC actor and the full evidence under ``job_telemetry.$.settle``, and re-reads every row
    so the outcome is the status it actually ended up with. ``job`` narrows to one family.

    Only five (verdict, runtime-state) readings settle; ``UNKNOWN`` and ``RUNNING_CONFIRMED``
    are always refused, and the plan says *why* each untouched family was left alone. Settle
    never deletes and never stamps ``ended_at``/``runtime_seconds`` — a row settled days later
    would report days of runtime. It leaves the run header to `registry.ops.close_runs`, which
    settling the job rows is precisely what unblocks; the report carries the hint.

    ``yes=`` rather than ``confirm=`` on purpose: `cancel`'s ``confirm=`` gates stopping running
    work in the cloud, ``yes=`` gates repairing a status field in our own registry — the same
    gate `registry.ops.close_runs` and `registry.ops.sweep_orphans` use.
    """
    from .probes.settle import settle_run

    return settle_run(
        run_id or self.run_id,
        job=job,
        yes=yes,
        reason=reason,
        settings=self._settings,
    )

simulate_scenarios

simulate_scenarios(
    scenarios,
    *,
    data=None,
    models=None,
    future_covariates_df=None,
    price_col=None,
    run_id=None,
)

Run counterfactual covariate shocks (scenarios.simulate_scenarios) across data (or BigQuery source series when run_id is provided, or synthetic sample data when data is omitted) using this config's settings.

Source code in src/scale_forecasting/sdk.py

def simulate_scenarios(
    self,
    scenarios: Mapping[str, Mapping[str, Any]],
    *,
    data: pd.DataFrame | None = None,
    models: Sequence[str] | None = None,
    future_covariates_df: pd.DataFrame | None = None,
    price_col: str | None = None,
    run_id: str | None = None,
) -> pd.DataFrame:
    """Run counterfactual covariate shocks (`scenarios.simulate_scenarios`) across ``data``
    (or BigQuery source series when ``run_id`` is provided, or synthetic sample data when
    ``data`` is omitted) using this config's settings.
    """
    from . import playground
    from .scenarios import simulate_scenarios

    panel = data
    if panel is None and run_id is not None:
        try:
            from .engines.bigquery_names import source_ref
            from .engines.bigquery_sql import series_filter
            from .registry.reads import resolve_settings

            resolved = resolve_settings(self._settings)
            source = source_ref(self._config, resolved.registry_dataset_ref)
            idc, datec = self._config.data.ts_id_col, self._config.data.date_col
            sfilter = series_filter(self._config, source, idc)
            where_clause = f" WHERE {sfilter}" if sfilter else ""
            sql = f"SELECT * FROM `{source}`{where_clause} ORDER BY {idc}, {datec}"
            client = bigquery_client(resolved.project_id)
            df = client.query(sql).to_dataframe()
            if not df.empty:
                panel = df
        except Exception:  # noqa: BLE001 - fall back to synthetic sample data when offline
            panel = None
    if panel is None:
        panel = playground.sample_data(
            n_series=min(self._config.data.series_limit or 3, 5),
            freq=self._config.data.freq,
            with_exog=True,
        )
    return simulate_scenarios(
        panel,
        self._config,
        scenarios,
        models=models,
        future_covariates_df=future_covariates_df,
        price_col=price_col,
    )

status

status(run_id=None)

The current registry status of a run (RUNNING/COMPLETED/FAILED/PARTIAL).

Reads the run's header (registry.header.header_status); returns None when this config has never run, and STAGED when it has only been staged — the artifacts and the commands exist, but nothing has launched them, so wait on such a run polls to its timeout. run_id defaults to this config's deterministic id, so forecaster.status() answers "did my config's run finish?" — the reattach path for a wait=False submit.

Source code in src/scale_forecasting/sdk.py

def status(self, run_id: str | None = None) -> str | None:
    """The current registry status of a run (``RUNNING``/``COMPLETED``/``FAILED``/``PARTIAL``).

    Reads the run's header (`registry.header.header_status`); returns ``None`` when this config
    has never run, and ``STAGED`` when it has only been staged — the artifacts and the commands
    exist, but nothing has launched them, so `wait` on such a run polls to its timeout.
    ``run_id`` defaults to this config's deterministic id, so
    ``forecaster.status()`` answers "did my config's run finish?" — the reattach path for a
    ``wait=False`` submit.
    """
    from .registry.header import header_status

    return header_status(run_id or self.run_id, settings=self._settings)

trace

trace(run_id=None, *, cell_limit=5000)

The run's execution timeline as a long-form frame — per-job spans + per-cell spans.

Reads the per-job trace (v_run_jobs via jobs' reader) and the per-cell wall-clock brackets (forecast_metadata via registry.reads.read_cell_timing, capped at cell_limit) for run_id (default: this config's id), then stacks them into one frame (columns kind/lane/label/start/end/duration_s/status/ runtime/model_type/ts_id) via build_trace_frame. Feed it to plot_trace for a Gantt/waterfall, or slice it directly (frame[frame.kind == "cell"]). Returns an empty frame (with the columns) when the run has no timed rows — e.g. an older run written before the trace columns existed. This is the "how did this run unfold over wall-clock time, and which worker ran what?" read that sits under the per-job jobs summary.

Source code in src/scale_forecasting/sdk.py

def trace(self, run_id: str | None = None, *, cell_limit: int = 5000) -> pd.DataFrame:
    """The run's execution timeline as a long-form frame — per-job spans + per-cell spans.

    Reads the per-job trace (``v_run_jobs`` via `jobs`' reader) and the per-cell wall-clock
    brackets (``forecast_metadata`` via `registry.reads.read_cell_timing`, capped at
    ``cell_limit``) for ``run_id`` (default: this config's id), then stacks them into one frame
    (columns ``kind``/``lane``/``label``/``start``/``end``/``duration_s``/``status``/
    ``runtime``/``model_type``/``ts_id``) via `build_trace_frame`. Feed it to `plot_trace` for a
    Gantt/waterfall, or slice it directly (``frame[frame.kind == "cell"]``). Returns an empty
    frame (with the columns) when the run has no timed rows — e.g. an older run written before
    the trace columns existed. This is the "how did this run unfold over wall-clock time, and
    which worker ran what?" read that sits under the per-job `jobs` summary.
    """
    from .registry.jobs import read_run_jobs
    from .registry.reads import read_cell_timing

    rid = run_id or self.run_id
    job_rows = read_run_jobs(rid, settings=self._settings)
    cell_rows = read_cell_timing(rid, limit=cell_limit, settings=self._settings)
    return build_trace_frame(job_rows, cell_rows)

variant_matrix

variant_matrix()

Return the resolved model-variant taxonomy and spec-hash matrix DataFrame (offline).

One row per model entry in config.models (including auto-expanded univariate baselines and ablation-ladder variants), detailing model_type, base_model, variant_tag, family, scope_capability, training_mode, cov_profile, is_univariate, covariates_used, and model_spec_hash.

Source code in src/scale_forecasting/sdk.py

def variant_matrix(self) -> pd.DataFrame:
    """Return the resolved model-variant taxonomy and spec-hash matrix ``DataFrame`` (offline).

    One row per model entry in ``config.models`` (including auto-expanded univariate baselines
    and ablation-ladder variants), detailing ``model_type``, ``base_model``, ``variant_tag``,
    ``family``, ``scope_capability``, ``training_mode``, ``cov_profile``, ``is_univariate``,
    ``covariates_used``, and ``model_spec_hash``.
    """
    return build_variant_matrix_frame(self._config)

wait

wait(run_id=None, *, timeout=3600.0, poll_seconds=15.0)

Block until a run reaches a terminal status and return it; raise on timeout/not-found.

Polls status every poll_seconds until the status is terminal (COMPLETED/FAILED/PARTIAL). Raises TimeoutError if timeout elapses first, or ConfigError if the run has no header yet (never submitted). Each poll re-reads the header through a fresh client, so a long wait re-authenticates rather than reusing a stale token.

Source code in src/scale_forecasting/sdk.py

def wait(
    self, run_id: str | None = None, *, timeout: float = 3600.0, poll_seconds: float = 15.0
) -> str:
    """Block until a run reaches a terminal status and return it; raise on timeout/not-found.

    Polls `status` every ``poll_seconds`` until the status is terminal
    (``COMPLETED``/``FAILED``/``PARTIAL``). Raises `TimeoutError` if ``timeout`` elapses first,
    or `ConfigError` if the run has no header yet (never submitted). Each poll re-reads the
    header through a fresh client, so a long wait re-authenticates rather than reusing a stale
    token.
    """
    from .errors import ConfigError

    rid = run_id or self.run_id
    deadline = time.monotonic() + timeout
    while True:
        status = self.status(rid)
        if status is None:
            raise ConfigError(f"no run found for run_id {rid}: nothing to wait on")
        if status in _TERMINAL_STATUSES:
            return status
        if time.monotonic() >= deadline:
            raise TimeoutError(f"run {rid} still {status} after {timeout:.0f}s")
        time.sleep(poll_seconds)

JobTrace dataclass

JobTrace(
    family,
    job_key,
    system_job_id,
    runtime,
    status,
    attempt,
    hardware,
    gpu_type,
    spark_mode,
    runtime_seconds,
)

One family's job on a run, read from v_run_jobs — the cross-system trace row.

job_key is the job's canonical id (run_jobs.job_id); system_job_id is the platform's own id for the same job (a Dataproc batch/job id, a Ray submission_id, or a BigQuery parent job id), so a status query can go straight to that platform. runtime says which platform to query (spark / ray / bigquery); hardware / gpu_type / spark_mode record the resolved placement. status is the job's registry/terminal status and runtime_seconds its wall-clock; attempt distinguishes a forced re-run's job from the original under one run_id.

ModelResult dataclass

ModelResult(
    model_type,
    ensemble_id,
    compute_engine,
    n_cells,
    no_artifact_rate,
    median_fit_seconds,
    mean_wape,
    mean_mae,
)

One model's outcome on a run, read from v_model_leaderboard.

ensemble_id is None for a base model and the ensemble-config digest for an ensemble pseudo-model. mean_wape / mean_mae are the mean decision metrics where a backtest scored them (None otherwise); no_artifact_rate near 1.0 flags a model that failed most cells. Rows come back best-first (lowest mean_wape).

Registry

Registry(*, settings=None)

The easy path for managing a registry — the operator counterpart to Forecaster.

Forecaster produces runs; this reads and prunes them. It is keyed on the registry the SF_* environment resolves to (Settings.registry_dataset_ref), not on a config, so it needs no config at all — which is exactly right for "what is in this registry, and clean up run X".

Like Forecaster, it adds no logic: every method delegates to registry.ops, the same code python -m scale_forecasting.registry.ops runs, so the notebook, the SDK, and the CLI can never diverge. The destructive verbs preview by default and execute only on yes=True.

>>> from scale_forecasting import Registry
>>> reg = Registry()
>>> print(reg.doctor())                 # read-only: counts, stuck runs, orphans
>>> reg.drop_run("abc123")              # preview — deletes nothing
>>> reg.drop_run("abc123", yes=True)    # artifacts, then models, then rows

There is deliberately no wipe-everything method: a full teardown is bq rm -r -f <dataset>.

Source code in src/scale_forecasting/sdk.py

def __init__(self, *, settings: Settings | None = None) -> None:
    self._settings = settings

dataset_ref property

dataset_ref

project.dataset for this registry, or None when SF_* can't be resolved.

close_runs

close_runs(*run_ids, yes=False)

Finalize abandoned RUNNING headers to what their job rows already imply.

No arguments means every stuck header in this registry. Preview unless yes. Deletes nothing and skips any run with a job row that is not yet terminal — probe those first. Forecaster.monitor (probe=True) is the usual way in, but it needs a config; probes.reconcile.probe_run takes a bare run_id, which is what you have here. See registry.ops.close_runs.

Source code in src/scale_forecasting/sdk.py

def close_runs(self, *run_ids: str, yes: bool = False) -> Any:
    """Finalize abandoned ``RUNNING`` headers to what their job rows already imply.

    No arguments means every stuck header in this registry. Preview unless ``yes``. Deletes
    nothing and skips any run with a job row that is not yet terminal — probe those first.
    `Forecaster.monitor` (``probe=True``) is the usual way in, but it needs a config;
    `probes.reconcile.probe_run` takes a bare ``run_id``, which is what you have here. See
    `registry.ops.close_runs`.
    """
    from .registry import ops

    return ops.close_runs(list(run_ids) or None, settings=self._settings, yes=yes)

detect_anomalies_df

detect_anomalies_df(
    run_id,
    *,
    model_name=None,
    anomaly_prob_threshold=0.95,
    cutoff_back_steps=None,
    anomalies_only=False,
    actuals_df=None,
    engine="auto",
    limit=50000,
)

Detect time-series anomalies on any completed run_id in this registry without needing its original config file. Delegates to :meth:Forecaster.detect_anomalies_df.

Source code in src/scale_forecasting/sdk.py

def detect_anomalies_df(
    self,
    run_id: str,
    *,
    model_name: str | None = None,
    anomaly_prob_threshold: float = 0.95,
    cutoff_back_steps: int | None = None,
    anomalies_only: bool = False,
    actuals_df: pd.DataFrame | None = None,
    engine: str = "auto",
    limit: int = 50000,
) -> pd.DataFrame:
    """Detect time-series anomalies on any completed ``run_id`` in this registry without
    needing its original config file. Delegates to :meth:`Forecaster.detect_anomalies_df`.
    """
    return self._forecaster_for_run(run_id).detect_anomalies_df(
        run_id,
        model_name=model_name,
        anomaly_prob_threshold=anomaly_prob_threshold,
        cutoff_back_steps=cutoff_back_steps,
        anomalies_only=anomalies_only,
        actuals_df=actuals_df,
        engine=engine,
        limit=limit,
    )

detect_change_points_df

detect_change_points_df(
    run_id,
    *,
    model_name=None,
    exclude_gap_edges=True,
    min_gap_steps=1,
    sigma_threshold=3.0,
    engine="auto",
    limit=50000,
)

Detect structural regime change points on any completed run_id in this registry (either on the run's raw source series or on a specific model_name's residuals). Delegates to :meth:Forecaster.detect_change_points_df.

Source code in src/scale_forecasting/sdk.py

def detect_change_points_df(
    self,
    run_id: str,
    *,
    model_name: str | None = None,
    exclude_gap_edges: bool = True,
    min_gap_steps: int = 1,
    sigma_threshold: float = 3.0,
    engine: str = "auto",
    limit: int = 50000,
) -> pd.DataFrame:
    """Detect structural regime change points on any completed ``run_id`` in this registry
    (either on the run's raw source series or on a specific ``model_name``'s residuals).
    Delegates to :meth:`Forecaster.detect_change_points_df`.
    """
    return self._forecaster_for_run(run_id).detect_change_points_df(
        run_id,
        model_name=model_name,
        exclude_gap_edges=exclude_gap_edges,
        min_gap_steps=min_gap_steps,
        sigma_threshold=sigma_threshold,
        engine=engine,
        limit=limit,
    )

doctor

doctor()

A read-only registry.ops.DoctorReport — row counts, stuck runs, orphaned artifacts.

Source code in src/scale_forecasting/sdk.py

def doctor(self) -> Any:
    """A read-only `registry.ops.DoctorReport` — row counts, stuck runs, orphaned artifacts."""
    from .registry import ops

    return ops.doctor(settings=self._settings)

drop_run

drop_run(*run_ids, yes=False, force=False)

Delete named run(s) — artifacts, then BQML models, then rows. Preview unless yes.

See registry.ops.drop_run for the ordering rule and the in-flight refusal force overrides.

Source code in src/scale_forecasting/sdk.py

def drop_run(self, *run_ids: str, yes: bool = False, force: bool = False) -> Any:
    """Delete named run(s) — artifacts, then BQML models, then rows. Preview unless ``yes``.

    See `registry.ops.drop_run` for the ordering rule and the in-flight refusal ``force``
    overrides.
    """
    from .registry import ops

    return ops.drop_run(list(run_ids), settings=self._settings, yes=yes, force=force)

ensemble_runs

ensemble_runs(
    source_run_ids, strategies=None, *, target_run_id=None
)

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

Delegates to ensemble_run.ensemble_cross_runs: validates that the source runs share the same source table, horizon, and backtest geometry, copies deduplicated base rows into target_run_id (default: make_run_id(merged_cfg)), and runs the requested calculated and/or learned ensemble strategies so all base models and cross-run ensembles appear together on v_model_leaderboard and review_run.

Source code in src/scale_forecasting/sdk.py

def ensemble_runs(
    self,
    source_run_ids: Sequence[str],
    strategies: list[str] | None = None,
    *,
    target_run_id: str | None = None,
) -> str:
    """Blend base models across multiple completed runs into a consolidated cross-run ensemble;
    return ``target_run_id``.

    Delegates to `ensemble_run.ensemble_cross_runs`: validates that the source runs share the
    same source table, horizon, and backtest geometry, copies deduplicated base rows into
    ``target_run_id`` (default: ``make_run_id(merged_cfg)``), and runs the requested calculated
    and/or learned ensemble strategies so all base models and cross-run ensembles appear
    together on ``v_model_leaderboard`` and `review_run`.
    """
    from .ensemble_run import ensemble_cross_runs

    return ensemble_cross_runs(
        source_run_ids,
        strategies=strategies,
        target_run_id=target_run_id,
        settings=self._settings,
    )

export

export(destination_root, *, fmt='PARQUET')

Export every registry table to GCS; returns {table: destination prefix}.

Source code in src/scale_forecasting/sdk.py

def export(self, destination_root: str, *, fmt: str = "PARQUET") -> dict[str, str]:
    """Export every registry table to GCS; returns ``{table: destination prefix}``."""
    from .registry import ops

    return ops.export(destination_root, settings=self._settings, fmt=fmt)

init

init(*, create_dataset=False)

Create this registry's tables + views (idempotent). See registry.ops.init.

Source code in src/scale_forecasting/sdk.py

def init(self, *, create_dataset: bool = False) -> str:
    """Create this registry's tables + views (idempotent). See `registry.ops.init`."""
    from .registry import ops

    return ops.init(settings=self._settings, create_dataset=create_dataset)

probe

probe(run_id, job=None)

Reconcile run_id against its live cloud runtime (probes.reconcile.probe_run).

Source code in src/scale_forecasting/sdk.py

def probe(self, run_id: str, job: str | None = None) -> Any:
    """Reconcile ``run_id`` against its live cloud runtime (`probes.reconcile.probe_run`)."""
    from .probes.reconcile import probe_run

    return probe_run(run_id, job=job, settings=self._settings)

reap_clusters

reap_clusters(
    *, regions=None, min_age_seconds=None, yes=False
)

Delete Ray clusters belonging to this registry whose run has already finished.

The compute counterpart of sweep_orphans, and the only protection against a launcher that was killed before its teardown could run — a Vertex Ray cluster has no idle TTL to fall back on the way a Dataproc one does. Preview unless yes. See ray_reaper.reap_clusters for the rules that decide a cluster is abandoned, and why an ambiguous one is left running.

Source code in src/scale_forecasting/sdk.py

def reap_clusters(
    self,
    *,
    regions: list[str] | None = None,
    min_age_seconds: float | None = None,
    yes: bool = False,
) -> Any:
    """Delete Ray clusters belonging to this registry whose run has already finished.

    The compute counterpart of `sweep_orphans`, and the only protection against a launcher that
    was killed before its teardown could run — a Vertex Ray cluster has no idle TTL to fall back
    on the way a Dataproc one does. Preview unless ``yes``. See `ray_reaper.reap_clusters` for
    the rules that decide a cluster is abandoned, and why an ambiguous one is left running.
    """
    from .ray_reaper import DEFAULT_MIN_AGE_SECONDS, reap_clusters

    return reap_clusters(
        settings=self._settings,
        regions=regions,
        min_age_seconds=(
            DEFAULT_MIN_AGE_SECONDS if min_age_seconds is None else min_age_seconds
        ),
        yes=yes,
    )

revision_stability_df

revision_stability_df(
    run_ids, *, jump_threshold_pct=0.3, limit=20000
)

Measure cross-run forecast revision stability ("whipsaw") across run_ids.

Source code in src/scale_forecasting/sdk.py

def revision_stability_df(
    self,
    run_ids: Sequence[str],
    *,
    jump_threshold_pct: float = 0.30,
    limit: int = 20000,
) -> pd.DataFrame:
    """Measure cross-run forecast revision stability ("whipsaw") across ``run_ids``."""
    import pandas as pd

    from .diagnostics import compute_revision_stability
    from .registry.reads import read_predictions

    by_run: dict[str, pd.DataFrame] = {}
    for rid in run_ids:
        rows = read_predictions(rid, limit=limit, settings=self._settings)
        by_run[str(rid)] = pd.DataFrame.from_records(rows) if rows else pd.DataFrame()
    return compute_revision_stability(by_run, jump_threshold_pct=jump_threshold_pct)

runs_df

runs_df(*, limit=25, status=None)

Return a DataFrame of the most recent runs in this registry from v_run_summary, newest first, optionally filtered by status.

Beside the header columns comes the run's time ledger, rolled up from its job rows: n_jobs, longest_job_seconds (the critical-path family), jobs_seconds (every job's wall added up — more than the run's own wall when families overlapped), jobs_span_seconds (first job start to last job end), and overhead_seconds / overhead_fraction (the span the slowest job did not account for — ensemble, native job, scheduler gaps; never negative). dcu_milli_seconds is the last Dataproc Serverless batch's usage, NULL on other runtimes. See registry.views.

Source code in src/scale_forecasting/sdk.py

def runs_df(self, *, limit: int = 25, status: str | None = None) -> pd.DataFrame:
    """Return a ``DataFrame`` of the most recent runs in this registry from ``v_run_summary``,
    newest first, optionally filtered by ``status``.

    Beside the header columns comes the run's time ledger, rolled up from its job rows:
    ``n_jobs``, ``longest_job_seconds`` (the critical-path family), ``jobs_seconds`` (every
    job's wall added up — more than the run's own wall when families overlapped),
    ``jobs_span_seconds`` (first job start to last job end), and ``overhead_seconds`` /
    ``overhead_fraction`` (the span the slowest job did not account for — ensemble, native
    job, scheduler gaps; never negative). ``dcu_milli_seconds`` is the last Dataproc
    Serverless batch's usage, NULL on other runtimes. See `registry.views`.
    """
    import pandas as pd

    from .registry.reads import RECENT_RUNS_COLUMNS, read_recent_runs

    cols = list(RECENT_RUNS_COLUMNS)
    rows = read_recent_runs(limit=limit, status=status, settings=self._settings)
    if not rows:
        return pd.DataFrame(columns=cols)
    return pd.DataFrame.from_records(rows, columns=cols)

score_realized

score_realized(
    run_id,
    actuals_df,
    *,
    metric="wape",
    drift_threshold=1.5,
    limit=20000,
)

Join run_id predictions against actuals_df and compute realized accuracy.

Source code in src/scale_forecasting/sdk.py

def score_realized(
    self,
    run_id: str,
    actuals_df: pd.DataFrame,
    *,
    metric: str = "wape",
    drift_threshold: float = 1.5,
    limit: int = 20000,
) -> pd.DataFrame:
    """Join ``run_id`` predictions against ``actuals_df`` and compute realized accuracy."""
    import pandas as pd

    from .diagnostics import compute_realized_accuracy
    from .registry.reads import read_oof, read_predictions

    pred_rows = read_predictions(run_id, limit=limit, settings=self._settings)
    oof_rows = read_oof(run_id, limit=limit, settings=self._settings)
    pred_df = pd.DataFrame.from_records(pred_rows) if pred_rows else pd.DataFrame()
    oof_df = pd.DataFrame.from_records(oof_rows) if oof_rows else None
    return compute_realized_accuracy(
        pred_df,
        actuals_df,
        oof_df=oof_df,
        metric=metric,
        drift_threshold=drift_threshold,
    )

snapshot

snapshot(suffix, *, into=None, expiration_days=None)

Table-snapshot every registry table; returns {table: snapshot ref}.

Source code in src/scale_forecasting/sdk.py

def snapshot(
    self, suffix: str, *, into: str | None = None, expiration_days: int | None = None
) -> dict[str, str]:
    """Table-snapshot every registry table; returns ``{table: snapshot ref}``."""
    from .registry import ops

    return ops.snapshot(
        suffix, settings=self._settings, into=into, expiration_days=expiration_days
    )

sweep_orphans

sweep_orphans(*, yes=False)

Delete artifacts under this registry with no run_registry row.

Preview unless yes. See registry.ops.sweep_orphans.

Source code in src/scale_forecasting/sdk.py

def sweep_orphans(self, *, yes: bool = False) -> Any:
    """Delete artifacts under this registry with no ``run_registry`` row.

    Preview unless ``yes``. See `registry.ops.sweep_orphans`.
    """
    from .registry import ops

    return ops.sweep_orphans(settings=self._settings, yes=yes)

RunResult dataclass

RunResult(
    run_id,
    dataset_ref,
    views,
    status=None,
    runtime_seconds=None,
)

A pointer to where a run's results live — the run_id plus how to query them.

dataset_ref is project.dataset (None when the GCP identity can't be resolved, e.g. an offline Forecaster.review); views are the registry view names to query under it (e.g. v_run_summary, v_model_leaderboard). Filter any of them by run_id. status and runtime_seconds carry the terminal outcome when returned by Forecaster.run or Forecaster.run_live.

build_best_params_frame

build_best_params_frame(rows)

Unpack read_best_params rows (with JSON best_params) into a tidy DataFrame (pure, offline).

Source code in src/scale_forecasting/sdk.py

def build_best_params_frame(rows: list[dict[str, Any]]) -> pd.DataFrame:
    """Unpack ``read_best_params`` rows (with JSON ``best_params``) into a tidy ``DataFrame``
    (pure, offline).
    """
    import json

    import pandas as pd

    cols = [
        "ts_id",
        "model_type",
        "compute_engine",
        "is_ensemble",
        "best_params_json",
        "best_params",
        "fit_seconds",
        "wape",
        "mae",
        "rmse",
        "mase",
    ]
    if not rows:
        return pd.DataFrame(columns=cols)
    records: list[dict[str, Any]] = []
    for r in rows:
        raw_bp = r.get("best_params")
        parsed: Any = None
        if isinstance(raw_bp, str) and raw_bp:
            try:
                parsed = json.loads(raw_bp)
            except ValueError:
                parsed = raw_bp
        elif isinstance(raw_bp, dict):
            parsed = raw_bp
        records.append(
            {
                "ts_id": str(r.get("ts_id") or ""),
                "model_type": str(r.get("model_type") or ""),
                "compute_engine": r.get("compute_engine"),
                "is_ensemble": r.get("ensemble_id") is not None,
                "best_params_json": (
                    json.dumps(parsed, sort_keys=True) if isinstance(parsed, dict) else str(raw_bp)
                ),
                "best_params": parsed,
                "fit_seconds": r.get("fit_seconds"),
                "wape": r.get("wape"),
                "mae": r.get("mae"),
                "rmse": r.get("rmse"),
                "mase": r.get("mase"),
            }
        )
    return pd.DataFrame.from_records(records, columns=cols)

build_explain_frame

build_explain_frame(cfg)

Build a per-family execution plan DataFrame for cfg — pure, offline, no GCP.

One row per node in dag.plan_dag (each active model family plus the downstream ensemble node when enabled), summarizing run_id, family, job_key, runtime, machine_type, workers, hardware, gpu_type, models, training_modes, n_series, n_folds, expected_cells, covariates, and depends_on.

Source code in src/scale_forecasting/sdk.py

def build_explain_frame(cfg: RunConfig) -> pd.DataFrame:
    """Build a per-family execution plan ``DataFrame`` for ``cfg`` — pure, offline, no GCP.

    One row per node in `dag.plan_dag` (each active model family plus the downstream ``ensemble``
    node when enabled), summarizing ``run_id``, ``family``, ``job_key``, ``runtime``,
    ``machine_type``, ``workers``, ``hardware``, ``gpu_type``, ``models``, ``training_modes``,
    ``n_series``, ``n_folds``, ``expected_cells``, ``covariates``, and ``depends_on``.
    """
    import pandas as pd

    from .dag import dag_nodes, plan_dag
    from .models import get_model

    run_dag = plan_dag(cfg)
    nodes = dag_nodes(run_dag)
    jobs_by_fam = {j.family: j for j in run_dag.jobs}
    workload = estimate_workload(cfg)

    cov_parts: list[str] = []
    if cfg.features.known_future_covariates:
        cov_parts.append(f"future({len(cfg.features.known_future_covariates)})")
    if cfg.features.past_covariates:
        cov_parts.append(f"past({len(cfg.features.past_covariates)})")
    if cfg.features.static_covariates:
        cov_parts.append(f"static({len(cfg.features.static_covariates)})")
    cov_summary = ", ".join(cov_parts) if cov_parts else "none"

    records: list[dict[str, Any]] = []
    for node in nodes:
        fjob = jobs_by_fam.get(node.family)
        comp = fjob.compute if fjob is not None else None
        # The three branches bind these to different shapes — SQL nodes have no machine; a family
        # node's shape may be unresolved (``None``) until the submitter sizes it.
        modes_str: str
        machine_type: str | None
        workers: int | None
        if node.family == "ensemble":
            models_str = ", ".join(f"ensemble_{s}" for s in cfg.ensemble.strategies)
            modes_str = cfg.compute.ensemble.mode
            n_cells = (
                workload.n_series * len(cfg.ensemble.strategies)
                if workload.n_series is not None
                else None
            )
            machine_type = "sql"
            workers = 1
        elif node.family == "native":
            models_str = ", ".join(node.models)
            modes_str = ", ".join(f"{m}:local" for m in node.models)
            n_cells = (
                workload.n_series * len(node.models) if workload.n_series is not None else None
            )
            machine_type = "sql"
            workers = 1
        else:
            models_str = ", ".join(node.models)
            modes_list: list[str] = []
            for m in node.models:
                m_cls = get_model(m)
                default_mode = "global" if m_cls.family == "automl" else "local"
                m_mode = str(cfg.model_params.get(m, {}).get("training_mode", default_mode))
                modes_list.append(f"{m}:{m_mode}")
            modes_str = ", ".join(modes_list)
            n_cells = (
                workload.n_series * len(node.models) if workload.n_series is not None else None
            )
            machine_type = comp.machine_type if comp is not None else "auto"
            workers = comp.workers if comp is not None else 1
            if (
                comp is not None
                and comp.workers is not None
                and (comp.runtime == "vertex" or (comp.runtime == "gke" and comp.gke_mode == "job"))
            ):
                from .engines.vertex_engine import effective_worker_count

                workers = effective_worker_count(
                    cfg,
                    node.models,
                    comp.workers,
                    n_series=workload.n_series,
                )

        records.append(
            {
                "run_id": run_dag.run_id,
                "family": node.family,
                "job_key": node.job_key,
                "runtime": node.runtime,
                "machine_type": machine_type,
                "workers": workers,
                "hardware": node.hardware or "sql",
                "gpu_type": node.gpu_type,
                "models": models_str,
                "training_modes": modes_str,
                "n_series": workload.n_series,
                "n_folds": workload.n_folds,
                "expected_cells": n_cells,
                "covariates": cov_summary,
                "depends_on": ", ".join(node.depends_on) if node.depends_on else "",
            }
        )
    return pd.DataFrame.from_records(records)

build_trace_frame

build_trace_frame(job_rows, cell_rows)

Stack per-job and per-cell timing rows into one long-form trace frame — pure, no I/O.

job_rows are v_run_jobs rows (see registry.jobs.read_run_jobs) and cell_rows are per-cell timing rows (see registry.reads.read_cell_timing); this is the pure assembly step Forecaster.trace calls after reading them, split out so it unit-tests without GCP. Each input row becomes one span record with the shared _TRACE_COLUMNS schema: a job lands on its family lane, a cell on its worker_id lane, so a plot can show the DAG over the workers that ran it. Rows without a start stamp are dropped (nothing to place on a timeline); a row with a start but no end (a job that never recorded completion) keeps a zero-width span at its start. Returns an empty frame carrying the columns when both inputs are empty.

Source code in src/scale_forecasting/sdk.py

def build_trace_frame(
    job_rows: list[dict[str, Any]], cell_rows: list[dict[str, Any]]
) -> pd.DataFrame:
    """Stack per-job and per-cell timing rows into one long-form trace frame — pure, no I/O.

    ``job_rows`` are ``v_run_jobs`` rows (see `registry.jobs.read_run_jobs`) and ``cell_rows`` are
    per-cell timing rows (see `registry.reads.read_cell_timing`); this is the pure assembly step
    `Forecaster.trace` calls after reading them, split out so it unit-tests without GCP. Each input
    row becomes one span record with the shared `_TRACE_COLUMNS` schema: a job lands on its
    ``family`` lane, a cell on its ``worker_id`` lane, so a plot can show the DAG over the workers
    that ran it. Rows without a start stamp are dropped (nothing to place on a timeline); a row with
    a start but no end (a job that never recorded completion) keeps a zero-width span at its start.
    Returns an empty frame carrying the columns when both inputs are empty.
    """
    import pandas as pd

    records: list[dict[str, Any]] = []
    for r in job_rows:
        start = r.get("started_at")
        if start is None:
            continue
        end = r.get("ended_at")
        records.append(
            {
                "kind": "job",
                "lane": r.get("family") or "",
                "label": r.get("family") or r.get("job_id") or "",
                "start": start,
                "end": end if end is not None else start,
                "duration_s": _duration_s(start, end, r.get("runtime_seconds")),
                "status": r.get("status"),
                "runtime": r.get("runtime"),
                "model_type": None,
                "ts_id": None,
            }
        )
    for r in cell_rows:
        start = r.get("cell_started_at")
        if start is None:
            continue
        end = r.get("cell_ended_at")
        model_type = r.get("model_type")
        ts_id = r.get("ts_id")
        records.append(
            {
                "kind": "cell",
                "lane": r.get("worker_id") or "",
                "label": f"{model_type}:{ts_id}",
                "start": start,
                "end": end if end is not None else start,
                "duration_s": _duration_s(start, end, None),
                "status": None,
                "runtime": r.get("compute_engine"),
                "model_type": model_type,
                "ts_id": ts_id,
            }
        )
    return pd.DataFrame.from_records(records, columns=list(_TRACE_COLUMNS))

build_variant_matrix_frame

build_variant_matrix_frame(cfg)

Build a per-variant taxonomy and spec-hash DataFrame for cfg — pure, offline.

Source code in src/scale_forecasting/sdk.py

def build_variant_matrix_frame(cfg: RunConfig) -> pd.DataFrame:
    """Build a per-variant taxonomy and spec-hash ``DataFrame`` for ``cfg`` — pure, offline."""
    import pandas as pd

    from .features import resolve_model_variant
    from .registry.ids import make_model_spec_hash

    cols = [
        "model_type",
        "base_model",
        "variant_tag",
        "family",
        "scope_capability",
        "training_mode",
        "cov_profile",
        "is_univariate",
        "covariates_used",
        "model_spec_hash",
    ]
    records: list[dict[str, Any]] = []
    for m in cfg.models:
        spec = resolve_model_variant(cfg, m)
        records.append(
            {
                "model_type": m,
                "base_model": spec.base_model,
                "variant_tag": spec.variant_tag,
                "family": str(spec.model_cls.family),
                "scope_capability": spec.scope_capability,
                "training_mode": spec.training_mode,
                "cov_profile": spec.cov_profile,
                "is_univariate": spec.is_univariate,
                "covariates_used": ", ".join(spec.covariates_used),
                "model_spec_hash": make_model_spec_hash(m, cfg),
            }
        )
    return pd.DataFrame.from_records(records, columns=cols)

plot_trace

plot_trace(frame, *, ax=None, title='run trace')

Render a build_trace_frame frame as a Gantt/waterfall and return the matplotlib Axes.

A convenience over the frame (the frame is the real deliverable): one horizontal bar per span, lanes stacked on the y-axis (jobs and cells kept apart so a family job and a worker_id cell never share a track), colored by kind (job vs cell). Pass an existing ax to compose into a larger figure, or let it create one. matplotlib imports lazily here so it never touches the near-instant import scale_forecasting path. An empty frame renders an empty titled axes rather than raising.

Source code in src/scale_forecasting/sdk.py

def plot_trace(frame: pd.DataFrame, *, ax: Any = None, title: str = "run trace") -> Any:
    """Render a `build_trace_frame` frame as a Gantt/waterfall and return the matplotlib ``Axes``.

    A convenience over the frame (the frame is the real deliverable): one horizontal bar per span,
    lanes stacked on the y-axis (jobs and cells kept apart so a ``family`` job and a ``worker_id``
    cell never share a track), colored by ``kind`` (job vs cell). Pass an existing ``ax`` to compose
    into a larger figure, or let it create one. matplotlib imports lazily here so it never touches
    the near-instant ``import scale_forecasting`` path. An empty frame renders an empty titled axes
    rather than raising.
    """
    from .errors import require_extra

    require_extra("notebook", purpose="Plotting")
    import matplotlib.dates as mdates
    import matplotlib.pyplot as plt

    if ax is None:
        _, ax = plt.subplots(figsize=(11, 6))
    ax.set_title(title if not frame.empty else f"{title} (no timed rows)")
    if frame.empty:
        return ax

    lane_keys = list(dict.fromkeys(zip(frame["kind"], frame["lane"], strict=True)))
    colors = {"job": "#4C72B0", "cell": "#DD8452"}
    for y, (kind, lane) in enumerate(lane_keys):
        grp = frame[(frame["kind"] == kind) & (frame["lane"] == lane)]
        spans = [
            (mdates.date2num(row.start), mdates.date2num(row.end) - mdates.date2num(row.start))
            for row in grp.itertuples(index=False)
        ]
        ax.broken_barh(spans, (y - 0.4, 0.8), facecolors=colors.get(kind, "#999999"))
    ax.set_yticks(range(len(lane_keys)))
    ax.set_yticklabels([f"{kind}:{lane}" for kind, lane in lane_keys])
    ax.xaxis_date()
    ax.set_xlabel("wall-clock time")
    return ax