时间序列数据局部异常值检测与替换方法咨询

针对你需要基于目标行前后局部范围（如上下50行）检测异常值、并将异常值替换为前10个值均值的需求，以下提供两种实用的局部异常检测方案：

方案一：滑动窗口Z-score法 #

该方法为每个数据点计算局部滑动窗口内的Z-score，避免全局Z-score受整体数据分布偏移的影响，更贴合局部异常的检测需求。

实现代码 #

import pandas as pd
import numpy as np
from scipy import stats

def local_zscore_outlier_detection(df, window_size=100, z_threshold=3):
    # 重置索引确保连续，避免原索引混乱
    df = df.reset_index(drop=True).copy()
    mw_series = df['MW']
    
    # 计算中心滑动窗口的均值与标准差（window_size=100对应上下各50行）
    rolling_mean = mw_series.rolling(window=window_size, center=True).mean()
    rolling_std = mw_series.rolling(window=window_size, center=True).std()
    
    # 计算局部Z-score，替换标准差为0的情况避免报错
    local_z = np.abs((mw_series - rolling_mean) / rolling_std.replace(0, np.nan))
    
    # 标记异常值，窗口边缘无足够数据的位置直接视为非异常
    outliers = local_z > z_threshold
    outliers = outliers.fillna(False)
    
    # 替换异常值为前10个值的均值
    for idx in df[outliers].index:
        # 处理索引越界情况，取现有前序所有数据（最多10条）
        start_idx = max(0, idx - 10)
        prev_values = mw_series.iloc[start_idx:idx]
        if not prev_values.empty:
            df.loc[idx, 'MW'] = prev_values.mean()
    
    return df

关键参数说明 #

• window_size=100：对应你需要的上下50行局部范围，若要改为仅向前/向后窗口，可设置center=False并调整窗口大小
• z_threshold=3：默认Z-score绝对值大于3视为异常，可根据数据波动程度调整阈值

方案二：滑动窗口IQR法 #

四分位数间距（IQR）是更稳健的异常检测方法，不受极端值干扰，适合局部范围的异常识别。

实现代码 #

import pandas as pd
import numpy as np

def local_iqr_outlier_detection(df, window_size=100, iqr_factor=1.5):
    df = df.reset_index(drop=True).copy()
    mw_series = df['MW']
    
    # 计算滑动窗口的四分位数与IQR
    rolling_q1 = mw_series.rolling(window=window_size, center=True).quantile(0.25)
    rolling_q3 = mw_series.rolling(window=window_size, center=True).quantile(0.75)
    rolling_iqr = rolling_q3 - rolling_q1
    
    # 计算异常值边界
    lower_bound = rolling_q1 - iqr_factor * rolling_iqr
    upper_bound = rolling_q3 + iqr_factor * rolling_iqr
    
    # 标记异常值，边缘无足够数据的位置视为非异常
    outliers = (mw_series < lower_bound) | (mw_series > upper_bound)
    outliers = outliers.fillna(False)
    
    # 替换异常值为前10个值的均值
    for idx in df[outliers].index:
        start_idx = max(0, idx - 10)
        prev_values = mw_series.iloc[start_idx:idx]
        if not prev_values.empty:
            df.loc[idx, 'MW'] = prev_values.mean()
    
    return df

关键参数说明 #

• iqr_factor=1.5：标准IQR异常判断系数，若需要更严格的检测可改为3
• 滑动窗口设置为100，确保覆盖目标行上下各50行的局部数据

使用建议 #

1. 根据数据的时间间隔调整window_size：若为分钟级数据，window_size=100对应50分钟的局部范围，匹配你的需求
2. 可先通过df[outliers]查看标记的异常值，验证是否覆盖图表中08:00附近的异常点
3. 若前10行存在连续异常值，可扩大取数范围或改用滑动窗口均值替代，适配实际数据情况

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