Skip to content

exponential_decay_weight

yohou.utils.weighting.exponential_decay_weight(half_life)

Generate exponential decay weights giving more weight to recent times.

Creates a callable that computes weights using exponential decay:

\[w(t) = \exp\left(-\ln(2) \cdot \frac{d(t)}{\text{half\_life}}\right)\]

where \(d(t)\) is the distance from the most recent time point.

Parameters

Name Type Description Default
half_life int, float, or timedelta

Time period over which weight decays to half. If int/float, interpreted as number of time steps. If timedelta, interpreted as time duration.

 required

Returns

Type Description
Callable[[Series], Series]

Function accepting time series (datetime) and returning weight series (float64). Most recent time has weight 1.0, older times decay exponentially.

References

[1] Hyndman, R.J., & Athanasopoulos, G. (2021). "Forecasting: principles and practice," 3rd edition, OTexts: Melbourne, Australia. OTexts.com/fpp3. Chapter 8.1.

See Also

Examples

>>> import polars as pl
>>> from datetime import datetime, timedelta
>>> times = pl.Series(
...     "time",
...     [
...         datetime(2024, 1, 1),
...         datetime(2024, 1, 2),
...         datetime(2024, 1, 3),
...     ],
... )
>>> weight_fn = exponential_decay_weight(half_life=1)
>>> weights = weight_fn(times)
>>> weights
shape: (3,)
Series: 'weight' [f64]
[
    0.25
    0.5
    1.0
]

Source Code

View on GitHub

Show/Hide source 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def exponential_decay_weight(
    half_life: int | float | timedelta,
) -> Callable[[pl.Series], pl.Series]:
    r"""Generate exponential decay weights giving more weight to recent times.

    Creates a callable that computes weights using exponential decay:

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

    where $d(t)$ is the distance from the most recent time point.

    Parameters
    ----------
    half_life : int, float, or timedelta
        Time period over which weight decays to half. If int/float, interpreted
        as number of time steps. If timedelta, interpreted as time duration.

    Returns
    -------
    Callable[[pl.Series], pl.Series]
        Function accepting time series (datetime) and returning weight series
        (float64). Most recent time has weight 1.0, older times decay exponentially.

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

    See Also
    --------
    - [`linear_decay_weight`][yohou.utils.weighting.linear_decay_weight] : Linear 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, timedelta
    >>> times = pl.Series(
    ...     "time",
    ...     [
    ...         datetime(2024, 1, 1),
    ...         datetime(2024, 1, 2),
    ...         datetime(2024, 1, 3),
    ...     ],
    ... )
    >>> weight_fn = exponential_decay_weight(half_life=1)
    >>> weights = weight_fn(times)
    >>> weights  # doctest: +NORMALIZE_WHITESPACE
    shape: (3,)
    Series: 'weight' [f64]
    [
        0.25
        0.5
        1.0
    ]

    """
    # Convert to timedelta if numeric (assume days for compatibility)
    half_life_td = half_life
    if isinstance(half_life, numbers.Number) and not isinstance(half_life, timedelta):
        # Type narrowing: exclude timedelta from union before converting
        half_life_td = timedelta(days=float(half_life))

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

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

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

        """
        # Compute distance from most recent time
        max_time = time.max()
        distances = (max_time - time).dt.total_seconds()

        # Convert half_life to seconds
        assert isinstance(half_life_td, timedelta)
        half_life_seconds = half_life_td.total_seconds()

        # Exponential decay: exp(-ln(2) * distance / half_life)
        weights = np.exp(-np.log(2) * distances / half_life_seconds)
        return pl.Series(weights, dtype=pl.Float64).alias("weight")

    return _weight_fn

Tutorials

The following example notebooks use this component:

  • How to Handle Long Series

    Data-Features

    Limit history with observation_horizon, weight recent errors with exponential decay, and downsample high-frequency data.

    View · Open in marimo

  • 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