linear_decay_weight

yohou.utils.weighting.linear_decay_weight(max_steps=None)

Generate linear decay weights giving more weight to recent times.

Creates a callable that computes weights using linear decay:

\[w(t) = \frac{\text{rank}(t)}{n - 1}\]

where \(\text{rank}(t) = 0\) for the oldest observation and \(n - 1\) for the most recent. When max_steps is set, observations older than max_steps receive weight 0.

Parameters

Name	Type	Description	Default
max_steps	int or None	Maximum number of steps to decay over. If None, decays linearly across entire time range. If specified, times older than max_steps from the most recent time receive weight 0.	None

Returns

Type	Description
Callable[[Series], Series]	Function accepting time series (datetime) and returning weight series (float64). Most recent time has weight 1.0, weights decrease linearly.

See Also

• exponential_decay_weight : Exponential decay weights for recent times.
• seasonal_emphasis_weight : Weights emphasizing seasonal positions.
• compose_weights : Compose multiple weight functions by multiplication.
• validate_callable_signature : Validate callable signature for time weighting.
• BaseReductionForecaster : Reduction forecaster supporting time_weight.

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),
...         datetime(2024, 1, 4),
...     ],
... )
>>> weight_fn = linear_decay_weight(max_steps=None)
>>> weights = weight_fn(times)
>>> weights
shape: (4,)
Series: 'weight' [f64]
[
    0.0
    0.333333
    0.666667
    1.0
]

Source Code

View on GitHub

Show/Hide source

def linear_decay_weight(
    max_steps: int | None = None,
) -> Callable[[pl.Series], pl.Series]:
    r"""Generate linear decay weights giving more weight to recent times.

    Creates a callable that computes weights using linear decay:

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

    where $\text{rank}(t) = 0$ for the oldest observation and $n - 1$
    for the most recent. When ``max_steps`` is set, observations older
    than ``max_steps`` receive weight 0.

    Parameters
    ----------
    max_steps : int or None, default=None
        Maximum number of steps to decay over. If None, decays linearly across
        entire time range. If specified, times older than max_steps from the
        most recent time receive weight 0.

    Returns
    -------
    Callable[[pl.Series], pl.Series]
        Function accepting time series (datetime) and returning weight series
        (float64). Most recent time has weight 1.0, weights decrease linearly.

    See Also
    --------
    - [`exponential_decay_weight`][yohou.utils.weighting.exponential_decay_weight] : Exponential decay weights for recent times.
    - [`seasonal_emphasis_weight`][yohou.utils.weighting.seasonal_emphasis_weight] : Weights emphasizing seasonal positions.
    - [`compose_weights`][yohou.utils.weighting.compose_weights] : Compose multiple weight functions by multiplication.
    - [`validate_callable_signature`][yohou.utils.weighting.validate_callable_signature] : Validate callable signature for time weighting.
    - [`BaseReductionForecaster`][yohou.base.reduction.BaseReductionForecaster] : Reduction forecaster supporting time_weight.

    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),
    ...         datetime(2024, 1, 4),
    ...     ],
    ... )
    >>> weight_fn = linear_decay_weight(max_steps=None)
    >>> weights = weight_fn(times)
    >>> weights  # doctest: +NORMALIZE_WHITESPACE
    shape: (4,)
    Series: 'weight' [f64]
    [
        0.0
        0.333333
        0.666667
        1.0
    ]

    """

    def _weight_fn(time: pl.Series) -> pl.Series:
        """Apply linear decay weighting to time series.

        Parameters
        ----------
        time : pl.Series
            Time series (datetime type).

        Returns
        -------
        pl.Series
            Weight series with linear decay from oldest to most recent.

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

        # Compute rank (0 = oldest, n-1 = newest) and convert to numpy
        ranks = (time.rank(method="ordinal") - 1).to_numpy()

        if max_steps is None:
            # Linear decay over entire range
            weights = ranks / (n - 1)
        else:
            # Linear decay over max_steps window
            # Older times (before the window) get weight 0
            # Recent max_steps times get linear increase from 0 to 1
            cutoff = n - max_steps  # Times with rank < cutoff get weight 0
            weights = np.where(ranks < cutoff, 0.0, (ranks - cutoff) / (max_steps - 1) if max_steps > 1 else 1.0)
            weights = np.minimum(1.0, weights)

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

    return _weight_fn

Tutorials

The following example notebooks use this component:

• How to Apply Time-Weighted Training

  ---

  Forecasting-Models

  Use time_weight and sample_weight_alignment to emphasise recent or seasonal training samples in PointReductionForecaster, with visualisation of weight curves and alignment strategy comparison.

  View · Open in marimo