FlatlineChecker#

class openstef_models.transforms.validation.flatline_checker.FlatlineChecker(**data: Any) None[source]

Bases: BaseConfig, TimeSeriesTransform

Transformer to detect flatliner patterns in time series load data.

A flatliner is a period where the load remains constant for a specified duration. This class can detect both zero and non-zero flatliners, depending on configuration.

Example

>>> from datetime import timedelta
>>> import numpy as np
>>> import pandas as pd
>>> from openstef_core.datasets import TimeSeriesDataset
>>> from openstef_models.transforms.validation import (
...     FlatlineChecker,
... )
>>> data = pd.DataFrame(
...     {
...         "load": [100, 110, 110, 110],
...     },
...     index=pd.date_range("2025-01-01", periods=4, freq="1h"),
... )
>>> dataset = TimeSeriesDataset(data, timedelta(hours=1))
>>> transform = FlatlineChecker(
...     flatliner_threshold=timedelta(hours=2),
...     detect_non_zero_flatliner=True,
...     relative_tolerance=1e-5
... )
>>> try:
...     transformed_data = transform.fit_transform(dataset)
... except FlatlinerDetectedError as e:
...     pass
Parameters:

data (Any)

load_column: str
flatliner_threshold: timedelta
detect_non_zero_flatliner: bool
absolute_tolerance: float
relative_tolerance: float
error_on_flatliner: bool
detect_ongoing_flatliner(data: Series) bool[source]

Detects if the latest measurements follow a flatliner pattern.

The following equation is used to test whether two floats are equivalent: absolute(measurement - flatliner_value) <= (atol + rtol * absolute(flatliner_value))

Parameters:
  • data (Series) – A timeseries of measured load with a DatetimeIndex.

  • data

Returns:

Boolean indicating whether or not there is a flatliner ongoing for the given data.

Return type:

bool

transform(data: TimeSeriesDataset) TimeSeriesDataset[source]

Returns the input data unchanged, optionally checking for flatliners.

This method can optionally run flatliner detection on new incoming data when check_on_transform=True, which is useful for real-time validation during forecasting.

Parameters:
Returns:

The unmodified input TimeSeriesDataset.

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