如何以Pythonic方式批量校验DataFrame中日期间计数差异

DevOps 小满

2026-6-18

问题背景

假设我有如下数据：

data = {'site': ['ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY', 'ACY'],
       'usage_date': ['2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-08-25', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-01'],
       'item_id': ['COR30013', 'PAC10463', 'COR30018', 'PAC10958', 'PAC11188', 'PAC20467', 'COR20275', 'PAC20702', 'COR30020', 'PAC10137', 'PAC10445', 'COR30029', 'COR30025', 'PAC10457', 'COR10746', 'PAC11136', 'COR10346', 'PAC11050', 'PAC11132', 'PAC11135', 'PAC10964', 'COR10439', 'PAC11131', 'COR10695', 'PAC11128', 'COR10433', 'COR10432', 'PAC11051', 'PAC10137', 'COR10695', 'COR30029', 'COR10346', 'COR10432', 'COR10746', 'COR10439', 'COR10433', 'COR20275', 'COR30020', 'COR30018', 'PAC11135', 'PAC10964', 'PAC11136', 'PAC10445', 'PAC11050', 'PAC11132', 'PAC20467', 'PAC11188', 'PAC10463', 'PAC20702', 'PAC10457', 'PAC10958', 'PAC11051', 'PAC11128', 'PAC11131'],
       'start_count':[400.0, 96000.0, 315.0, 45000.0, 2739.0, 2232.0, 2800.0, 283500.0, 280.0, 200000.0, 96000.0, 481.0, 600.0, 18000.0, 400.0, 5500.0, 1200.0, 5850.0, 5500.0, 5500.0, 36000.0, 600.0, 5500.0, 550.0, 300.0, 4800.0, 1800.0, 1800.0, 108000.0, 500.0, 481.0, 1200.0, 1800.0, 400.0, 600.0, 3300.0, 2800.0, 455.0, 315.0, 5500.0, 36000.0, 5500.0, 96000.0, 5400.0, 5500.0, 2232.0, 2739.0, 96000.0, 283500.0, 18000.0, 72000.0, 1800.0, 300.0, 5500.0],
       'received_total': [0.0, 0.0, 0.0, 0.0, 3168.0, 0.0, 0.0, 0.0, 280.0, 0.0, 0.0, 0.0, 0.0, 0.0, 400.0, 0.0, 1800.0, 0.0, 0.0, 0.0, 0.0, 400.0, 0.0, 0.0, 0.0, 0.0, 0.0, 3600.0, 0.0, 0.0, 0.0, 1800.0, 2400.0, 400.0, 400.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1800.0, 0.0, 0.0, 3168.0, 0.0, 0.0, 0.0, 45000.0, 3600.0, 0.0, 0.0],
       'end_count': [240.0, 84000.0, 280.0, 27000.0, 3432.0, 2160.0, 2000.0, 90000.0, 455.0, 108000.0, 96000.0, 437.0, 500.0, 9000.0, 600.0, 5500.0, 1950.0, 4950.0, 5500.0, 5500.0, 36000.0, 600.0, 5500.0, 550.0, 270.0, 3300.0, 1200.0, 4200.0, 192000.0, 450.0, 350.0, 1890.0, 3600.0, 600.0, 525.0, 2835.0, 1600.0, 420.0, 187.0, 5500.0, 36000.0, 5500.0, 96000.0, 6750.0, 5500.0, 1992.0, 1881.0, 84000.0, 58500.0, 9000.0, 85500.0, 3300.0, 252.0, 5500.0]}

df_sample = pd.DataFrame(data=data)

校验规则：针对每个item_id，校验当前日期（如2019-09-01）的end_count是否大于上一日期（如2019-08-25）的end_count，且当前received_total为0，满足则判定为计数异常。

现有实现

已有可运行代码，但不够Pythonic：

def check_end_count(df):
    l = []
    for loc, df_loc in df.groupby(['site', 'item_id']):
        try:
            ending_count_previous = df_loc['end_count'].iloc[0]
            ending_count_current = df_loc['end_count'].iloc[1]
            received_total_current = df_loc['received_total'].iloc[1]
            
            if ending_count_current > ending_count_previous and received_total_current == 0:
                l.append("Ending count discrepancy")
                l.append("Ending count discrepancy")
            else:
                l.append("Good Row")
                l.append("Good Row")
        except:
            l.append("Nothing to compare")

    df['ending_count_check'] = l
    return df

df_sample = check_end_count(df_sample)

扩展需求

需要处理一组滑动窗口日期对，示例如下：

print(sliding_window_dates[:3])

[array(['2019-08-25', '2019-09-01'], dtype=object),
 array(['2019-09-01', '2019-09-08'], dtype=object),
 array(['2019-09-08', '2019-09-15'], dtype=object)]

当前用两层循环实现批量校验：

df_list = []
for date1, date2 in sliding_window_dates:
    df_check = df_test[(df_test['usage_date'] == date1) | (df_test['usage_date'] == date2)]
    for loc, df_loc in df_check.groupby(['sort_center', 'item_id']):
        df_list.append(check_end_count(df_loc))

但这种方式效率较低，寻求更Pythonic、更高效的实现方案。

优化方案

1. 单个日期对的高效处理

利用pandas的groupby+shift方法，替代手动循环分组：

def check_end_count_optimized(df):
    # 按site、item_id、usage_date排序，确保日期顺序正确
    df_sorted = df.sort_values(['site', 'item_id', 'usage_date']).reset_index(drop=True)
    
    # 分组后获取上一日期的end_count和日期标识
    df_sorted['prev_end_count'] = df_sorted.groupby(['site', 'item_id'])['end_count'].shift(1)
    df_sorted['has_prev'] = df_sorted.groupby(['site', 'item_id'])['usage_date'].shift(1).notna()
    
    # 构造校验条件
    anomaly_condition = (df_sorted['end_count'] > df_sorted['prev_end_count']) & \
                        (df_sorted['received_total'] == 0) & \
                        df_sorted['has_prev']
    
    # 赋值校验结果
    df_sorted['ending_count_check'] = df_sorted.apply(
        lambda row: "Ending count discrepancy" if anomaly_condition.loc[row.name] 
        else ("Good Row" if row['has_prev'] else "Nothing to compare"),
        axis=1
    )
    
    return df_sorted

该方案用pandas内置的向量化操作替代手动遍历，大幅降低Python层级的循环开销。

2. 滑动窗口日期对的批量处理

无需逐个筛选日期对，先一次性计算全量相邻日期的校验结果，再筛选出属于滑动窗口的记录：

def batch_check_sliding_windows(df, sliding_window_dates):
    # 转换日期格式，确保排序准确性
    df['usage_date'] = pd.to_datetime(df['usage_date'])
    # 按site、item_id、usage_date排序
    df_sorted = df.sort_values(['site', 'item_id', 'usage_date']).reset_index(drop=True)
    
    # 计算相邻日期的校验字段
    df_sorted['prev_end_count'] = df_sorted.groupby(['site', 'item_id'])['end_count'].shift(1)
    df_sorted['prev_date'] = df_sorted.groupby(['site', 'item_id'])['usage_date'].shift(1)
    df_sorted['has_prev'] = df_sorted['prev_date'].notna()
    
    # 构造异常校验条件
    anomaly_condition = (df_sorted['end_count'] > df_sorted['prev_end_count']) & \
                        (df_sorted['received_total'] == 0) & \
                        df_sorted['has_prev']
    
    # 赋值校验结果
    df_sorted['ending_count_check'] = df_sorted.apply(
        lambda row: "Ending count discrepancy" if anomaly_condition.loc[row.name] 
        else ("Good Row" if row['has_prev'] else "Nothing to compare"),
        axis=1
    )
    
    # 转换滑动窗口日期对为datetime格式，方便匹配
    window_pairs = [(pd.to_datetime(d1), pd.to_datetime(d2)) for d1, d2 in sliding_window_dates]
    
    # 筛选出滑动窗口内的记录（包含当前日期和对应的上一日期）
    current_window_mask = df_sorted.apply(
        lambda row: (row['prev_date'], row['usage_date']) in window_pairs,
        axis=1
    )
    prev_window_mask = df_sorted.apply(
        lambda row: (row['usage_date'], df_sorted.loc[row.name+1, 'usage_date']) in window_pairs 
        if row.name < len(df_sorted)-1 else False,
        axis=1
    )
    
    result_df = df_sorted[current_window_mask | prev_window_mask].reset_index(drop=True)
    return result_df

该方案的核心是一次计算全量相邻日期的校验结果，避免了多次循环筛选和分组操作，大幅提升处理效率。

关键优化点

用pandas向量化操作替代手动循环，减少Python层级的遍历开销
提前统一排序并处理全量数据，避免重复分组计算
减少多次切片筛选数据的操作，降低内存IO开销

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

如何以Pythonic方式批量校验DataFrame中日期间计数差异

问题背景 #

现有实现 #

扩展需求 #

优化方案 #

1. 单个日期对的高效处理 #

2. 滑动窗口日期对的批量处理 #

关键优化点 #