Transforms¶

Composable transforms with a fit / apply lifecycle.

See the transforms guide for how to write your own.

transforms ¶

Composable series transforms with a fit / apply lifecycle.

Every transform here satisfies the Transform protocol — a single apply method. Transforms that learn state from data (Standardize, MinMaxScale) additionally satisfy Fittable and raise NotFittedError if used before fit.

Because the protocols are structural, Pipeline happily composes your own transforms alongside the built-in ones — no base class to inherit from, no registry to join:

from dataclasses import dataclass

from mypackage import Series, chain, Standardize


@dataclass(frozen=True, slots=True)
class Square:
    def apply(self, series: Series) -> list[float]:
        return [value**2 for value in series]


pipeline = chain(Standardize(), Square())

Standardize `dataclass` ¶

Standardize(ddof: int = 1)

Centre a series on zero and scale it to unit standard deviation.

zᵢ = (xᵢ − x̄) / s, with x̄ the mean and s the standard deviation.

The mean and standard deviation are learned by fit and reused by every later apply, which is what makes this safe to fit on a training split and apply to a held-out one.

Attributes:

Name	Type	Description
`ddof`	`int`	Delta degrees of freedom for the standard deviation.

Examples:

>>> scaler = Standardize().fit([10.0, 12.0, 14.0])
>>> [round(value, 4) for value in scaler.apply([10.0, 12.0, 14.0])]
[-1.0, 0.0, 1.0]

The learned statistics carry over to unseen data:

>>> round(scaler.apply([16.0])[0], 4)
2.0

inverse round-trips:

>>> [round(v, 6) for v in scaler.inverse(scaler.apply([11.0, 13.0]))]
[11.0, 13.0]

fit ¶

fit(series: Series) -> Self

Learn the mean and standard deviation of series.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The reference observations.	required

Returns:

Type	Description
`Self`	`self`, so `fit` chains into `apply`.

Raises:

Type	Description
`EmptySeriesError`	If `series` is empty.
`ValidationError`	If `series` is constant, leaving `std == 0`.

Source code in src/mypackage/transforms.py

def fit(self, series: Series) -> Self:
    """Learn the mean and standard deviation of ``series``.

    Args:
        series: The reference observations.

    Returns:
        ``self``, so ``fit`` chains into ``apply``.

    Raises:
        EmptySeriesError: If ``series`` is empty.
        ValidationError: If ``series`` is constant, leaving ``std == 0``.
    """
    summary = summarize(series, ddof=self.ddof)
    if summary.std == 0.0:
        msg = "cannot standardise a constant series (std == 0)"
        raise ValidationError(msg)
    self.mean_ = summary.mean
    self.std_ = summary.std
    return self

apply ¶

apply(series: Series) -> list[float]

Standardise series with the learned statistics.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The observations to transform.	required

Returns:

Type	Description
`list[float]`	A new list of standardised values.

Raises:

Type	Description
`NotFittedError`	If `fit` has not been called.

Source code in src/mypackage/transforms.py

def apply(self, series: Series) -> list[float]:
    """Standardise ``series`` with the learned statistics.

    Args:
        series: The observations to transform.

    Returns:
        A new list of standardised values.

    Raises:
        NotFittedError: If `fit` has not been called.
    """
    mean, std = self._check_fitted()
    return [(value - mean) / std for value in as_floats(series)]

inverse ¶

inverse(series: Series) -> list[float]

Map standardised values back to the original scale.

Parameters:

Name	Type	Description	Default
`series`	`Series`	Standardised observations.	required

Returns:

Type	Description
`list[float]`	A new list on the original scale.

Raises:

Type	Description
`NotFittedError`	If `fit` has not been called.

Source code in src/mypackage/transforms.py

def inverse(self, series: Series) -> list[float]:
    """Map standardised values back to the original scale.

    Args:
        series: Standardised observations.

    Returns:
        A new list on the original scale.

    Raises:
        NotFittedError: If `fit` has not been called.
    """
    mean, std = self._check_fitted()
    return [value * std + mean for value in as_floats(series)]

fit_apply ¶

fit_apply(series: Series) -> list[float]

Fit on series and immediately transform it.

Source code in src/mypackage/transforms.py

def fit_apply(self, series: Series) -> list[float]:
    """Fit on ``series`` and immediately transform it."""
    return self.fit(series).apply(series)

MinMaxScale `dataclass` ¶

MinMaxScale(feature_range: tuple[float, float] = (0.0, 1.0))

Linearly rescale a series into feature_range.

yᵢ = a + (b − a) (xᵢ − min x) / (max x − min x),

with (a, b) the feature_range.

Attributes:

Name	Type	Description
`feature_range`	`tuple[float, float]`	Target `(low, high)` interval.

Examples:

>>> MinMaxScale().fit_apply([0.0, 5.0, 10.0])
[0.0, 0.5, 1.0]
>>> MinMaxScale(feature_range=(-1.0, 1.0)).fit_apply([0.0, 5.0, 10.0])
[-1.0, 0.0, 1.0]

fit ¶

