如何实现支持单列/全列的DataFrame异常值检测函数？

我现在有一个仅支持处理单列的find_outliers函数，用来检测DataFrame中的异常值，代码如下：

import pandas as pd
import datetime
from IPython.display import display

# 创建5-6行间隔1分钟的时间序列列表
date_list = [datetime.datetime(2023, 1, 1, 0, 0) + datetime.timedelta(minutes=i) for i in range(5, 11)]

# 构造包含时间列和两个数值列的DataFrame
data = {
    'bucket': date_list,
    'ophPumping': [1, 2, 3, 30, 20, 5],
    'ophNonPumping': [1, 2, 3, 20, 4, 5]
}

df = pd.DataFrame(data)

# 将'bucket'设为索引
df.set_index('bucket', inplace=True)


# 移除无效数据
df = df.dropna()
outliers = pd.DataFrame()

# 仅支持处理指定单列的函数
def find_outliers(df, columns:str) -> pd.DataFrame:
    start = df.shape[0]
    while (True):    
        df['DifferencePumping'] = pd.to_timedelta(df['ophPumping'].diff(), unit="h")
        df['DifferencePumpingShifted'] = df['DifferencePumping'].shift(-1)
        df['bucket'] = df.index

        df['DifferenceBucket'] = df["bucket"].diff()
        df['DifferenceBucketShifted'] = df['DifferenceBucket'].shift(-1)
    
        df["decreasing"] = df["DifferencePumping"] < pd.Timedelta(0)
        df["rapid_increase"] = df["DifferenceBucketShifted"] > df["DifferencePumpingShifted"] - pd.Timedelta(seconds=6)

        df["potential_outliers"] = df.decreasing | df.rapid_increase   

        try:
            outliers = pd.concat([outliers, df.loc[df.potential_outliers].head(1)])
            df = df.drop(df.loc[df.potential_outliers].index[0])
        
        except IndexError as index_error:
            print("Filtered all outliers.")
            break

    # 移除临时生成的列
    df = df.drop(columns=['DifferencePumping', 'DifferencePumpingShifted', "bucket", "DifferenceBucket", "DifferenceBucketShifted", "decreasing", "rapid_increase", "potential_outliers"])

    display(df)
    print()
    print("Removed outliers:")
    display(outliers)
    print()
    print('Head of cleaned df')
    display(df.head(5))
    print()
    print(f"Deleted {start - df.shape[0]} rows.")
    return df

我希望能通过df = remove_outliers(df, 指定列或全部列)的方式调用函数，让它支持指定单列，或者自动遍历DataFrame的全部列进行异常值检测。我尝试用for循环实现，但没能正确移除异常值，代码如下：

from IPython.display import display
import pandas as pd

def remove_outliers(data_frame, columns=None) -> pd.DataFrame:
    if columns is None:
        # 如果未指定列，使用所有列
        columns = data_frame.columns.tolist()

    # 移除无效数据
    data_frame = data_frame.dropna(subset=columns)
    outliers = pd.DataFrame()

    start = data_frame.shape[0]
    while True:
        for column in columns:
            data_frame['DifferencePumping'] = pd.to_timedelta(data_frame[column].diff(), unit="h")
            data_frame['DifferencePumpingShifted'] = data_frame['DifferencePumping'].shift(-1)
            data_frame['Bucket'] = data_frame.index
            data_frame['DifferenceBucket'] = data_frame["Bucket"].diff()
            data_frame['DifferenceBucketShifted'] = data_frame['DifferenceBucket'].shift(-1)

            data_frame["decreasing"] = data_frame["DifferencePumping"] < pd.Timedelta(0)
            data_frame["rapid_increase"] = data_frame["DifferenceBucketShifted"] < data_frame["DifferencePumpingShifted"] - pd.Timedelta(seconds=6)

            data_frame["potential_outliers"] = data_frame.decreasing | data_frame.rapid_increase

            try:
                outliers = pd.concat([outliers, data_frame.loc[data_frame.potential_outliers].head(1)])
                data_frame = data_frame.drop(data_frame.loc[data_frame.potential_outliers].index[0])

            except IndexError as index_error:
                print("Filtered all outliers.")
                break

        # 移除临时生成的列
        data_frame = data_frame.drop(columns=['DifferencePumping', 'DifferencePumpingShifted', "Bucket", "DifferenceBucket", "DifferenceBucketShifted", "decreasing", "rapid_increase", "potential_outliers"])

        display(data_frame)
        print()
        print("Removed outliers:")
        display(outliers)
        print()
        print('Head of cleaned df')
        display(data_frame.head(5))
        print()
        print(f"Deleted {start - data_frame.shape[0]} rows.")

        return data_frame

