Shifter#

class openstef_models.transforms.general.Shifter(**data: Any) → None[source]

Bases: BaseConfig, TimeSeriesTransform

Transform that shifts features to align their aggregation interval with the target.

When source features are aggregated over a different interval than the target variable, their timestamps represent a different center point in time. This transform corrects the phase misalignment by shifting the source features and linearly interpolating back onto the original time grid.

The shift is computed as:

shift = source_aggregation_period / 2 - target_aggregation_period / 2

Timestamps are assumed to be at the end of the aggregation interval. For example, a timestamp of 12:00 with a 60-minute aggregation period represents the average over [11:00, 12:00], centered at 11:30. For instantaneous features or target, use an aggregation period of zero.

Example: Aligning hourly radiation with 15-minute load

Hourly radiation (source_aggregation_period=60 min) has its center 30 min before the timestamp, while 15-minute load (target_aggregation_period=15 min) has its center 7.5 min before the timestamp. The required backward shift for radiation is 22.5 minutes.

>>> import pandas as pd
>>> from datetime import timedelta
>>> from openstef_core.datasets import TimeSeriesDataset
>>> from openstef_models.transforms.general import Shifter
>>> from openstef_models.utils.feature_selection import FeatureSelection
>>>
>>> # Hourly radiation interpolated onto a 15-minute grid
>>> index = pd.date_range('2025-01-01', periods=8, freq='15min')
>>> data = pd.DataFrame({
...     'load': range(8),
...     'radiation': [200, 220, 240, 260, 280, 300, 320, 340],
... }, index=index)
>>> dataset = TimeSeriesDataset(data, timedelta(minutes=15))
>>>
>>> shifter = Shifter(
...     selection=FeatureSelection(include=['radiation']),
...     source_aggregation_period=timedelta(minutes=60),
...     target_aggregation_period=timedelta(minutes=15),
...     fill_edges=True,
... )
>>> result = shifter.transform(dataset)
>>> result.data['radiation'].tolist()
[230.0, 250.0, 270.0, 290.0, 310.0, 330.0, 340.0, 340.0]
Parameters:

data (Any)

selection: FeatureSelection
source_aggregation_period: timedelta
target_aggregation_period: timedelta
fill_edges: bool
model_post_init(context: Any) → None[source]

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Parameters:

context (Any)

Return type:

None

transform(data: TimeSeriesDataset) → TimeSeriesDataset[source]

Transform the input data.

This method should apply a transformation to the input data and return a new instance.

Parameters:
Returns:

A new instance of the transformed data.

Raises:

NotFittedError – If the transform has not been fitted yet.

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].