ExponentialMovingAverage¶

class ExponentialMovingAverage(BaseActualTransformer):
    """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

    Unlike the other window transformers, this transformer is stateless: it has
    no internal buffer, so ``observe``/``rewind`` fall through to the base no-op.
    Each ``transform`` call must therefore receive the full history needed by the
    EWMA computation.

    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`][yohou.preprocessing.window.RollingStatisticsTransformer] : Fixed-window rolling statistics.
    - [`SlidingWindowFunctionTransformer`][yohou.preprocessing.window.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