请问该如何解决这个问题？

你的代码主要存在三个核心问题：

• 循环逻辑混乱：while True与for column嵌套顺序错误，导致仅处理一轮列就直接返回，无法持续迭代检测异常值
• 临时列覆盖：多列处理时重复使用同一临时列名，前一列的检测结果被后一列覆盖，无法合并多列异常标记
• 移除逻辑错误：每处理一列就立即删除异常行，导致后续列检测基于不完整数据，结果偏差

以下是修正后的完整代码：

import pandas as pd
import datetime
from IPython.display import display

def remove_outliers(data_frame, columns=None) -> pd.DataFrame:
    # 处理列参数：默认自动选择数值列，兼容单列字符串输入
    if columns is None:
        columns = data_frame.select_dtypes(include=['int64', 'float64']).columns.tolist()
    elif isinstance(columns, str):
        columns = [columns]

    # 复制原数据避免修改原始输入
    df = data_frame.copy().dropna(subset=columns)
    outliers = pd.DataFrame()
    start_rows = df.shape[0]

    while True:
        # 初始化全局异常标记
        df['is_outlier'] = False

        for column in columns:
            # 临时计算当前列的检测指标，不写入DataFrame避免污染
            diff_val = pd.to_timedelta(df[column].diff(), unit="h")
            diff_val_shifted = diff_val.shift(-1)
            diff_bucket = df.index.diff()
            diff_bucket_shifted = diff_bucket.shift(-1)

            # 标记当前列的异常值，并合并到全局标记
            decreasing = diff_val < pd.Timedelta(0)
            rapid_increase = diff_bucket_shifted > diff_val_shifted - pd.Timedelta(seconds=6)
            df['is_outlier'] = df['is_outlier'] | decreasing | rapid_increase

        # 检查本轮是否有异常值
        current_outliers = df[df['is_outlier']]
        if current_outliers.empty:
            print("已过滤所有异常值。")
            break

        # 移除第一个异常值并记录
        first_outlier = current_outliers.head(1)
        outliers = pd.concat([outliers, first_outlier])
        df = df.drop(first_outlier.index)
        df = df.drop(columns=['is_outlier'])

    # 输出结果
    display(df)
    print("\n移除的异常值：")
    display(outliers)
    print("\n清洗后数据的前5行：")
    display(df.head(5))
    print(f"\n共删除 {start_rows - df.shape[0]} 行数据。")

    return df

# 测试示例
date_list = [datetime.datetime(2023, 1, 1, 0, 0) + datetime.timedelta(minutes=i) for i in range(5, 11)]
data = {
    'bucket': date_list,
    'ophPumping': [1, 2, 3, 30, 20, 5],
    'ophNonPumping': [1, 2, 3, 20, 4, 5]
}
df = pd.DataFrame(data).set_index('bucket')

# 调用方式1：处理所有数值列
cleaned_df = remove_outliers(df)
# 调用方式2：处理指定单列
# cleaned_df = remove_outliers(df, columns='ophPumping')

关键改动说明 #

1. 数据隔离：复制输入DataFrame，避免修改原始数据
2. 参数兼容：支持单列字符串、列列表输入，默认自动识别数值列
3. 异常值统一标记：每轮循环先遍历所有列，汇总全局异常标记后再处理移除，避免数据中途变化影响检测
4. 临时计算优化：列检测指标临时计算，不写入DataFrame，避免列名冲突和数据污染
5. 循环逻辑修正：while True持续迭代直到无异常值，匹配原函数的迭代移除逻辑

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