"""Exponentially weighted momentum descriptor."""
# Copyright (c) 2023-2026
# Author: Hugo Delatte <hugo.delatte@skfoliolabs.com>
# SPDX-License-Identifier: BSD-3-Clause
from __future__ import annotations
import numpy as np
from skfolio.containers import AssetPanel
from skfolio.descriptor._base import BaseDescriptor
from skfolio.typing import FloatArray
from skfolio.utils.tools import (
_validate_non_negative_integer,
_validate_positive_integer,
_validate_positive_real,
half_life_to_decay_factor,
)
from skfolio.utils.validation import validate_asset_panel
_FITTED_ATTR = "momentum_"
[docs]
class EWMomentum(BaseDescriptor):
r"""Exponentially weighted momentum descriptor.
Computes an EWMA of log returns with an optional skip period to exclude the most
recent observations:
The skip period separates medium-term momentum from short-term reversal. The classic
"12-1" momentum signal uses a skip of approximately one month [1]_.
.. math::
:nowrap:
\[
\begin{aligned}
x(t)
&= \log(1 + r(t)) \\[0.75em]
S(t)
&= \lambda \cdot S(t-1)
+ (1 - \lambda) \cdot x(t - \text{skip}) \\[0.75em]
\text{momentum}(t)
&=
\begin{cases}
\exp(S(t)) - 1 & \text{if } \texttt{exponentiate=True} \\
S(t) & \text{otherwise}
\end{cases}
\end{aligned}
\]
where :math:`\lambda = \exp(-\ln(2) / \text{half\_life})` is the EWMA decay factor.
At observation :math:`t`, the EWMA input is the log return from :math:`t - \text{skip}`.
Therefore, :math:`\text{half\_life}` is measured on the delayed input series. An
EWMA with this half-life is comparable to a fixed window of about
:math:`2 \cdot \text{half\_life} / \ln 2` delayed observations, with the most recent
:math:`\text{skip}` observations excluded.
Parameters
----------
half_life : float, default=87.0
Controls how fast old returns decay in the EWMA of
:math:`\log(1 + r(t - \text{skip}))`. The default of 87 approximately matches a
:class:`RollingMomentum` window of 252 observations (~1 year of daily data). To
match a different window :math:`W`, use
:math:`\text{half\_life} \approx W \cdot \ln(2) / 2 \approx 0.35 \, W`. Adjust
for other frequencies (e.g., `half_life=6` for monthly data).
skip : int, default=21
Number of most recent observations to exclude before the EWMA window starts.
Used to separate medium-term momentum from short-term reversal. The default
assumes daily data and skips approximately one month. Set to `0` for short-term
momentum.
min_periods : int, optional
Minimum number of valid delayed returns required for each asset. Until an asset
reaches this count, its output is NaN. This warm-up period avoids exposing early
EWMA values before the estimate has sufficiently converged from its zero
initialization. If `None`, defaults to :math:`\lceil\text{half\_life}\rceil`,
with a minimum of 1.
exponentiate : bool, default=False
If True, output is :math:`\exp(S(t)) - 1` (return units). If False, output is
:math:`S(t)` (EWMA of log returns; log space). Cross-sectional ranking is
unchanged.
Attributes
----------
n_assets_ : int
Number of assets seen during fitting.
asset_names_ : ndarray of shape (n_assets,)
Asset names seen during fitting.
momentum_ : ndarray of shape (n_assets,)
Last exponentially weighted momentum value for each asset.
Notes
-----
The EWMA is initialized to zero (no bias correction).
NaNs are allowed as missing observations. Non-missing `returns` values must be
finite and greater than `-1`, so :math:`\log(1 + r)` is finite. The EWMA state is
updated only for finite delayed log returns, and each asset's valid-observation
count controls when its output starts. The `active_mask` property of the input
:class:`~skfolio.containers.AssetPanel` distinguishes holidays from delistings.
See Also
--------
RollingMomentum : Fixed-window (equal-weighted) momentum.
References
----------
.. [1] "Returns to buying winners and selling losers: Implications for stock market
efficiency" The Journal of Finance. Jegadeesh, N., & Titman, S. (1993).
Examples
--------
>>> from skfolio.datasets import make_synthetic_characteristics
>>> from skfolio.descriptor import EWMomentum
>>>
>>> X = make_synthetic_characteristics()
>>>
>>> # 12-1 momentum with daily data (default)
>>> descriptor = EWMomentum()
>>> momentum = descriptor.fit_transform(X)
>>>
>>> # Short-term momentum (no skip)
>>> descriptor = EWMomentum(half_life=10, skip=0)
>>> short_term_momentum = descriptor.fit_transform(X)
>>>
>>> # Log-space output
>>> descriptor = EWMomentum(exponentiate=False)
>>> momentum_log = descriptor.fit_transform(X)
"""
momentum_: FloatArray
def __init__(
self,
half_life: float = 87.0,
skip: int = 21,
min_periods: int | None = None,
exponentiate: bool = False,
):
self.half_life = half_life
self.skip = skip
self.min_periods = min_periods
self.exponentiate = exponentiate
def _reset(self):
if hasattr(self, _FITTED_ATTR):
delattr(self, _FITTED_ATTR)
def _validate_params(self) -> None:
"""Validate constructor parameters."""
_validate_positive_real(self.half_life, "half_life")
_validate_non_negative_integer(self.skip, "skip")
if self.min_periods is not None:
_validate_positive_integer(self.min_periods, "min_periods")
def _initialize(self) -> None:
"""Initialize EWMA state and delay buffer."""
n_assets = self.n_assets_
self.decay_ = half_life_to_decay_factor(self.half_life)
# Minimum valid delayed returns before output
if self.min_periods is None:
self.min_periods_ = max(1, int(np.ceil(self.half_life)))
else:
self.min_periods_ = int(self.min_periods)
# EWMA accumulator (one per asset, initialized to zero)
self._ewma = np.zeros(n_assets, dtype=float)
self._n_valid = np.zeros(n_assets, dtype=int)
# Delay ring buffer for skip (filled with NaN for warm-up)
if self.skip > 0:
self._skip_buffer = np.full((self.skip, n_assets), np.nan, dtype=float)
self._skip_write_idx = 0
# Counter of total observations processed
self._n_seen = 0
