基于相对时间区间统计用户交易活动次数并广播结果的高效实现

问题背景 #

现有一个百万级行的DataFrame，person_id唯一，结构如下：

+-----------+------------+----------+
| person_id |    date    | activity |
+-----------+------------+----------+
| A         | 31/03/2022 | Sell     |
| A         | 02/03/2023 | Buy      |
| A         | 29/08/2023 | Buy      |
| A         | 13/05/2023 | Buy      |
| A         | 28/02/2023 | Sell     |
| A         | 02/01/2024 | Sell     |
+-----------+------------+----------+

需求：

• 以系统当前日期为基准，按person_id分组
• 统计每个用户过去3、6、9、12个月内Buy和Sell活动的次数，新增列：buy_3m、buy_6m、buy_9m、buy_12m、sell_3m、sell_6m、sell_9m、sell_12m
• 将分组统计结果广播到原DataFrame每一行，保持行数与输入一致
• 原方案用relativedelta、多次join和groupby，代码冗长且效率低，需要更高效的实现方式

高效实现思路 #

核心是避免多次分组/连接，通过向量化操作和一次分组聚合完成统计，再合并回原表。

步骤1：预处理日期列 #

先把date列转为datetime类型，计算当前日期，生成各个周期的时间阈值：

import pandas as pd
from datetime import datetime

# 加载原DataFrame（示例）
df = pd.DataFrame({
    'person_id': ['A']*6,
    'date': ['31/03/2022', '02/03/2023', '29/08/2023', '13/05/2023', '28/02/2023', '02/01/2024'],
    'activity': ['Sell', 'Buy', 'Buy', 'Buy', 'Sell', 'Sell']
})

# 转换日期格式（根据实际格式调整dayfirst参数）
df['date'] = pd.to_datetime(df['date'], dayfirst=True)
current_date = datetime.today()

# 定义时间周期（月），生成对应的起始日期阈值
periods = [3,6,9,12]
date_thresholds = {f'{p}m': current_date - pd.DateOffset(months=p) for p in periods}

步骤2：生成活动类型的哑变量 #

用pd.get_dummies把activity转为二进制列，方便后续批量统计：

df_dummies = pd.get_dummies(df['activity'], prefix='', prefix_sep='')
df = pd.concat([df, df_dummies], axis=1)

步骤3：单次分组聚合完成所有周期统计 #

按person_id分组，对每个周期阈值统计符合条件的活动次数，用apply一次性生成所有统计项：

# 定义聚合函数：统计单个用户所有周期的活动次数
def count_period_activities(group):
    stats = {}
    for period, threshold in date_thresholds.items():
        # 筛选当前周期内的记录
        mask = group['date'] >= threshold
        # 统计Buy和Sell的累计次数
        stats[f'buy_{period}'] = group.loc[mask, 'Buy'].sum()
        stats[f'sell_{period}'] = group.loc[mask, 'Sell'].sum()
    return pd.Series(stats)

# 分组聚合得到每个用户的统计结果
user_stats = df.groupby('person_id').apply(count_period_activities).reset_index()

步骤4：合并统计结果回原DataFrame #

用merge将统计结果广播到原表的每一行，保持原数据行数：

# 合并后保留原表所有行
result = df.merge(user_stats, on='person_id', how='left')

# 可选：删除中间生成的Buy/Sell哑变量列
result = result.drop(['Buy', 'Sell'], axis=1)

优化点说明 #

• 向量化操作：用哑变量替代多次条件判断，减少循环开销
• 单次分组聚合：避免多次groupby和join，大幅降低百万级数据的处理耗时
• 日期处理优化：用pd.DateOffset替代relativedelta，更适配pandas的datetime类型，计算效率更高

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