SeasonalDifferencing¶

class SeasonalDifferencing(BaseActualTransformer):
    r"""Seasonal differencing time series transformer.

    Computes the difference between each value and its value at a seasonal
    lag:

    $$\Delta_s y_t = y_t - y_{t-s}$$

    where $s$ is the ``seasonality``. This removes seasonal patterns and is
    a common stationarization technique.

    Parameters
    ----------
    seasonality : int >= 1, default=1
        Seasonality for the differencing.

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

    Notes
    -----
    This transformer is stateful with ``observation_horizon = seasonality``.
    The first ``seasonality`` rows are dropped in the output since they lack
    sufficient history for differencing.

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

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.stationarity import SeasonalDifferencing

    >>> X = pl.DataFrame({
    ...     "time": [datetime(2020, 1, i) for i in range(1, 8)],
    ...     "value": [10.0, 12.0, 15.0, 13.0, 11.0, 14.0, 16.0],
    ... })

    >>> # First-order differencing (seasonality=1)
    >>> transformer = SeasonalDifferencing(seasonality=1)
    >>> transformer.fit(X)  # doctest: +ELLIPSIS
    SeasonalDifferencing(...)
    >>> X_diff = transformer.transform(X)
    >>> len(X_diff)  # One row dropped
    6

    See Also
    --------
    - [`SeasonalLogDifferencing`][yohou.stationarity.transformers.SeasonalLogDifferencing] : Log transform followed by seasonal differencing.
    - [`SeasonalReturn`][yohou.stationarity.transformers.SeasonalReturn] : Compute seasonal returns ((x_t - x_{t-s}) / x_{t-s}).

    """

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

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

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

    @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_t = X.select(~cs.by_name("time")).select(pl.all().diff(self.seasonality))[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."""
        return _inverse_seasonal_diff(self, X_t, X_p, self.seasonality)

    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"diff_s_{self.seasonality}") for col in input_features]

        return feature_names