TransformPipeline#

class openstef_core.mixins.transform.TransformPipeline(**data: Any) → None[source]

Bases: BaseModel, Transform, Generic

Sequential pipeline of transformations.

Applies multiple transforms in order, fitting each transform on the intermediate outputs of the previous transforms. Ensures proper error handling and state management across the pipeline.

Invariants

  • Transforms are called in order, receiving the output of the previous transform.

  • Pipeline is considered fitted only when all transforms are fitted

Example

Creating and using a transformation pipeline

>>> from openstef_core.datasets import TimeSeriesDataset
>>> # Create an empty pipeline
>>> pipeline = TransformPipeline[TimeSeriesDataset](transforms=[])
>>>
>>> # The pipeline can be used even when empty
>>> # processed_data = pipeline.transform(data)
Parameters:

data (Any)

transforms: Sequence[Transform[TypeVar(T), TypeVar(T)]]
__reduce__() → tuple[Callable[[], TransformPipeline[Any]], tuple[()], Any][source]

Support pickling of generic TransformPipeline instances.

When TransformPipeline is parameterized (e.g., TransformPipeline[TimeSeriesDataset]), Python creates a dynamic type that pickle cannot find by its module path. This method provides custom pickling support by reducing to the base class and reconstructing through __setstate__.

Return type:

tuple[Callable[[], TransformPipeline[Any]], tuple, Any]

Returns:

Tuple of (callable, args, state) for pickle reconstruction.

property is_fitted: bool

Check if all transforms in the pipeline are fitted.

Returns:

True if all transforms are fitted, False otherwise.

fit(data: T) → None[source]

Fit all transforms in the pipeline sequentially.

Parameters:

  • data (TypeVar(T)) – Input data to fit the transforms on.

  • data

Return type:

None

transform(data: T) → T[source]

Transform data using all fitted transforms in sequence.

Parameters:

  • data (TypeVar(T)) – Input data to transform.

  • data

Returns:

Transformed data after applying all transforms.

Return type:

TypeVar(T)

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