Skip to content
epftoolbox2

Data Pipeline Overview

The DataPipeline class combines data sources, transformers, and validators into a cohesive data processing workflow.

Components

The pipeline runs in three steps, each backed by a kind of component:

  1. Sources fetch data in UTC. Every source returns a DataFrame with a UTC DatetimeIndex. Two sources producing the same column name raise ValueError naming both sources, so give one a prefix.

  2. Transformers modify the data, applied in order to the joined source output (resample, create lags, convert timezone, …).

  3. Validators check quality without modifying the DataFrame. validators run on the final frame, and source_validators run on the raw, joined source output, before any transformer. Every result is kept in pipeline.validation_results. DataPipeline(on_invalid="raise") raises DataValidationError on a failure instead of the default ("warn", log and continue). See Serialization.

Basic Usage

from epftoolbox2.pipelines import DataPipeline
from epftoolbox2.data.sources import EntsoeSource, CalendarSource
from epftoolbox2.data.transformers import ResampleTransformer, TimezoneTransformer
from epftoolbox2.data.validators import CoverageValidator


pipeline = (
    DataPipeline(timezone="Europe/Warsaw")
    .add_source(EntsoeSource(country_code="PL", api_key="...", type=["load", "price"]))
    .add_source(CalendarSource(country="PL", holidays="binary"))
    .add_transformer(ResampleTransformer(freq="1h"))
    # target_tz is left unset: it inherits DataPipeline.timezone above.
    .add_transformer(TimezoneTransformer())
    .add_validator(CoverageValidator(columns=["load_actual", "price"], min_coverage=1.0))
)


df = pipeline.run(start="2024-01-01", end="2024-06-01", cache=True)

Date Format Options

The start and end parameters accept multiple formats:

df = pipeline.run(start="2024-01-01", end="2024-06-01")

end is inclusive: a date-only end (a string, a datetime.date, or a keyword such as "today") includes that whole day, the local day when timezone is set and the UTC day otherwise. A datetime.date is what YAML makes of an unquoted end: 2024-01-31, so a workflow file gives the same range with or without quotes. A pd.Timestamp or datetime.datetime is an exact instant, and an end given that way is exclusive.

A date-only start begins at that day’s midnight, and naive timestamps are taken as UTC. Every source’s output is clipped to the resolved range, whether it comes from the API or from the cache (see Data Caching).

Timezone (local calendar days)

Pass a market time zone to make those days local calendar days:

pipeline = DataPipeline(timezone="Europe/Warsaw", sources=[...], transformers=[...])


# 2024-01-01 00:00 local .. 2024-01-03 23:00 local (both days complete, inclusive)
df = pipeline.run(start="2024-01-01", end="2024-01-03")

With timezone set:

  • start begins at local midnight, and end covers its whole local day, correctly sized (23 or 25 hours) on a DST-transition day.
  • "today" / "now_d±N" keywords resolve using the current date in that zone.
  • Naive timestamps passed to run() are read as local instants, and timezone-aware timestamps are used as given.

Without timezone (the default), the day is a UTC one: date strings and keywords mean midnight UTC, and naive timestamps are UTC.

Available Components

Data Sources

EntsoeSource ENTSOE Transparency Platform
OpenMeteoSource Weather forecasts
CalendarSource Holidays & calendar
CsvSource CSV files
EpftoolboxBenchmarkSource Lago et al. (2021) benchmark datasets

Transformers

ResampleTransformer Regular frequency
LagTransformer Lagged features
TimezoneTransformer Convert timezone
DSTNormalizeTransformer 24 slots per local day
ImputeTransformer Leak-safe gap filling
DailyProfileTransformer Spread a day's values across its rows
DailyStatsTransformer Per-day min/max/mean/last
LambdaTransformer Custom column arithmetic

Validators

DSTStructureValidator Rows per local day
CoverageValidator Index and per-column outages
PlausibilityValidator Ranges and frozen values
EdaValidator EDA statistics