ExponentialMovingAverage

yohou.preprocessing.window.ExponentialMovingAverage

Bases: BaseTransformer

Exponentially Weighted Moving Average (EWMA) transformer.

Computes the exponentially weighted moving average for time series data. The EWMA gives more weight to recent observations with exponentially decreasing weights for older observations.

Parameters

Name	Type	Description	Default
alpha	float	Smoothing factor (0 < alpha <= 1). Higher values give more weight to recent observations.	required
adjust	bool	If True, uses adjusted weights (divide by decaying adjustment factor). If False, uses standard exponential decay.	True
ignore_nulls	bool	If True, ignore null values when computing EWMA. If False, propagate null values.	True

Attributes

Name	Type	Description
n_features_in_	int	Number of features seen during fit.
feature_names_in_	list of str	Names of features seen during fit.

Notes

The EWMA is commonly used for: - Smoothing noisy time series - Technical indicators (e.g., EMA in finance) - Adaptive feature engineering

Examples

>>> import polars as pl
>>> from datetime import datetime
>>> from yohou.preprocessing import ExponentialMovingAverage

>>> times = pl.datetime_range(
...     start=datetime(2020, 1, 1), end=datetime(2020, 1, 10), interval="1d", eager=True
... )
>>> X = pl.DataFrame({"time": times, "value": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]})

>>> transformer = ExponentialMovingAverage(alpha=0.5)
>>> transformer.fit(X)
ExponentialMovingAverage(...)
>>> X_t = transformer.transform(X)
>>> len(X_t) == len(X)
True

See Also

RollingStatisticsTransformer : Fixed-window rolling statistics. SlidingWindowFunctionTransformer : Custom window functions.

Source Code

View on GitHub

Show/Hide source

class ExponentialMovingAverage(BaseTransformer):
    """Exponentially Weighted Moving Average (EWMA) transformer.

    Computes the exponentially weighted moving average for time series data.
    The EWMA gives more weight to recent observations with exponentially
    decreasing weights for older observations.

    Parameters
    ----------
    alpha : float
        Smoothing factor (0 < alpha <= 1). Higher values give more weight
        to recent observations.
    adjust : bool, default=True
        If True, uses adjusted weights (divide by decaying adjustment factor).
        If False, uses standard exponential decay.
    ignore_nulls : bool, default=True
        If True, ignore null values when computing EWMA.
        If False, propagate null values.

    Attributes
    ----------
    n_features_in_ : int
        Number of features seen during fit.
    feature_names_in_ : list of str
        Names of features seen during fit.

    Notes
    -----
    The EWMA is commonly used for:
    - Smoothing noisy time series
    - Technical indicators (e.g., EMA in finance)
    - Adaptive feature engineering

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from yohou.preprocessing import ExponentialMovingAverage

    >>> times = pl.datetime_range(
    ...     start=datetime(2020, 1, 1), end=datetime(2020, 1, 10), interval="1d", eager=True
    ... )
    >>> X = pl.DataFrame({"time": times, "value": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]})

    >>> transformer = ExponentialMovingAverage(alpha=0.5)
    >>> transformer.fit(X)  # doctest: +ELLIPSIS
    ExponentialMovingAverage(...)
    >>> X_t = transformer.transform(X)
    >>> len(X_t) == len(X)
    True

    See Also
    --------
    `RollingStatisticsTransformer` : Fixed-window rolling statistics.
    `SlidingWindowFunctionTransformer` : Custom window functions.

    """

    _parameter_constraints: dict = {
        "alpha": [Interval(numbers.Real, 0.0, 1.0, closed="right")],
        "adjust": ["boolean"],
        "ignore_nulls": ["boolean"],
    }

    def __init__(
        self,
        alpha: float,
        adjust: bool = True,
        ignore_nulls: bool = True,
    ):
        self.alpha = alpha
        self.adjust = adjust
        self.ignore_nulls = ignore_nulls

    def _transform(self, X: pl.DataFrame) -> pl.DataFrame:
        """Transform X by computing EWMA.

        Parameters
        ----------
        X : pl.DataFrame
            Validated input time series.

        Returns
        -------
        pl.DataFrame
            Transformed time series with a ``"time"`` column and transformed
            value columns.

        """
        # Get data columns
        data_cols = [c for c in X.columns if c != "time"]

        # Build EWMA expressions
        exprs = [pl.col("time")]
        for col_name in data_cols:
            ewma_expr = pl.col(col_name).ewm_mean(
                alpha=self.alpha,
                adjust=self.adjust,
                ignore_nulls=self.ignore_nulls,
            )
            exprs.append(ewma_expr.alias(f"{col_name}_ewma"))

        return X.select(exprs)

    def get_feature_names_out(self, input_features: list[str] | None = None) -> list[str]:
        """Get output feature names for transformation.

        Parameters
        ----------
        input_features : list of str or None, default=None
            Column names of the input features.  If ``None``, uses the
            feature names seen during ``fit``.

        Returns
        -------
        list of str
            Output feature names after transformation.

        """
        input_features = _check_feature_names_in(self, input_features)
        feature_names = [f"{col}_ewma" for col in input_features]
        arr: list[str] = np.asarray(feature_names, dtype=object).tolist()
        return arr

Methods

get_feature_names_out(input_features=None)

Get output feature names for transformation.

Parameters

Name	Type	Description	Default
input_features	list of str or None	Column names of the input features. If None, uses the feature names seen during fit.	None

Returns

Type	Description
list of str	Output feature names after transformation.

Source Code

View on GitHub

Show/Hide source

def get_feature_names_out(self, input_features: list[str] | None = None) -> list[str]:
    """Get output feature names for transformation.

    Parameters
    ----------
    input_features : list of str or None, default=None
        Column names of the input features.  If ``None``, uses the
        feature names seen during ``fit``.

    Returns
    -------
    list of str
        Output feature names after transformation.

    """
    input_features = _check_feature_names_in(self, input_features)
    feature_names = [f"{col}_ewma" for col in input_features]
    arr: list[str] = np.asarray(feature_names, dtype=object).tolist()
    return arr

Tutorials

The following example notebooks use this component:

• How to Compose Features with FeatureUnion

  ---

  Data-Features

  Combine lag features, rolling statistics, EMA, and scaling in parallel with FeatureUnion and automatic observation horizon resolution.

  View · Open in marimo

• How to Apply Window Transformations

  ---

  Data-Features

  Feature engineering with LagTransformer, RollingStatisticsTransformer, SlidingWindowFunctionTransformer, and ExponentialMovingAverage on time series data.

  View · Open in marimo