指数移动平均平滑器

功能说明

指数移动平均对近期数据赋予更高权重，适合监控指标、行情曲线和传感器读数的去噪。与简单移动平均相比，它不需要保存整个窗口，只需保留上一期的平滑值，因此可以在流式管道里常数内存运行。

EmaTracker 在初始化时根据周期计算平滑系数：周期越小，系数越大，曲线越贴近原始值；周期越大，曲线越平滑但滞后越明显。首次更新直接采用样本值作为起点，避免从零开始导致曲线长期偏低。

关键代码

alpha 等于二除以周期加一，这个取值让指数平均的等效窗口接近给定周期。update 在 value 为 None 时赋初值，否则按 alpha 乘新样本加上一减 alpha 乘旧值递推。reset 清空状态，便于在数据源切换或重启后重新预热。simple_moving_average 在前 window 减一个位置填充 None，对齐时间轴，便于和 EMA 并排比较。

使用注意

首次回填会让曲线在前几期偏向第一个样本，若指标对启动偏差敏感，可以先用历史数据预热或丢弃前若干期。周期必须是正整数，零或负数会直接报错。缺失值不要用零填充，否则会把曲线拉低；应先跳过或插值。EMA 对异常值仍有一定敏感度，极端尖峰可以配合截断或中位数滤波。多个序列需要各自维护独立的 tracker 实例。

验证方式

对 10、12、11、15、14、16 六个数、周期为三时，alpha 为 0.5，EMA 依次为 10、11、11、13、13.5、14.75。简单平均在前两个位置输出 None，第三个位置输出 11.0，说明两个序列已按同一时间轴对齐。可以把手动计算的第一期与打印值比对，确认平滑系数正确。

代码示例

from collections.abc import Iterable
from dataclasses import dataclass, field


@dataclass
class EmaTracker:
    period: int
    value: float | None = None
    alpha: float = field(init=False, default=0.0)

    def __post_init__(self) -> None:
        if self.period < 1:
            raise ValueError("period must be at least 1")
        self.alpha = 2.0 / (self.period + 1)

    def update(self, sample: float) -> float:
        if self.value is None:
            self.value = sample
        else:
            self.value = self.alpha * sample + (1 - self.alpha) * self.value
        return self.value

    def reset(self) -> None:
        self.value = None


def exponential_moving_average(values: Iterable[float], period: int) -> list[float]:
    tracker = EmaTracker(period)
    return [tracker.update(value) for value in values]


def simple_moving_average(values: list[float], window: int) -> list[float | None]:
    if window < 1:
        raise ValueError("window must be at least 1")
    smoothed: list[float | None] = []
    for index in range(len(values)):
        if index + 1 < window:
            smoothed.append(None)
        else:
            chunk = values[index + 1 - window : index + 1]
            smoothed.append(sum(chunk) / window)
    return smoothed


def main() -> None:
    samples = [10.0, 12.0, 11.0, 15.0, 14.0, 16.0]
    print("ema:", [round(value, 3) for value in exponential_moving_average(samples, 3)])
    print("sma:", simple_moving_average(samples, 3))


if __name__ == "__main__":
    main()