Pandas中高效实现组内时间窗口事件检测

本文详细介绍了如何在pandas dataframe中，针对每个分组（如“团队”），高效地检测特定事件是否在指定时间窗口（如7秒）内发生。通过结合`groupby.rolling`、时间偏移以及数据帧操作，我们能够灵活地在时间序列数据中查找符合条件的未来事件，并生成相应的布尔标志列。

在处理时间序列数据时，我们经常需要分析特定事件在某个时间点之后的一段时间内是否发生。例如，在一个包含事件、团队和时间戳的数据集中，我们可能需要判断在每个事件发生后的7秒内，该团队是否发生了特定类型的事件（如事件2）。Pandas提供了强大的工具来高效地完成这类任务，特别是groupby.rolling功能。

1. 数据准备

首先，我们创建一个示例DataFrame，并确保时间戳列被正确解析为Pandas的datetime类型，这对于时间窗口操作至关重要。

import pandas as pd

data = {'event': [1, 1, 3, 2, 3, 1, 2, 3, 4, 5],
        'team': ['A', 'A', 'B', 'A', 'B', 'C', 'C', 'C', 'D', 'D'],
        'timeStamp': ['2023-07-23 14:57:13.357', '2023-07-23 14:57:14.357',
                      '2023-07-23 14:57:15.357', '2023-07-23 14:57:16.357',
                      '2023-07-23 14:57:20.357', '2023-07-23 14:57:13.357',
                      '2023-07-23 14:57:18.357', '2023-07-23 14:57:23.357',
                      '2023-07-23 14:57:23.357', '2023-07-23 14:57:25.357']}
df = pd.DataFrame(data)

# 将 'timeStamp' 列转换为 datetime 类型
df['timeStamp'] = pd.to_datetime(df['timeStamp'])

print("原始DataFrame:")
print(df)

2. 目标：在7秒时间窗口内查找特定事件

我们的目标是为DataFrame中的每一行，判断在当前行的时间戳之后的7秒内，该行所属的team是否发生了event等于2的事件。结果将存储在一个新的布尔列is_2_in_7_sec中。

3. 解决方案：使用 groupby.rolling 实现前瞻性时间窗口检测

Pandas的rolling函数默认是回顾性的（即窗口包含当前点及之前的数据）。为了实现前瞻性（即窗口包含当前点及之后的数据），我们可以采用一个巧妙的技巧：先将DataFrame反转，然后应用回顾性rolling窗口，最后再将结果合并回原DataFrame。

3.1 方案一：不包含当前行进行判断

如果目标是查找当前行之后（严格大于当前时间戳）的7秒内是否存在事件2，我们需要在rolling窗口中排除当前行。

ViiTor实时翻译

AI实时多语言翻译专家！强大的语音识别、AR翻译功能。

116

查看详情

# 1. 创建一个布尔列，标记 'event' 是否为 2
# 2. 将DataFrame反转 `[::-1]`，这样默认的回顾性rolling就变成了前瞻性
# 3. 按 'team' 分组，并在 'timeStamp' 列上应用 '7s' 的滚动窗口
# 4. 对每个窗口，使用 `shift(1)` 排除当前行，然后 `max()` 检查是否有 True
# 5. `eq(1)` 将结果转换为布尔类型
# 6. `reset_index()` 将多级索引转换为列，方便后续合并
# 7. `merge` 回原始DataFrame，使用临时重置的索引进行对齐
# 8. 恢复原始索引和顺序
out_exclude_current = (df.reset_index()
                       .merge(df.assign(is_2_in_7_sec=df['event'].eq(2))[::-1]
                               .groupby(df['team'])
                               .rolling('7s', on='timeStamp')
                               ['is_2_in_7_sec'].apply(lambda x: x.shift(1).max()).eq(1)
                               .reset_index(), how='left'
                             )
                       .set_index('index').reindex(df.index)
                      )

print("\n方案一：不包含当前行，在7秒内查找事件2:")
print(out_exclude_current)

代码解析：

