Pseudocode: All the code listed is pseudocode to illustrate how interfaces are used and might not necessarily be python compliant. The actual function signatures may change during implementation.

OpenSTEF-models

High Level Overview

openstef-models/
├── feature_engineering/
│   ├── validation_transforms/     # Flatliner checking, completeness checks
│   ├── temporal_transforms/       # Lags, holidays, cyclic features
│   ├── forecasting_transforms/    # Normalization, trend, seasonality
│   ├── weather_transforms/        # Radiation, humidity calculations
│   └── energy_domain_transforms/  # Wind/solar energy specific features
├── models/
│   ├── forecasting/              # Core forecasting models
│   └── component_splitting/      # Component decomposition models
├── anomaly_detection/            # Flatliner, polarity switch detection
├── pipelines/                    # Model training and inference pipelines
├── postprocessing/               # Quantile processing and calibration
└── explainability/               # Model interpretation tools

PredictPipeline / TrainModelPipeline: Below we can see a data flow diagram for both training and prediction processes within OpenSTEF 4.0 models. These pipelines are comparable to the current pipelines in OpenSTEF 3.0, although these ones follow the new design decisions.

Note that openstef-models contains more pipelines than the two in the diagram. The other ones are constructed in a similar fashion, as making additional diagrams for those would introduce too much redundancy.

Interfaces

TimeSeriesDataset: This dataset, spefically for openstef-models is for validation and aggregation of input data. There are various frequently used input types that have strict requirements and should be validated and fail early on.

class TimeSeriesDataset[Type = Generic]:
  data: pd.DataFrame
  sample_interval: timedelta
  metadata: dict[str, str | number | datetime | timedelta]
  
  def __init__(*params) -> None:
    # Validate the data and index
    
  @classmethod
  def required_columns(cls) -> list[str]:
    if cls.type == Generic: return []
    elif cls.type == ModelInput: return ['load', 'sample_weights']
    elif cls.type == ComponentsModelInput: 
      return ['solar', 'wind', 'other', 'sample_weights']
    elif cls.type == Forecast: return ['quantile_P*']
    # Other predefined types.
    # Custom validation can be implemented by extenders.
    
  def to_parquet(self) -> None:
    ... # Saves the dataset with metadata embedded in parquet
  
  @classmethod
  def from_parquet(cls, path) -> Self:
    ...

ModelState: encodes all the stateful information of a model. It could be self or the actual learned weights. What works best depends on the type of storage and what works best.

class ForecastingModel:
  def __init__(**config, **hyperparams) -> None
    ...

  def fit(self, input_data: TimeSeriesDataset) -> None:
    ...
  
  def predict(
    self, input_data: TimeSeriesDataset
  ) -> TimeSeriesDataset[Forecast]:
    ...
    
  def get_state(self) -> ModelState:
    ...
    
  def set_state(self, state: ModelState) -> None:
    ...


class ComponentSplitModel(Model):
  def fit(self, input_data: TimeSeriesDataset) -> None:
    ...
  
  def predict(
      self, input_data: TimeSeriesDataset
  ) -> TimeSeriesDataset[ComponentsForecast]:
    ...
    
  def get_state(self) -> ModelState:
    ...
    
  def set_state(self, state: ModelState) -> None:
    ...

Transforms: Transforms are quite generic to work across the data. They can be used for both:

Data Transformation: The transform, can add or remove columns, or transform the data in one or more columns. The goal is to do the smallest possible change to keep data consistency.
Data Validation: Sometimes data is invalid by chance (flatliners) or by error (bug in code / data). Transforms can handle this by throwing one of the predefined errors. These errors can be caught in the user or pipeline code to take alternate paths such as training a flatliner or fallback model.

Errors = ValidationError | MissingDataError | NotFitError


class FeatureTransform:
  def __init__(**configuration) -> None
    ...

  def fit(self, input_data: TimeSeriesDataset) -> None:
    ...
  
  def transform(self, input_data: TimeSeriesDataset) -> TimeSeriesDataset:
    ...  
    
  def get_state(self) -> TransformState:
    ...
    
  def set_state(self, state: TransformState) -> None:
    ...


