CustomComponentSplitWorkflow

class openstef_models.workflows.custom_component_split_workflow.CustomComponentSplitWorkflow(**data: Any) → None

Bases: BaseModel

Complete component splitting workflow with model management and lifecycle hooks.

Orchestrates the full component splitting process by combining a ComponentSplittingModel with callback execution and optional model persistence. Provides the main interface for production component splitting systems where models need to be trained, saved, loaded, and used for prediction with monitoring.

Invariants

• Callbacks are executed at appropriate lifecycle stages

• Model fitting and prediction delegate to the underlying ComponentSplittingModel

• Storage operations (if configured) maintain model persistence

Example

Basic workflow with callbacks

>>> import pandas as pd
>>> import numpy as np
>>> from datetime import timedelta
>>> from openstef_core.datasets import TimeSeriesDataset
>>> from openstef_models.models import ComponentSplittingModel
>>>
>>> # Create sample data
>>> dataset = TimeSeriesDataset(
...     data=pd.DataFrame({
...         "load": np.random.default_rng(42).standard_normal(size=48),
...     }, index=pd.date_range("2025-01-01", periods=48, freq="h")),
...     sample_interval=timedelta(hours=1),
... )
>>>
>>> # Create model and workflow
>>> model = ComponentSplittingModel(...)
>>>
>>> class LoggingCallback(ComponentSplitCallback):
...     def on_fit_end(self, workflow, data):
...         print("Model training completed")
>>>
>>> workflow = CustomComponentSplitWorkflow(
...     model=model, model_id="my_model", callbacks=LoggingCallback()
... )
>>> workflow.fit(dataset)
Model training completed
>>> components = workflow.predict(dataset)

Loading from storage with fallback

>>> workflow = ComponentSplitWorkflow.from_storage(
...     model_id="production_model_v1",
...     storage=my_storage,
...     default_model_factory=lambda: create_default_model()
... )

Parameters:

data (Any)

model: ComponentSplittingModel

callbacks: ComponentSplitCallback

model_id: TypeAliasType

fit(data: TimeSeriesDataset, data_val: TimeSeriesDataset | None = None) → None

Train the component splitting model with callback execution.

Executes the complete training workflow including pre-fit callbacks, model training, and post-fit callbacks.

Parameters:

• data (TimeSeriesDataset) – Training dataset for the component splitting model.

• data_val (TimeSeriesDataset | None) – Optional validation dataset for monitoring during training.

• data

• data_val

Return type:

None

predict(data: TimeSeriesDataset) → EnergyComponentDataset

Generate component predictions with callback execution.

Executes the complete prediction workflow including pre-prediction callbacks, model prediction, and post-prediction callbacks.

Parameters:

• data (TimeSeriesDataset) – Input dataset for generating component predictions.

• data

Returns:

Generated energy component dataset.

Raises:

NotFittedError – If the underlying model hasn’t been trained.

Return type:

EnergyComponentDataset

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'protected_namespaces': (), 'ser_json_inf_nan': 'null'}

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