• df.assign(is_2_in_7_sec=df['event'].eq(2)): 创建一个临时列 is_2_in_7_sec，如果 event 是 2，则为 True，否则为 False。
• [::-1]: 这一步是关键。它将整个DataFrame的行顺序反转。这样，当我们应用默认的回顾性 rolling 窗口时，它实际上会检查原始DataFrame中当前行“之后”的数据。
• .groupby(df['team']): 按照 team 分组，确保每个团队的事件独立处理。注意这里 df['team'] 是原始的 team 列，groupby 会根据其值进行分组。
• .rolling('7s', on='timeStamp'): 在每个 team 组内，基于 timeStamp 列创建时间窗口。'7s' 表示7秒的窗口。on='timeStamp' 指定了时间列。
• ['is_2_in_7_sec'].apply(lambda x: x.shift(1).max()):
  • x 代表当前滚动窗口中的 is_2_in_7_sec Series。
  • x.shift(1)：将窗口内的值向下移动一位。由于DataFrame是反转的，并且rolling是回顾性的，shift(1)实际上排除了在原始DataFrame中与当前行对应的那个元素，从而实现了“严格大于当前时间”的判断。
  • .max()：如果窗口内（排除当前行后）有任何 True 值，则返回 True。
• .eq(1)：将 max() 返回的布尔值（True/False）转换为 bool 类型。
• .reset_index(): groupby.rolling 会生成一个多级索引，reset_index() 将其转换为普通列，方便后续 merge。
• .merge(...): 将计算出的 is_2_in_7_sec 列合并回原始DataFrame。how='left' 确保所有原始行都被保留。
• .set_index('index').reindex(df.index): 恢复原始DataFrame的索引和行顺序。

3.2 方案二：包含当前行进行判断

如果需要包含当前行，即判断在当前时间戳及之后7秒内是否存在事件2，则无需使用 shift(1)。

# 1. 创建一个布尔列，标记 'event' 是否为 2
# 2. 将DataFrame反转 `[::-1]`
# 3. 按 'team' 分组，并在 'timeStamp' 列上应用 '7s' 的滚动窗口
# 4. 对每个窗口，直接使用 `max()` 检查是否有 True (包含当前行)
# 5. `astype(bool)` 确保结果为布尔类型
# 6. `reset_index()` 将多级索引转换为列
# 7. `merge` 回原始DataFrame
# 8. 恢复原始索引和顺序
out_include_current = (df.reset_index()
                       .merge(df.assign(is_2_in_7_sec=df['event'].eq(2))[::-1]
                               .groupby(df['team'])
                               .rolling('7s', on='timeStamp')
                               ['is_2_in_7_sec'].max().astype(bool)
                               .reset_index(), how='left'
                             )
                       .set_index('index').reindex(df.index)
                      )

print("\n方案二：包含当前行，在7秒内查找事件2:")
print(out_include_current)

代码解析（与方案一不同点）：

• ['is_2_in_7_sec'].max().astype(bool): 直接在窗口内查找 max()。由于 rolling 默认包含当前元素，此操作会检查当前行及“之后”（在原始DataFrame中）的7秒窗口内是否存在 True。.astype(bool) 确保结果是布尔类型。

4. 注意事项与总结

1. 时间戳类型: 确保 timeStamp 列是 datetime 类型，否则 rolling 无法正确处理时间窗口。
2. 前瞻性窗口: 通过 [::-1] 反转DataFrame，结合 rolling 的默认回顾性行为，可以巧妙地实现前瞻性时间窗口。
3. 排除当前行: shift(1) 是在反转后的窗口中排除原始DataFrame中当前行对应元素的关键。
4. 性能: 对于非常大的数据集，groupby.rolling 通常比自定义 apply 函数更高效，因为它在C语言级别进行了优化。
5. 灵活性: rolling 除了 max() 之外，还可以配合其他聚合函数（如 min, sum, count, mean）来执行更复杂的窗口计算。
6. 索引管理: 在使用 groupby.rolling 后，通常需要 reset_index() 来扁平化多级索引，并通过 merge 和 reindex 将结果正确地合并回原始DataFrame的结构和顺序。

通过上述方法，我们可以灵活且高效地在Pandas中处理复杂的组内时间窗口事件检测任务，这在金融分析、日志分析、传感器数据处理等领域都非常有用。

以上就是Pandas中高效实现组内时间窗口事件检测的详细内容，更多请关注php中文网其它相关文章！