使用pandas筛选传感器时序数据中u_code序列前后值不同的记录

核心逻辑说明 #

输入为已按sensor_id分组、组内按时间戳ts升序排列的DataFrame，包含sensor_id/u_code/ts三个字段，按以下规则筛选：

• 仅在相同sensor_id范围内识别连续u_code取值为1、2、3的序列段，跨传感器记录不做关联
• 对每个目标序列段，匹配紧邻的前序第一条非1/2/3的u_code值、紧邻的后序第一条非1/2/3的u_code值
• 若目标段前后非目标u_code值相等，过滤该段及关联的前后两条非目标记录；若值不等（含段位于序列首/尾无对应前/后记录的情况），保留前序非目标记录、整个目标连续段、后序非目标记录

实现代码 #

import pandas as pd
import numpy as np

def filter_sensor_data(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    # 初始化保留标记，默认所有记录不保留
    df['keep_flag'] = False

    # 按传感器分组逐组处理，避免跨传感器关联
    for sensor_id, group in df.groupby('sensor_id', sort=False):
        group_idx = group.index
        record_count = len(group)
        # 标记单条记录是否属于目标u_code范围
        is_target = group['u_code'].isin([1, 2, 3]).values
        # 为连续目标段分配唯一段ID
        segment_id = np.zeros(record_count, dtype=int)
        current_seg_id = 0
        in_target_seg = False
        for i in range(record_count):
            if is_target[i] and not in_target_seg:
                current_seg_id += 1
                in_target_seg = True
            if not is_target[i]:
                in_target_seg = False
            segment_id[i] = current_seg_id if in_target_seg else 0
        
        # 逐段校验保留规则
        for seg_id in range(1, current_seg_id + 1):
            seg_pos = np.where(segment_id == seg_id)[0]
            seg_start = seg_pos[0]
            seg_end = seg_pos[-1]
            # 定位前后紧邻非目标记录位置
            prev_record_pos = seg_start - 1
            next_record_pos = seg_end + 1

            # 边界情况（段在序列首/尾无对应前后记录）用不可能相等的哨兵值填充
            prev_u_code = group.iloc[prev_record_pos]['u_code'] if prev_record_pos >= 0 else None
            next_u_code = group.iloc[next_record_pos]['u_code'] if next_record_pos < record_count else None

            # 前后值不相等则标记所有关联记录为保留
            if prev_u_code != next_u_code:
                if prev_record_pos >= 0:
                    df.loc[group_idx[prev_record_pos], 'keep_flag'] = True
                df.loc[group_idx[seg_start:seg_end+1], 'keep_flag'] = True
                if next_record_pos < record_count:
                    df.loc[group_idx[next_record_pos], 'keep_flag'] = True
    
    # 过滤结果并删除辅助列
    result = df[df['keep_flag']].drop(columns=['keep_flag']).reset_index(drop=True)
    return result

效果验证 #

对应规则示例构造测试数据：

test_df = pd.DataFrame([
    # 首段单条u=1，前后u_code均为5，需过滤
    {'sensor_id':'abcd', 'u_code':5, 'ts':1},
    {'sensor_id':'abcd', 'u_code':1, 'ts':2},
    {'sensor_id':'abcd', 'u_code':5, 'ts':3},
    # 后续连续3条u=1，前序u=6、后序u=8，需保留
    {'sensor_id':'abcd', 'u_code':6, 'ts':4},
    {'sensor_id':'abcd', 'u_code':1, 'ts':5},
    {'sensor_id':'abcd', 'u_code':1, 'ts':6},
    {'sensor_id':'abcd', 'u_code':1, 'ts':7},
    {'sensor_id':'abcd', 'u_code':8, 'ts':8},
])

print(filter_sensor_data(test_df))

运行后仅返回ts为4-8的5条记录，首段3条记录被过滤，与规则预期一致。u_code为2、3的连续序列会按相同逻辑自动校验。

超大数据集场景可将逐段循环逻辑替换为pandas向量化操作提升运行效率，上述代码优先保证规则匹配准确性与可读性。

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