Skip to content

BaseConformityScorer

yohou.metrics.conformity_base.BaseConformityScorer

Bases: BaseScorer

Base class for conformal prediction conformity scorers.

Conformity scorers quantify how "unusual" a prediction is compared to the calibration set. Used in conformal prediction to construct valid prediction intervals with coverage guarantees.

See Also

Source Code

View on GitHub

Show/Hide source 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class BaseConformityScorer(BaseScorer, metaclass=abc.ABCMeta):
    """Base class for conformal prediction conformity scorers.

    Conformity scorers quantify how "unusual" a prediction is compared to the
    calibration set. Used in conformal prediction to construct valid prediction
    intervals with coverage guarantees.

    See Also
    --------
    - [`Residual`][yohou.metrics.conformity.Residual] : Concrete conformity scorer.
    - [`AbsoluteResidual`][yohou.metrics.conformity.AbsoluteResidual] : Concrete conformity scorer.
    - [`SplitConformalForecaster`][yohou.interval.split_conformal.SplitConformalForecaster] : Uses conformity scores.

    """

    def __sklearn_tags__(self) -> Tags:
        """Get estimator tags.

        Returns
        -------
        Tags
            Estimator tags with conformity scorer attributes.

        """
        tags = super().__sklearn_tags__()
        assert tags.scorer_tags is not None
        tags.scorer_tags.prediction_type = "conformity"  # ty: ignore[invalid-assignment]
        return tags

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(self, y_train: pl.DataFrame, *, forecaster=None, **params) -> "BaseConformityScorer":
        """Fit the scorer on training data if needed."""
        # Conformity scorers typically don't aggregate results in the same way,
        # so they don't use aggregation_method, but they must implement fit.
        return super().fit(y_train, forecaster=forecaster, **params)

    @staticmethod
    def _compute_assymetric_quantiles(conformity_scores: pl.DataFrame, coverage_rate: float) -> tuple[float, float]:
        """Compute lower and upper quantiles for asymmetric intervals.

        Parameters
        ----------
        conformity_scores : pl.DataFrame
            Conformity scores from calibration.

        coverage_rate : float
            Target coverage rate.

        Returns
        -------
        lower_quantile : float
            Lower quantile value.

        upper_quantile : float
            Upper quantile value.

        """
        # Convert to numpy array for quantile computation
        if isinstance(conformity_scores, pl.DataFrame):
            scores_array = conformity_scores.to_numpy()
        else:
            scores_array = conformity_scores

        # Check if array is empty
        if hasattr(scores_array, "size") and scores_array.size == 0:
            raise ValueError(
                "Cannot compute quantile: conformity_scores is empty. "
                "This typically happens when the calibration set is too small. "
                "Increase calibration_size or reduce forecasting_horizon."
            )

        alpha = 1.0 - coverage_rate

        if coverage_rate == 0:
            median_val: float = np.quantile(scores_array, 0.5, method="lower")
            return median_val, median_val

        lower_quantile: float = np.quantile(scores_array, alpha / 2.0, method="lower")

        upper_quantile: float = np.quantile(scores_array, 1.0 - alpha / 2.0, method="higher")

        return lower_quantile, upper_quantile

    @staticmethod
    def _compute_symetric_quantiles(conformity_scores: pl.DataFrame, coverage_rate: float) -> float:
        """Compute quantile for symmetric intervals.

        Parameters
        ----------
        conformity_scores : pl.DataFrame
            Conformity scores from calibration.

        coverage_rate : float
            Target coverage rate.

        Returns
        -------
        float
            Quantile value for symmetric intervals.

        Raises
        ------
        ValueError
            If conformity_scores is empty.

        """
        # Convert to numpy array for quantile computation
        conformity_array = conformity_scores.to_numpy()

        # Check if array is empty
        if conformity_array.size == 0:
            raise ValueError(
                "Cannot compute quantile: conformity_scores is empty. "
                "This typically happens when the calibration set is too small. "
                "Increase calibration_size or reduce forecasting_horizon."
            )

        quantile: float = np.quantile(conformity_array, coverage_rate, method="lower")

        return quantile

    @staticmethod
    def _format_y_pred_interval(
        lower_bound: pl.DataFrame, upper_bound: pl.DataFrame, coverage_rate: float
    ) -> pl.DataFrame:
        """Format lower and upper bounds into interval DataFrame.

        Parameters
        ----------
        lower_bound : pl.DataFrame
            Lower bound predictions.

        upper_bound : pl.DataFrame
            Upper bound predictions.

        coverage_rate : float
            Coverage rate for labeling columns.

        Returns
        -------
        pl.DataFrame
            Formatted prediction intervals.

        """
        lower_bound.columns = [f"{col}_lower_{coverage_rate}" for col in lower_bound.columns]
        upper_bound.columns = [f"{col}_upper_{coverage_rate}" for col in upper_bound.columns]

        y_pred_interval = pl.concat([lower_bound, upper_bound], how="horizontal")

        return y_pred_interval

    @abc.abstractmethod
    def inverse_score(
        self, y_pred: pl.DataFrame, conformity_scores: pl.DataFrame, coverage_rate: float
    ) -> pl.DataFrame:
        """Transform conformity scores into prediction intervals.

        Parameters
        ----------
        y_pred : pl.DataFrame
            Point predictions.

        conformity_scores : pl.DataFrame
            Conformity scores from calibration.

        coverage_rate : float
            Target coverage probability.

        Returns
        -------
        pl.DataFrame
            Prediction intervals.

        """

Methods

__sklearn_tags__()

Get estimator tags.

Returns
Type Description
Tags

Estimator tags with conformity scorer attributes.
Source Code

View on GitHub

Show/Hide source 
30
31
32
33
34
35
36
37
38
39
40
41
42
def __sklearn_tags__(self) -> Tags:
    """Get estimator tags.

    Returns
    -------
    Tags
        Estimator tags with conformity scorer attributes.

    """
    tags = super().__sklearn_tags__()
    assert tags.scorer_tags is not None
    tags.scorer_tags.prediction_type = "conformity"  # ty: ignore[invalid-assignment]
    return tags

fit(y_train, *, forecaster=None, **params)

Fit the scorer on training data if needed.

Source Code

View on GitHub

Show/Hide source 
44
45
46
47
48
49
@_fit_context(prefer_skip_nested_validation=True)
def fit(self, y_train: pl.DataFrame, *, forecaster=None, **params) -> "BaseConformityScorer":
    """Fit the scorer on training data if needed."""
    # Conformity scorers typically don't aggregate results in the same way,
    # so they don't use aggregation_method, but they must implement fit.
    return super().fit(y_train, forecaster=forecaster, **params)

inverse_score(y_pred, conformity_scores, coverage_rate) abstractmethod

Transform conformity scores into prediction intervals.

Parameters
Name Type Description Default
y_pred DataFrame

Point predictions.

 required
conformity_scores DataFrame

Conformity scores from calibration.

 required
coverage_rate float

Target coverage probability.

 required
Returns
Type Description
DataFrame

Prediction intervals.
Source Code

View on GitHub

Show/Hide source 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@abc.abstractmethod
def inverse_score(
    self, y_pred: pl.DataFrame, conformity_scores: pl.DataFrame, coverage_rate: float
) -> pl.DataFrame:
    """Transform conformity scores into prediction intervals.

    Parameters
    ----------
    y_pred : pl.DataFrame
        Point predictions.

    conformity_scores : pl.DataFrame
        Conformity scores from calibration.

    coverage_rate : float
        Target coverage probability.

    Returns
    -------
    pl.DataFrame
        Prediction intervals.

    """