如何在Pandas时序DataFrame中识别符合特定时长规则的事件

需求规则 #

• 事件指数值非零持续超过30秒的时段；
• 事件内允许包含0值，但连续0值的时长不能超过30秒；
• 若某段非零时长不足30秒且被0值完全包围，不视为事件；
• 事件结束于后续连续0值时长≥30秒的最后一个非零值。

实现思路 #

给定数据是固定5秒频率，所以时长可以直接用连续行数换算：30秒对应6行（5*6=30）。核心思路是先把连续的非零/零值分成独立块，再根据每个块的时长、前后块的类型和时长，判断哪些块属于事件，最后把标记映射回原DataFrame。

完整代码实现 #

import pandas as pd

# 构建示例数据
Timestamp = pd.date_range("11-30-2023 23:54:00", periods=63, freq="5s")
Value = [0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.0,0.0,0.0,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.0,0.0,0.5,0.5,0.5,0.5,0.5,0.5,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.5,0.5,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.5,0.0,0.0,0.0,0.0]
df = pd.DataFrame({"Timestamp": Timestamp, "Value": Value})

# 1. 标记非零值
df['is_non_zero'] = df['Value'] != 0

# 2. 给连续的非零/零值分配块ID（同一连续块ID相同）
df['block_id'] = (df['is_non_zero'] != df['is_non_zero'].shift()).cumsum()

# 3. 计算每个块的类型（非零/零）和持续时长（秒）
block_stats = df.groupby('block_id').agg(
    is_non_zero_block=('is_non_zero', 'first'),
    duration=('Timestamp', lambda x: (x.iloc[-1] - x.iloc[0]).seconds + 5)  # 加上频率5秒，确保首尾时间都计入
).reset_index()

# 4. 标记符合条件的事件非零块
block_stats['is_event'] = False

for idx in range(len(block_stats)):
    # 跳过零块，只处理非零块
    if not block_stats.loc[idx, 'is_non_zero_block']:
        continue
    
    current_duration = block_stats.loc[idx, 'duration']
    # 条件1：自身时长超过30秒，直接标记为事件块
    if current_duration > 30:
        block_stats.loc[idx, 'is_event'] = True
        continue
    
    # 找前后最近的非零块ID
    prev_non_zero_idx = None
    for j in range(idx-1, -1, -1):
        if block_stats.loc[j, 'is_non_zero_block']:
            prev_non_zero_idx = j
            break
    
    next_non_zero_idx = None
    for j in range(idx+1, len(block_stats)):
        if block_stats.loc[j, 'is_non_zero_block']:
            next_non_zero_idx = j
            break
    
    # 条件3：被零包围且时长不足30秒，跳过
    if prev_non_zero_idx is not None and next_non_zero_idx is not None and current_duration <= 30:
        continue
    
    # 检查前后是否有事件块，且中间零块时长≤30秒
    has_valid_prev = False
    if prev_non_zero_idx is not None:
        # 中间的零块ID
        zero_block_idx = prev_non_zero_idx + 1
        if zero_block_idx < idx:
            zero_duration = block_stats.loc[zero_block_idx, 'duration']
            if zero_duration <= 30 and block_stats.loc[prev_non_zero_idx, 'is_event']:
                has_valid_prev = True
    
    has_valid_next = False
    if next_non_zero_idx is not None:
        zero_block_idx = idx + 1
        if zero_block_idx < next_non_zero_idx:
            zero_duration = block_stats.loc[zero_block_idx, 'duration']
            if zero_duration <= 30:
                # 后续非零块要么是事件块，要么自身时长超30秒
                if block_stats.loc[next_non_zero_idx, 'duration'] > 30 or block_stats.loc[next_non_zero_idx, 'is_event']:
                    has_valid_next = True
    
    if has_valid_prev or has_valid_next:
        block_stats.loc[idx, 'is_event'] = True

# 5. 将事件块标记映射回原DataFrame
df = df.merge(block_stats[['block_id', 'is_event']], on='block_id', how='left')

# 6. 处理事件内的零块：前后都是事件非零块且零块时长≤30秒，标记为事件
df['Events'] = 0
# 先标记事件非零块的行
df.loc[df['is_event'], 'Events'] = 1

# 遍历零块，补充标记符合条件的行
for idx in df[df['Value'] == 0].index:
    # 找前面最近的非零行的事件标记
    prev_event = None
    for j in range(idx-1, -1, -1):
        if df.loc[j, 'Value'] != 0:
            prev_event = df.loc[j, 'Events']
            break
    
    # 找后面最近的非零行的事件标记
    next_event = None
    for j in range(idx+1, len(df)):
        if df.loc[j, 'Value'] != 0:
            next_event = df.loc[j, 'Events']
            break
    
    # 检查条件：前后都是事件，且当前零块时长≤30秒
    if prev_event == 1 and next_event == 1:
        current_block_id = df.loc[idx, 'block_id']
        zero_duration = block_stats[block_stats['block_id'] == current_block_id]['duration'].iloc[0]
        if zero_duration <= 30:
            df.loc[idx, 'Events'] = 1

# 验证结果是否与示例一致
print("结果匹配验证：", all(df['Events'] == [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,1,1,1,1,1,1,1,1,1,1,0,0,0,0]))

关键步骤说明 #

1. 分块标记：通过shift()比较当前行与前一行的非零状态，生成连续块的ID，把同一连续的零/非零值归为一个块，方便后续统计时长。
2. 块属性统计：计算每个块的类型和持续时长，这里用时间戳的差值加频率5秒，确保块的首尾时间都被计入。
3. 事件块判断：遍历每个非零块，分别检查自身时长、前后块的关联情况，排除被零包围的短非零块。
4. 零块补全标记：事件内允许包含短零块，所以需要把前后都是事件非零块且时长≤30秒的零块也标记为事件。
5. 结果验证：最后对比生成的Events列和示例，确保逻辑正确。

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