SeasonalLogDifferencing¶

class SeasonalLogDifferencing(SeasonalDifferencing, LogTransformer):
    """Seasonal log differencing time series transformer.

    Applies log transform followed by seasonal differencing. This combines
    variance stabilization with seasonal stationarization.

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

    offset : float >= 0.0, default=0.0
        Offset to apply to the input time series before the log transform.

    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).
    log_transform_ : LogTransformer
        Fitted log transform component (internal; used by
        ``inverse_transform`` only).
    seasonal_diff_transform_ : SeasonalDifferencing
        Fitted seasonal differencing component (internal; used by
        ``inverse_transform`` only).

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.stationarity import SeasonalLogDifferencing
    >>> X = pl.DataFrame({
    ...     "time": [datetime(2024, 1, i) for i in range(1, 6)],
    ...     "value": [1.0, 2.0, 4.0, 8.0, 16.0],
    ... })
    >>> transformer = SeasonalLogDifferencing(seasonality=1)
    >>> transformer.fit(X)  # doctest: +ELLIPSIS
    SeasonalLogDifferencing(...)
    >>> X_t = transformer.transform(X)
    >>> "time" in X_t.columns
    True

    Notes
    -----
    This is equivalent to computing ``log(x_t + offset) - log(x_{t-s} + offset)``
    which equals ``log((x_t + offset) / (x_{t-s} + offset))``.

    This transformer is stateful with ``observation_horizon = seasonality``.
    The first ``seasonality`` rows are dropped in the output. Inverse
    transform requires the ``X_p`` prior-context argument (the untransformed
    rows preceding the differenced output).

    References
    ----------
    [1] Box, G.E.P., & Cox, D.R. (1964). "An analysis of
        transformations." Journal of the Royal Statistical Society:
        Series B, 26(2), 211-252.
    [2] Hyndman, R.J., & Athanasopoulos, G. (2021). "Forecasting:
        principles and practice," 3rd edition, OTexts: Melbourne, Australia.
        OTexts.com/fpp3. Chapter 9.1.

    See Also
    --------
    - [`SeasonalDifferencing`][yohou.stationarity.transformers.SeasonalDifferencing] : Simple seasonal differencing without log transform.
    - [`LogTransformer`][yohou.stationarity.transformers.LogTransformer] : Log transform without differencing.
    - [`SeasonalReturn`][yohou.stationarity.transformers.SeasonalReturn] : Percentage returns instead of log differences.

    """

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

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

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

    def _fit(self, X: pl.DataFrame, y: pl.DataFrame | None = None) -> None:
        """Fit the internal model.

        ``log_transform_`` and ``seasonal_diff_transform_`` are kept solely as
        vehicles for ``_inverse_transform``; the forward ``_transform`` applies
        the log and seasonal-difference steps inline rather than delegating to
        them.
        """
        self.log_transform_ = LogTransformer(offset=self.offset).fit(X=X, y=y)
        self.seasonal_diff_transform_ = SeasonalDifferencing(seasonality=self.seasonality).fit(X=X, y=y)

    def _transform(self, X: pl.DataFrame) -> pl.DataFrame:
        """Transform the input time series."""
        # Apply log transform
        X_t = LogTransformer._transform(self, X)

        # Apply seasonal differencing manually (skip validate_data since columns are transformed)
        time = X_t.select(cs.by_name("time"))[self.seasonality :]
        X_diff = X_t.select(~cs.by_name("time")).select(pl.all().diff(self.seasonality))[self.seasonality :]
        feature_names = self.get_feature_names_out()
        X_diff = X_diff.rename(dict(zip(X_diff.columns, feature_names, strict=False)))
        X_t = pl.concat([time, X_diff], 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,
        )

        X_p = self.log_transform_.transform(X=X_p)
        X = self.seasonal_diff_transform_.inverse_transform(X_t=X_t, X_p=X_p)
        X = self.log_transform_.inverse_transform(X_t=X, X_p=None)

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

        return feature_names