fit(series: Series) -> Self

Learn the minimum and maximum of series.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The reference observations.	required

Returns:

Type	Description
`Self`	`self`.

Raises:

Type	Description
`EmptySeriesError`	If `series` is empty.
`ValidationError`	If `series` is constant.

Source code in src/mypackage/transforms.py

def fit(self, series: Series) -> Self:
    """Learn the minimum and maximum of ``series``.

    Args:
        series: The reference observations.

    Returns:
        ``self``.

    Raises:
        EmptySeriesError: If ``series`` is empty.
        ValidationError: If ``series`` is constant.
    """
    values = require_non_empty(series)
    minimum, maximum = min(values), max(values)
    if minimum == maximum:
        msg = "cannot rescale a constant series (max == min)"
        raise ValidationError(msg)
    self.minimum_ = minimum
    self.maximum_ = maximum
    return self

apply ¶

apply(series: Series) -> list[float]

Rescale series with the learned bounds.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The observations to transform.	required

Returns:

Type	Description
`list[float]`	A new list of rescaled values.

Raises:

Type	Description
`NotFittedError`	If `fit` has not been called.

Source code in src/mypackage/transforms.py

def apply(self, series: Series) -> list[float]:
    """Rescale ``series`` with the learned bounds.

    Args:
        series: The observations to transform.

    Returns:
        A new list of rescaled values.

    Raises:
        NotFittedError: If `fit` has not been called.
    """
    if self.minimum_ is None or self.maximum_ is None:
        msg = "MinMaxScale is not fitted; call .fit(series) first"
        raise NotFittedError(msg)
    low, high = self.feature_range
    span = self.maximum_ - self.minimum_
    return [
        low + (high - low) * (value - self.minimum_) / span
        for value in as_floats(series)
    ]

fit_apply ¶

fit_apply(series: Series) -> list[float]

Fit on series and immediately transform it.

Source code in src/mypackage/transforms.py

def fit_apply(self, series: Series) -> list[float]:
    """Fit on ``series`` and immediately transform it."""
    return self.fit(series).apply(series)

Clip `dataclass` ¶

Clip(lower: float | None = None, upper: float | None = None)

Clamp every value into [lower, upper].

Stateless: fit is a no-op that returns self, so Clip drops into a Pipeline next to fitted transforms without special casing.

Attributes:

Name	Type	Description
`lower`	`float \| None`	Lower bound, or `None` for unbounded below.
`upper`	`float \| None`	Upper bound, or `None` for unbounded above.

Examples:

>>> Clip(lower=0.0).apply([-2.0, 0.5, 3.0])
[0.0, 0.5, 3.0]
>>> Clip(lower=0.0, upper=1.0).apply([-2.0, 0.5, 3.0])
[0.0, 0.5, 1.0]

fit ¶

fit(series: Series) -> Clip

Return self — Clip learns nothing from data.

Source code in src/mypackage/transforms.py

def fit(self, series: Series) -> Clip:
    """Return ``self`` — `Clip` learns nothing from data."""
    return self

apply ¶

apply(series: Series) -> list[float]

Clamp every value of series into the configured bounds.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The observations to clamp.	required

Returns:

Type	Description
`list[float]`	A new list of clamped values.

Source code in src/mypackage/transforms.py

def apply(self, series: Series) -> list[float]:
    """Clamp every value of ``series`` into the configured bounds.

    Args:
        series: The observations to clamp.

    Returns:
        A new list of clamped values.
    """
    values = as_floats(series)
    if self.lower is not None:
        values = [max(self.lower, value) for value in values]
    if self.upper is not None:
        values = [min(self.upper, value) for value in values]
    return values

fit_apply ¶

fit_apply(series: Series) -> list[float]

Fit on series and immediately transform it.

Source code in src/mypackage/transforms.py

def fit_apply(self, series: Series) -> list[float]:
    """Fit on ``series`` and immediately transform it."""
    return self.apply(series)

MovingAverage `dataclass` ¶

MovingAverage(window: int, padding: Padding | str = EDGE)

Transform adapter around moving_average.

Stateless, so it composes freely inside a Pipeline.

Attributes:

Name	Type	Description
`window`	`int`	Number of samples averaged per output point.
`padding`	`Padding \| str`	Boundary strategy, see `Padding`.

Examples:

>>> MovingAverage(window=3).apply([1.0, 2.0, 3.0, 4.0, 5.0])[1:4]
[2.0, 3.0, 4.0]

fit ¶

fit(series: Series) -> MovingAverage

Return self — MovingAverage learns nothing from data.

Source code in src/mypackage/transforms.py

def fit(self, series: Series) -> MovingAverage:
    """Return ``self`` — `MovingAverage` learns nothing from data."""
    return self

apply ¶

apply(series: Series) -> list[float]

Smooth series with the configured window.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The observations to smooth.	required

Returns:

Type	Description
`list[float]`	A new list of smoothed values.

Source code in src/mypackage/transforms.py

