PinballLoss¶

class PinballLoss(BaseIntervalScorer):
    r"""Pinball Loss (Quantile Score) for prediction intervals.

    Evaluates quantile forecasts with asymmetric penalty. Each interval bound
    corresponds to a quantile: lower bound = (1-α)/2, upper bound = (1+α)/2.

    The pinball loss for quantile τ is:

    $$\\text{Pinball}(\\tau, y, q) = \\begin{cases}
    \\tau(y - q) & \\text{if } y \\geq q \\\\
    (1-\\tau)(q - y) & \\text{if } y < q
    \\end{cases}$$

    For interval forecasts at coverage α, the total pinball loss is the sum of
    losses for lower (τ=(1-α)/2) and upper (τ=(1+α)/2) bounds.

    Parameters
    ----------
    aggregation_method : list of str or str, default="all"
        Dimensions to collapse when aggregating scores. Orthogonal modes:

        - "stepwise": Collapse the forecasting-step dimension.
        - "vintagewise": Collapse the vintage/observed-time dimension.
        - "componentwise": Collapse components, return per-timestep scores.
        - "groupwise": Collapse panel groups (panel data only).
        - "coveragewise": Collapse coverage rates (return average pinball loss).

        - "all": Collapse all dimensions (returns scalar).
    coverage_rates : list of float, dict of float to float, or None, default=None
        Coverage rate filter (list) or filter with weights (dict).
    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
        True for pinball loss (lower is better).

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.metrics import PinballLoss
    >>> y_true = pl.DataFrame({
    ...     "time": [datetime(2020, 1, 1), datetime(2020, 1, 2)],
    ...     "value": [10.0, 20.0]
    ... })
    >>> y_pred = pl.DataFrame({
    ...     "vintage_time": [datetime(2019, 12, 31)] * 2,
    ...     "time": [datetime(2020, 1, 1), datetime(2020, 1, 2)],
    ...     "value_lower_0.9": [8.0, 18.0],
    ...     "value_upper_0.9": [12.0, 22.0]
    ... })
    >>> loss = PinballLoss()
    >>> _ = loss.fit(y_true)
    >>> loss.score(y_true, y_pred)  # doctest: +ELLIPSIS
    0.2...

    Notes
    -----
    - Lower is better
    - Penalizes under-prediction differently than over-prediction
    - Scale-dependent
    - More relevant for quantile regression forecasters
    - Useful when asymmetric errors have different costs

    See Also
    --------
    - [`IntervalScore`][yohou.metrics.interval.IntervalScore] : Symmetric penalty for coverage violations
    - [`EmpiricalCoverage`][yohou.metrics.interval.EmpiricalCoverage] : Coverage-only metric

    """

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

    _metric_name = "loss"

    def __init__(
        self,
        aggregation_method: list[str] | str = "all",
        coverage_rates: list[float] | dict[float, float] | None = None,
        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:
        agg_list = aggregation_method
        if aggregation_method == "all":
            agg_list = ["stepwise", "vintagewise", "componentwise", "groupwise", "coveragewise"]

        super().__init__(
            aggregation_method=agg_list,
            coverage_rates=coverage_rates,
            groups=groups,
            components=components,
            time_weighter=time_weighter,
            step_weighter=step_weighter,
            vintage_weighter=vintage_weighter,
        )

    def _compute_raw_scores(self, y_truth, y_pred, coverage_rates, target_columns):
        """Compute per-row pinball loss values."""
        frames = []
        for rate in coverage_rates:
            tau_lower = (1 - rate) / 2
            tau_upper = (1 + rate) / 2
            rate_data = {}
            for col in target_columns:
                lower_col = f"{col}_lower_{rate}"
                upper_col = f"{col}_upper_{rate}"
                if lower_col in y_pred.columns and upper_col in y_pred.columns:
                    loss_lower = _pinball_loss(tau_lower, y_truth[col], y_pred[lower_col])
                    loss_upper = _pinball_loss(tau_upper, y_truth[col], y_pred[upper_col])
                    loss_lower_series = y_pred.select(
                        loss_lower.alias("loss_lower")  # ty: ignore[unresolved-attribute]
                    )["loss_lower"]
                    loss_upper_series = y_pred.select(
                        loss_upper.alias("loss_upper")  # ty: ignore[unresolved-attribute]
                    )["loss_upper"]
                    rate_data[col] = loss_lower_series + loss_upper_series
            frames.append(pl.DataFrame(rate_data).with_columns(pl.lit(rate).alias("coverage_rate")))
        return pl.concat(frames)