如何将IPR异常值检测结果同步至百万行级原DataFrame

原代码存在的问题 #

1. 效率极低：双重循环遍历product和brick的唯一值，在百万行数据场景下会产生大量重复索引查找，运行速度极慢
2. 位置匹配错误：outliers[0]是当前分组内的相对行号（如0、1、2...），并非原DataFrame的索引，用df.index.isin(outliers[0])无法匹配到正确行
3. 异常隐藏：空except会掩盖所有错误（比如分组数据量不足无法拟合回归的情况），难以排查问题

优化实现：用groupby替代循环 #

借助pandas的groupby做分组处理，既能保证效率，又能自动将异常值标记映射回原DataFrame，无需手动处理索引匹配。

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression

def mark_group_outliers(group):
    units_data = group['units'].values.reshape(-1, 1)
    # 分组数据量不足2个时，无法做线性回归，直接标记为非异常值
    if len(units_data) < 2:
        group['is_outlier'] = False
        return group
    
    # 标准化数据
    scaler = StandardScaler()
    scaled_data = scaler.fit_transform(units_data)
    
    # 拟合线性回归模型
    X = np.arange(len(scaled_data)).reshape(-1, 1)
    reg_model = LinearRegression()
    reg_model.fit(X, scaled_data)
    
    # 计算残差
    residuals = scaled_data - reg_model.predict(X)
    
    # 2倍标准差作为异常值阈值
    threshold = 2 * residuals.std()
    # 标记异常值
    group['is_outlier'] = np.abs(residuals) > threshold
    
    return group

# 按product和brick分组处理，自动合并结果到原DataFrame
df = df.groupby(['product', 'brick'], group_keys=False).apply(mark_group_outliers)

关键优势 #

• 高效处理大规模数据：groupby内部采用向量化操作，比手动循环快10-100倍，完全适配百万行级别的DataFrame
• 自动映射索引：分组函数内直接操作原DataFrame的子集（保留原索引），处理后apply自动合并，无需手动匹配行位置
• 透明的异常处理：明确判断分组数据量，避免无意义的报错，同时保留可排查的空间

内容的提问来源于stack exchange，提问作者Maksim .Levin