LookupWeighter¶

class LookupWeighter(BaseWeighter):
    r"""Explicit per-key weights from a mapping.

    Each key is looked up in ``mapping``; keys absent from ``mapping`` receive
    ``default``. Replaces the former ``{key: weight}`` dict input (the ``"*"``
    wildcard is replaced by the tunable ``default`` parameter, though a ``"*"``
    entry in ``mapping`` is still honored for compatibility).

    Parameters
    ----------
    mapping : dict or None, default=None
        Mapping from key value to weight. ``None`` is treated as an empty
        mapping, so every key receives ``default``.
    default : float, default=1.0
        Weight for keys absent from ``mapping``. If ``mapping`` contains a
        ``"*"`` key, that entry takes precedence over this parameter as the
        fallback weight.

    See Also
    --------
    - [`TableWeighter`][yohou.weighting.weighters.TableWeighter] : DataFrame-driven weights.
    - [`CompositeWeighter`][yohou.weighting.weighters.CompositeWeighter] : Combine weighters by product or mean.

    Examples
    --------
    >>> import polars as pl
    >>> steps = pl.Series("forecasting_step", [1, 2, 3])
    >>> LookupWeighter(mapping={1: 1.0}, default=0.0).compute_weights(
    ...     steps
    ... )  # doctest: +NORMALIZE_WHITESPACE
    shape: (3,)
    Series: 'weight' [f64]
    [
        1.0
        0.0
        0.0
    ]

    """

    _parameter_constraints: dict = {
        "mapping": [dict, None],
        "default": [Interval(numbers.Real, 0, None, closed="left")],
    }

    def __init__(self, mapping: dict | None = None, default: float = 1.0) -> None:
        self.mapping = mapping
        self.default = default

    def compute_weights(self, key: pl.Series, group_name: str | None = None) -> pl.Series:
        """Compute lookup weights for ``key``."""
        self._validate_params()
        mapping = self.mapping or {}
        effective_default = mapping.get("*", self.default)
        lookup = {k: v for k, v in mapping.items() if k != "*"}
        if lookup:
            weights = key.replace_strict(
                list(lookup.keys()),
                list(lookup.values()),
                default=effective_default,
                return_dtype=pl.Float64,
            )
        else:
            weights = pl.Series(np.full(len(key), effective_default, dtype=np.float64))
        return weights.alias("weight")