`macroforecast.pipeline`#

A comprehensive pseudo-out-of-sample (POOS) forecasting pipeline that orchestrates the existing macroforecast building blocks. You declare a set of arms (each a full preprocessing/features/model configuration) and the pipeline runs them through forecasting.run, evaluates every contender against a benchmark with relative RMSE, Diebold-Mariano, Clark-West and the Model Confidence Set, builds forecast combinations, and assembles a single PipelineReport. Targets are resolved from the FRED-MD/QD t-code, and ML arms can be interpreted (SHAP/ALE/PDP) without re-running.

Terminology#

The pipeline uses four terms consistently.

cell – the execution unit the pipeline manages: one arm applied to one target over the window for a horizon-group. The pipeline enumerates and schedules cells, and each cell is executed by exactly one run() call. The serial backend groups all horizons of a (target, arm) into one cell; the parallel backend splits one horizon per cell.
run() – the atomic forecasting function (forecasting.run) that executes one cell (one model, one target, over the window). “run” is the function and the verb; the execution unit it produces is a cell, not “a run”.
arm – a target-agnostic configuration (preprocessing + features + a single model). It is not a cell by itself; applied to a target and a horizon it forms a cell, and in the evaluation it appears as one contender.
contender – an arm as it appears in the evaluation comparison, the named entity compared via DM/CW/MCS. One arm is one contender; forecast combinations are additional contenders.

In one line: arm × target × horizon → cell = one run(); arm = contender.

Entry points#

Symbol	Summary
`pipeline_spec(...)`	Validating generator that builds a `PipelineSpec`.
`model_arms(models, ...)`	Build one `Arm` per model for a pure model comparison.
`run_pipeline(spec)`	Run arms, evaluate, and return a `PipelineReport`.
`interpret_pipeline(report, *, methods, which_fit, arms)`	Deferred multi-method ML interpretation.
`run_arms(spec)`	Execute every cell into the master forecast frame (lower-level; name retained for back-compat).
`evaluate(master, spec)`	Accuracy + DM/CW + MCS + combinations on a master frame.
`apply_combinations(master, spec)`	Append cross-arm combination contenders.
`resolve_target(target, *, data, tcode, tcode_map, reduce_i2)`	Resolve a target’s (policy, transform).
`contender_names(arm)`	The contender label for an arm. A contender IS exactly an arm (one model per arm), so this returns `[arm.name]`.

Configuration objects#

PipelineSpec – validated, frozen run configuration.
Arm – a target-agnostic configuration with exactly ONE model: name, model, preprocessing, features, params, interpret. An arm is one contender in the evaluation; applied to a target and horizon it forms a cell. Comparing models means multiple arms identical except for model; comparing feature cases means arms differing in features. A list/mapping of models in one arm is rejected.
TargetSpec – a target and how its forecast object is defined (transform, policy, reduce_i2).
ResolvedTarget – a target with its forecast policy and transform resolved.
InterpretSpec – ML interpretation request (methods, which_fit, background, top_k).
EvalSpec – benchmark, metrics, tests, primary_axis, MCS settings, subsamples.
CombinationContender – a forecast combination that becomes an additional contender.
PipelineReport – output: forecasts, accuracy, significance, mcs, provenance, leakage_audit, interpretation, failed_cells. failed_cells lists any (target, arm, horizon-group) cell whose run() raised; the rest of the cells still ran and the failures are also mirrored into leakage_audit["failed_cells"].
TCODE_TARGET_MAP – FRED t-code to (forecast_policy, target_transform) mapping.

t-code to target mapping#

The forecast object is the h-period cumulation (direct_average) of the transformation implied by the target’s t-code, not the raw single-period transform. I(2) price/level series are reduced to the first-difference object (reduce_i2).

t-code	forecast policy	transform
1	direct	level
2	direct_average	change
3	direct_average	change
4	direct	level
5	direct_average	log_growth
6	direct_average	log_growth
7	direct_average	growth

Example#

import macroforecast as mf
from macroforecast.pipeline import Arm, EvalSpec, CombinationContender, pipeline_spec, run_pipeline

spec = pipeline_spec(
    data=bundle,                                  # carries FRED transform_codes
    targets=["INDPRO", "CPIAUCSL"],               # resolved from t-codes
    horizons=[1, 3, 12],
    window=mf.window.from_cutoffs(test_start="1990-01", horizon=1, embargo=0),
    arms=[
        Arm("AR", model="ar"),                    # benchmark
        Arm("RF", model="random_forest", interpret=("shap", "ale")),
        Arm("MARX", model="ridge", preprocessing=marx_spec),   # transformation comparison
    ],
    evaluation=EvalSpec(benchmark="AR", tests=["dm", "cw", "mcs"]),
    combinations=[CombinationContender("POOL", method="constrained_ls")],
    n_jobs=8,                                     # fan (arm × target × horizon) cells across 8 processes
)
report = run_pipeline(spec)
report.accuracy        # contender x target x horizon relative RMSE (common sample)
report.significance    # Diebold-Mariano and Clark-West p-values vs the benchmark
report.mcs             # Model Confidence Set membership
report.leakage_audit   # window.validate() warnings (per horizon)

