BaseSimilarity¶

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 _to_weights(
        distances: np.ndarray,
    ) -> np.ndarray[tuple[int, int], np.dtype[np.floating[Any]]]:
        r"""Convert a distance matrix to calibration weights.

        Applies a numerically-stable softmax of negative distances and
        reserves uniform mass for the (hypothetical) test point over the
        calibration axis:

        $$w_{ji} = \frac{\exp(-(d_{ji} - \min_k d_{jk}))}
        {1 + \sum_k \exp(-(d_{jk} - \min_k d_{jk}))}$$

        Each output row is non-negative and sums to a value strictly less
        than 1; the remainder ``1 / (1 + \sum_k raw)`` is the mass reserved
        for the new test point, following the non-exchangeable conformal
        construction (Barber et al., 2023).

        Parameters
        ----------
        distances : numpy.ndarray
            Distance matrix of shape ``(n_pred, n_calibration)``.

        Returns
        -------
        numpy.ndarray
            Weight matrix of shape ``(n_pred, n_calibration)``.

        """
        # Stable softmax of the exponent -d: subtract its row max (= -min(d)),
        # so every exponent is <= 0 and exp() cannot overflow.
        neg_d = -distances
        neg_d = neg_d - neg_d.max(axis=1, keepdims=True)
        return BaseSimilarity._reserve_mass(np.exp(neg_d))

    @staticmethod
    def _reserve_mass(
        raw_weights: np.ndarray,
    ) -> np.ndarray[tuple[int, int], np.dtype[np.floating[Any]]]:
        r"""Normalize non-negative weights, reserving uniform mass per row.

        Returns ``raw / (\sum_k raw + 1)`` so each row is non-negative and
        sums to a value strictly less than 1; the remainder is reserved for
        the test point. Shared by the distance softmax (:meth:`_to_weights`)
        and the
        [`CompositeSimilarity`][yohou.interval.similarity.CompositeSimilarity]
        ``"multiply"`` combination.

        Parameters
        ----------
        raw_weights : numpy.ndarray
            Non-negative weight matrix of shape ``(n_pred, n_calibration)``.

        Returns
        -------
        numpy.ndarray
            Row-normalized weight matrix of the same shape.

        """
        return raw_weights / (raw_weights.sum(axis=1, keepdims=True) + 1.0)

    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

    @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