DistanceSimilarity¶

class DistanceSimilarity(BaseSimilarity):
    r"""Distance-based similarity using scipy metrics for weighting observations.

    Computes observation weights by measuring the distance between new
    predictions and historical predictions in feature space. Closer
    historical observations receive higher weights, which are then used
    by interval forecasters to weight conformity scores when constructing
    prediction intervals.

    The weight for the *i*-th historical observation given prediction
    *j* is computed with a numerically-stable softmax of negative
    distances that reserves uniform mass for the (hypothetical) test
    point over the calibration axis:

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

    where $d_{ji} = d(x_j, x_i)$ for the chosen distance metric. The
    ``+1`` in the denominator reserves mass for the new test point
    (Barber et al., 2023), so each row sums to strictly less than 1.

    Parameters
    ----------
    metric : str, default="euclidean"
        Distance metric to use (e.g., ``"euclidean"``, ``"cityblock"``,
        ``"cosine"``). Any metric supported by
        ``scipy.spatial.distance.cdist`` is accepted.

    metric_params : dict or None, default=None
        Additional keyword arguments forwarded to the distance metric
        function.

    Notes
    -----
    The distance-to-weight conversion uses the softmax of negative
    distances, so distant observations contribute exponentially less
    than nearby ones. The weights are further normalised so that each
    prediction row sums to a value in (0, 1).

    References
    ----------
    [1] Lei, J., G'Sell, M., Rinaldo, A., Tibshirani, R.J., &
        Wasserman, L. (2018). "Distribution-free predictive inference for
        regression." Journal of the American Statistical Association,
        113(523), 1094-1111.
        https://doi.org/10.1080/01621459.2017.1307116
    [2] Barber, R.F., Candes, E.J., Ramdas, A., & Tibshirani, R.J.
        (2023). "Conformal prediction beyond exchangeability." Annals of
        Statistics, 51(2), 816-845.
        https://doi.org/10.1214/23-AOS2276

    See Also
    --------
    - [`BaseSimilarity`][yohou.interval.base.BaseSimilarity] : Abstract similarity base class.
    - [`BaseIntervalForecaster`][yohou.interval.base.BaseIntervalForecaster] :
        Interval forecaster that can consume similarity weights.

    Examples
    --------
    >>> from datetime import datetime
    >>> import polars as pl
    >>> import numpy as np
    >>> from yohou.interval.similarity import DistanceSimilarity
    >>>
    >>> # Create training data
    >>> time_train = pl.datetime_range(
    ...     start=datetime(2021, 12, 16), end=datetime(2021, 12, 16, 0, 0, 7), interval="1s", eager=True
    ... )
    >>> y_train = pl.DataFrame({"time": time_train, "value": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0]})
    >>> y_pred_train = pl.DataFrame({"time": time_train, "value": [1.1, 2.1, 2.9, 4.2, 4.8, 6.1, 7.0, 8.1]})
    >>>
    >>> # Fit similarity model
    >>> similarity = DistanceSimilarity(metric="euclidean")
    >>> _ = similarity.fit(y_train, y_pred_train)
    >>>
    >>> # Create new predictions to compute similarities for
    >>> time_test = pl.datetime_range(
    ...     start=datetime(2021, 12, 16, 0, 0, 8),
    ...     end=datetime(2021, 12, 16, 0, 0, 9),
    ...     interval="1s",
    ...     eager=True,
    ... )
    >>> y_pred_test = pl.DataFrame({"time": time_test, "value": [8.5, 9.2]})
    >>>
    >>> # Compute similarity weights
    >>> weights = similarity.predict(y_pred_test)
    >>> weights.shape
    (2, 8)
    >>> isinstance(weights, np.ndarray)
    True

    """

    _parameter_constraints: dict = {
        "metric": [str],
        "metric_params": [dict, None],
    }

    def __init__(
        self,
        metric: str = "euclidean",
        metric_params: dict[str, object] | None = None,
    ) -> None:
        self.metric = metric
        self.metric_params = metric_params

    def _get_X(
        self,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None,
    ) -> pl.DataFrame:
        """Combine predictions and features into single feature matrix.

        Drops the ``"time"`` column from ``X`` before concatenation to
        avoid duplicate columns.  Validates that no column (except
        ``"time"``) contains null or NaN values.

        Parameters
        ----------
        y_pred : pl.DataFrame
            Predictions.

        X_actual : pl.DataFrame or None
            Exogenous features.

        Returns
        -------
        pl.DataFrame
            Combined feature matrix.

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

        """
        if X_actual is not None:
            X_no_time = X_actual.drop("time", strict=False)
            result = pl.concat([y_pred, X_no_time], how="horizontal")
        else:
            result = y_pred

        value_cols = [col for col in result.columns if col != "time"]
        if value_cols:
            bad_mask = result.select(
                (pl.col(col).is_null() | pl.col(col).cast(pl.Float64, strict=False).is_nan()).any().alias(col)
                for col in value_cols
            )
            bad_cols = [col for col in value_cols if bad_mask[col][0]]
            if bad_cols:
                raise ValueError(
                    f"Columns {bad_cols} contain null or NaN values. DistanceSimilarity requires complete data."
                )
        return result

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "DistanceSimilarity":
        """Store the calibration feature matrix for distance computation.

        Combines ``y_pred`` and ``X_actual`` (if provided) via ``_get_X``
        and saves the result as ``_X_observed``. Subsequent ``predict``
        calls compute distances from new predictions to this stored matrix.

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

        y_pred : pl.DataFrame
            Point forecasts time series.

        X_actual : pl.DataFrame or None, default=None
            Exogenous feature time series.

        Returns
        -------
        self

        """
        X_features = self._get_X(y_pred, X_actual)
        self._X_observed = X_features

        return self

    def observe(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "DistanceSimilarity":
        """Observe new data and update similarity model.

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

        y_pred : pl.DataFrame
            New predictions.

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

        Returns
        -------
        self

        """
        check_is_fitted(self, "_X_observed")
        X_features = self._get_X(y_pred, X_actual)

        self._X_observed = pl.concat([self._X_observed, X_features])

        return self

    def rewind(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "DistanceSimilarity":
        """Rewind the most recently observed data.

        Removes the last ``len(y)`` rows from the internal reference
        matrix, reversing the effect of the corresponding ``observe()``
        call.

        Parameters
        ----------
        y : pl.DataFrame
            Target observations to rewind (used only for row count).

        y_pred : pl.DataFrame
            Predictions to rewind (used only for row count).

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

        Returns
        -------
        self

        """
        check_is_fitted(self, "_X_observed")
        n_rewind = len(y)
        self._X_observed = self._X_observed[: len(self._X_observed) - n_rewind]
        return self

    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 new predictions.

        Parameters
        ----------
        y_pred : pl.DataFrame
            New predictions to compute similarities for.

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

        Returns
        -------
        np.ndarray
            Similarity weight matrix.

        """
        check_is_fitted(self, "_X_observed")
        X_features = self._get_X(y_pred, X_actual)

        XA = X_features.select(pl.exclude("time")).to_numpy()
        XB = self._X_observed.select(pl.exclude("time")).to_numpy()
        distances: np.ndarray = cdist(XA, XB, metric=self.metric, **(self.metric_params or {}))  # ty: ignore[no-matching-overload]
        return self._to_weights(distances)