基于窗口的Pandas动态异常值检测：循环实现优化问询

我太懂你用for循环做滑动窗口异常检测时卡到怀疑人生的感觉了！这种逐元素遍历的方式在数据量稍微大一点的时候，速度直接拉胯。下面给你三个不同方向的优化方案，根据你的数据规模和复杂度选就行：

Pandas本身就自带rolling()方法可以生成滑动窗口，我们只需要把中位数/MAD的计算逻辑封装好，用apply()批量处理就行，完全不用自己写循环。

针对你给的示例，代码可以这么写：

import pandas as pd
import numpy as np

s = pd.Series([0,1,2,3,4])
window_size = 3
threshold = 3.5  # 异常判断阈值

# 1. 计算每个窗口的中位数和MAD，平移1位对应"用前3个元素判断第4个"的逻辑
rolling_median = s.rolling(window_size).median().shift(1)
rolling_mad = s.rolling(window_size).apply(lambda x: np.median(np.abs(x - np.median(x)))).shift(1)

# 2. 计算异常值边界
lower_bound = rolling_median - threshold * rolling_mad
upper_bound = rolling_median + threshold * rolling_mad

# 3. 标记异常值（前window_size个元素无足够窗口数据，默认标记为False）
is_outlier = pd.Series(False, index=s.index)
is_outlier[window_size:] = (s[window_size:] < lower_bound[window_size:]) | (s[window_size:] > upper_bound[window_size:])

print(is_outlier)

这个方案的好处是代码直观，不用折腾底层逻辑，但如果数据量超过10万条，apply()的速度可能还是会有点跟不上，这时候就需要下面的向量化方案。

利用Numpy的as_strided生成滑动窗口的视图（注意：是视图不是复制，内存占用极低），然后批量计算所有窗口的中位数和MAD，完全规避循环，速度能提升几十倍甚至上百倍。

代码示例：

import numpy as np
import pandas as pd

s = pd.Series([0,1,2,3,4])
window_size = 3
threshold = 3.5

# 转成numpy数组方便操作
arr = s.values

# 生成滑动窗口视图：每个窗口对应前window_size个元素，待检测元素是第window_size位及以后的
window_count = len(arr) - window_size
windows = np.lib.stride_tricks.as_strided(
    arr,
    shape=(window_count, window_size),
    strides=(arr.strides[0], arr.strides[0])
)

# 批量计算所有窗口的中位数和MAD
medians = np.median(windows, axis=1)
mads = np.median(np.abs(windows - medians[:, np.newaxis]), axis=1)

# 计算异常边界并标记
lower = medians - threshold * mads
upper = medians + threshold * mads
test_elements = arr[window_size:]

is_outlier = pd.Series(False, index=s.index)
is_outlier[window_size:] = (test_elements < lower) | (test_elements > upper)

print(is_outlier)

这个方案是速度最快的，但要注意as_strided的使用：如果数据量极大，要确保内存足够（不过因为是视图，内存压力其实很小）。如果想用均值+标准差代替中位数+MAD，只需要把np.median换成np.mean，mads换成np.std就行。

如果你的异常检测逻辑比单纯的中位数/MAD更复杂，没法用上面的窗口批量处理，那可以用Numba把原来的for循环编译成机器码，速度也能提升一个数量级。

比如原来的慢循环可以改成这样：

from numba import njit
import pandas as pd
import numpy as np

# 用njit装饰器编译函数，第一次运行会编译，之后调用极快
@njit
def detect_outliers_fast(arr, window_size, threshold):
    is_outlier = np.zeros(len(arr), dtype=np.bool_)
    for i in range(window_size, len(arr)):
        window = arr[i-window_size:i]
        median = np.median(window)
        mad = np.median(np.abs(window - median))
        lower = median - threshold * mad
        upper = median + threshold * mad
        is_outlier[i] = arr[i] < lower or arr[i] > upper
    return is_outlier

# 使用示例
s = pd.Series([0,1,2,3,4])
is_outlier = pd.Series(detect_outliers_fast(s.values, 3, 3.5), index=s.index)
print(is_outlier)

这个方案的逻辑和你原来的循环完全一致，不需要改太多代码，适合逻辑复杂、没法批量处理的场景。

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