MaxAbsoluteError¶

class MaxAbsoluteError(BasePointScorer):
    r"""Maximum Absolute Error metric for point forecasts.

    Computes the maximum of absolute differences between predictions and actual
    values. This metric captures worst case prediction error, providing a bound
    on how far off the forecast can be.

    The MaxAE is defined as:

    $$\text{MaxAE} = \max_{i=1}^{n}|y_i - \hat{y}_i|$$

    where $y_i$ is the actual value, $\hat{y}_i$ is the predicted value, and
    $n$ is the number of observations.

    Parameters
    ----------
    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).
    time_weighter : BaseWeighter or None, default=None
        Weighter applied along the time axis (observed timestamps). If None,
        all timestamps contribute equally.
    step_weighter : BaseWeighter or None, default=None
        Weighter applied along the forecasting-step axis. If None, all
        forecasting steps contribute equally.
    vintage_weighter : BaseWeighter or None, default=None
        Weighter applied along the vintage-time axis. If None, all vintages
        contribute equally.

    Attributes
    ----------
    lower_is_better : bool
        Always True for MaxAE.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.metrics import MaxAbsoluteError
    >>> y_true = pl.DataFrame({
    ...     "time": [datetime(2020, 1, 1), datetime(2020, 1, 2), datetime(2020, 1, 3)],
    ...     "value": [10.0, 20.0, 30.0],
    ... })
    >>> y_pred = pl.DataFrame({
    ...     "vintage_time": [datetime(2019, 12, 31)] * 3,
    ...     "time": [datetime(2020, 1, 1), datetime(2020, 1, 2), datetime(2020, 1, 3)],
    ...     "value": [12.0, 19.0, 25.0],
    ... })
    >>> max_ae = MaxAbsoluteError()
    >>> _ = max_ae.fit(y_true)
    >>> max_ae.score(y_true, y_pred)
    5.0

    Notes
    -----
    - MaxAE captures the worst case prediction error in a forecast
    - Highly sensitive to outliers by design
    - Interpretable in the same units as the target variable
    - Row collapse uses ``max`` (not ``mean``), while component and group collapse
      use weighted ``mean`` (consistent with the pipeline convention)

    See Also
    --------
    - [`MeanAbsoluteError`][yohou.metrics.point.MeanAbsoluteError] : Mean Absolute Error, average case measure
    - [`MedianAbsoluteError`][yohou.metrics.point.MedianAbsoluteError] : Median Absolute Error, robust central tendency measure

    """

    _metric_name = "max_ae"

    def __init__(
        self,
        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,
        )

    def _compute_raw_errors(self, y_truth: pl.DataFrame, y_pred: pl.DataFrame) -> pl.DataFrame:
        """Compute per-row absolute errors."""
        return (y_truth - y_pred).select(pl.all().abs())

    def _collapse_rows(
        self,
        df: pl.DataFrame,
        context: ScoringContext | None,
        dims: set[str],
    ) -> pl.DataFrame:
        """Collapse row dimensions using max instead of mean."""
        return self._collapse_rows_with(df, context, dims, agg_fn="max")