MeanDirectionalAccuracy¶

class MeanDirectionalAccuracy(BasePointScorer):
    r"""Mean Directional Accuracy metric for point forecasts.

    Computes the proportion of time steps where the predicted direction of
    change matches the actual direction of change. This metric evaluates
    whether the forecast correctly predicts upward or downward movements.

    The MDA is defined as:

    $$\text{MDA} = \frac{1}{n-1}\sum_{i=2}^{n}\mathbf{1}[\text{sign}(\Delta y_i) = \text{sign}(\Delta \hat{y}_i)]$$

    where $\Delta y_i = y_i - y_{i-1}$ and $\Delta \hat{y}_i = \hat{y}_i - \hat{y}_{i-1}$.

    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"].
    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 MDA
        is non-linear, only zero-weight timestamps are dropped; non-zero weights
        do not otherwise reweight the score. 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 False for MDA. Higher values indicate better directional prediction.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.metrics import MeanDirectionalAccuracy
    >>> y_true = pl.DataFrame({
    ...     "time": [
    ...         datetime(2020, 1, 1),
    ...         datetime(2020, 1, 2),
    ...         datetime(2020, 1, 3),
    ...         datetime(2020, 1, 4),
    ...         datetime(2020, 1, 5),
    ...     ],
    ...     "value": [10.0, 15.0, 12.0, 18.0, 20.0],
    ... })
    >>> y_pred = pl.DataFrame({
    ...     "vintage_time": [datetime(2019, 12, 31)] * 5,
    ...     "time": [
    ...         datetime(2020, 1, 1),
    ...         datetime(2020, 1, 2),
    ...         datetime(2020, 1, 3),
    ...         datetime(2020, 1, 4),
    ...         datetime(2020, 1, 5),
    ...     ],
    ...     "value": [10.0, 14.0, 15.0, 17.0, 19.0],
    ... })
    >>> mda = MeanDirectionalAccuracy()
    >>> _ = mda.fit(y_true)
    >>> mda.score(y_true, y_pred)
    0.75

    Notes
    -----
    - MDA = 1.0 means all directional changes were predicted correctly
    - MDA = 0.5 is equivalent to random guessing for direction
    - MDA = 0.0 means all directional predictions were wrong
    - Requires at least 2 time steps (N-1 comparisons from ``.diff()``)
    - Returns 0.0 when fewer than 2 rows are available and
      ``aggregation_method='all'``; raises ``ValueError`` for all other
      aggregation settings
    - Overrides ``score()`` because computing direction requires ``.diff()``
      on the full columns, not per-row errors

    See Also
    --------
    - [`MeanAbsoluteError`][yohou.metrics.point.MeanAbsoluteError] : Error magnitude metric (not directional)
    - [`R2Score`][yohou.metrics.point.R2Score] : Variance explained metric

    """

    _metric_name = "mda"

    _lower_is_better = False

    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:
        """Not reachable. MDA fully overrides ``score()`` and never calls this hook."""
        raise NotImplementedError("MeanDirectionalAccuracy overrides score(); _compute_raw_errors is never called.")

    def score(
        self,
        y_truth: pl.DataFrame,
        y_pred: pl.DataFrame,
        /,
        **params,
    ) -> float | pl.DataFrame:
        """Compute Mean Directional Accuracy.

        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
            MDA score between 0 and 1. 1.0 for perfect directional prediction.

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

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

        if len(y_truth) < 2:
            # Direction requires at least 2 rows. Only the fully-collapsed
            # (scalar) aggregation has a well-defined default; for any
            # DataFrame-shaped aggregation the return-type contract cannot be
            # honored with too few rows, so fail loudly instead.
            dims = self._normalize_agg_methods(self.aggregation_method)
            scalar_dims = {"stepwise", "vintagewise", "componentwise", "groupwise"}
            if scalar_dims.issubset(dims):
                return 0.0
            raise ValueError(
                "MeanDirectionalAccuracy requires at least 2 rows to compute a "
                f"directional change, but received {len(y_truth)}. Provide more "
                "data or use aggregation_method='all'."
            )

        # Resolve all weighters: drop zero-weight time/step rows and store
        # vintage weights in context (mirrors the base point pipeline). MDA 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,
        )

        def _compute_mda(yt_slice: pl.DataFrame, yp_slice: pl.DataFrame) -> pl.DataFrame | None:
            """Compute per-column mean directional accuracy."""
            if len(yt_slice) < 2:
                return None
            mda_values = {}
            for col in yt_slice.columns:
                truth_diff = np.diff(yt_slice[col].to_numpy().astype(np.float64))
                pred_diff = np.diff(yp_slice[col].to_numpy().astype(np.float64))
                matches = (np.sign(truth_diff) == np.sign(pred_diff)).astype(np.float64)
                mda_values[col] = float(np.mean(matches))
            return pl.DataFrame(mda_values).select(yt_slice.columns)

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