实现与Pandas ewm()一致的EMA算法遇异常，求错误排查

我仔细看了你的代码，找到了导致窗口超出初始数据集后和Pandas ewm()结果不一致的两个核心问题：

1. 错误截断了历史数据的使用范围 #

你用index = min(len(c), self.lookback_period+1)限制了计算时仅使用最近6个数据（因为lookback_period=5，+1后为6），但Pandas的ewm(span=5, adjust=True)是基于所有历史数据计算加权平均的——只有adjust=False时才会用递归方式近似依赖最近span个数据，adjust=True会严格累积所有历史权重，不会截断。

2. 预先生成的权重列表长度固定 #

你在__init__里生成的lookback_alphas只包含(1-α)^1到(1-α)^5的权重，当数据量超过6个时，需要用到(1-α)^6及更高次的权重，你的列表里没有对应值，自然无法正确计算所有历史数据的加权和。

修复后的代码 #

我调整了实现逻辑，去掉固定长度的权重列表，改为动态计算每个历史数据点的权重，严格匹配Pandas ewm(span=5, adjust=True)的计算规则：

import numpy as np
import pandas as pd

class MovingAverages(object):
    def __init__(self, **kwargs):
        self.measures = []
        self.lookback_period = 5
        # 保存EMA的alpha系数，对应Pandas span参数的alpha=2/(span+1)
        self.alpha = 2 / (self.lookback_period + 1)
        self.one_minus_alpha = 1 - self.alpha

    def insert_bar(self):
        self.measures.insert(0, 0)

    def on_calculate(self, c):
        # c是按「最新到最早」排列的数据列表
        n = len(c)
        weighted_sum = 0.0
        weight_sum = 0.0
        # 遍历所有历史数据，计算加权和与权重总和
        for i in range(n):
            weight = self.one_minus_alpha ** i
            weighted_sum += c[i] * weight
            weight_sum += weight
        # 加权平均得到EMA值
        self.measures[0] = weighted_sum / weight_sum

if __name__ == "__main__":
    data = [5.00,7.00,4.00,3.00,4.00, 5.00,6.00,7.00,9.00,13.00,
            15.00,14.00,14.00,12.00, 11.00,10.00,9.00,8.00,
            8.00,8.00,10.00,11.00, 13.00,16.00,18.00,20.00]

    ma_online = MovingAverages()
    series = []
    for d in data:
        series.insert(0, d)
        ma_online.insert_bar()
        ma_online.on_calculate(series)

    # Pandas参考计算
    df = pd.DataFrame({'close': data})
    ma_pd = list(df.close.ewm(span=ma_online.lookback_period, adjust=True, ignore_na=True).mean())

    # 对比结果
    ma_online.measures.reverse()
    for i in range(len(data)):
        # 保留两位小数对比
        assert round(ma_pd[i], 2) == round(ma_online.measures[i], 2), f"Index {i} mismatch: Pandas {ma_pd[i]:.2f}, Your code {ma_online.measures[i]:.2f}"
    print("All matches!")

匹配Pandas的核心逻辑 #

Pandas ewm(adjust=True)的核心公式是：

EMAₜ = (xₜ + (1-α)xₜ₋₁ + (1-α)²xₜ₋₂ + ... + (1-α)ᵗ⁻¹x₁) / (1 + (1-α) + (1-α)² + ... + (1-α)ᵗ⁻¹)

其中α=2/(span+1)，修复后的代码完全按照这个公式实现：遍历所有历史数据，计算每个数据点的权重(1-α)^i（i是数据点距离当前的步数），然后求加权和除以权重总和，和Pandas的计算逻辑完全一致。

内容的提问来源于stack exchange，提问作者Ian Ash