Pandas中基于时间窗口关联和聚合数据的技巧：以交易与浏览记录为例（为例.关联.聚合.窗口.技巧...）

本文详细介绍了如何在Pandas中，从两个DataFrame（如交易记录和浏览历史）中，高效地识别并聚合出在特定时间窗口（例如交易前7天）内相关联的数据。教程提供了两种实现方法：一种是利用pyjanitor库的conditional_join函数进行性能优化，另一种是纯Pandas的merge结合条件筛选，并最终通过groupby.agg完成数据聚合，旨在帮助用户处理复杂的时序数据关联需求。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的循环或笛卡尔积合并。

实现步骤

1. 创建辅助列 start_date： 在trade DataFrame中添加一个start_date列，表示每笔交易发生日期前7天的日期。这将作为我们时间窗口的下限。
2. 执行 conditional_join： 使用conditional_join函数连接trade和view。
   • 时间条件： 确保view_dates在start_date和trade.date之间（即start_date <= view_dates <= trade.date）。
   • 匹配条件： 确保person和code相等。
   • 重命名列： 在view DataFrame中，将date和value列重命名为view_dates和view_values，以避免与trade中的列名冲突，并明确其来源。
3. 清理和格式化： 删除不再需要的start_date辅助列，并将view_dates格式化为字符串。
4. 聚合结果： 根据原始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（笛卡尔积或近似笛卡尔积）。

实现步骤

1. 进行笛卡尔合并： 首先，基于person和code列对trade和view进行合并。这将为每个交易条目与每个匹配的浏览条目创建一个组合。
2. 应用时间条件筛选： 在合并后的DataFrame上，应用两个时间条件：
   • 浏览日期必须早于交易日期 (trade_date > view_dates)。
   • 浏览日期必须在交易日期前7天之内 (trade_date - 7_days <= view_dates)。
3. 格式化和聚合： 同样，将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)来收集所有匹配的条目。

以上就是Pandas中基于时间窗口关联和聚合数据的技巧：以交易与浏览记录为例的详细内容，更多请关注知识资源分享宝库其它相关文章！