如何替换Dataframe各组件分组下Level列误报值且保留对应首行原值

核心思路 #

按CompName字段分组独立处理每个设备的时间序列：

• 每组首行Level值直接保留，不参与异常替换逻辑
• 对组内剩余数据，识别连续相同Level的片段，将长度≤2的短片段判定为误报，替换为相邻长片段的取值

可运行代码 #

import pandas as pd

# 示例数据构造，实际使用时替换为你自己的DataFrame即可
df = pd.DataFrame({
    'Timestamp': pd.date_range(start='2024-01-01', periods=10, freq='1min'),
    'CompName': ['A']*5 + ['B']*5,
    'Level': ['Off', 'Off', 'Off', 'On', 'Off', 'On', 'Off', 'On', 'On', 'On']
})

def clean_level_sequence(group):
    # 单独保留分组首行，不参与替换
    first_row = group.iloc[[0]]
    rest_rows = group.iloc[1:].copy()
    if len(rest_rows) == 0:
        return first_row
    
    # 标记连续相同Level的区块
    rest_rows['block_id'] = (rest_rows['Level'] != rest_rows['Level'].shift()).cumsum()
    # 计算每个连续区块的长度
    block_length = rest_rows.groupby('block_id')['Level'].transform('count')
    
    # 标记长度<=2的孤立异常区块
    rest_rows.loc[block_length <= 2, 'Level'] = pd.NA
    # 先向前填充再向后填充，覆盖所有异常位置
    rest_rows['Level'] = rest_rows['Level'].ffill().bfill()
    
    # 清理临时字段，合并首行和处理后的剩余行
    rest_rows = rest_rows.drop(columns=['block_id'])
    return pd.concat([first_row, rest_rows], ignore_index=False)

# 按设备分组处理后，恢复原始时间排序
df_result = df.groupby('CompName', group_keys=False)\
              .apply(clean_level_sequence)\
              .sort_values('Timestamp')\
              .reset_index(drop=True)

参数调整说明 #

• 若需要调整异常判定的长度阈值，修改block_length <= 2中的数值即可，比如要过滤最长3个的孤立值，改为block_length <=3
• 填充逻辑采用先向前再向后的顺序，可同时覆盖序列开头、中间、结尾的异常片段，不会产生空值
• 全程保留原始Timestamp的顺序，不会打乱原有时间序列结构

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