Workflow

Workflow ties a DataPipeline YAML, a ModelPipeline YAML, and all run parameters together into a single experiment file.

Quick Example

from epftoolbox2.pipelines import Workflow

# Phase 1: save config (once, on your laptop)
Workflow(
    data_pipeline="data_pipeline.yaml",
    model_pipeline="model_pipeline.yaml",
    data_start="2023-05-01",
    data_end="2024-08-01",
    data_cache=True,
    model_test_start="2024-06-01",
    model_test_end="2024-07-01",
    model_target="price",
    model_horizon=6,
    max_processes=32,
    threads_per_process=8,
    cache_path="/tmp/scratch/$SLURM_JOB_ID/cache",
).save("experiment.yaml")

# Phase 2: run (cluster job)
report = Workflow.load("experiment.yaml").run()
print(report.summary())

YAML Format

environment:
  max_processes: 32
  threads_per_process: 8
  cache_path: /tmp/scratch/$SLURM_JOB_ID/cache   # $VAR expanded at runtime

data_pipeline:
  path: data_pipeline.yaml    # relative to this file's directory
  start: "2023-05-01"
  end: "2024-08-01"
  cache: true

model_pipeline:
  path: model_pipeline.yaml
  test_start: "2024-06-01"
  test_end: "2024-07-01"
  target: price
  horizon: 6
  freq: 1h

Dates may be quoted or not. YAML reads an unquoted end: 2024-08-01 as a date and the workflow treats it like the string "2024-08-01", so the whole day is included.

Forecast-only variant: add forecast_only: true and use date keywords so the file runs daily without editing:

data_pipeline:
  path: data_pipeline.yaml
  start: "now_d-735"       # ~2 years of history, relative to today
  end: "today_d+6"         # whole horizon-6 day (local day, given `timezone` in data_pipeline.yaml)
  cache: true

model_pipeline:
  path: model_pipeline.yaml
  target: price
  horizon: 6
  freq: 1h                 # set to 15min for 15-minute forecasting
  forecast_only: true      # test_start/test_end default to "today", skips evaluators and exporters

Constructor Parameters

Parameter	Type	Default	Description
`data_start`	str	Required	Data fetch start date
`data_end`	str	Required	Data fetch end date
`model_test_start`	str	Required	Test period start
`model_test_end`	str	Required	Test period end
`data_pipeline`	str \| Path	`None`	Path to `data_pipeline.yaml`
`model_pipeline`	str \| Path	`None`	Path to `model_pipeline.yaml`
`data_cache`	bool \| str	`False`	Data caching (True, False, or path)
`model_target`	str	`"price"`	Target column name
`model_horizon`	int	`7`	Max forecast horizon (days)
`model_freq`	str	`"1h"`	Forecasting frequency (e.g., `"15min"`)
`max_processes`	int	`None`	Worker process count
`threads_per_process`	int	`None`	Threads per process
`cache_path`	str	`None`	Cache directory (overrides `data_cache` if set)
`model_index`	int	`None`	Run only `models[model_index]`, all models run if omitted. Requires `model_save_dir`
`model_forecast_only`	bool	`False`	Skip evaluators/exporters. `test_start`/`test_end` default to `"today"` if omitted
`model_save_dir`	str	`None`	Directory for one resumable JSONL per model (YAML: `model_pipeline.save_dir`), resolved from the YAML’s directory
`model_allow_manifest_mismatch`	bool	`False`	Resume into `model_save_dir` even if its manifest disagrees with this run’s configuration (overwrites it) (YAML: `model_pipeline.allow_manifest_mismatch`)
`model_require_manifest`	bool	`False`	Refuse to resume from a `model_save_dir` that has results but no manifest, instead of warning (YAML: `model_pipeline.require_manifest`)

Environment Variables

Workflow.run() always sets these before spawning any worker processes:

Variable	Value	Why
`OMP_NUM_THREADS`	`1`	Prevents BLAS oversubscription with multiple processes
`MKL_NUM_THREADS`	`1`	Pins Intel MKL threads
`OPENBLAS_NUM_THREADS`	`1`	Pins OpenBLAS threads
`BLAS_NUM_THREADS`	`1`	Generic BLAS thread pin
`PYTHON_GIL`	`0`	Enables free-threading on Python 3.13t+
`MAX_PROCESSES`	from config	Applied if `max_processes` is set
`THREADS_PER_PROCESS`	from config	Applied if `threads_per_process` is set

Inline Component Overrides

Pipeline sections can be specified directly in the workflow YAML, overriding the corresponding section from the referenced path. If no path is given, the pipeline is built entirely from inline sections.

data_pipeline:
  path: data_pipeline.yaml
  start: "2023-05-01"
  end: "2024-08-01"
  cache: true
  transformers:                    # replaces transformers from data_pipeline.yaml
    - class: LagTransformer
      params:
        columns: [load_actual, price]
        lags: [-7, -6, -5, -4, -3, -2, -1, 0]
        freq: day

model_pipeline:
  path: model_pipeline.yaml
  test_start: "2024-06-01"
  test_end: "2024-07-01"
  models:                          # replaces models from model_pipeline.yaml
    - class: OLSModel
      params:
        predictors: [load_actual, price_d-1]
        training_window: 180
        name: OLS_short

