CompositeWeighter¶

class CompositeWeighter(BaseWeighter, _BaseComposition):
    r"""Combine multiple named weighters into a single weight series.

    Multiplies or averages the ``compute_weights`` outputs of its components.
    Sub-weighters are named ``(name, weighter)`` tuples, so their parameters are
    addressable as ``<name>__<param>`` (sklearn ``_BaseComposition``) and tunable
    through a forecaster, for example ``time_weighter__decay__half_life``. Mirrors
    ``CompositeSimilarity`` and the other yohou compositors (``FeaturePipeline``,
    ``ColumnForecaster``, the voting ensembles, …).

    Parameters
    ----------
    weighters : list of (str, BaseWeighter) tuples
        Named component weighters to combine, e.g.
        ``[("decay", ExponentialDecayWeighter(7)), ("seasonal", SeasonalEmphasisWeighter(7))]``.
        Must be non-empty.
    combination : {"multiply", "mean"}, default="multiply"
        How to combine the component weight series.

        ``"multiply"``
            Element-wise product with optional exponents:
            ``w = prod(w_i ** alpha_i)``.
        ``"mean"``
            Weighted average: ``w = sum(alpha_i * w_i) / sum(alpha_i)``.
    weights : list of float or None, default=None
        Per-component exponents (under ``"multiply"``) or mixing coefficients
        (under ``"mean"``), aligned with ``weighters``. If None, every component
        contributes equally (1.0 each). Named ``weights`` to match
        ``CompositeSimilarity``; it configures the *composition*, not the
        produced weight series.

    See Also
    --------
    - [`ExponentialDecayWeighter`][yohou.weighting.weighters.ExponentialDecayWeighter] : Exponential recency weighting.
    - [`SeasonalEmphasisWeighter`][yohou.weighting.weighters.SeasonalEmphasisWeighter] : Seasonal emphasis weighting.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> times = pl.Series("time", [datetime(2024, 1, d) for d in range(1, 4)])
    >>> w = CompositeWeighter([
    ...     ("decay", ExponentialDecayWeighter(2)),
    ...     ("seasonal", SeasonalEmphasisWeighter(2, 1.5)),
    ... ])
    >>> _ = w.compute_weights(times)

    """

    _parameter_constraints: dict = {
        "weighters": [list, None],
        "combination": [StrOptions({"multiply", "mean"})],
        "weights": [list, None],
    }

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

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

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

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

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

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

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

        Returns
        -------
        self

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

    def _check_weighters(self) -> None:
        """Validate the composition parameters (not the sklearn ``_validate_params``)."""
        if not self.weighters:
            raise ValueError("CompositeWeighter requires at least one weighter")
        for item in self.weighters:
            if not (isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)):
                raise ValueError(f"Each entry in `weighters` must be a (name, weighter) tuple, got {item!r}")
        if self.weights is not None and len(self.weights) != len(self.weighters):
            raise ValueError(
                f"weights length ({len(self.weights)}) must match weighters length ({len(self.weighters)})"
            )

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

    def compute_weights(self, key: pl.Series, group_name: str | None = None) -> pl.Series:
        """Combine all component weights for ``key`` by product or mean."""
        self._validate_params()
        self._check_weighters()
        alphas = self._resolved_weights()
        series = [weighter.compute_weights(key, group_name) for _name, weighter in self.weighters]  # ty: ignore[not-iterable]

        if self.combination == "multiply":
            result = pl.Series(np.ones(len(key)), dtype=pl.Float64)
            for s, alpha in zip(series, alphas, strict=True):
                result = result * s.pow(alpha)
        else:  # mean
            total_alpha = sum(alphas)
            if total_alpha == 0:
                raise ValueError(
                    "CompositeWeighter with combination='mean' requires the mixing "
                    f"coefficients to sum to a nonzero value, got weights={alphas} "
                    "(sum is zero). A zero sum would silently produce all-zero weights."
                )
            result = pl.Series(np.zeros(len(key)), dtype=pl.Float64)
            for s, alpha in zip(series, alphas, strict=True):
                result = result + s * alpha
            result = result / total_alpha

        return result.alias("weight")