Model Pipeline Overview

The ModelPipeline class trains and evaluates forecasting models with rolling-window validation.

Basic Usage

from epftoolbox2.pipelines import ModelPipeline
from epftoolbox2.models import OLSModel, LassoCVModel
from epftoolbox2.evaluators import MAEEvaluator
from epftoolbox2.exporters import ExcelExporter, TerminalExporter

pipeline = (
    ModelPipeline()
    .add_model(OLSModel(predictors=predictors, name="OLS"))
    .add_model(LassoCVModel(predictors=predictors, name="LassoCV"))
    .add_evaluator(MAEEvaluator())
    .add_exporter(TerminalExporter())
    .add_exporter(ExcelExporter("results.xlsx"))
)

report = pipeline.run(
    data=df,
    test_start="2024-02-01",
    test_end="2024-03-01",
    target="price",
    horizon=7,
    freq="1h",                   # Optional. Allows sub-hourly testing like "15min"
    save_dir="results",
)

Pipeline Components

Models

OLSModel Ordinary Least Squares

LassoCVModel Lasso with cross-validation

Evaluators

MAEEvaluator Mean Absolute Error

Exporters

TerminalExporter Console output

ExcelExporter Excel file

---

Predictor Specification

Predictors can be specified in four ways:

Static Columns

predictors = [
    "load_actual",
    "load_actual_d-1",
    "price_d-1",
]

Templates

Use {horizon} placeholder: replaced with 1, 2, …, horizon at runtime:

predictors = [
    "warsaw_temperature_2m_d+{horizon}",
    "is_monday_d+{horizon}",
    "is_holiday_d+{horizon}",
    "daylight_hours_d+{horizon}",
]

Tip

String templates are the only form compatible with YAML serialization. Use them instead of lambdas when saving pipelines with save() or Workflow.

Callables

Callables receive horizon (1 param) or (horizon, hour) (2 params):

# horizon-only
predictors = [
    lambda horizon: f"weather_d+{horizon}",
]

# horizon + hour: different feature per hour of day
predictors = [
    lambda horizon, hour: f"load_h{hour:02d}_d+{horizon}",
]

Caution

Callables cannot be serialized to YAML. If you need to save the pipeline with save() or use Workflow, replace callables with "{horizon}" string templates.

Loop Generation

Use list comprehensions for many features:

predictors = [
    "load_actual",
    *[f"load_actual_h-{i}" for i in range(1, 169)],
    *[f"price_d-{i}" for i in range(1, 8)],
]

---

Parallelism

Models use a process pool with inner thread pools. Each worker process handles one forecast day (24 hours × horizon), keeping results consistent and progress tracking stable.

Variable	Default	Description
MAX_PROCESSES	cpu_count // THREADS_PER_PROCESS	Worker processes
THREADS_PER_PROCESS	16	Threads per process

import os

# Must be set before any imports
os.environ["PYTHON_GIL"] = "0"            # Enable free-threading (Python 3.13t+)
os.environ["MAX_PROCESSES"] = "4"
os.environ["THREADS_PER_PROCESS"] = "16"  # → 4 × 16 = 64 cores
os.environ["OMP_NUM_THREADS"] = "1"       # Prevent BLAS oversubscription

See Installation for full setup including platform notes.

---

Feature Scaling

Models scale each task’s data with a scaler fitted on that task’s training rows only. By default this is StandardScaler:

1. Numeric features are standardized (mean=0, std=1)

2. Binary features (0/1) are auto-detected and skipped

3. Target variable is scaled, predictions are inverse-scaled

Set a model’s scaler parameter to "asinh" or "npit" to use a variance-stabilising transformation instead, which damps price spikes and handles negative prices. See Scalers and VSTs.

---

Run Parameters

Parameter	Type	Default	Description
data	DataFrame	Required	Input data with DatetimeIndex
test_start	str	Required	Test period start
test_end	str	Required	Test period end
target	str	"price"	Target column name
horizon	int	7	Max forecast horizon (days)
freq	str	"1h"	Forecasting frequency (e.g., "15min")
save_dir	str	None	Directory for incremental results
forecast_only	bool	False	Skip evaluators/exporters. test_start/test_end default to "today" if omitted

Model names must be unique: they name the result files in save_dir.

---

Data Checks

Before any model is fitted, run() checks every model’s input and raises ValueError with a description of the problem:

• Resolution: rows must be freq apart, with one row per local day and period (hour, or quarter-hour for freq="15min"). Only the hour repeated when clocks fall back may occur twice. Data finer than freq would put the issue day’s prices into the training set. Resample it first (ResampleTransformer) or pass the matching freq.
• Timestamps: no duplicates. Rows are sorted by time.
• Names: the data must not contain the model’s target columns ({target}_d+1 … {target}_d+{horizon}, e.g. from a LagTransformer with negative lags on the target) or columns named hour or day, which the model uses internally (use CalendarSource(prefix=...) for its hour column).
• Predictors: every expanded predictor (for every hour and horizon) must exist, be numeric, and must not be one of the model’s targets.

Only the target and the predictors are used, so other columns (e.g. text columns such as holiday_name) can stay in the data.

---

Information Set

The checks above don’t know when a forecast is issued, so they can’t tell a legitimate lag from look-ahead (same-day load_actual passes them, even though it isn’t known until after the hour it describes). Pass issue_time to ModelPipeline to add that check:

pipeline = ModelPipeline(issue_time="11:45", issue_timezone="Europe/Brussels")

This raises InformationSetError before any fit if a predictor (or the newest training target) is not known at issue time, using built-in availability rules for the shipped sources (ENTSO-E prices/load/generation, Open-Meteo weather, calendar). See Information Set for the full rule reference, custom rules, and the empirical leakage_probe.

---

EvaluationReport

The pipeline returns an EvaluationReport:

report.summary()           # Overall metrics, with a `forecasts` count per model
report.forecast_counts()   # Planned vs produced forecasts and skip reasons
report.by_hour()           # Breakdown by hour
report.by_horizon()        # Breakdown by horizon
report.by_hour_horizon()   # Combined breakdown

See Working with Results for what the counts mean.

---

YAML Serialization

Save and reload a complete pipeline configuration:

pipeline.save("model_pipeline.yaml")
pipeline = ModelPipeline.load("model_pipeline.yaml")

See ModelPipeline Serialization for the full YAML format and serialization rules.

---

Learn More

Examples Complete pipeline examples

Results Working with evaluation results

Caching Incremental training

Workflow YAML-based experiment orchestration

Forecasting Forecast-only mode and date keywords

Information Set Check predictors against the forecast issue time