class PreprocessingTransform:
  def fit(self, input_data: TimeSeriesDataset[Forecast]) -> None:
    ...
  
  def transform(
    self, input_data: TimeSeriesDataset[Forecast]
  ) -> TimeSeriesDataset[Forecast]:
    ...  
    
  def get_state(self) -> TransformState:
    ...
    
  def set_state(self, state: TransformState) -> None:
    ...

ModelStorage: provides a unified interface for storing the models in a model registry or similar (for example MLFlowModelStorage).

Note: ModelStorage is only for storage. If as part of MLFlow run some info or inbetween artifacts need to be stored, Callback should be used instead, as it provides hook points within the pipeline to execute custom logic.


class ModelStorage:
  def save_model(
    self,
    id: ModelIdentifier,
    state: tuple[TransformState, ModelState]
  ):
    ...
  
  def load_model(
    self,
    id: ModelIdentifier
  ) -> tuple[TransformState, ModelState]:
    ...
    
    

class Callback:
  def on_data_prep_start(...):
    ...
  
  def on_data_prep_complete(...):
    ...

  def on_training_start(...):
    ...
  
  def on_training_complete(...):
    ...
    
  def on_predict_start(...):
    ...
  
  def on_predict_end(...):
    ...

OpenSTEF-BEAM

High Level Overview

openstef-beam/
├── core/
│   ├── dataset/                  # Advanced dataset operations
│   └── utils/                    # Shared utilities
├── backtesting/                  # Backtesting framework with retrain strategies
├── benchmarking/                 # Benchmarking framework combining the steps
├── evaluation/                   # Model evaluation and comparison pipelines     
├── analysis/                     # Analysis plotting and reporting
└── metrics/                      # Comprehensive forecasting metrics

The diagram below illustrates full benchmarking flow of a model. stef-beam serves as a generic evaluation framework, as such, no assumptions are made about the model which is abstracted away behind a model interface.

The pipeline does backtesting, evaluation and analysis, which in other scenarios can be run separates on output of OpenSTEF for example.

stef-beam provides some pre-implemented metrics and visualizations for common energy forecasting use cases, which can be extended by the user with their own.

The benchmarks provider is responsible for providing benchmark specific information, which in production would be supplied by the forecasting stack. This data can be configured by the user, or some preconfigured benchmarks can be used as part of OpenSTEF or user’s team.

Interfaces

TimeSeriesDataset: This dataset is different from openstef-models dataset. Mainly in fact that stef-beam is responsible for evaluation and doesn’t run on instant data, but instead on different version of data collected over time. This is useful because availability changes over time:

Data can arrive with a lag and might be available from a certain date
Data is an estimation which may get revisions causing it to become more accurate
Data is available for the future because the data is a forecast

This data availability can be resolved by making two columns required:

timestamp: for when a datapoint applies
available_at: the earliest time this datapoint became visible

Moreover during production data is queryable from an api, to make the simulation realistic, a similar interface is exposed through get_window function.

class TimeSeriesDataset:
  data: pd.DataFrame
  index: pd.TimeSeriesIndex
  sample_interval: datetime
  
  def __init__(*args) -> None:
    # This dataset requires 'timestamp' and 'available_at' columns to 
    # account for data availablity during forecasting and evaluation
  
  
  def get_window(
    start: datetime, end: datetime, available_at: datetime | None
  ) -> pd.DataFrame:
    # Check data time range against the index
    # Return data based on available_at constraint
    # Data points that have no data are filled with NaN
    
class HorizonDataset(TimeSeriesDataset):
  horizon: datetime
  # Adds additional validation that available_at <= horizon so no data 
  # in the future can be read.

ModelInterface: Is a generic interface for various forecasting libraries. It can be openstef-models, it can be openstef-pretrained-models or it can be some custom code using sklearn or similar. openstef-beam makes no assumptions on what forecasting library is used, so long data is output in an expected format.

class ModelInterface:
  def fit(self, context, data: HorizonDataset) -> None:
    # Dataset acts as data repository, allowing the process to request data
    # given a point of observation.
    ...
  
  def predict(self, context, data: HorizonDataset) -> TimeSeriesDataset[Forecast]:
    ...

Both MetricProvider and VisualizationProvider provide an interface for computing metrics and creating visualizations on the evaluation outputs, making it possible for custom configuration and custom implementations.

class MetricProvider:
  def __init__(**configuration):
    ...

  def calculate(self, data: ...) -> dict[Quantile | global, dict[str, number]]:
    ...
  

