Skip to content

BaseSimilarity

yohou.interval.base.BaseSimilarity

Bases: BaseEstimator

Base class for similarity measures used in interval forecasting.

Similarity measures assign weights to calibration residuals based on how similar past prediction contexts are to the current one.

Notes

Used by SplitConformalForecaster to produce adaptive (locally weighted) prediction intervals. When similarity=None, uniform weights are used.

See Also

Source Code

View on GitHub

Show/Hide source 
 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
189
190
191
192
193
194
195
196
197
class BaseSimilarity(BaseEstimator, metaclass=abc.ABCMeta):
    """Base class for similarity measures used in interval forecasting.

    Similarity measures assign weights to calibration residuals based
    on how similar past prediction contexts are to the current one.

    Notes
    -----
    Used by ``SplitConformalForecaster`` to produce adaptive (locally
    weighted) prediction intervals.  When ``similarity=None``, uniform
    weights are used.

    See Also
    --------
    - [`DistanceSimilarity`][yohou.interval.similarity.DistanceSimilarity] : Distance-based similarity measure.
    - [`SplitConformalForecaster`][yohou.interval.split_conformal.SplitConformalForecaster] : Conformal forecaster that uses similarities.

    """

    _parameter_constraints: dict = {}

    @staticmethod
    def _validate_no_nulls(df: pl.DataFrame, method_name: str) -> None:
        """Raise if any column contains null or NaN values.

        Parameters
        ----------
        df : pl.DataFrame
            DataFrame to validate.
        method_name : str
            Name of the calling method (for the error message).

        Raises
        ------
        ValueError
            If any column contains null or NaN values.

        """
        null_cols = [col for col in df.columns if df[col].is_null().any()]
        nan_cols = [col for col in df.select(cs.numeric()).columns if df[col].is_nan().any()]
        bad_cols = sorted(set(null_cols + nan_cols))
        if bad_cols:
            raise ValueError(f"{method_name}() received data with null or NaN values in columns: {bad_cols}")

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

        Returns
        -------
        Tags
            Estimator tags with similarity-specific attributes.

        """
        tags = Tags(estimator_type="similarity", requires_fit=True)

        # Most similarity measures are symmetric and require predictions
        assert tags.similarity_tags is not None
        tags.similarity_tags.symmetric = True
        tags.similarity_tags.requires_predictions = True
        tags.similarity_tags.produces_weights = True

        return tags

    @property
    def discarded_time_stamps(self) -> None:
        """Get discarded timestamps (placeholder property).

        Returns
        -------
        None

        """
        return None

    @abc.abstractmethod
    def fit(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "BaseSimilarity":
        """Fit the similarity measure.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series.

        y_pred : pl.DataFrame
            Point predictions.

        X_actual : pl.DataFrame or None, default=None
            Exogenous features.

        Returns
        -------
        self

        """

    @abc.abstractmethod
    def observe(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "BaseSimilarity":
        """Observe new data and update the similarity measure.

        Parameters
        ----------
        y : pl.DataFrame
            New target observations.

        y_pred : pl.DataFrame
            New predictions.

        X_actual : pl.DataFrame or None, default=None
            New exogenous features.

        Returns
        -------
        self

        """

    @abc.abstractmethod
    def predict(
        self,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> np.ndarray[tuple[int, int], np.dtype[np.floating[Any]]]:
        """Compute similarity weights for predictions.

        Parameters
        ----------
        y_pred : pl.DataFrame
            Predictions to compute similarities for.

        X_actual : pl.DataFrame or None, default=None
            Exogenous features.

        Returns
        -------
        np.ndarray
            Similarity weights.

        """

    def rewind(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "BaseSimilarity":
        """Rewind observed data from the similarity measure.

        Default implementation is a no-op. Concrete subclasses that
        track observed data should override this to remove the most
        recently observed rows.

        Parameters
        ----------
        y : pl.DataFrame
            Target observations to rewind.

        y_pred : pl.DataFrame
            Predictions to rewind.

        X_actual : pl.DataFrame or None, default=None
            Exogenous features to rewind.

        Returns
        -------
        self

        """
        return self

Methods

discarded_time_stamps property

Get discarded timestamps (placeholder property).

Returns
Type Description
None

__sklearn_tags__()

Get estimator tags.

Returns
Type Description
Tags

Estimator tags with similarity-specific attributes.
Source Code

View on GitHub

Show/Hide source 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __sklearn_tags__(self) -> Tags:
    """Get estimator tags.

    Returns
    -------
    Tags
        Estimator tags with similarity-specific attributes.

    """
    tags = Tags(estimator_type="similarity", requires_fit=True)

    # Most similarity measures are symmetric and require predictions
    assert tags.similarity_tags is not None
    tags.similarity_tags.symmetric = True
    tags.similarity_tags.requires_predictions = True
    tags.similarity_tags.produces_weights = True

    return tags

fit(y, y_pred, X_actual=None) abstractmethod

Fit the similarity measure.

Parameters
Name Type Description Default
y DataFrame

Target time series.

 required
y_pred DataFrame

Point predictions.

 required
X_actual DataFrame or None

Exogenous features.

 None
Returns
Type Description
self
Source Code

View on GitHub

Show/Hide source 
 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
@abc.abstractmethod
def fit(
    self,
    y: pl.DataFrame,
    y_pred: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
) -> "BaseSimilarity":
    """Fit the similarity measure.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series.

    y_pred : pl.DataFrame
        Point predictions.

    X_actual : pl.DataFrame or None, default=None
        Exogenous features.

    Returns
    -------
    self

    """

observe(y, y_pred, X_actual=None) abstractmethod

Observe new data and update the similarity measure.

Parameters
Name Type Description Default
y DataFrame

New target observations.

 required
y_pred DataFrame

New predictions.

 required
X_actual DataFrame or None

New exogenous features.

 None
Returns
Type Description
self
Source Code

View on GitHub

Show/Hide source 
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
@abc.abstractmethod
def observe(
    self,
    y: pl.DataFrame,
    y_pred: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
) -> "BaseSimilarity":
    """Observe new data and update the similarity measure.

    Parameters
    ----------
    y : pl.DataFrame
        New target observations.

    y_pred : pl.DataFrame
        New predictions.

    X_actual : pl.DataFrame or None, default=None
        New exogenous features.

    Returns
    -------
    self

    """

predict(y_pred, X_actual=None) abstractmethod

Compute similarity weights for predictions.

Parameters
Name Type Description Default
y_pred DataFrame

Predictions to compute similarities for.

 required
X_actual DataFrame or None

Exogenous features.

 None
Returns
Type Description
ndarray

Similarity weights.
Source Code

View on GitHub

Show/Hide source 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@abc.abstractmethod
def predict(
    self,
    y_pred: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
) -> np.ndarray[tuple[int, int], np.dtype[np.floating[Any]]]:
    """Compute similarity weights for predictions.

    Parameters
    ----------
    y_pred : pl.DataFrame
        Predictions to compute similarities for.

    X_actual : pl.DataFrame or None, default=None
        Exogenous features.

    Returns
    -------
    np.ndarray
        Similarity weights.

    """

rewind(y, y_pred, X_actual=None)

Rewind observed data from the similarity measure.

Default implementation is a no-op. Concrete subclasses that track observed data should override this to remove the most recently observed rows.

Parameters
Name Type Description Default
y DataFrame

Target observations to rewind.

 required
y_pred DataFrame

Predictions to rewind.

 required
X_actual DataFrame or None

Exogenous features to rewind.

 None
Returns
Type Description
self
Source Code

View on GitHub

Show/Hide source 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def rewind(
    self,
    y: pl.DataFrame,
    y_pred: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
) -> "BaseSimilarity":
    """Rewind observed data from the similarity measure.

    Default implementation is a no-op. Concrete subclasses that
    track observed data should override this to remove the most
    recently observed rows.

    Parameters
    ----------
    y : pl.DataFrame
        Target observations to rewind.

    y_pred : pl.DataFrame
        Predictions to rewind.

    X_actual : pl.DataFrame or None, default=None
        Exogenous features to rewind.

    Returns
    -------
    self

    """
    return self