LinearDecayWeighter¶

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

    Computes rank-based linear decay. When ``max_steps`` is None:

    $$w(k) = \frac{\text{rank}(k)}{n - 1}$$

    where $\text{rank}(k) = 0$ for the oldest key and $n - 1$ for the most
    recent. When ``max_steps`` is set, the denominator becomes
    ``max_steps - 1`` and the window is shifted:

    $$w(k) = \max\!\left(0,\ \frac{\text{rank}(k) - (n - \text{max\_steps})}{\text{max\_steps} - 1}\right)$$

    so only the ``max_steps - 1`` most-recent keys receive a strictly positive
    weight; the boundary key at rank ``n - max_steps`` maps to exactly 0.0, as
    do all older keys. Rank-based, so usable for datetime or integer keys
    without a scale parameter.

    Parameters
    ----------
    max_steps : int or None, default=None
        Window over which to decay. If None, decays linearly across the whole
        range. If set, keys at rank ``< n - max_steps`` receive weight 0, and so
        does the boundary key at rank ``n - max_steps``; only the
        ``max_steps - 1`` most-recent keys receive strictly positive weights.

    See Also
    --------
    - [`ExponentialDecayWeighter`][yohou.weighting.weighters.ExponentialDecayWeighter] : Exponential 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, d) for d in range(1, 5)])
    >>> LinearDecayWeighter().compute_weights(times)  # doctest: +NORMALIZE_WHITESPACE
    shape: (4,)
    Series: 'weight' [f64]
    [
        0.0
        0.333333
        0.666667
        1.0
    ]

    """

    _parameter_constraints: dict = {
        "max_steps": [Interval(numbers.Integral, 1, None, closed="left"), None],
    }

    def __init__(self, max_steps: int | None = None) -> None:
        self.max_steps = max_steps

    def compute_weights(self, key: pl.Series, group_name: str | None = None) -> pl.Series:
        """Compute linear decay weights for ``key``."""
        self._validate_params()
        n = len(key)
        if n == 1:
            return pl.Series([1.0], dtype=pl.Float64).alias("weight")

        # Cast to float: polars rank() yields UInt32, and an unsigned array
        # underflows (OverflowError) on ``ranks - cutoff`` when cutoff is
        # negative (max_steps > len(key)).
        ranks = (key.rank(method="ordinal") - 1).to_numpy().astype(np.float64)

        if self.max_steps is None:
            weights = ranks / (n - 1)
        else:
            cutoff = n - self.max_steps
            # The most-recent key (rank n - 1) maps to exactly 1.0, so no
            # additional upper clamp is required.
            weights = np.where(
                ranks < cutoff,
                0.0,
                (ranks - cutoff) / (self.max_steps - 1) if self.max_steps > 1 else 1.0,
            )

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