Python DataFrame缺失数据识别的代码优化方案咨询
2026-7-26
问题
现有如下结构的DataFrame：
location_id device_id phase timestamp date day_no hour quarter_hour data_received data_sent 10001 1001 Phase 1 2023-01-30 00:00:00 2023-01-30 1 0 00:00:00 150 98 10001 1001 Phase 1 2023-01-30 00:15:00 2023-01-30 1 0 00:15:00 130 101 10001 1001 Phase 1 2023-01-30 00:45:00 2023-01-30 1 0 00:45:00 121 75 10001 1001 Phase 1 2023-01-30 01:00:00 2023-01-30 1 1 01:00:00 104 110 10001 1001 Phase 1 2023-01-30 01:15:00 2023-01-30 1 1 01:15:00 85 79 10001 1001 Phase 1 2023-01-30 01:30:00 2023-01-30 1 1 01:45:00 127 123 . . . . . . . . . . 10001 1001 Phase 1 2023-02-03 23:30:00 2023-02-03 5 23 23:30:00 100 83 10001 1001 Phase 1 2023-02-03 23:45:00 2023-02-03 5 23 23:45:00 121 75 10001 1005 Phase 2 2023-02-15 02:15:00 2023-02-15 1 2 02:15:00 90 101 10001 1005 Phase 2 2023-02-15 02:30:00 2023-02-15 1 2 02:30:00 111 98 . . . . . . . . . . 10001 1005 Phase 2 2023-02-19 23:15:00 2023-02-19 5 23 23:15:00 154 76 10001 1005 Phase 2 2023-02-19 23:30:00 2023-02-19 5 23 23:30:00 97 101 10003 1010 Phase 2 2023-01-14 00:00:00 2023-01-14 1 0 00:00:00 112 87 10003 1010 Phase 2 2023-01-14 00:15:00 2023-01-14 1 0 00:15:00 130 101 10003 1010 Phase 2 2023-01-14 00:30:00 2023-01-14 1 0 00:30:00 89 91 . . . . . . . . . . 10003 1010 Phase 2 2023-01-18 23:45:00 2023-01-18 5 23 23:45:00 123 117
需求说明：
- 每个地点包含多个按顺序推进的阶段，每个阶段分配对应设备且持续5天，期间每15分钟生成一条设备状态数据。
- 需要识别DataFrame中的缺失行，但要排除设备未开启时段的缺失；同时需识别整个阶段数据未上传的情况。
现有代码可实现需求，但运行速度极慢，性能瓶颈为以下代码行：
day_quarter_hour_data_check = quarter_hourly_data_df[(quarter_hourly_data_df['location_id'].isin([location_id])) & (quarter_hourly_data_df['phase'].isin([phase])) & (quarter_hourly_data_df['day_no'].isin([cur_day])) & (quarter_hourly_data_df['quarter_hour'].isin([quarter_hour]))]['timestamp']
完整代码如下：
quarter_hourly_data_df = pd.read_csv('location_quarter_hourly_data.csv') quarter_hourly_data_df = quarter_hourly_data_df.astype({'timestamp':'datetime64[ns]'}) quarter_hourly_data_df['quarter_hour'] = quarter_hourly_data_df['timestamp'].dt.time start_times_data_df = quarter_hourly_data_df.groupby(['location_id','device_id','phase']).agg(start_time=pd.NamedAgg(column="timestamp", aggfunc="min"),day_no=pd.NamedAgg(column="day_no", aggfunc="min")).reset_index() start_times_data_df['start_time'] = start_times_data_df['start_time'].astype('datetime64[ns]') start_times_data_df['quarter_hour'] = start_times_data_df['start_time'].dt.time location_ids = start_times_data_df['location_id'].unique() location_ids.sort() phases = start_times_data_df['phase'].unique() phases.sort() location_phases_df = quarter_hourly_data_df[['location_id','phase']].drop_duplicates() location_last_phase_df = location_phases_df.groupby(['location_id']).agg(last_phase=pd.NamedAgg(column='phase', aggfunc='max')).reset_index() location_site_df = location_df[['site_id','location_id']].drop_duplicates() location_site_df.rename(columns={"site_id": "site_no"}, inplace = True) quarter_hours= pd.date_range("00:00", "23:45", freq="15min").time location_phase_quarter_hour_missing_data = [] for location_id in location_ids: location_last_phase = location_last_phase_df[(location_last_phase_df['location_id'].isin([location_id]))]['last_phase'].values[0] for phase in phases: if int(phase[6]) <= int(location_last_phase[6]): # check if phase exists or not location_phase_data_check = start_times_data_df[(start_times_data_df['location_id'].isin([location_id])) & (start_times_data_df['phase'].isin([phase]))]['start_time'] # if phase exists get start time of phase if not location_phase_data_check.empty: location_start_day = start_times_data_df[(start_times_data_df['location_id'].isin([location_id])) & (start_times_data_df['phase'].isin([phase]))]['day_no'].values[0] location_start_time = start_times_data_df[(start_times_data_df['location_id'].isin([location_id])) & (start_times_data_df['phase'].isin([phase]))]['quarter_hour'].values[0] for day_no in range(5): cur_day = day_no+1 for quarter_hour in quarter_hours: if ((quarter_hour >= location_start_time) and (cur_day >= location_start_day)) : #check if data exists for every quarter hour after start time day_quarter_hour_data_check = quarter_hourly_data_df[(quarter_hourly_data_df['location_id'].isin([location_id])) & (quarter_hourly_data_df['phase'].isin([phase])) & (quarter_hourly_data_df['day_no'].isin([cur_day])) & (quarter_hourly_data_df['quarter_hour'].isin([quarter_hour]))]['timestamp'] # if data doesn't exist add row to missing data if day_quarter_hour_data_check.empty: location_phase_quarter_hour_missing_data.append({'location_id':location_id, 'phase':phase, 'day_no':cur_day, 'hour':pd.to_datetime(quarter_hour, format='%H:%M:%S').hour, 'quarter_hour':quarter_hour}) else: #if phase doesn't exist add all rows to missing data for day_no in range(5): cur_day = day_no+1 for quarter_hour in quarter_hours: location_phase_quarter_hour_missing_data.append({'location_id':location_id, 'phase':phase, 'day_no':cur_day, 'hour':pd.to_datetime(quarter_hour, format='%H:%M:%S').hour, 'quarter_hour':quarter_hour}) location_phase_missing_data_df = pd.DataFrame(location_phase_quarter_hour_missing_data) location_phase_missing_data_df.to_csv('location_phase_missing_rows.csv', index=False)
请求：是否有更快的缺失行识别方法，或可优化现有代码的方案？
优化方案
核心思路
原代码性能瓶颈在于三层嵌套循环+循环内多次全表切片查询，时间复杂度极高。优化方向是用向量化操作和预构建完整预期序列替代循环查询，利用Pandas的索引与合并快速定位缺失行。
具体优化措施
- 预构建完整预期序列：为每个有效(location_id, phase)组合生成应有的所有15分钟时间点记录（排除未开启时段），再与原始数据左连接，空值行即为缺失。
- 复合索引提速：为原始数据的关键查询列设置复合索引，大幅降低合并与查找耗时。
- 避免重复查询：将阶段起始信息整合进DataFrame，避免循环中重复提取数据。
优化后代码
import pandas as pd # 1. 读取并预处理数据 quarter_hourly_data_df = pd.read_csv('location_quarter_hourly_data.csv') quarter_hourly_data_df['timestamp'] = pd.to_datetime(quarter_hourly_data_df['timestamp']) quarter_hourly_data_df['quarter_hour'] = quarter_hourly_data_df['timestamp'].dt.time # 2. 获取阶段起始信息（按location_id和phase分组，无需device_id） start_times_data_df = quarter_hourly_data_df.groupby(['location_id', 'phase']).agg( start_day=pd.NamedAgg(column='day_no', aggfunc='min'), start_quarter_hour=pd.NamedAgg(column='quarter_hour', aggfunc='min') ).reset_index() # 3. 过滤每个地点的有效阶段（不超过该地点的最大阶段） location_phases_df = quarter_hourly_data_df[['location_id', 'phase']].drop_duplicates() location_last_phase_df = location_phases_df.groupby('location_id').agg( last_phase=pd.NamedAgg(column='phase', aggfunc='max') ).reset_index() # 提取阶段编号用于数值比较 location_last_phase_df['last_phase_num'] = location_last_phase_df['last_phase'].str.extract(r'(\d+)').astype(int) location_phases_df['phase_num'] = location_phases_df['phase'].str.extract(r'(\d+)').astype(int) # 生成所有需要检查的(location_id, phase)组合 valid_location_phases = pd.merge( location_phases_df, location_last_phase_df, on='location_id' ).query('phase_num <= last_phase_num')[['location_id', 'phase']] # 4. 生成全量15分钟时间点与5天的组合 quarter_hours = pd.date_range("00:00", "23:45", freq="15min").time days = pd.DataFrame({'day_no': range(1, 6)}) time_day_comb = pd.DataFrame( [(qh, d) for qh in quarter_hours for d in days['day_no']], columns=['quarter_hour', 'day_no'] ) time_day_comb['hour'] = time_day_comb['quarter_hour'].apply(lambda x: pd.to_datetime(str(x)).hour) # 5. 生成每个有效组合的预期完整序列 expected_full_df = valid_location_phases.merge(time_day_comb, how='cross') expected_full_df = expected_full_df.merge( start_times_data_df, on=['location_id', 'phase'], how='left' ) # 6. 过滤设备未开启的时段 def filter_inactive(row): if pd.isna(row['start_day']): return True # 阶段无数据，全时段算缺失 if row['day_no'] > row['start_day']: return True elif row['day_no'] == row['start_day']: return row['quarter_hour'] >= row['start_quarter_hour'] else: return False expected_full_df['is_valid'] = expected_full_df.apply(filter_inactive, axis=1) expected_full_df = expected_full_df[expected_full_df['is_valid']].drop(columns=['start_day', 'start_quarter_hour', 'is_valid']) # 7. 为原始数据设置复合索引，提升合并效率 quarter_hourly_data_df = quarter_hourly_data_df.set_index(['location_id', 'phase', 'day_no', 'quarter_hour']) # 8. 左连接找到缺失行 missing_df = expected_full_df.merge( quarter_hourly_data_df.reset_index()[['location_id', 'phase', 'day_no', 'quarter_hour']], on=['location_id', 'phase', 'day_no', 'quarter_hour'], how='left', indicator=True ).query('_merge == "left_only"').drop(columns=['_merge']) # 9. 处理完全缺失的阶段（无任何数据的阶段） existing_phases = quarter_hourly_data_df.index.get_level_values(['location_id', 'phase']).unique() missing_phases = valid_location_phases[ ~valid_location_phases.set_index(['location_id', 'phase']).index.isin(existing_phases) ] missing_phases_full = missing_phases.merge(time_day_comb, how='cross') missing_phases_full['hour'] = missing_phases_full['quarter_hour'].apply(lambda x: pd.to_datetime(str(x)).hour) # 10. 合并两种缺失结果并保存 final_missing_df = pd.concat([missing_df, missing_phases_full], ignore_index=True) final_missing_df.to_csv('location_phase_missing_rows.csv', index=False)
效果说明
- 彻底消除三层嵌套循环，时间复杂度从O(NMK)降至O(N)级别，处理大数据集时速度提升显著。
- 复合索引将合并操作的查找效率提升数倍。
- 逻辑更清晰，减少了重复查询和条件判断的冗余操作。
内容的提问来源于stack exchange，提问作者Ram
相关产品推荐
相关产品推荐
