Transforms

A transform is anything with an apply method that maps a series to a new series. That is the entire contract:

from mypackage import Series

class Square:
    def apply(self, series: Series) -> list[float]:
        return [value**2 for value in series]

Square inherits from nothing and registers with nothing, yet it composes with the built-ins, satisfies isinstance checks, and works inside a Pipeline:

import mypackage as mp

isinstance(Square(), mp.Transform)                 # True
mp.chain(mp.Standardize(), Square()).fit_apply(data)

This works because Transform is a runtime_checkable Protocol — structural typing. The static checker verifies the shape; isinstance verifies the method exists at runtime.

Stateless versus fitted¶

Stateless transforms implement fit as a no-op returning self, so they drop into a pipeline next to fitted ones without any special casing.

Fitted transforms raise NotFittedError when used before fit — never a silent identity, never a None-propagating nan:

mp.Standardize().apply([1.0, 2.0])
# NotFittedError: Standardize is not fitted; call .fit(series) first

Leakage¶

The reason fit and apply are separate steps at all:

train = [10.0, 12.0, 14.0]
test = [16.0, 18.0]

scaler = mp.Standardize().fit(train)   # statistics come from train only
scaler.apply(train)                    # [-1.0, 0.0, 1.0]
scaler.apply(test)                     # [2.0, 3.0]  — same scale, not re-fitted

Calling fit_apply(test) instead would re-learn the mean and standard deviation from the test data, quietly leaking information about it into the transform. The shortcut exists — fit_apply — precisely so that the leaking version has to be typed out, rather than being what you get by default.

Standardize also provides inverse, which maps standardised values back to the original scale:

scaler.inverse(scaler.apply(train)) == train   # True (to floating point)

Pipelines¶

Pipeline applies steps left to right. Fitting is sequential: each step is fitted on the output of the previous step, exactly as it will see the data at apply time.

pipeline = mp.Pipeline([
    mp.Clip(upper=50.0),
    mp.Standardize(),
    mp.MovingAverage(window=3),
])

clean = pipeline.fit_apply(raw)

pipeline = mp.Pipeline([mp.Clip(upper=10.0), mp.Standardize()])
pipeline.fit([0.0, 10.0, 1000.0])
pipeline[1].mean_          # 6.666..., not 336.666...

Composition¶

Pipelines compose with |. The operands are never mutated — you always get a new pipeline:

base = mp.chain(mp.Clip(lower=0.0), mp.Standardize())
extended = base | mp.MovingAverage(3)
nested = base | mp.chain(mp.Clip(-3.0, 3.0), Square())

len(base), len(extended), len(nested)   # (2, 3, 4)

They are also sequences — iterable, indexable, sized:

for step in pipeline:
    print(type(step).__name__)

pipeline[0]      # Clip(lower=None, upper=50.0)
len(pipeline)    # 3

An empty pipeline is the identity, which makes Pipeline(steps) safe to build from a config file that happens to list no steps.

Writing your own¶

Make it a frozen dataclass, take configuration in __init__, validate there, and keep apply pure:

from dataclasses import dataclass

from mypackage import Series, ValidationError

@dataclass(frozen=True, slots=True)
class Detrend:
    """Subtract a rolling baseline estimated by a median filter."""

    window: int

    def __post_init__(self) -> None:
        if self.window < 1:
            raise ValidationError(f"'window' must be positive, got {self.window}")

    def fit(self, series: Series) -> "Detrend":
        return self

    def apply(self, series: Series) -> list[float]:
        baseline = mp.median_filter(series, self.window)
        return [x - b for x, b in zip(series, baseline, strict=True)]

Checklist for a well-behaved transform:

• apply returns a new list and never mutates its input

• configuration is validated at construction, not at apply time

• fit returns self so it chains

• stateful transforms raise NotFittedError before fit

• errors derive from MypackageError

API¶

Full signatures live in the transforms API reference.