MedianAbsoluteError¶

class MedianAbsoluteError(BasePointScorer):
    r"""Median Absolute Error metric for point forecasts.

    Computes the median of absolute differences between predictions and actual values.
    This metric is highly robust to outliers and provides a more stable measure of
    typical error magnitude compared to mean-based metrics.

    The MedianAE is defined as:

    $$\\text{MedianAE} = \\text{median}(|y_i - \\hat{y}_i|)$$

    where $y_i$ is the actual value and $\\hat{y}_i$ is the predicted value.

    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). Because the
        median is non-linear, only zero-weight timestamps are dropped; non-zero
        weights do not otherwise reweight the median. If None, all timestamps
        contribute equally.
    step_weighter : BaseWeighter or None, default=None
        Weighter applied along the forecasting-step axis. Only zero-weight steps
        are dropped (see ``time_weighter``). 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 MedianAE.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.metrics import MedianAbsoluteError
    >>> 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, 28.0],
    ... })
    >>> medae = MedianAbsoluteError()
    >>> _ = medae.fit(y_true)
    >>> medae.score(y_true, y_pred)
    2.0

    Notes
    -----
    - MedianAE is highly robust to outliers and extreme errors
    - Provides a better measure of typical error when error distribution is skewed
    - Less sensitive to a few very large prediction errors compared to MAE
    - Interpretable in the same units as the target variable
    - Suitable when outliers should not dominate the evaluation

    See Also
    --------
    - [`MeanAbsoluteError`][yohou.metrics.point.MeanAbsoluteError] : Mean-based absolute error, more sensitive to outliers
    - [`MaxAbsoluteError`][yohou.metrics.point.MaxAbsoluteError] : Maximum absolute error, worst-case measure

    """

    _parameter_constraints: dict = {
        **BasePointScorer._parameter_constraints,
    }

    _metric_name = "median_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 for median aggregation."""
        return (y_truth - y_pred).select(pl.all().abs())

    def score(
        self,
        y_truth: pl.DataFrame,
        y_pred: pl.DataFrame,
        /,
        **params,
    ) -> float | pl.DataFrame:
        """Compute median absolute error.

        Parameters
        ----------
        y_truth : pl.DataFrame
            True values with "time" column.
        y_pred : pl.DataFrame
            Predicted values with "time" column.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        float or pl.DataFrame
            Aggregated median absolute error.

        """
        check_is_fitted(self, ["_is_fitted"])

        y_truth, y_pred, context = validate_scorer_data(
            self,
            y_truth,
            y_pred,
        )

        # Resolve all weighters: drop zero-weight time/step rows and store
        # vintage weights in context (mirrors the base point pipeline). Median
        # is non-linear, so continuous time/step weights are not multiplied
        # into the errors; only zero-weight rows are removed.
        y_truth, y_pred, context, _, _, _ = self._pre_filter_zero_weights(
            y_truth,
            y_pred,
            context,
            self.time_weighter,
            self.step_weighter,
            self.vintage_weighter,
        )

        dims = self._normalize_agg_methods(self.aggregation_method)
        collapse_steps = "stepwise" in dims
        collapse_vintages = "vintagewise" in dims

        if not collapse_steps and not collapse_vintages:
            # Componentwise/groupwise only: keep per-row scores, let
            # _aggregate_per_vintage_scores handle component/group collapse.
            abs_errors = self._compute_raw_errors(y_truth, y_pred)
            # Per-row median across columns (components)
            result = abs_errors.select(pl.concat_list(pl.all()).alias("_err")).select(
                pl.col("_err").list.eval(pl.element().median()).list.first().alias("score")
            )
            time_values = context.time_values if context is not None else None
            if time_values is not None:
                result = result.with_columns(pl.Series("time", time_values).cast(pl.Datetime))
                result = result.select(["time"] + [c for c in result.columns if c != "time"])
            result = self._rename_metric_columns(result)
            return result

        def _compute_median(yt_slice: pl.DataFrame, yp_slice: pl.DataFrame) -> pl.DataFrame:
            """Compute per-column median absolute error."""
            errors = (yt_slice - yp_slice).select(pl.all().abs())
            return errors.select(pl.all().median())

        result = self._map_per_vintage(y_truth, y_pred, context, _compute_median)
        return self._aggregate_per_vintage_scores(result, context)