R2Score¶

class R2Score(BasePointScorer):
    r"""R-squared (Coefficient of Determination) metric for point forecasts.

    Computes the proportion of variance in the true values that is explained
    by the predictions. A score of 1.0 indicates perfect prediction, 0.0
    indicates performance equivalent to predicting the mean, and negative
    values indicate worse performance than predicting the mean.

    The R² is defined as:

    $$R^2 = 1 - \frac{\sum_{i=1}^{n}(y_i - \hat{y}_i)^2}{\sum_{i=1}^{n}(y_i - \bar{y})^2}$$

    where $y_i$ is the actual value, $\hat{y}_i$ is the predicted value,
    $\bar{y}$ is the mean of actual values, 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"].
    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 R²
        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 R². Higher values indicate better fit.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.metrics import R2Score
    >>> 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, 18.0, 31.0],
    ... })
    >>> r2 = R2Score()
    >>> _ = r2.fit(y_true)
    >>> r2.score(y_true, y_pred)  # doctest: +ELLIPSIS
    0.955

    Notes
    -----
    - R² = 1.0 means perfect prediction
    - R² = 0.0 means predictions are as good as predicting the mean
    - R² < 0 means predictions are worse than predicting the mean
    - When SS_tot = 0 (constant true values), returns 0.0 by convention
    - Overrides ``score()`` because computing the denominator (SS_tot) requires
      access to the full ``y_truth`` column, not just per-row errors

    See Also
    --------
    - [`MeanSquaredError`][yohou.metrics.point.MeanSquaredError] : Mean Squared Error, the numerator component of R²
    - [`MeanAbsoluteError`][yohou.metrics.point.MeanAbsoluteError] : Mean Absolute Error, alternative regression metric

    """

    _metric_name = "r2"

    _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. R² fully overrides ``score()`` and never calls this hook."""
        raise NotImplementedError("R2Score overrides score(); _compute_raw_errors is never called.")

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

        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
            R² score. 1.0 for perfect predictions, 0.0 for mean-level predictions.

        """
        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). R² 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_r2(yt_slice: pl.DataFrame, yp_slice: pl.DataFrame) -> pl.DataFrame:
            """Compute per-column R² score."""
            r2_values = {}
            for col in yt_slice.columns:
                truth = yt_slice[col].to_numpy().astype(np.float64)
                pred = yp_slice[col].to_numpy().astype(np.float64)
                ss_res = np.sum((truth - pred) ** 2)
                ss_tot = np.sum((truth - np.mean(truth)) ** 2)
                r2_values[col] = 1.0 - ss_res / ss_tot if ss_tot != 0 else 0.0
            return pl.DataFrame(r2_values).select(yt_slice.columns)

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