Python多条件Shift列表推导式优化：30万行数据集提速方案

问题背景 #

我是Python新手，编写了如下代码用于统计传感器故障次数：

data = {'deviceTimestamp' : ['2022-08-21 00:00:04.019345800','2022-08-21 00:00:09.666780800','2022-08-21 00:00:15.193316200','2022-08-21 00:00:20.708187','2022-08-21 00:00:26.212871600','2022-08-21 00:00:31.792147900','2022-08-21 00:00:37.345935900','2022-08-21 00:00:42.855598900','2022-08-21 00:00:48.380273','2022-08-21 00:00:53.894567100','2022-08-21 00:00:59.421452300','2022-08-21 00:01:04.994086700','2022-08-21 00:01:10.549778100','2022-08-21 00:01:16.067442300','2022-08-21 00:01:21.585345300','2022-08-21 00:01:27.123348','2022-08-21 00:01:32.657100400','2022-08-21 00:01:38.681526500','2022-08-21 00:01:44.254180800','2022-08-21 00:01:49.832247900','2022-08-21 00:01:55.404809800','2022-08-21 00:02:00.935254300','2022-08-21 00:02:06.456698200','2022-08-21 00:02:12.023842500','2022-08-21 00:02:17.601805600'], 'Failure_Sensor' : [0,1,1,1,0,0,0,1,0,0,0,1,1,1,0,0,0,0,1,0,1,1,1,0,0]}
df_failure_sensor = pd.DataFrame(data)
df_failure_sensor['Failure_Sensor_Transition'] = [(df_failure_sensor['Failure_Sensor'][i] if df_failure_sensor['Failure_Sensor'].shift(1)[i] != df_failure_sensor['Failure_Sensor'][i] else (df_failure_sensor['Failure_Sensor'][i] if((df_failure_sensor['Failure_Sensor'].shift(-1)[i] != df_failure_sensor['Failure_Sensor'][i])) else "")) for i in range(0, len(df_failure_sensor))]
df_failure_sensor = df_failure_sensor[df_failure_sensor['Failure_Sensor_Transition'] != ""].reset_index(drop=True)
df_failure_sensor["Start"] = [(str(df_failure_sensor['deviceTimestamp'][i]) if df_failure_sensor['Failure_Sensor'].shift(-1)[i] == df_failure_sensor['Failure_Sensor'][i] else "") for i in range(0, len(df_failure_sensor))]
number_of_faults = df_failure_sensor['Failure_Sensor'][(df_failure_sensor['Start'] != "") & (df_failure_sensor['Failure_Sensor'] == True)].count()

代码逻辑为：先通过shift方法分析Failure_Sensor列的状态转换，筛选转换节点；再生成Start列标记故障起始时间；最终统计故障次数。该代码在小数据集上运行正常，但在30万行数据集及多故障列场景下运行过慢，请问是否有高效的优化或替代实现方案？

优化思路与实现 #

原始代码的核心问题是用列表推导式逐行遍历DataFrame，完全浪费了Pandas的向量化操作优势，在大数据集下会产生极大的性能开销。下面是基于向量化操作的优化方案：

方案1：极简版（仅统计故障次数+起始时间） #

直接定位从0→1的状态转换点（即故障开始节点），无需生成中间过渡列：

import pandas as pd

data = {'deviceTimestamp' : ['2022-08-21 00:00:04.019345800','2022-08-21 00:00:09.666780800','2022-08-21 00:00:15.193316200','2022-08-21 00:00:20.708187','2022-08-21 00:00:26.212871600','2022-08-21 00:00:31.792147900','2022-08-21 00:00:37.345935900','2022-08-21 00:00:42.855598900','2022-08-21 00:00:48.380273','2022-08-21 00:00:53.894567100','2022-08-21 00:00:59.421452300','2022-08-21 00:01:04.994086700','2022-08-21 00:01:10.549778100','2022-08-21 00:01:16.067442300','2022-08-21 00:01:21.585345300','2022-08-21 00:01:27.123348','2022-08-21 00:01:32.657100400','2022-08-21 00:01:38.681526500','2022-08-21 00:01:44.254180800','2022-08-21 00:01:49.832247900','2022-08-21 00:01:55.404809800','2022-08-21 00:02:00.935254300','2022-08-21 00:02:06.456698200','2022-08-21 00:02:12.023842500','2022-08-21 00:02:17.601805600'], 'Failure_Sensor' : [0,1,1,1,0,0,0,1,0,0,0,1,1,1,0,0,0,0,1,0,1,1,1,0,0]}
df = pd.DataFrame(data)

# 标记所有从0→1的转换点（故障开始）
fault_starts = (df['Failure_Sensor'] == 1) & (df['Failure_Sensor'].shift(1) == 0)

# 统计故障次数
number_of_faults = fault_starts.sum()

# 提取所有故障起始时间
fault_start_times = df.loc[fault_starts, 'deviceTimestamp'].tolist()

方案2：保留状态转换节点（如需分析故障结束时间） #

如果需要同时查看故障的起止节点，可以先标记所有状态变化的行，再筛选故障起始点：

import pandas as pd

df = pd.DataFrame(data)

# 标记所有状态变化的行：当前行与上一行/下一行状态不同
df['state_change'] = (df['Failure_Sensor'] != df['Failure_Sensor'].shift(1)) | (df['Failure_Sensor'] != df['Failure_Sensor'].shift(-1))

# 筛选状态转换节点
transition_df = df[df['state_change']].copy().reset_index(drop=True)

# 标记故障起始节点：当前状态为1，且下一个状态仍为1
transition_df['is_fault_start'] = (transition_df['Failure_Sensor'] == 1) & (transition_df['Failure_Sensor'] == transition_df['Failure_Sensor'].shift(-1))

# 统计故障次数
number_of_faults = transition_df['is_fault_start'].sum()

多故障列场景扩展 #

如果存在多个故障监测列（如Failure_Sensor1、Failure_Sensor2），可以批量处理：

# 筛选所有故障列
fault_columns = [col for col in df.columns if col.startswith('Failure_Sensor')]

fault_stats = {}
for col in fault_columns:
    # 统计该列的故障次数与起始时间
    fault_starts = (df[col] == 1) & (df[col].shift(1) == 0)
    fault_stats[col] = {
        'count': fault_starts.sum(),
        'start_times': df.loc[fault_starts, 'deviceTimestamp'].tolist()
    }

# 输出结果
for col, stats in fault_stats.items():
    print(f"{col} 故障次数：{stats['count']}")
    print(f"故障起始时间：{stats['start_times']}\n")

性能提升原因 #

• 向量化操作：所有计算基于Pandas列级运算，避免了Python循环的逐行遍历开销，30万行数据的处理时间可从分钟级降至毫秒级。
• 减少中间数据：极简版直接定位目标节点，无需生成冗余过渡列和表格，内存占用更低。

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