RollingAggregatesAdder#

class openstef_models.transforms.time_domain.rolling_aggregates_adder.RollingAggregatesAdder(**data: Any) → None[source]

Bases: BaseConfig, TimeSeriesTransform

Transform that adds rolling aggregate features to time series data.

Computes rolling aggregate statistics (e.g., mean, median, min, max) over a specified rolling window and adds these as new features to the dataset. It is useful for capturing recent trends and patterns in the data. Handles missing target data during inference via a fallback strategy:

  1. Forward-fill from last computed aggregate

  2. Use last valid aggregate from training

Example

>>> import pandas as pd
>>> from datetime import timedelta
>>> from openstef_core.datasets import TimeSeriesDataset
>>> from openstef_models.transforms.time_domain import RollingAggregatesAdder
>>>
>>> # Create sample dataset
>>> data = pd.DataFrame({
...     'load': [100, 120, 110, 130, 125],
...     'temperature': [20, 22, 21, 23, 24]
... }, index=pd.date_range('2025-01-01', periods=5, freq='1h'))
>>> dataset = TimeSeriesDataset(data, timedelta(hours=1))
>>>
>>> # Initialize and apply transform
>>> transform = RollingAggregatesAdder(
...     feature='load',
...     rolling_window_size=timedelta(hours=2),
...     aggregation_functions=["mean", "max"],
...     horizons=[LeadTime.from_string("PT36H")],
... )
>>> transform.fit(dataset)
>>> transformed_dataset = transform.transform(dataset)
>>> result = transformed_dataset.data[['rolling_mean_load_PT2H', 'rolling_max_load_PT2H']]
>>> print(result.round(1).head(3))
                     rolling_mean_load_PT2H  rolling_max_load_PT2H
timestamp
2025-01-01 00:00:00                   100.0                  100.0
2025-01-01 01:00:00                   110.0                  120.0
2025-01-01 02:00:00                   115.0                  120.0
Parameters:

data (Any)

feature: str
horizons: list[LeadTime]
rolling_window_size: timedelta
aggregation_functions: list[TypeAliasType]
property is_fitted: bool

Check if the transform has been fitted.

fit(data: TimeSeriesDataset) → None[source]

Compute and store last valid aggregates from training data for fallback.

Parameters:

data (TimeSeriesDataset)

Return type:

None

transform(data: TimeSeriesDataset) → TimeSeriesDataset[source]

Add rolling aggregate features, using fallbacks for missing values.

Returns:

Dataset with rolling aggregate feature columns added.

Raises:

NotFittedError – If fit() has not been called.

Parameters:

data (TimeSeriesDataset)

Return type:

TimeSeriesDataset

features_added() → list[str][source]

List of feature names added by this transform.

Return type:

list[str]

Returns:

A list of strings representing the names of features added to the dataset by this transform. Default is an empty list.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': False, 'extra': 'ignore', 'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self (BaseModel) – The BaseModel instance.

  • context (Any) – The context.

  • self

  • context

Return type:

None