CompositeSimilarity¶

class CompositeSimilarity(BaseSimilarity, _BaseComposition):
    r"""Combine multiple named similarity measures into a single weight vector.

    Delegates ``fit``, ``observe``, ``rewind``, and ``predict`` to each
    sub-similarity and then combines their weight matrices using either
    element-wise multiplication or weighted averaging. Sub-similarities are
    named ``(name, similarity)`` tuples, so their parameters are tunable via
    the ``similarities__<name>__<param>`` syntax (sklearn ``_BaseComposition``).

    Parameters
    ----------
    similarities : list of (str, BaseSimilarity) tuples
        At least two named sub-similarities to combine, e.g.
        ``[("dist", DistanceSimilarity()), ("seasonal", SeasonalSimilarity([7.0]))]``.
    combination : {"multiply", "mean"}, default="multiply"
        How to combine the individual weight matrices.

        ``"multiply"``
            Element-wise product with optional exponents:
            ``w_combined = prod(w_i ** alpha_i)``, then re-normalised with the
            shared per-row mass reservation.
        ``"mean"``
            Weighted average: ``w_combined = sum(alpha_i * w_i) / sum(alpha_i)``.
            Unlike ``"multiply"``, the ``"mean"`` path does not re-apply the
            per-row mass reservation; each row of the result sums to the
            weighted average of the sub-similarity row sums, which may exceed
            the strict ``(0, 1)`` mass-reservation guarantee of the
            ``"multiply"`` path.

    weights : list of float or None, default=None
        Per-similarity exponents (multiply) or mixing coefficients
        (mean), aligned with ``similarities``. If ``None``, all
        similarities contribute equally (exponents/coefficients of 1.0).

    Attributes
    ----------
    similarities_ : list of (str, BaseSimilarity) tuples
        Fitted copies of the named sub-similarities (set after ``fit``).

    See Also
    --------
    - [`DistanceSimilarity`][yohou.interval.similarity.DistanceSimilarity] : Value-based distance similarity.
    - [`SeasonalSimilarity`][yohou.interval.similarity.SeasonalSimilarity] : Seasonal-phase Fourier feature similarity.

    Examples
    --------
    >>> from datetime import datetime, timedelta
    >>> import polars as pl
    >>> import numpy as np
    >>> from yohou.interval.similarity import (
    ...     CompositeSimilarity,
    ...     DistanceSimilarity,
    ...     SeasonalSimilarity,
    ... )
    >>>
    >>> dates = [datetime(2021, 1, 1) + timedelta(days=i) for i in range(28)]
    >>> y = pl.DataFrame({"time": dates, "value": np.random.randn(28)})
    >>> y_pred = pl.DataFrame({"time": dates, "value": np.random.randn(28)})
    >>>
    >>> comp = CompositeSimilarity(
    ...     similarities=[
    ...         ("dist", DistanceSimilarity(metric="euclidean")),
    ...         ("seasonal", SeasonalSimilarity(seasonality=[7.0])),
    ...     ],
    ...     combination="multiply",
    ... )
    >>> _ = comp.fit(y, y_pred)
    >>> new_date = [datetime(2021, 1, 29)]
    >>> y_pred_new = pl.DataFrame({"time": new_date, "value": [0.5]})
    >>> weights = comp.predict(y_pred_new)
    >>> weights.shape
    (1, 28)

    """

    _parameter_constraints: dict = {
        "similarities": [list],
        "combination": [str],
        "weights": [list, None],
    }

    def __init__(
        self,
        similarities: list[tuple[str, BaseSimilarity]] | None = None,
        combination: Literal["multiply", "mean"] = "multiply",
        weights: list[float] | None = None,
    ) -> None:
        self.similarities = similarities
        self.combination = combination
        self.weights = weights

    def get_params(self, deep: bool = True) -> dict[str, Any]:
        """Get parameters, including nested sub-similarity parameters.

        Parameters
        ----------
        deep : bool, default=True
            If True, include sub-similarity parameters as
            ``similarities__<name>__<param>``.

        Returns
        -------
        dict
            Parameter names mapped to values.

        """
        return self._get_params("similarities", deep=deep)

    def set_params(self, **params: Any) -> "CompositeSimilarity":
        """Set parameters, routing ``similarities__<name>__<param>`` to sub-similarities.

        Parameters
        ----------
        **params : dict
            Parameters to set.

        Returns
        -------
        self

        """
        self._set_params("similarities", **params)
        return self

    def _check_similarities(self) -> None:
        """Validate the composition parameters (not the sklearn ``_validate_params``)."""
        if self.similarities is None or len(self.similarities) < 2:
            raise ValueError(
                "CompositeSimilarity requires at least 2 sub-similarities, "
                f"got {0 if self.similarities is None else len(self.similarities)}"
            )
        for item in self.similarities:
            if not (isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)):
                raise ValueError(f"Each entry in `similarities` must be a (name, similarity) tuple, got {item!r}")
        if self.combination not in ("multiply", "mean"):
            raise ValueError(f"combination must be 'multiply' or 'mean', got {self.combination!r}")
        if self.weights is not None and len(self.weights) != len(self.similarities):
            raise ValueError(
                f"weights length ({len(self.weights)}) must match similarities length ({len(self.similarities)})"
            )

    def _resolved_weights(self) -> list[float]:
        """Return per-similarity weights, defaulting to 1.0 each."""
        if self.weights is not None:
            return self.weights
        return [1.0] * len(self.similarities)  # ty: ignore[invalid-argument-type]

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "CompositeSimilarity":
        """Fit all sub-similarities on the calibration data.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series.
        y_pred : pl.DataFrame
            Point forecast time series.
        X_actual : pl.DataFrame or None, default=None
            Exogenous features.

        Returns
        -------
        self

        """
        self._check_similarities()
        self.similarities_ = [
            (name, clone(sim).fit(y=y, y_pred=y_pred, X_actual=X_actual))
            for name, sim in self.similarities  # ty: ignore[not-iterable]
        ]
        return self

    def observe(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "CompositeSimilarity":
        """Forward observation to all sub-similarities.

        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

        """
        check_is_fitted(self, "similarities_")
        for _name, sim in self.similarities_:
            sim.observe(y=y, y_pred=y_pred, X_actual=X_actual)
        return self

    def rewind(
        self,
        y: pl.DataFrame,
        y_pred: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
    ) -> "CompositeSimilarity":
        """Forward rewind to all sub-similarities.

        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

        """
        check_is_fitted(self, "similarities_")
        for _name, sim in self.similarities_:
            sim.rewind(y=y, y_pred=y_pred, X_actual=X_actual)
        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]]]:
        """Combine sub-similarity weights into a single weight matrix.

        Parameters
        ----------
        y_pred : pl.DataFrame
            Predictions to compute similarities for.
        X_actual : pl.DataFrame or None, default=None
            Exogenous features.

        Returns
        -------
        np.ndarray
            Combined weight matrix of shape
            ``(n_predictions, n_calibration)``.

        """
        check_is_fitted(self, "similarities_")
        alphas = self._resolved_weights()
        weight_matrices = [sim.predict(y_pred=y_pred, X_actual=X_actual) for _name, sim in self.similarities_]

        if self.combination == "multiply":
            combined = np.ones_like(weight_matrices[0])
            for w, alpha in zip(weight_matrices, alphas, strict=True):
                combined *= np.power(w, alpha)
            # Shared per-row mass reservation (drops the former arbitrary feature-count factor).
            combined = self._reserve_mass(combined)
        else:  # mean
            combined = np.zeros_like(weight_matrices[0])
            for w, alpha in zip(weight_matrices, alphas, strict=True):
                combined += alpha * w
            total_alpha = sum(alphas)
            if total_alpha != 0:
                combined /= total_alpha

        return combined