增量计算分ID的7日滚动平均值的可行性与效率咨询

样例DataFrame #

avg列是7日滚动平均值，当前使用以下代码计算整列：

index["avg"] = index.groupby("ID")["data"].transform(lambda x: x.rolling(7).mean())

由于数据是每日增量添加，希望了解：

1. 是否可以仅针对每个ID，计算包含最新添加数据在内的近7日数据的最新日期avg值？
2. 这种方法是否比重新计算整列更高效？

1. 可以仅计算最新的滚动平均值 #

完全可以只针对每个ID的最新日期计算对应的7日滚动平均值，无需重新计算整列。核心思路是过滤每个ID的最新记录，取其最近7天的数据计算平均值后回填。

代码实现示例

首先确保date列为datetime类型（若不是先转换）：

index['date'] = pd.to_datetime(index['date'], format='%d/%m/%Y')
index = index.sort_values(["ID", "date"])

场景1：已有数据中补充最新行的avg值

# 提取每个ID的最新日期
latest_dates = index.groupby("ID")["date"].max().reset_index()

# 遍历计算并回填
for _, row in latest_dates.iterrows():
    id_val = row["ID"]
    latest_date = row["date"]
    # 获取该ID近7天的数据（含最新日期）
    window_data = index[(index["ID"] == id_val) & (index["date"] >= latest_date - pd.Timedelta(days=6))]["data"]
    # 赋值给对应行的avg列
    index.loc[(index["ID"] == id_val) & (index["date"] == latest_date), "avg"] = window_data.mean()

场景2：每日新增数据时直接计算
假设新增数据存储在new_data中：

# 合并新数据到原表（若未合并）
index = pd.concat([index, new_data]).sort_values(["ID", "date"])

# 处理每条新增记录
for _, row in new_data.iterrows():
    id_val = row["ID"]
    current_date = row["date"]
    window_data = index[(index["ID"] == id_val) & (index["date"] >= current_date - pd.Timedelta(days=6))]["data"]
    index.loc[(index["ID"] == id_val) & (index["date"] == current_date), "avg"] = window_data.mean()

2. 增量计算远优于全量重算 #

这种方法比重新计算整列高效得多，原因如下：

• 计算范围极小：全量重算要对每个ID的所有历史数据做滚动窗口计算，增量方法仅针对每个ID的最近7条数据计算一次
• 无重复计算：全量重算每次都会覆盖所有avg值，增量方法只处理新增行，完全不触碰已计算好的历史数据
• 数据量越大优势越明显：当数据达到上万甚至百万行时，增量计算的时间成本会远低于全量重算

注意事项 #

• 必须确保date列为datetime类型，否则无法正确筛选时间窗口
• 若同一ID在同一天有多条数据，需调整为按ID+日期分组后计算平均值
• 若要求严格按自然日的7天窗口（而非最近7条记录），必须用时间范围过滤，避免日期不连续导致窗口错误

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