如何在Pandas中按分组实现3天滚动频次统计？

核心需求拆解 #

• 按name分组，对每个用户的时间序列数据，计算每个date_id对应的前1-3天（即当前日期往前推1、2、3天，不包含当前日期）的滚动统计：
  1. 总记录数（3 days roll count）
  2. 完成状态为Y的记录数（3 day completion count）
• 若窗口范围早于用户数据的起始日期，超出部分计为0。

解决方案步骤 #

1. 数据预处理

首先确保日期列是datetime类型，并按用户和日期排序，保证时间序列有序：

import pandas as pd

# 构造示例数据（匹配需求场景）
data = {
    'name': ['John', 'John', 'John', 'John', 'John', 'John', 'Jane', 'Jane'],
    'date_id': ['2022-01-01', '2022-01-02', '2022-01-03', '2022-01-04', '2022-01-05', '2022-01-08', '2022-01-03', '2022-01-06'],
    'completion': ['N', 'Y', 'N', 'N', 'N', 'N', 'Y', 'N']
}
df = pd.DataFrame(data)

# 转换日期类型
df['date_id'] = pd.to_datetime(df['date_id'])
# 按用户和日期排序
df = df.sort_values(['name', 'date_id']).reset_index(drop=True)

2. 方法一：使用Pandas原生rolling时间窗口（高效简洁）

利用rolling的时间窗口功能，指定窗口为3天，且不包含当前日期（closed='left'），自动处理时间范围：

# 计算3天滚动总次数
df['3 days roll count'] = df.groupby('name').rolling(
    window='3D', on='date_id', closed='left'
).size().reset_index(level=0, drop=True).astype(int)

# 计算3天滚动完成次数：先将Y/N转为数值，再滚动求和
df['completion_num'] = df['completion'].map({'Y': 1, 'N': 0})
df['3 day completion count'] = df.groupby('name').rolling(
    window='3D', on='date_id', closed='left'
)['completion_num'].sum().reset_index(level=0, drop=True).astype(int)

# 删除临时列
df = df.drop('completion_num', axis=1)

3. 方法二：自定义窗口逻辑（灵活适配特殊规则）

如果需要更精确控制窗口（比如强制窗口不早于用户数据起始日），可以用groupby.apply自定义统计逻辑：

def calculate_rolling_stats(group):
    group = group.sort_values('date_id').reset_index(drop=True)
    # 初始化结果列
    group['3 days roll count'] = 0
    group['3 day completion count'] = 0
    
    for idx, row in group.iterrows():
        current_date = row['date_id']
        # 窗口起始日：取当前日期减3天和用户最早日期的较大值
        start_date = max(group['date_id'].min(), current_date - pd.Timedelta(days=3))
        # 窗口范围：start_date <= date_id < current_date
        mask = (group['date_id'] >= start_date) & (group['date_id'] < current_date)
        
        # 统计总次数和完成次数
        group.loc[idx, '3 days roll count'] = mask.sum()
        group.loc[idx, '3 day completion count'] = group.loc[mask, 'completion'].eq('Y').sum()
    
    return group

# 应用自定义函数
df_result = df.groupby('name', group_keys=False).apply(calculate_rolling_stats)

验证结果 #

以你提到的示例场景为例：

• John在2022-01-05的3天滚动总次数为4（对应窗口内的4条符合条件的记录）
• John在2022-01-08的3天滚动总次数为4（对应窗口内的4条符合条件的记录）

两种方法都能满足需求，方法一效率更高，适合大数据量；方法二更灵活，适配特殊规则。

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