RootMeanSquaredScaledError¶

class RootMeanSquaredScaledError(BasePointScorer):
    r"""Root Mean Squared Scaled Error metric for point forecasts.

    Computes RMSE scaled by the in-sample naive seasonal forecast error. This provides
    a scale-independent metric that enables comparison across time series with different
    magnitudes. Requires training data to compute scaling factors.

    The RootMeanSquaredScaledError is defined as:

    $$\\text{RMSSE} = \\sqrt{\\frac{1}{h}\\sum_{t=1}^{h}\\frac{(y_t - \\hat{y}_t)^2}{\\text{scale}_j}}$$

    where the scale is the mean squared seasonal-naive error computed from
    training data as:

    $$\\text{scale}_j = \\frac{1}{T-m}\\sum_{t=m+1}^{T}(y_{t,j} - y_{t-m,j})^2$$

    with $m$ = seasonality, $T$ = training length, $h$ = forecast horizon, and $j$ = column index.
    Per-column RootMeanSquaredScaledError values are averaged to produce the final score.

    Parameters
    ----------
    seasonality : int, default=1
        Seasonal period for computing scaling factors. Must be at least 1.
        Common values: 1 (non-seasonal), 7 (weekly), 12 (monthly), 24 (hourly daily pattern).

    aggregation_method : list of str or str, default="all"
        Dimensions to aggregate over. Options:
        - "stepwise": Aggregate across forecasting steps.
        - "vintagewise": Aggregate across vintages (observed times).
        - "componentwise": Aggregate across components, return per-timestep DataFrame
        - "groupwise": Aggregate across panel groups (panel data only)
        - "all": Aggregate across all dimensions (returns scalar). Same as
          ["stepwise", "vintagewise", "componentwise", "groupwise"].
        Example outputs:
        - ["stepwise", "vintagewise"]: Per-component (and per-group) DataFrame.
        - "componentwise" or ["componentwise"]: Per-timestep (and per-group) DataFrame.
        - "groupwise" or ["groupwise"]: Per-component per-timestep DataFrame (panel aggregated).
        - ["stepwise", "vintagewise", "componentwise"]: Scalar (global) or per-group DataFrame (panel).
        - "all": Scalar float (hierarchically aggregated for panel data).
    groups : list of str, dict of str to float, or None, default=None
        Panel group filter (list) or filter with weights (dict).
    components : list of str, dict of str to float, or None, default=None
        Component filter (list) or filter with weights (dict).

    Attributes
    ----------
    lower_is_better : bool
        Always True for RootMeanSquaredScaledError.
    scales_ : dict[str, float]
        Fitted per-column scaling factors. Computed during fit() from training data
        naive seasonal forecast errors.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime, timedelta
    >>> from yohou.metrics import RootMeanSquaredScaledError
    >>> # Training data
    >>> y_train = pl.DataFrame({
    ...     "time": [datetime(2020, 1, 1) + timedelta(days=i) for i in range(10)],
    ...     "value": [10.0, 12.0, 11.0, 13.0, 12.0, 14.0, 13.0, 15.0, 14.0, 16.0],
    ... })
    >>> # Test predictions
    >>> y_true = pl.DataFrame({
    ...     "time": [datetime(2020, 1, 11), datetime(2020, 1, 12)],
    ...     "value": [15.0, 17.0],
    ... })
    >>> y_pred = pl.DataFrame({
    ...     "vintage_time": [datetime(2020, 1, 10)] * 2,
    ...     "time": [datetime(2020, 1, 11), datetime(2020, 1, 12)],
    ...     "value": [15.5, 16.5],
    ... })
    >>> rmsse = RootMeanSquaredScaledError(seasonality=2)
    >>> rmsse.fit(y_train)
    RootMeanSquaredScaledError(seasonality=2)
    >>> rmsse.score(y_true, y_pred)
    0.5

    Notes
    -----
    - RootMeanSquaredScaledError values are scale-independent, enabling comparison across different time series
    - Values < 1 indicate better performance than naive seasonal forecast on training data
    - Values > 1 indicate worse performance than naive seasonal baseline
    - Requires training data with length > seasonality
    - Per-column scaling factors are stored and applied independently

    See Also
    --------
    - [`RootMeanSquaredError`][yohou.metrics.point.RootMeanSquaredError] : Root Mean Squared Error, non-scaled version
    - [`MeanAbsoluteError`][yohou.metrics.point.MeanAbsoluteError] : Mean Absolute Error, non-scaled alternative
    - [`MeanSquaredError`][yohou.metrics.point.MeanSquaredError] : Mean Squared Error, squared version

    """

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

    _metric_name = "rmsse"

    def __init__(
        self,
        seasonality: int = 1,
        aggregation_method: list[str] | str = "all",
        groups: list[str] | dict[str, float] | None = None,
        components: list[str] | dict[str, float] | None = None,
        time_weighter: BaseWeighter | None = None,
        step_weighter: BaseWeighter | None = None,
        vintage_weighter: BaseWeighter | None = None,
    ) -> None:
        super().__init__(
            aggregation_method=aggregation_method,
            groups=groups,
            components=components,
            time_weighter=time_weighter,
            step_weighter=step_weighter,
            vintage_weighter=vintage_weighter,
        )
        self.seasonality = seasonality

    def __sklearn_tags__(self):
        """Get estimator tags.

        Returns
        -------
        Tags
            Estimator tags with requires_calibration=True.

        """
        tags = super().__sklearn_tags__()
        if tags.scorer_tags is not None:
            tags.scorer_tags.requires_calibration = True
        return tags

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(self, y_train: pl.DataFrame, *, forecaster=None, **params) -> RootMeanSquaredScaledError:
        """Fit the scorer by computing per-column scaling factors.

        Parameters
        ----------
        y_train : pl.DataFrame
            Training set target values with "time" column.
        forecaster : BaseForecaster or None, default=None
            If provided, metadata is extracted directly from the fitted
            forecaster instead of being re-inferred from ``y_train``.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        self

        Raises
        ------
        ValueError
            If y_train is None or seasonality > len(y_train) - 1.

        Warns
        -----
        UserWarning
            If any training column has zero variance for the given
            seasonality; a scale floor of 1e-10 is used instead.

        """
        # Call parent fit() to validate parameters (aggregation_method, groups, etc.)
        super().fit(y_train, forecaster=forecaster, **params)

        # Validate training data and remove time column
        y_train_values, _, _ = validate_scorer_data(scorer=self, y_true=y_train, y_pred=None, reset=True)

        # Compute per-column scaling factors using seasonal naive forecast errors
        self.scales_ = {}
        for col in y_train_values.columns:
            # Compute seasonal differences: y_t - y_{t-seasonality}
            col_data = y_train_values[col].to_numpy()
            seasonal_errors = col_data[self.seasonality :] - col_data[: -self.seasonality]

            # Scale is mean squared error of seasonal naive forecast
            scale = float(np.mean(seasonal_errors**2))

            if scale == 0:
                warnings.warn(
                    f"Training data for column '{col}' has zero scale "
                    f"(constant values with seasonality={self.seasonality}). "
                    "RMSSE scores for this column will use a scale floor of 1e-10.",
                    UserWarning,
                    stacklevel=2,
                )

            # Store non-zero scale (avoid division by zero in score())
            self.scales_[col] = max(scale, 1e-10)

        return self

    def _compute_raw_errors(self, y_truth: pl.DataFrame, y_pred: pl.DataFrame) -> pl.DataFrame:
        """Compute per-row scaled squared errors for RMSSE."""
        missing = [col for col in y_truth.columns if col not in self.scales_]
        if missing:
            raise ValueError(
                f"Columns {missing} were not seen during fit(); RMSSE has no scale "
                f"for them. Fitted columns: {list(self.scales_)}."
            )
        diff = y_truth - y_pred
        return diff.select((pl.col(col) / float(np.sqrt(self.scales_[col]))).pow(2) for col in y_truth.columns)

    def _transform_scores(self, df: pl.DataFrame) -> pl.DataFrame:
        """Apply square root to aggregated scaled squared errors."""
        return df.select(pl.all().sqrt())