class VisualizationProvider:
  def create(self, data: ...) -> PlotWrapper:
    ...

BenchmarkStorage: Used for storing outputs of the benchmark pipeline. The storage can be local or a custom implementation such as S3 to preserve the research / benchmark results to a datalake.

Similarly BenchmarkCallback can be used to hook in at various stages of benchmarking, to handle errors gracefully, to monitor execution or for example send a teams message if a long running benchmark has finished.

class BenchmarkStorage:
  def save_(backtest|evaluation|analytics)_output(...):
    ...
    
  def load_(backtest|evaluation|analytics)_output(...) -> ...:
    ...
    
  def has_(backtest|evaluation|analytics)_output(...) -> bool:
    ...
  

class BenchmarkCallback:
  def on_(backtest|evaluation|analysis)_(start|end)(...):
    ...
    
  def on_error(self, error: Exception):
    raise error

Deployment

The diagram below addresses combining OpenSTEF 4.0 libraries into a production deployment use case.

We demonstrate it using a hypothetical generic deployment example within openstef-examples. There is a service that is responsible for storing the measurements, predictors, forecasting targets and forecasts (for example PostgresQL) and there is a service for storing the models (for example MLFlow). The compute layer inbetween orchestrates calling of OpenSTEF 4.0 pipelines to train, forecast and evaluate models.

Configuration: The one of the core principles of OpenSTEF 4.0 is extensibility. Unfortunately it is not possible to maintain extensibility while maintaining one single configuration file (like PredictionJob in OpenSTEF 3.0).

OpenSTEF 4.0 two options to do configuration per component instead:

Configure each component individually by constructing them and providing them to the pipeline. This is quite a low level approach fit for big organizations that have a high need for customizability. The WorkflowPreset code would in this case be inside the user’s deployment code.
Use a pre-existing preset (illustrated as WorkflowPreset inside openstef-models). This way one curated config can be used to configure multiple subcomponents, thus making WorkflowPreset inevitably opinionated for a specific configuration (just like current PredictionJob). This is not a bad thing, so long underlying code is easily extensible giving user the ability to create custom workflow preset variants or to switch to option 1.

Example for constructing a pipeline using lower level components (pseudocode)

forecast_pipeline = PredictPipeline(
  transform_pipeline=TransformPipeline(
    steps=[
      DNITransform(column="radiation")
      LagsTransform(column="load", lags=[timedelta(days=7), ...]),
      PCATransform(n_components=5),
    ]
  )
  model=XGBoostModel(
    quantiles=[0.1, 0.5, 0.9]
    num_steps=1000,
    ...,
  )
  model_storage=MLFlowModelStorage(
    url="file://./local_dir/mlflow_storage"
  )
)

try:
  predict_data = TimeSeriesDataset(
    data=input_dataframe, 
    sample_interval=timedelta(minutes=15),
    timestamp_column=IndexColumn,
    available_at_column=None
  ).mask_from(datetime.fromisoformat("2025-07-01T00:00:00"))
  
  forecast: DataFrame = forecast_pipeline.run(
    model_id="model_for_location_A",
    data=predict_data,
  )
except ModelNotFound:
  print("Could not make a forecast")

Example of higher level configuration creating purpose built workflows (pseudocode)

class CongestionForecastWorkflowConfig:
  model_type: Literal["xgboost", "gblinear"]
  quantiles: list[float]
  apply_pca: bool

class TransportForecastWorkflow:
  config: TransportForecastWorkflowConfig
  model_storage: ModelStorageInterface

  ...

  def predict(data: TimeSeriesDataset, model_id: str) -> DataFrame:
    model = XGBoostModel(...) if self.config.model_type == "xgboost" \
      else ...
  
    return PredictPipeline(
      model=model,
      ...
    ).run(data, model_id)

# Usage
workflow = TransportForecastWorkflow(
  config=CongestionForecastWorkflowConfig(
    model_type="xgboost",
    ...
  ),
  ...
)

workflow.train(data=train_data, model_id="model_for_location_A")

forecast = workflow.predict(data=predict_data, model_id="model_for_location_A")

OpenSTEF 4.0 Architecture Document

High Level Overview

Interfaces

OpenSTEF-BEAM

High Level Overview

Interfaces

Deployment