ExponentialDecayWeighter¶

class ExponentialDecayWeighter(BaseWeighter):
    r"""Exponential decay weights giving more weight to recent keys.

    Computes weights via exponential decay from the most recent key:

    $$w(k) = \exp\left(-\ln(2) \cdot \frac{d(k)}{\text{half\_life}}\right)$$

    where $d(k)$ is the distance from the most recent key. With
    ``scale="elapsed"`` the distance is the real elapsed time to the most recent
    datetime; with ``scale="position"`` it is the rank-index distance (usable for
    integer forecasting steps or regularly-spaced time).

    Parameters
    ----------
    half_life : int, float, or datetime.timedelta
        Distance over which weight decays to half. With ``scale="elapsed"`` an
        int/float is interpreted as a number of days and a ``timedelta`` as a
        duration. With ``scale="position"`` it is a number of steps (a
        ``timedelta`` is invalid and raises ``ValueError``).
    scale : {"elapsed", "position"} or None, default=None
        Decay basis. ``"elapsed"`` decays by real elapsed time (datetime keys);
        ``"position"`` decays by rank index. When None, inferred from the key
        dtype (any temporal dtype, i.e. Date, Datetime, Duration, or Time ->
        ``"elapsed"``; numeric -> ``"position"``).

    Notes
    -----
    The most recent (largest) key receives weight 1.0; smaller keys decay.

    References
    ----------
    [1] Hyndman, R.J., & Athanasopoulos, G. (2021). "Forecasting:
        principles and practice," 3rd edition, OTexts. Chapter 8.1.

    See Also
    --------
    - [`LinearDecayWeighter`][yohou.weighting.weighters.LinearDecayWeighter] : Linear recency weighting.
    - [`SeasonalEmphasisWeighter`][yohou.weighting.weighters.SeasonalEmphasisWeighter] : Seasonal emphasis weighting.
    - [`CompositeWeighter`][yohou.weighting.weighters.CompositeWeighter] : Combine weighters by product or mean.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> times = pl.Series(
    ...     "time",
    ...     [datetime(2024, 1, 1), datetime(2024, 1, 2), datetime(2024, 1, 3)],
    ... )
    >>> ExponentialDecayWeighter(half_life=1).compute_weights(times)  # doctest: +NORMALIZE_WHITESPACE
    shape: (3,)
    Series: 'weight' [f64]
    [
        0.25
        0.5
        1.0
    ]

    """

    _parameter_constraints: dict = {
        "half_life": [Interval(numbers.Real, 0, None, closed="neither"), timedelta],
        "scale": [StrOptions({"elapsed", "position"}), None],
    }

    def __init__(self, half_life: int | float | timedelta = 1, scale: str | None = None) -> None:
        self.half_life = half_life
        self.scale = scale

    def _resolve_scale(self, key: pl.Series) -> str:
        """Resolve the effective decay scale, inferring from dtype if unset."""
        if self.scale is not None:
            return self.scale
        return "elapsed" if key.dtype.is_temporal() else "position"

    def compute_weights(self, key: pl.Series, group_name: str | None = None) -> pl.Series:
        """Compute exponential decay weights for ``key``."""
        self._validate_params()
        scale = self._resolve_scale(key)

        if scale == "elapsed":
            distances = (key.max() - key).dt.total_seconds().to_numpy().astype(np.float64)
            if isinstance(self.half_life, timedelta):
                half_life_seconds = self.half_life.total_seconds()
            else:
                half_life_seconds = timedelta(days=float(self.half_life)).total_seconds()
            weights = np.exp(-np.log(2) * distances / half_life_seconds)
        else:  # position
            if isinstance(self.half_life, timedelta):
                raise ValueError("timedelta half_life requires scale='elapsed', got scale='position'")
            ranks = (key.rank(method="ordinal") - 1).to_numpy().astype(np.float64)
            distances = ranks.max() - ranks
            weights = np.exp(-np.log(2) * distances / float(self.half_life))

        return pl.Series(weights, dtype=pl.Float64).alias("weight")