1. 问题描述与数据准备
在数据分析中，我们经常需要根据时间范围将两个数据集关联起来。一个常见的场景是，我们有两个dataframe：一个记录了用户的交易信息，另一个记录了用户的浏览历史。我们的目标是为每笔交易找出其发生前7天内的所有相关浏览记录，并聚合这些记录的日期和值。这里的“相关”意味着交易和浏览记录必须属于同一个用户（person）和同一个商品代码（code）。
考虑以下两个示例DataFrame：
交易记录 (trade)
import pandas as pd
import janitor # 导入 janitor 库
trade = pd.DataFrame({'date': ['2019-08-31', '2019-09-01', '2019-09-04'],
'person': [1, 1, 2],
'code': [123, 123, 456],
'value1': [1, 2, 3]})浏览历史 (view)
view = pd.DataFrame({'date': ['2019-08-29', '2019-08-29', '2019-08-30', '2019-08-31', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-02', '2019-09-03'],
'person': [1, 1, 1, 2, 1, 2, 2, 1, 2],
'code': [123, 456, 123, 456, 123, 123, 456, 123, 456],
'value': [1, 2, 3, 4, 5, 6, 7, 8, 9]
})在进行时间相关的操作之前，将日期列转换为Pandas的datetime类型是至关重要的。
trade['date'] = pd.to_datetime(trade['date']) view['date'] = pd.to_datetime(view['date'])
我们期望的输出结果是，在原始交易信息的基础上，增加两列：view_dates和view_values，它们分别是以列表形式存储的、在每笔交易前7天内匹配到的浏览记录日期和值。
2. 传统 merge_asof 的局限性
Pandas的merge_asof函数常用于近似合并时序数据。然而，它主要设计用于“最近匹配”场景，通常每个右侧（right）DataFrame的条目最多只能匹配一次。在本例中，我们希望每个交易条目能够匹配到多个浏览历史条目，且这些浏览历史条目可以被不同的交易重复匹配。因此，merge_asof无法直接满足我们的需求。例如，如果使用merge_asof，它可能会将浏览记录[1, 3]分配给第一笔交易，但对于第二笔交易，它可能只会分配[5]，而忽略了同样在7天窗口内的[1, 3]。
3. 方法一：使用 pyjanitor.conditional_join (推荐)
pyjanitor库提供了一个强大的conditional_join函数，它允许我们基于多个自定义条件进行DataFrame的连接，包括非等值条件（如时间范围）。这使得它非常适合处理此类复杂的时序数据关联问题，尤其是在处理大规模数据时，其效率通常高于纯Pandas的循环或笛卡尔积合并。
实现步骤
- 创建辅助列 start_date： 在trade DataFrame中添加一个start_date列，表示每笔交易发生日期前7天的日期。这将作为我们时间窗口的下限。
-
执行 conditional_join： 使用conditional_join函数连接trade和view。
- 时间条件： 确保view_dates在start_date和trade.date之间（即start_date
- 匹配条件： 确保person和code相等。
- 重命名列： 在view DataFrame中，将date和value列重命名为view_dates和view_values，以避免与trade中的列名冲突，并明确其来源。
- 清理和格式化： 删除不再需要的start_date辅助列，并将view_dates格式化为字符串。
- 聚合结果： 根据原始trade DataFrame的列进行分组，然后将匹配到的view_dates和view_values聚合成列表。
示例代码
# 1. 创建辅助列 'start_date'
# 2. 执行 conditional_join
# 3. 清理和格式化
# 4. 聚合结果
out_janitor = (trade
.assign(start_date=lambda d: d['date'].sub(pd.DateOffset(days=7))) # 计算7天前的日期
.conditional_join(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view列
('start_date', 'view_dates', '<='), # 条件1: start_date <= view_dates
('date', 'view_dates', '>='), # 条件2: trade.date >= view_dates
('person', 'person', '=='), # 条件3: person 匹配
('code', 'code', '=='), # 条件4: code 匹配
right_columns=['view_dates', 'view_values'] # 保留右侧DataFrame的哪些列
)
.drop(columns='start_date') # 删除辅助列
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade), as_index=False).agg(list) # 按原始trade列分组并聚合列表
)
print("使用 pyjanitor.conditional_join 的结果:")
print(out_janitor)输出结果
date person code value1 view_dates view_values 0 2019-08-31 1 123 1 [2019-08-29, 2019-08-30] [1, 3] 1 2019-09-01 1 123 2 [2019-08-29, 2019-08-30, 2019-09-01] [1, 3, 5] 2 2019-09-04 2 456 3 [2019-08-31, 2019-09-01, 2019-09-03] [4, 7, 9]
4. 方法二：纯 Pandas 解决方案
如果不想引入额外的库，也可以纯粹使用Pandas的merge和条件筛选来实现。这种方法在概念上更直观，但对于非常大的数据集，其性能可能不如pyjanitor.conditional_join，因为它首先会生成一个潜在的较大中间DataFrame（笛卡尔积或近似笛卡尔积）。
实现步骤
- 进行笛卡尔合并： 首先，基于person和code列对trade和view进行合并。这将为每个交易条目与每个匹配的浏览条目创建一个组合。
-
应用时间条件筛选： 在合并后的DataFrame上，应用两个时间条件：
- 浏览日期必须早于交易日期 (trade_date > view_dates)。
- 浏览日期必须在交易日期前7天之内 (trade_date - 7_days
- 格式化和聚合： 同样，将view_dates格式化为字符串，然后按原始trade DataFrame的列进行分组，并聚合列表。
示例代码
# 1. 进行笛卡尔合并
# 2. 应用时间条件筛选
# 3. 格式化和聚合
out_pandas = (trade
.merge(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view列并合并
on=['person', 'code'])
.loc[lambda d: d['date'].gt(d['view_dates']) & # 筛选条件1: 交易日期 > 浏览日期
d['date'].sub(pd.DateOffset(days=7)).le(d['view_dates']) # 筛选条件2: 交易日期 - 7天 <= 浏览日期
]
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade), as_index=False).agg(list) # 按原始trade列分组并聚合列表
)
print("\n使用纯 Pandas 解决方案的结果:")
print(out_pandas)输出结果
date person code value1 view_dates view_values 0 2019-08-31 1 123 1 [2019-08-29, 2019-08-30] [1, 3] 1 2019-09-01 1 123 2 [2019-08-29, 2019-08-30, 2019-09-01] [1, 3, 5] 2 2019-09-04 2 456 3 [2019-08-31, 2019-09-01, 2019-09-03] [4, 7, 9]
5. 总结与选择
本教程展示了两种在Pandas中实现基于时间窗口的数据关联和聚合的方法：
- pyjanitor.conditional_join： 这种方法在处理复杂的非等值连接（如时间范围）时表现出色，尤其在性能方面优于纯Pandas的笛卡尔合并加筛选，因为它避免了生成过大的中间DataFrame。当处理大型数据集或需要更灵活的连接条件时，强烈推荐使用此方法。
- 纯 Pandas merge + 筛选： 这种方法易于理解和实现，不需要额外库。但在数据量较大时，由于可能产生庞大的中间合并结果，其内存消耗和计算效率可能较低。对于中小型数据集或对性能要求不极致的场景，这是一个可行的选择。
选择哪种方法取决于你的具体需求、数据集大小以及对性能和依赖库的偏好。无论选择哪种，关键都在于正确地定义时间窗口条件并使用groupby.agg(list)来收集所有匹配的条目。
