如何用Pandas计算过去24小时滚动累计唯一交易ID数？

我有一份包含user_account（用户账户）、transaction_id（交易ID）、transaction_date（交易日期）三列的交易数据，需要按user_account分组，计算过去24小时时间窗口内的滚动累计唯一transaction_id数量，示例如下：

示例说明：

• 第一行的transaction_id "G0000006"对应的计数为1，因为在2024-04-03 14:03:46的过去24小时内无其他交易ID；
• 第三行的transaction_id "G0000004"对应的计数为2，因为在2024-04-22 22:56:43的过去24小时内存在两个唯一交易ID（G0000004和G0000005）。

当前使用Pandas的apply方法实现，但处理300万+行数据时速度极慢，代码如下：

def count_unique_id(x):
    condition = (data['datetime'].between(x['datetime'] - dt.timedelta(days=1), x['datetime'])) & (data['user_account'] == x['user_account'])
    return data[condition]['transaction_id'].nunique()


data['count_unique_id'] = data.swifter.apply(count_unique_id, axis=1)

原代码速度慢的核心原因是时间复杂度为O(n²)：每一行都需要遍历整个数据集筛选符合条件的行，对于300万行数据来说完全不可行。以下是两种线性时间复杂度（O(n)）的高效实现方案：

方法1：哈希表维护窗口内唯一ID #

对每个用户分组后，遍历交易记录时用哈希表记录每个ID的最后出现时间，同时移除超过24小时的ID，直接通过哈希表长度得到当前窗口的唯一ID数量：

import pandas as pd
import datetime as dt

# 预处理：转换日期格式+按用户和时间排序
data['transaction_date'] = pd.to_datetime(data['transaction_date'])
data = data.sort_values(['user_account', 'transaction_date']).reset_index(drop=True)

def rolling_unique_count(group):
    id_last_seen = {}  # 存储每个transaction_id的最后出现时间
    counts = []
    for _, row in group.iterrows():
        current_time = row['transaction_date']
        current_id = row['transaction_id']
        
        # 更新当前ID的最后出现时间
        id_last_seen[current_id] = current_time
        # 移除窗口外（超过24小时）的ID
        cutoff_time = current_time - dt.timedelta(days=1)
        id_last_seen = {tid: t for tid, t in id_last_seen.items() if t >= cutoff_time}
        
        counts.append(len(id_last_seen))
    
    group['cumulative_distinct_count'] = counts
    return group

# 分组应用函数
data = data.groupby('user_account', group_keys=False).apply(rolling_unique_count)

方法2：双指针维护滑动窗口（更快） #

通过双指针固定窗口的左右边界，用集合维护窗口内的唯一ID，仅在窗口超出24小时时移动左指针，避免重复遍历哈希表，进一步提升效率：

import pandas as pd
import numpy as np

# 预处理步骤同方法1
data['transaction_date'] = pd.to_datetime(data['transaction_date'])
data = data.sort_values(['user_account', 'transaction_date']).reset_index(drop=True)

def rolling_unique_count_fast(group):
    times = group['transaction_date'].values
    ids = group['transaction_id'].values
    n = len(times)
    counts = [0] * n
    id_set = set()
    left = 0
    
    for right in range(n):
        # 将当前ID加入集合
        id_set.add(ids[right])
        # 移动左指针，确保窗口内所有时间都在24小时范围内
        cutoff = times[right] - np.timedelta64(1, 'D')
        while times[left] < cutoff:
            # 仅当该ID在窗口右侧无重复时，才从集合中移除
            if ids[left] not in ids[left+1:right+1]:
                id_set.remove(ids[left])
            left += 1
        
        counts[right] = len(id_set)
    
    group['cumulative_distinct_count'] = counts
    return group

# 分组应用函数
data = data.groupby('user_account', group_keys=False).apply(rolling_unique_count_fast)

两种方法均为线性时间复杂度，处理300万行数据的速度会比原apply方法提升几个数量级。其中方法2的双指针实现避免了每次遍历哈希表过滤，速度会略优于方法1。

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