SeasonalEmphasisWeighter¶

class SeasonalEmphasisWeighter(BaseWeighter):
    r"""Weights emphasizing keys in phase with the most recent seasonal position.

    Gives higher weight to keys whose ordinal rank shares the same value modulo
    the seasonal period as the most-recent key. This coincides with "same
    weekday" (for ``seasonality=7``) only when the series is contiguous and its
    length is an exact multiple of the period; in general it is a rank-phase
    match, not a calendar match. Rank-based, so usable for datetime or integer
    keys.

    Parameters
    ----------
    seasonality : int or list of int, default=1
        Seasonal period(s). If a list, keys matching ANY period receive emphasis.
        With ``seasonality=1`` every key is in phase (all receive ``emphasis``),
        making the weighter equivalent to a constant weighter.
    emphasis : float, default=2.0
        Weight multiplier for in-phase keys (out-of-phase keys receive 1.0).

    See Also
    --------
    - [`ExponentialDecayWeighter`][yohou.weighting.weighters.ExponentialDecayWeighter] : Exponential recency weighting.
    - [`LinearDecayWeighter`][yohou.weighting.weighters.LinearDecayWeighter] : Linear recency 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, 8), datetime(2024, 1, 9)],
    ... )
    >>> SeasonalEmphasisWeighter(seasonality=7).compute_weights(times)  # doctest: +NORMALIZE_WHITESPACE
    shape: (4,)
    Series: 'weight' [f64]
    [
        1.0
        1.0
        1.0
        2.0
    ]

    """

    _parameter_constraints: dict = {
        "seasonality": [Interval(numbers.Integral, 1, None, closed="left"), list],
        "emphasis": [Interval(numbers.Real, 0, None, closed="neither")],
    }

    def __init__(self, seasonality: int | list[int] = 1, emphasis: float = 2.0) -> None:
        self.seasonality = seasonality
        self.emphasis = emphasis

    def compute_weights(self, key: pl.Series, group_name: str | None = None) -> pl.Series:
        """Compute seasonal emphasis weights for ``key``."""
        self._validate_params()
        seasonalities = [self.seasonality] if isinstance(self.seasonality, int) else self.seasonality

        n = len(key)
        if n == 1:
            return pl.Series([1.0], dtype=pl.Float64).alias("weight")

        ranks = key.rank(method="ordinal") - 1
        in_phase = np.zeros(n, dtype=bool)
        for s in seasonalities:
            # The most-recent key has the highest ordinal rank (n - 1),
            # regardless of the positional order of ``key``. Using ranks[-1]
            # (the last positional element) is only correct when ``key`` is
            # already sorted ascending.
            most_recent_phase = (n - 1) % s
            phases = ranks.to_numpy() % s
            in_phase = in_phase | (phases == most_recent_phase)

        weights = np.where(in_phase, self.emphasis, 1.0)
        return pl.Series(weights, dtype=pl.Float64).alias("weight")