SeasonalReturn

yohou.stationarity.transformers.SeasonalReturn

Bases: BaseTransformer

Seasonal percentage return time series transformer.

Computes the percentage return relative to the value from seasonality time steps ago:

\[r_t = \frac{X_t + \text{offset}}{X_{t-s} + \text{offset}} - 1\]

This is useful for modeling relative changes in time series with seasonal patterns, such as year-over-year percentage growth.

Parameters

Name	Type	Description	Default
seasonality	int > 1	Seasonality lag for computing returns.	1
offset	float >= 0.0	Offset to apply to avoid division by zero. Should be positive if data contains zeros or near-zero values.	0.0

Attributes

Name	Type	Description
n_features_in_	int	Number of features seen during fit.
feature_names_in_	list of str	Names of features seen during fit (excluding "time" column).

Examples

>>> import polars as pl
>>> from datetime import datetime
>>> from yohou.stationarity import SeasonalReturn
>>> X = pl.DataFrame({
...     "time": [datetime(2024, 1, i) for i in range(1, 6)],
...     "value": [100.0, 110.0, 105.0, 115.0, 120.0],
... })
>>> transformer = SeasonalReturn(seasonality=2, offset=0.0)
>>> transformer.fit(X)
SeasonalReturn(...)
>>> X_t = transformer.transform(X)
>>> len(X_t) == len(X) - 2  # First 2 rows dropped
True

References

[1] Hyndman, R.J., & Athanasopoulos, G. (2021). "Forecasting: principles and practice," 3^rd edition, OTexts: Melbourne, Australia. OTexts.com/fpp3. Chapter 9.1.

See Also

• AbsoluteSeasonalReturn : Absolute difference instead of percentage return.
• SeasonalDifferencing : Simple differencing without percentage computation.
• SeasonalLogDifferencing : Log-differencing for multiplicative relationships.

Source Code

View on GitHub

Show/Hide source