data_pipeline:
  start: "2023-05-01"
  end: "2024-08-01"
  cache: true
  sources:
    - class: EntsoeSource
      params:
        country_code: PL
        api_key: YOUR_KEY
        type: [load, price]
  transformers:
    - class: ResampleTransformer
      params: {freq: 1h}
  validators:
    - class: CoverageValidator
      params:
        columns: [load_actual, price]
        min_coverage: 1.0

Override semantics: inline sections replace the entire corresponding section from the loaded file, not merge. Sources + validators from path are kept unless also overridden.

Full Workflow Setup

Build and save pipeline configs

import os
os.environ["PYTHON_GIL"] = "0"

from epftoolbox2.pipelines import DataPipeline, ModelPipeline
from epftoolbox2.data.sources import EntsoeSource, OpenMeteoSource, CalendarSource
from epftoolbox2.data.transformers import TimezoneTransformer, ResampleTransformer, LagTransformer
from epftoolbox2.data.validators import CoverageValidator
from epftoolbox2.models import OLSModel, LassoCVModel
from epftoolbox2.evaluators import MAEEvaluator
from epftoolbox2.exporters import ExcelExporter, TerminalExporter

predictors = [
    "is_monday_d+{horizon}", "is_tuesday_d+{horizon}",
    "is_wednesday_d+{horizon}", "is_thursday_d+{horizon}",
    "is_friday_d+{horizon}", "is_saturday_d+{horizon}",
    "is_sunday_d+{horizon}", "is_holiday_d+{horizon}",
    "daylight_hours_d+{horizon}",
    "load_actual_d-1", "load_actual_d-2",
    "price_d-1", "price_d-2",
    "warsaw_temperature_2m_d+{horizon}",
]

(
    DataPipeline(timezone="Europe/Warsaw")
    .add_source(EntsoeSource(country_code="PL", api_key=os.environ["ENTSOE_API_KEY"], type=["load", "price"]))
    .add_source(OpenMeteoSource(latitude=52.2297, longitude=21.0122, prefix="warsaw"))
    .add_source(CalendarSource(country="PL", holidays="binary", daylight=True, weekday="onehot"))
    .add_transformer(TimezoneTransformer())
    .add_transformer(ResampleTransformer(freq="1h"))
    .add_transformer(LagTransformer(columns=["load_actual", "price"], lags=[1, 2, 7], freq="day"))
    .add_transformer(LagTransformer(
        lags=range(-7, 1),
        freq="day",
        columns=[
            "is_monday", "is_tuesday", "is_wednesday", "is_thursday", "is_friday",
            "is_saturday", "is_sunday", "is_holiday", "daylight_hours",
        ],
    ))
    .add_validator(CoverageValidator(columns=["load_actual", "price"], min_coverage=1.0))
    .save("data_pipeline.yaml")
)

(
    ModelPipeline()
    .add_model(OLSModel(predictors=predictors, training_window=365, name="OLS"))
    .add_model(LassoCVModel(predictors=predictors, training_window=365, cv=5, name="LassoCV"))
    .add_evaluator(MAEEvaluator())
    .add_exporter(TerminalExporter())
    .add_exporter(ExcelExporter("results.xlsx"))
    .save("model_pipeline.yaml")
)

Save workflow config

from epftoolbox2.pipelines import Workflow

Workflow(
    data_pipeline="data_pipeline.yaml",
    model_pipeline="model_pipeline.yaml",
    data_start="2023-05-01",
    data_end="2024-08-01",
    data_cache=True,
    model_test_start="2024-06-01",
    model_test_end="2024-07-01",
    model_target="price",
    model_horizon=6,
    max_processes=32,
    threads_per_process=8,
    cache_path="/tmp/cache",
).save("experiment.yaml")

Run job

import os
os.environ["PYTHON_GIL"] = "0"

from epftoolbox2.pipelines import Workflow

if __name__ == "__main__":
    report = Workflow.load("experiment.yaml").run()
    print(report.summary())

Run only one model (SLURM arrays)

model_index selects a single model by its 0-based position in the model list. When omitted, all models run. This enables submitting one job per model that allows further parallelization.

In this array mode each task writes its model’s results to model_pipeline.save_dir (one JSONL per model, resumable if the task is restarted) and skips the evaluators and exporters: relative metrics such as rMAE and the exporters need every model. Once all tasks have finished, merge the results and run the evaluators and exporters once with epf report experiment.yaml (or Workflow.load("experiment.yaml").report()). A negative or out-of-range model_index raises, and so does array mode without save_dir.

Workflow YAML for example SLURM cluster:

environment:
  max_processes: 8
  threads_per_process: 8
  model_index: $SLURM_ARRAY_TASK_ID   # 0-based, expanded at load() time

model_pipeline:
  path: model_pipeline.yaml
  save_dir: results                   # shared by all tasks; relative to this YAML
  test_start: "2024-01-01"
  test_end: "2024-12-31"

sbatch --array=0-3 run_task.sh        # each task: epf run experiment.yaml
# after the array has finished (e.g. sbatch --dependency=afterok:<array job id>):
epf report experiment.yaml

report() raises if a model has no results file. report(allow_missing=True) (epf report --allow-missing) reports the models that have one. Check report.forecast_counts() before comparing models.

Learn More

CLI Reference Run experiments from the command line without Python

DataPipeline Serialization YAML format for DataPipeline

ModelPipeline Serialization YAML format for ModelPipeline

Data Pipeline Sources, transformers, and validators

Model Pipeline Models, evaluators, and exporters