TargetProvider

class openstef_beam.benchmarking.TargetProvider(**data: Any) → None

Bases: BaseConfig, Generic

Abstract interface for loading benchmark targets and their associated datasets.

Guarantees consistent access to targets, measurements, and predictor data across different benchmark implementations. All returned datasets maintain temporal alignment and consistent sampling intervals.

Examples

Complete custom provider implementation:

>>> from pathlib import Path
>>> from datetime import datetime, timedelta
>>> from openstef_beam.benchmarking.models.benchmark_target import BenchmarkTarget
>>> from openstef_beam.evaluation.metric_providers import RMAEProvider
>>> class EnergyTargetProvider(TargetProvider[BenchmarkTarget, None]):
...     def __init__(self, data_path: Path, region: str):
...         super().__init__()
...         self.data_path = data_path
...         self.region = region
...
...     def get_targets(self, filter_args=None):
...         # Load from database or config files
...         return [
...             BenchmarkTarget(
...                 name=f"substation_{i:03d}",
...                 description=f"Energy load for substation {i}",
...                 group_name=self.region,
...                 latitude=52.0 + i * 0.001,
...                 longitude=4.0 + i * 0.001,
...                 limit=100.0 + i * 10,
...                 benchmark_start=datetime(2024, 1, 1),
...                 benchmark_end=datetime(2024, 3, 1),
...                 train_start=datetime(2022, 1, 1)
...             ) for i in range(1, 11)
...         ]
...
...     def get_measurements_for_target(self, target):
...         # Load actual load data from parquet files
...         return VersionedTimeSeriesDataset.read_parquet(
...             self.data_path / f"{target.group_name}/{target.name}_load.parquet"
...         )
...
...     def get_predictors_for_target(self, target):
...         # Combine weather, profiles, and building characteristics
...         datasets = [
...             self.load_weather_data(target),
...             self.load_building_profiles(target),
...             self.load_price_data()
...         ]
...         return concat_featurewise(datasets, mode="inner")
...
...     def get_metrics_for_target(self, target):
...         # Target-specific metrics based on building type
...         base_metrics = [RMAEProvider(), RCRPSProvider()]
...         if target.limit > 500:  # Large buildings get additional metrics
...             base_metrics.append(MAPEProvider())
...         return base_metrics

Parameters:

data (Any)

target_column: str

abstractmethod get_targets(filter_args: F | None = None) → list[T]

Load all available benchmark targets.

Parameters:

• filter_args (Optional[TypeVar(F)]) – Provider-specific filtering criteria.

• filter_args

Returns:

Complete list of targets with validated time constraints and metadata.

Raises:

• FileNotFoundError – When target data source is inaccessible.

• ValidationError – When target definitions violate constraints.

Return type:

list[TypeVar(T, bound= BenchmarkTarget)]

abstractmethod get_measurements_for_target(target: T) → VersionedTimeSeriesDataset

Load ground truth measurements for benchmark evaluation.

Parameters:

• target (TypeVar(T, bound= BenchmarkTarget)) – Must have valid time ranges and exist in data source.

• target

Returns:

Dataset with measurements covering at least the benchmark period.

Raises:

FileNotFoundError – When measurement data is missing for target.

Return type:

VersionedTimeSeriesDataset

abstractmethod get_predictors_for_target(target: T) → VersionedTimeSeriesDataset

Load predictor features for model training and inference.

Parameters:

• target (TypeVar(T, bound= BenchmarkTarget)) – Must have valid time ranges and location coordinates.

• target

Returns:

Dataset with features covering training and benchmark periods.

Raises:

FileNotFoundError – When predictor data is incomplete for target.

Return type:

VersionedTimeSeriesDataset

abstractmethod get_metrics_for_target(target: T) → list[MetricProvider]

Returns the list of metrics to use for evaluation of a target.

Parameters:

• target (TypeVar(T, bound= BenchmarkTarget)) – The target to get metrics for

• target

Returns:

A list of metric providers to use for evaluating predictions for this target

Return type:

list[MetricProvider]

abstractmethod get_evaluation_mask_for_target(target: T) → DatetimeIndex | None

Get the evaluation mask for a target.

Parameters:

• target (TypeVar(T, bound= BenchmarkTarget)) – The target to get the evaluation mask for

• target

Returns:

A DatetimeIndex representing the evaluation mask, or None if no mask is defined

Return type:

DatetimeIndex | None

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