from macroforecast.pipeline import interpret_pipeline
interpret_pipeline(report, methods=("shap", "ale"))   # deferred, no re-forecast

Comparing models (`model_arms`)#

Since an Arm is exactly ONE model, a pure model comparison is “several arms identical except model”. Writing one Arm per model by hand is verbose, so model_arms(...) builds them for you:

from macroforecast.pipeline import model_arms, pipeline_spec, EvalSpec

arms = model_arms(["ar", "random_forest", "far"], features=feats, preprocessing=pp)
spec = pipeline_spec(..., arms=arms, evaluation=EvalSpec(benchmark="ar"))

Each model becomes one Arm (one contender; each of its (target, horizon) is one cell run by one run()), and the contender is the arm name. Arm names default to the model name (str(model) / ModelSpec.name / callable.__name__); pass a Mapping[name -> model] or names=[...] to label them explicitly. All arms share the given preprocessing, features, and evaluation config.

params and model_selection are shared by every arm unless given as a Mapping whose keys are exactly the arm names, in which case each entry is applied to its own arm. A plain shared dict of hyperparameters (whose keys are hyperparameter names, not arm names) is therefore unambiguously shared. nested_in_benchmark is a bool shared by all, or a set/sequence of arm names that nest the benchmark.

This shares all config, so it is for pure model comparison. Comparing feature cases (arms differing in features or preprocessing) still needs explicit Arm objects built by hand.

Parallel execution (`n_jobs`)#

pipeline_spec(..., n_jobs=N) parallelises the pseudo-out-of-sample replication natively, so a fan-out across cores no longer needs hand-rolled shell processes.

n_jobs=1 (default) keeps the sequential path byte-for-byte: each (target, arm) is one cell run as a consolidated multi-horizon run() that shares a single per-origin preprocessing cache across horizons (EM imputation is computed once per origin).
n_jobs>1 splits the work into (arm × target × horizon) cells and runs them across a ProcessPoolExecutor with min(n_jobs, n_cells) workers. Each cell is a single-horizon run(), so it trades the cross-horizon EM-sharing for wall-clock parallelism: every cell recomputes its own preprocessing. On a many-core machine this finishes in a fraction of the sequential wall-clock.
The parallel path is deterministic: every cell uses spec.seed and the per-cell computation is independent of sibling cells, so the assembled forecast frame and every downstream accuracy table are numerically identical to n_jobs=1.
Workers coordinate through the existing per-(target, arm, horizon) checkpoints (checkpoint_dir), which are already namespaced so parallel processes never collide. Each worker caps nested BLAS/OpenMP threads to one to avoid oversubscribing cores.
Memory scales with n_jobs: each worker holds its own copy of the data panel, so peak memory is roughly n_jobs × the single-process footprint.

`n_jobs="auto"`#

pipeline_spec(..., n_jobs="auto") removes the need to hand-tune the worker count. It inspects the core budget (the affinity count, len(os.sched_getaffinity(0)), which respects cgroup/taskset pinning) and the work structure (len(targets) × len(arms) × len(horizons) cells), then splits the cores between cell workers and per-cell model-internal threads so the CPU is saturated without oversubscription (cell_workers × model_threads ≤ cores). Cell-level parallelism comes first (one worker per cell up to the core count); leftover cores become model-internal threads handed to the parallelizable models (random_forest, gradient_boosting, xgboost, lightgbm) inside each worker. The single-threaded models (ar, ols, ridge, lasso, elastic_net, far) ignore the thread budget and are unaffected. The resolved cell-worker count is stored as PipelineSpec.n_jobs and the per-cell thread budget as PipelineSpec.model_threads.

This also fixes a latent oversubscription: previously a tree model inside a parallel worker defaulted its internal n_jobs to 'auto' (= full CPU count), so N workers each spawned cpu_count threads. Each worker now pins its model-internal threads to model_threads. The split changes only the number of internal threads, not the numerical result (tree training is deterministic in random_state regardless of the thread count), so a n_jobs="auto" run is byte-for-byte identical to n_jobs=1.

Notes#

Accuracy uses a common sample: every contender is scored on the origins where all contenders and the realised target are observed, consistent with the MCS.
The leakage audit validates the window for each horizon, surfacing the multi-horizon pseudo-out-of-sample embargo convention.
Models are saved by default so interpret_pipeline runs without re-forecasting.
Arm.nested_in_benchmark (default False) marks an arm whose model nests the benchmark, for example a data-poor autoregressive benchmark nested in a factor-augmented or penalised model. The evaluator reports a Clark-West statistic only for such contenders, since Clark-West is valid only when the benchmark is nested within the contender. Arms that do not nest the benchmark receive Diebold-Mariano only, and forecast combinations are never treated as nested.

macroforecast.pipeline#