PinballLoss

yohou.metrics.interval.PinballLoss

Bases: BaseIntervalScorer

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

Name	Type	Description	Default
aggregation_method	list of str or str	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).	"all"
coverage_rates	list of float, dict of float to float, or None	Coverage rate filter (list) or filter with weights (dict).	None
groups	list of str, dict of str to float, or None	Panel group filter (list) or filter with weights (dict).	None
components	list of str, dict of str to float, or None	Component filter (list) or filter with weights (dict).	None

Attributes

Name	Type	Description
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)
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 : Symmetric penalty for coverage violations
• EmpiricalCoverage : Coverage-only metric

Source Code

View on GitHub

Show/Hide source

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,
    ) -> 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,
        )

    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)

Tutorials

The following example notebooks use this component:

• How to Evaluate Interval Forecasts

  ---

  Evaluation-Search

  Evaluate prediction intervals with EmpiricalCoverage, IntervalScore, MeanIntervalWidth, PinballLoss, and CalibrationError across coverage levels.

  View · Open in marimo