class SeasonalReturn(BaseTransformer):
    r"""Seasonal percentage return time series transformer.

    Computes the percentage return relative to the value from ``seasonality``
    time steps ago:

    $$r_t = \frac{X_t + \text{offset}}{X_{t-s} + \text{offset}} - 1$$

    This is useful for modeling relative changes in time series with
    seasonal patterns, such as year-over-year percentage growth.

    Parameters
    ----------
    seasonality : int > 1, default=1
        Seasonality lag for computing returns.

    offset : float >= 0.0, default=0.0
        Offset to apply to avoid division by zero. Should be positive if
        data contains zeros or near-zero values.

    Attributes
    ----------
    n_features_in_ : int
        Number of features seen during fit.
    feature_names_in_ : list of str
        Names of features seen during fit (excluding "time" column).

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.stationarity import SeasonalReturn
    >>> X = pl.DataFrame({
    ...     "time": [datetime(2024, 1, i) for i in range(1, 6)],
    ...     "value": [100.0, 110.0, 105.0, 115.0, 120.0],
    ... })
    >>> transformer = SeasonalReturn(seasonality=2, offset=0.0)
    >>> transformer.fit(X)  # doctest: +ELLIPSIS
    SeasonalReturn(...)
    >>> X_t = transformer.transform(X)
    >>> len(X_t) == len(X) - 2  # First 2 rows dropped
    True

    References
    ----------
    [1] Hyndman, R.J., & Athanasopoulos, G. (2021). "Forecasting:
        principles and practice," 3rd edition, OTexts: Melbourne, Australia.
        OTexts.com/fpp3. Chapter 9.1.

    See Also
    --------
    - [`AbsoluteSeasonalReturn`][yohou.stationarity.transformers.AbsoluteSeasonalReturn] : Absolute difference instead of percentage return.
    - [`SeasonalDifferencing`][yohou.stationarity.transformers.SeasonalDifferencing] : Simple differencing without percentage computation.
    - [`SeasonalLogDifferencing`][yohou.stationarity.transformers.SeasonalLogDifferencing] : Log-differencing for multiplicative relationships.

    """

    _parameter_constraints: dict = {
        "seasonality": [Interval(numbers.Integral, 1, None, closed="left")],
        "offset": [Interval(numbers.Real, 0, None, closed="left")],
    }

    _tags = {"stateful": True, "invertible": True}

    def __init__(self, seasonality: StrictInt = 1, offset: StrictFloat = 0.0):
        self.seasonality = seasonality
        self.offset = offset

    @property
    def observation_horizon(self) -> int:  # noqa: D102
        """Return the number of past observations needed."""
        return self.seasonality

    def _transform(self, X: pl.DataFrame) -> pl.DataFrame:
        """Transform the input time series."""
        time = X.select(cs.by_name("time"))[self.seasonality :]
        X_numeric = X.select(~cs.by_name("time")) + self.offset

        # Compute return: (X_t / X_{t-seasonality}) - 1
        X_lagged = X_numeric.shift(self.seasonality)
        X_t = (X_numeric / X_lagged - 1)[self.seasonality :]

        feature_names = self.get_feature_names_out()
        X_t = X_t.rename(dict(zip(X_t.columns, feature_names, strict=False)))
        X_t = pl.concat([time, X_t], how="horizontal")

        return X_t

    def _inverse_transform(self, X_t: pl.DataFrame, X_p: pl.DataFrame | None = None) -> pl.DataFrame:
        """Inverse-transform the time series."""
        X_t, X_p = validate_transformer_data(
            self,
            X=X_t,
            reset=False,
            inverse=True,
            X_p=X_p,
            observation_horizon=self.observation_horizon,
            stateful=True,
        )

        time = X_t.select(cs.by_name("time"))

        # Shift X_p by offset for computation
        X_p_shifted = X_p.select(~cs.by_name("time")) + self.offset
        X_t_numeric = X_t.select(~cs.by_name("time"))
        X_t_numeric.columns = X_p_shifted.columns

        # Inverse: X_t = (return_t + 1) * X_{t-seasonality} - offset
        # We need to reconstruct step by step
        X_combined = pl.concat([X_p_shifted, X_t_numeric])

        # Get the columns and their dtypes (excluding "time")
        cols_and_dtypes = list(zip(X_combined.columns, X_combined.dtypes, strict=False))

        def inverse_return_col(series: pl.Series) -> pl.Series:
            """Reverse seasonal return for a single series."""
            arr = series.to_numpy().copy()
            for i in range(len(X_p), len(arr)):
                # X_t = (return_t + 1) * X_{t-seasonality}
                arr[i] = (arr[i] + 1) * arr[i - self.seasonality]
            return pl.Series(arr)

        X = (
            X_combined.with_columns([
                pl.col(col).map_batches(inverse_return_col, return_dtype=dtype) for col, dtype in cols_and_dtypes
            ])[len(X_p) :]
            - self.offset
        )

        X.columns = self.feature_names_in_
        X = pl.concat([time, X], how="horizontal")

        return X

    def get_feature_names_out(self, input_features: list[str] | None = None) -> list[str]:
        """Get output feature names for transformation.

        Parameters
        ----------
        input_features : array-like of str or None, default=None
            Column names of the input features.  If ``None``, uses the
            feature names seen during ``fit``.

        Returns
        -------
        list of str
            Output feature names after transformation.

        """
        input_features = _check_feature_names_in(self, input_features)
        feature_names = [
            panel_aware_prefix(col, f"return_s_{self.seasonality}_off_{self.offset}") for col in input_features
        ]

        return feature_names

Methods

observation_horizon property

Return the number of past observations needed.

get_feature_names_out(input_features=None)

Get output feature names for transformation.

Parameters

Name	Type	Description	Default
input_features	array-like of str or None	Column names of the input features. If None, uses the feature names seen during fit.	None

Returns

Type	Description
list of str	Output feature names after transformation.

Source Code

View on GitHub

Show/Hide source

def get_feature_names_out(self, input_features: list[str] | None = None) -> list[str]:
    """Get output feature names for transformation.

    Parameters
    ----------
    input_features : array-like of str or None, default=None
        Column names of the input features.  If ``None``, uses the
        feature names seen during ``fit``.

    Returns
    -------
    list of str
        Output feature names after transformation.

    """
    input_features = _check_feature_names_in(self, input_features)
    feature_names = [
        panel_aware_prefix(col, f"return_s_{self.seasonality}_off_{self.offset}") for col in input_features
    ]

    return feature_names

Tutorials

The following example notebooks use this component:

• How to Apply Stationarity Transforms

  ---

  Data-Features

  Catalogue of variance-stabilising and detrending transforms: LogTransformer, BoxCox, SeasonalDifferencing, SeasonalReturn, and ASinh with inverse verification.

  View · Open in marimo