To detect anomalies using EWMA, compute the exponentially weighted moving average (EWMA) as a smooth estimate of the signal, then compute residual = value - EWMA, rolling std of residuals (also exponentially or windowed), and flag points where |residual| > k * rolling_std.python
import pandas as pd
import numpy as np
def detect_ewma(df, value_col, span=20, k=3, std_span=None, min_periods=1):
"""
Adds columns to df: ewma, residual, rolling_std, anomaly.
- span: EWMA span (pandas uses alpha = 2/(span+1))
- k: threshold multiplier for std
- std_span: span for EW running std; if None uses same span
- min_periods: minimum observations to compute stats
"""
df = df.copy()
if std_span is None:
std_span = span
# EWMA estimate
df['ewma'] = df[value_col].ewm(span=span, adjust=False).mean()
# Residuals
df['residual'] = df[value_col] - df['ewma']
# Rolling standard deviation of residuals using EWMA (gives more weight to recent)
# unbiased estimation not necessary for anomaly flagging; use ewm of squared residuals then sqrt(var)
var_ewm = df['residual'].ewm(span=std_span, adjust=False).mean()**2
# Better: use EWM of squared residuals minus square of EWM => variance
s1 = df['residual'].ewm(span=std_span, adjust=False).mean()
s2 = (df['residual']**2).ewm(span=std_span, adjust=False).mean()
df['rolling_std'] = np.sqrt((s2 - s1**2).clip(lower=0))
# fallback: if rolling_std is zero or NaN, avoid false positives
df['rolling_std'].fillna(0, inplace=True)
df['anomaly'] = df['rolling_std'] > 0
df.loc[df['rolling_std'] > 0, 'anomaly'] = (df.loc[df['rolling_std'] > 0, 'residual'].abs()
> k * df.loc[df['rolling_std'] > 0, 'rolling_std'])
df['anomaly'] = df['anomaly'].astype(bool)
return df
