如何计算当前行日期前分组的sales_amt移动平均值？

原始数据 #

import pandas as pd
import numpy as np

product_type = ['A','B']
df = pd.DataFrame({
    'prod_id':np.repeat(np.arange(start=2,stop=5,step=1),59),
    'prod_type': np.random.choice(np.array(product_type), 177),
    'sales_time': pd.date_range(start ='1-1-2018', end ='3-30-2018', freq ='12H'),
    'sale_amt':np.random.randint(4,100,size = 177)
})

样例数据：

需求 #

按prod_id、prod_type分组，计算**当前记录日期之前（不含当日）**所有交易的sale_amt平均值，期望输出：

已掌握分组忽略当前行的移动平均方法，但无法过滤当日记录：

sf = df.sort_values('sales_time')
sf['avg_sale_amt'] = sf.groupby(['prod_id', 'prod_type'])['sale_amt'].transform(lambda gr: gr.expanding().mean().shift())

解决方法 #

方法一：遍历分组计算（直观易懂） #

先提取日期列，再在每个分组内筛选出当前日期之前的记录计算平均值：

# 按时间排序，保证计算顺序正确
sf = df.sort_values('sales_time').reset_index(drop=True)
# 提取日期部分，方便按自然日筛选
sf['sales_date'] = sf['sales_time'].dt.date

def get_historical_avg(group):
    avg_list = []
    for idx, row in group.iterrows():
        # 筛选同组内日期早于当前记录的交易金额
        prev_amts = group[group['sales_date'] < row['sales_date']]['sale_amt']
        avg_list.append(prev_amts.mean() if len(prev_amts) > 0 else 0)
    group['avg_sale_amt'] = avg_list
    return group

# 分组应用函数
result = sf.groupby(['prod_id', 'prod_type'], group_keys=False).apply(get_historical_avg)
# 整理列顺序，删除临时日期列
result = result[['prod_id', 'prod_type', 'sales_time', 'sale_amt', 'avg_sale_amt']]

方法二：按日聚合优化（适合大数据集） #

先按日聚合统计，再计算累计均值，最后映射回原表，效率更高：

# 按时间排序
sf = df.sort_values('sales_time').reset_index(drop=True)
sf['sales_date'] = sf['sales_time'].dt.date

# 按分组+日期聚合，计算每日总金额和交易数
daily_stats = sf.groupby(['prod_id', 'prod_type', 'sales_date']).agg(
    total_amt=('sale_amt', 'sum'),
    count=('sale_amt', 'size')
).reset_index()

# 计算分组内的累计总和与累计计数（shift排除当前日期）
daily_stats['cum_total'] = daily_stats.groupby(['prod_id', 'prod_type'])['total_amt'].cumsum().shift()
daily_stats['cum_count'] = daily_stats.groupby(['prod_id', 'prod_type'])['count'].cumsum().shift()

# 计算历史平均，无前置记录时填充0
daily_stats['avg_sale_amt'] = (daily_stats['cum_total'] / daily_stats['cum_count']).fillna(0)

# 将结果映射回原表
result = sf.merge(daily_stats[['prod_id', 'prod_type', 'sales_date', 'avg_sale_amt']],
                  on=['prod_id', 'prod_type', 'sales_date'],
                  how='left')
# 整理列顺序
result = result[['prod_id', 'prod_type', 'sales_time', 'sale_amt', 'avg_sale_amt']]

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