Pandas中基于多条件和时间窗口匹配关联数据的策略

本教程探讨如何在Pandas中高效地将一个DataFrame中的事件与另一个DataFrame中特定时间窗口（例如7天内）内的相关事件进行匹配和聚合。针对merge_asof的局限性，我们将介绍两种主要方法：利用pyjanitor库的conditional_join功能实现多条件高效连接，以及纯Pandas的merge结合筛选和聚合方案。文章将通过具体示例代码，详细阐述两种方法的实现步骤、适用场景及其性能考量。

问题场景与挑战

在数据分析中，我们经常需要关联两个数据集，其中一个数据集的记录（例如交易记录）需要匹配另一个数据集（例如浏览历史）中发生在特定时间窗口内的所有相关记录。考虑以下两个pandas dataframe：

1. trade DataFrame：包含交易日期、人员ID、商品代码和交易价值。

   import pandas as pd
   import 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]})

2. view DataFrame：包含浏览日期、人员ID、商品代码和浏览价值。

   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]})

我们的目标是为每一笔交易，找出其发生前7天内（含交易当天）所有匹配person和code的浏览记录，并将这些浏览记录的日期和价值聚合为列表。

直接使用Pandas的merge_asof函数在这种场景下存在局限性。merge_asof设计用于“最近匹配”，它会为左DataFrame的每一行找到右DataFrame中最近的匹配行，且通常只匹配一次。这不符合我们“获取所有在时间窗口内的匹配项”的需求。例如，merge_asof可能会将view中的[1, 3]分配给第一笔交易，但对于第二笔交易，它可能只会分配[5]，而忽略了同样在时间窗口内的[1, 3]。

方案一：使用pyjanitor.conditional_join进行高效多条件连接

pyjanitor库提供了一个强大的conditional_join函数，它允许基于多个自定义条件进行连接，包括非等值条件（如时间范围）。这使得它成为解决此类问题的理想选择，尤其是在处理大型数据集时，其效率通常高于纯Pandas的通用合并再筛选方案。

实现步骤

1. 日期列转换：确保所有涉及日期比较的列都是Pandas的datetime类型。
2. 创建辅助时间列：在trade DataFrame中，为每笔交易计算一个“开始日期”，即交易日期减去7天。这将作为时间窗口的下限。
3. 执行条件连接：使用conditional_join连接trade和view DataFrame。连接条件包括：
   • view的日期在trade的start_date之后或等于。
   • view的日期在trade的date之前或等于。
   • person列相等。
   • code列相等。
4. 数据聚合：连接完成后，结果DataFrame中每行将是trade记录与一个匹配的view记录的组合。我们需要根据原始trade的列进行分组，然后将匹配到的view_dates和view_values聚合成列表。
5. 格式化输出：将聚合后的日期列表转换回字符串格式。

代码示例

# 确保日期列为datetime类型
trade['date'] = pd.to_datetime(trade['date'])
view['date'] = pd.to_datetime(view['date'])
# 方案一：使用janitor.conditional_join
out_janitor = (trade
.assign(start_date=lambda d: d['date'].sub(pd.DateOffset(days=7))) # 计算时间窗口的开始日期
.conditional_join(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view的列以避免冲突
# 定义连接条件：(左列, 右列, 操作符)
('start_date', 'view_dates', '<='), # view_dates >= start_date
('date', 'view_dates', '>='),       # view_dates <= trade_date
('person', 'person', '=='),         # person相等
('code', 'code', '=='),             # code相等
right_columns=['view_dates', 'view_values'] # 只保留view中需要的列
)
.drop(columns='start_date') # 移除辅助列
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade.columns), as_index=False).agg(list) # 按原始trade列分组并聚合为列表
)
print("--- 使用 janitor.conditional_join 的结果 ---")
print(out_janitor)

优势分析

conditional_join在内部实现了优化的算法来处理非等值连接，特别是涉及到范围查询时，其性能通常优于先进行笛卡尔积式合并再筛选的纯Pandas方法，尤其适用于大数据集。

方案二：纯Pandas的通用合并与筛选

虽然conditional_join更高效，但也可以纯粹使用Pandas的merge和loc进行筛选来实现相同的功能。这种方法在概念上更直接，但可能在性能上有所牺牲，因为它首先会生成一个更大的中间DataFrame。

实现步骤

1. 日期列转换：同方案一，确保日期列为datetime类型。
2. 通用合并：使用merge函数基于person和code列进行内连接。这将生成trade和view中所有person和code匹配的组合。
3. 时间窗口筛选：使用loc根据时间条件（view_dates在trade_date前7天内）对合并后的结果进行筛选。
4. 数据聚合：同方案一，根据原始trade的列进行分组，然后将匹配到的view_dates和view_values聚合成列表。
5. 格式化输出：将聚合后的日期列表转换回字符串格式。

代码示例

# 确保日期列为datetime类型
trade['date'] = pd.to_datetime(trade['date'])
view['date'] = pd.to_datetime(view['date'])
# 方案二：纯Pandas解决方案
out_pandas = (trade
.merge(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view的列
on=['person', 'code']) # 基于person和code进行合并
.loc[lambda d: d['date'].gt(d['view_dates']) & # 交易日期必须晚于浏览日期
d['date'].sub(pd.DateOffset(days=7)).le(d['view_dates']) # 浏览日期必须在交易日期前7天内
]
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade.columns), as_index=False).agg(list) # 按原始trade列分组并聚合为列表
)
print("\n--- 纯Pandas解决方案的结果 ---")
print(out_pandas)

局限性分析

纯Pandas方案首先会生成一个包含所有person和code匹配组合的中间DataFrame。如果trade和view中存在大量相同person和code的记录，这个中间DataFrame可能会非常大，导致内存消耗增加和计算时间延长。随后进行的时间条件筛选会减少数据量，但前期合并的开销是不可避免的。

总结与最佳实践

两种方法都能达到预期的结果：

        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]

• 对于小型数据集或对性能要求不高的场景：纯Pandas解决方案简单直观，易于理解和实现。
• 对于大型数据集或对性能有严格要求的场景：强烈推荐使用pyjanitor.conditional_join。它通过更优化的算法处理非等值连接，能够显著提高效率并减少内存消耗。

注意事项：

• 日期类型一致性：在进行任何日期比较或计算之前，务必确保所有日期列都已正确转换为Pandas datetime类型。
• 时间窗口定义：根据具体业务需求精确定义时间窗口（例如，pd.DateOffset(days=7)表示7天，可以根据需要调整为hours、minutes等）。
• 列名冲突：在合并前，如果两个DataFrame中有相同名称但含义不同的列，请务必重命名以避免混淆。在示例中，我们重命名了view中的date和value列。
• 安装pyjanitor：如果选择使用pyjanitor，请确保已通过pip install pyjanitor安装该库。

通过上述两种方法，我们可以灵活地在Pandas中处理复杂的基于多条件和时间窗口的数据关联任务，选择最适合当前项目需求和数据规模的方案。