def apply(self, series: Series) -> list[float]:
    """Smooth ``series`` with the configured window.

    Args:
        series: The observations to smooth.

    Returns:
        A new list of smoothed values.
    """
    return moving_average(series, self.window, padding=self.padding)

fit_apply ¶

fit_apply(series: Series) -> list[float]

Fit on series and immediately transform it.

Source code in src/mypackage/transforms.py

def fit_apply(self, series: Series) -> list[float]:
    """Fit on ``series`` and immediately transform it."""
    return self.apply(series)

Pipeline `dataclass` ¶

Pipeline(steps: list[Transform] = list())

Apply a sequence of transforms left to right.

Fitting is sequential: each step is fitted on the output of the previous one, exactly as it will see the data at apply time.

Attributes:

Name	Type	Description
`steps`	`list[Transform]`	The transforms to apply, in order.

Examples:

>>> pipeline = Pipeline([Standardize(), Clip(lower=-1.0, upper=1.0)])
>>> [round(v, 4) for v in pipeline.fit_apply([10.0, 12.0, 14.0, 40.0])]
[-0.6385, -0.4966, -0.3547, 1.0]

Pipelines compose with |, and nest:

>>> combined = pipeline | MovingAverage(window=3)
>>> len(combined)
3
>>> [round(v, 4) for v in combined.fit_apply([10.0, 12.0, 14.0, 40.0])]
[-0.5912, -0.4966, 0.0495, 0.5484]

Empty pipelines are the identity:

>>> Pipeline([]).fit_apply([1.0, 2.0])
[1.0, 2.0]

fit ¶

fit(series: Series) -> Self

Fit every step in turn on the output of the previous step.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The reference observations.	required

Returns:

Type	Description
`Self`	`self`.

Source code in src/mypackage/transforms.py

def fit(self, series: Series) -> Self:
    """Fit every step in turn on the output of the previous step.

    Args:
        series: The reference observations.

    Returns:
        ``self``.
    """
    current = as_floats(series)
    for step in self.steps:
        fit = getattr(step, "fit", None)
        if callable(fit):
            fit(current)
        current = step.apply(current)
    return self

apply ¶

apply(series: Series) -> list[float]

Apply every step in turn.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The observations to transform.	required

Returns:

Type	Description
`list[float]`	A new list holding the output of the final step.

Raises:

Type	Description
`NotFittedError`	If any stateful step has not been fitted.

Source code in src/mypackage/transforms.py

def apply(self, series: Series) -> list[float]:
    """Apply every step in turn.

    Args:
        series: The observations to transform.

    Returns:
        A new list holding the output of the final step.

    Raises:
        NotFittedError: If any stateful step has not been fitted.
    """
    current = as_floats(series)
    for step in self.steps:
        current = step.apply(current)
    return current

fit_apply ¶

fit_apply(series: Series) -> list[float]

Fit every step and return the fully transformed series.

Source code in src/mypackage/transforms.py

def fit_apply(self, series: Series) -> list[float]:
    """Fit every step and return the fully transformed series."""
    return self.fit(series).apply(series)

chain ¶

chain(*transforms: Transform) -> Pipeline

Build a Pipeline from positional transforms.

A small convenience over Pipeline([...]) that reads better inline.

Parameters:

Name	Type	Description	Default
`*transforms`	`Transform`	Transforms to apply, in order.	`()`

Returns:

Type	Description
`Pipeline`	A new pipeline.

Raises:

Type	Description
`ValidationError`	If any argument does not implement `apply`.

Examples:

>>> pipeline = chain(MinMaxScale(), MovingAverage(window=3))
>>> [round(v, 4) for v in pipeline.fit_apply([0.0, 5.0, 10.0])]
[0.1667, 0.5, 0.8333]

Source code in src/mypackage/transforms.py

def chain(*transforms: Transform) -> Pipeline:
    """Build a [`Pipeline`][mypackage.Pipeline] from positional transforms.

    A small convenience over ``Pipeline([...])`` that reads better inline.

    Args:
        *transforms: Transforms to apply, in order.

    Returns:
        A new pipeline.

    Raises:
        ValidationError: If any argument does not implement ``apply``.

    Examples:
        >>> pipeline = chain(MinMaxScale(), MovingAverage(window=3))
        >>> [round(v, 4) for v in pipeline.fit_apply([0.0, 5.0, 10.0])]
        [0.1667, 0.5, 0.8333]
    """
    return Pipeline(list(transforms))

Transforms¶

transforms ¶

Standardize dataclass ¶

fit ¶

apply ¶

inverse ¶

fit_apply ¶

MinMaxScale dataclass ¶

fit ¶

apply ¶

fit_apply ¶

Clip dataclass ¶

fit ¶

apply ¶

fit_apply ¶

MovingAverage dataclass ¶

fit ¶

apply ¶

fit_apply ¶

Pipeline dataclass ¶

fit ¶

apply ¶

fit_apply ¶

chain ¶

Standardize `dataclass` ¶

MinMaxScale `dataclass` ¶

Clip `dataclass` ¶

MovingAverage `dataclass` ¶

Pipeline `dataclass` ¶