SeasonalNaive¶

class SeasonalNaive(BasePointForecaster):
    """Seasonal naive forecaster that repeats values from previous season.

    Parameters
    ----------
    seasonality : int, default=1
        The seasonal period length. For example, 7 for weekly seasonality
        in daily data, or 12 for monthly seasonality in monthly data.
    panel_strategy : {"global", "multivariate"}, default="global"
        How to handle panel data. See `BaseForecaster` for details.

    Attributes
    ----------
    interval_ : str
        Detected time interval of the training data.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.point import SeasonalNaive
    >>>
    >>> df = pl.DataFrame({
    ...     "time": pl.datetime_range(
    ...         start=datetime(2021, 1, 1),
    ...         end=datetime(2021, 1, 10),
    ...         interval="1d",
    ...         eager=True,
    ...     ),
    ...     "value": [1.0, 2.0, 3.0, 1.0, 2.0, 3.0, 1.0, 2.0, 3.0, 1.0],
    ... })
    >>> forecaster = SeasonalNaive(seasonality=3)
    >>> _ = forecaster.fit(y=df, forecasting_horizon=3)
    >>> y_pred = forecaster.predict(forecasting_horizon=3)
    >>> len(y_pred)
    3

    Notes
    -----
    Predictions repeat the last ``seasonality`` observed values
    cyclically.  For example, with ``seasonality=7`` the forecast for
    each day equals the observation from the same weekday in the last
    observed week.

    See Also
    --------
    - [`MeanSeasonalNaive`][yohou.point.naive.MeanSeasonalNaive] : Averages multiple past seasonal cycles.
    - [`PointReductionForecaster`][yohou.point.reduction.PointReductionForecaster] : ML-based point forecaster.

    """

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

    def __init__(
        self,
        seasonality: StrictInt = 1,
        panel_strategy: Literal["global", "multivariate"] = "global",
    ):
        BasePointForecaster.__init__(
            self,
            actual_transformer=None,
            target_transformer=None,
            target_as_feature=None,
            panel_strategy=panel_strategy,
        )

        self.seasonality = seasonality

    def __sklearn_tags__(self) -> Tags:
        """Get estimator tags.

        Returns
        -------
        Tags
            Estimator tags with yohou-specific attributes.

        """
        tags = super().__sklearn_tags__()
        assert tags.forecaster_tags is not None
        tags.forecaster_tags.requires_exogenous = False
        tags.forecaster_tags.stateful = True
        return tags

    @property
    def _observation_horizon(self) -> int:
        """Size of the rolling observation buffer.

        ``SeasonalNaive`` retains the last ``seasonality`` rows so that
        ``_predict_one`` can tile the full seasonal pattern over the
        forecast horizon.
        """
        return self.seasonality

    def _predict_one(
        self,
        groups: list[str],
        **params,
    ) -> pl.DataFrame:
        """Predict ``fit_forecasting_horizon_`` steps from the observation horizon.

        Parameters
        ----------
        groups : list of str
            Panel group names to predict for.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Predicted time series.

        """
        # Non-panel data
        if self.groups_ is None:
            assert isinstance(self._y_observed, pl.DataFrame)
            pattern = self._y_observed.select(~cs.by_name("time"))
            y_pred = _tile_to_horizon(pattern, self.fit_forecasting_horizon_, self.seasonality)

        # Panel data
        else:
            assert isinstance(self._y_observed, dict)
            y_pred = _build_panel_prediction(
                self._y_observed,
                groups,
                self.fit_forecasting_horizon_,
                self.seasonality,
                lambda values: values,
            )

        y_pred = self._add_time_columns(y_pred)

        return y_pred