异常检测算法:从统计到机器学习的实战选择
异常检测,说白了就是在一堆数据里找出「不对劲」的那些点。体量异常场景下,这个「不对劲」可能是流量突增、响应时间飙升、错误率爆表。我做了这么多年稳定性保障,发现很多人一上来就上机器学习模型,结果效果还不如一个简单的 3-sigma 阈值。为什么?因为选对方法比用对方法更重要。
这一章,我把异常检测算法分成三个梯队来讲:统计阈值、时间序列、机器学习。每个梯队我都会给出适用场景、代码实现和避坑经验。
核心观点:没有最好的算法,只有最合适的算法。90% 的体量异常,用统计或时间序列方法就能搞定。机器学习是最后的手段,不是首选。
一、统计阈值检测:3-sigma 与 MAD
统计方法是最直观的异常检测手段。它的核心假设是:正常数据服从某种分布,异常点会偏离这个分布。
1. 3-sigma 法则
3-sigma 基于正态分布。在正态分布中,99.7% 的数据落在均值 ±3 个标准差范围内。超出这个范围的点,我们标记为异常。
import numpy as np
def three_sigma_detection(data, threshold=3):
"""
3-sigma 异常检测
data: 一维数组
threshold: 标准差倍数,默认 3
"""
mean = np.mean(data)
std = np.std(data)
upper_bound = mean + threshold * std
lower_bound = mean - threshold * std
anomalies = []
for i, val in enumerate(data):
if val < lower_bound or val > upper_bound:
anomalies.append((i, val))
return anomalies, (upper_bound, lower_bound)
# 示例
data = [10, 12, 11, 13, 12, 100, 11, 12, 13, 11]
anomalies, bounds = three_sigma_detection(data)
print(f"异常点: {anomalies}")
print(f"阈值范围: {bounds}")
我曾经踩过的坑:3-sigma 对离群值非常敏感。如果数据本身包含几个极端异常点,均值和标准差会被「拉偏」,导致正常点被误判为异常,异常点反而被放过。我遇到过线上流量数据里混了几个 10 倍峰值,结果 3-sigma 阈值被拉到天上,后续的突增全漏报了。
2. MAD(Median Absolute Deviation)
MAD 用中位数代替均值,用偏差中位数代替标准差。说白了,它更「抗揍」,不受极端值影响。
def mad_detection(data, threshold=3.5):
"""
MAD 异常检测
threshold: 常用 3.5(对应 3-sigma 的近似)
"""
median = np.median(data)
abs_dev = np.abs(data - median)
mad = np.median(abs_dev)
# 修正因子 1.4826,使 MAD 与标准差在正态分布下等价
modified_z_scores = 0.6745 * abs_dev / mad if mad != 0 else abs_dev
anomalies = []
for i, val in enumerate(data):
if modified_z_scores[i] > threshold:
anomalies.append((i, val))
return anomalies, median, mad
data = [10, 12, 11, 13, 12, 100, 11, 12, 13, 11]
anomalies, median, mad = mad_detection(data)
print(f"异常点: {anomalies}")
print(f"中位数: {median}, MAD: {mad}")
我的个人习惯:在监控系统中,我通常同时跑 3-sigma 和 MAD。如果两者结果一致,基本可以确认异常。如果结果冲突,我会优先相信 MAD——它更稳健。
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 3-sigma | 计算简单,解释性强 | 对离群值敏感,需正态分布 | 数据质量好、分布稳定的场景 |
| MAD | 抗离群值,稳健 | 计算稍复杂,对对称分布效果更好 | 数据含噪声、有离群值的场景 |
二、时间序列检测:移动平均与指数平滑
统计阈值的问题是:它不关心时间顺序。你想想看,流量在凌晨 3 点和下午 3 点的正常范围完全不同。这时候就需要时间序列方法了。
1. 移动平均(Moving Average)
移动平均用最近 N 个点的均值作为当前点的预测值。如果实际值偏离预测值超过阈值,就标记为异常。
def moving_average_detection(data, window=5, threshold=2.0):
"""
移动平均异常检测
window: 滑动窗口大小
threshold: 标准差倍数
"""
anomalies = []
for i in range(window, len(data)):
window_data = data[i-window:i]
mean = np.mean(window_data)
std = np.std(window_data)
if std == 0:
continue
z_score = (data[i] - mean) / std
if abs(z_score) > threshold:
anomalies.append((i, data[i], z_score))
return anomalies
# 模拟带趋势的数据
import numpy as np
np.random.seed(42)
trend = np.linspace(0, 10, 100)
noise = np.random.normal(0, 1, 100)
data = trend + noise
data[50] = 30 # 注入异常
anomalies = moving_average_detection(data, window=10, threshold=2.5)
print(f"检测到 {len(anomalies)} 个异常点")
2. 指数平滑(Exponential Smoothing)
移动平均有个问题:窗口内的点权重一样。但直觉上,越近的数据越能反映当前状态。指数平滑就解决了这个问题——它给近期的数据更高的权重。
def ewma_detection(data, alpha=0.3, threshold=3.0):
"""
指数加权移动平均(EWMA)异常检测
alpha: 平滑因子,0-1 之间,越大越关注近期数据
"""
ewma = [data[0]]
residuals = []
for i in range(1, len(data)):
# EWMA 递推公式
ewma_val = alpha * data[i] + (1 - alpha) * ewma[-1]
ewma.append(ewma_val)
residuals.append(data[i] - ewma_val)
# 计算残差的标准差
residuals = np.array(residuals)
std_residual = np.std(residuals)
anomalies = []
for i in range(1, len(data)):
residual = data[i] - ewma[i-1]
if abs(residual) > threshold * std_residual:
anomalies.append((i, data[i], residual))
return anomalies, ewma
data = [10, 12, 11, 13, 12, 50, 11, 12, 13, 11]
anomalies, ewma = ewma_detection(data, alpha=0.3)
print(f"异常点: {anomalies}")
我建议:alpha 值不要拍脑袋定。我一般用网格搜索,选使残差最小的 alpha。对于秒级监控数据,alpha 在 0.1-0.3 之间效果不错。对于分钟级数据,可以适当调高到 0.3-0.5。
三、机器学习检测:孤立森林与 LOF
当数据维度高、模式复杂时,统计和时间序列方法就力不从心了。这时候该机器学习上场了。但记住我的话:机器学习是最后的手段,不是首选。
1. 孤立森林(Isolation Forest)
孤立森林的思路很巧妙:异常点因为「与众不同」,更容易被孤立。它通过随机切分特征空间,异常点需要的切分次数更少。
from sklearn.ensemble import IsolationForest
import numpy as np
def isolation_forest_detection(data, contamination=0.1):
"""
孤立森林异常检测
contamination: 预期异常比例,默认 10%
"""
# 数据需要是二维的
X = np.array(data).reshape(-1, 1)
model = IsolationForest(
contamination=contamination,
random_state=42,
n_estimators=100
)
# -1 表示异常,1 表示正常
predictions = model.fit_predict(X)
anomalies = []
for i, pred in enumerate(predictions):
if pred == -1:
anomalies.append((i, data[i]))
return anomalies, model
# 生成数据:正常点 + 异常点
np.random.seed(42)
normal = np.random.normal(0, 1, 1000)
anomaly = np.random.uniform(5, 10, 20)
data = np.concatenate([normal, anomaly])
anomalies, model = isolation_forest_detection(data, contamination=0.02)
print(f"检测到 {len(anomalies)} 个异常点")
我曾经犯过的错:孤立森林的 contamination 参数很关键。我一开始用默认的 0.1,结果把很多正常点标成了异常。后来我改用异常比例估计 + 交叉验证,效果才稳定下来。记住:contamination 不是你想设多少就设多少,它应该反映你数据中真实的异常比例。
2. LOF(Local Outlier Factor)
LOF 看的是「局部密度」。一个点的密度远低于它邻居的密度,那它就很可能是异常。这个方法的优势在于:它能处理密度不均匀的数据。
from sklearn.neighbors import LocalOutlierFactor
def lof_detection(data, n_neighbors=20, contamination=0.1):
"""
LOF 异常检测
n_neighbors: 邻居数量
contamination: 预期异常比例
"""
X = np.array(data).reshape(-1, 1)
model = LocalOutlierFactor(
n_neighbors=n_neighbors,
contamination=contamination
)
predictions = model.fit_predict(X)
anomalies = []
for i, pred in enumerate(predictions):
if pred == -1:
anomalies.append((i, data[i]))
return anomalies, model
data = [10, 12, 11, 13, 12, 100, 11, 12, 13, 11]
anomalies, model = lof_detection(data, n_neighbors=5, contamination=0.1)
print(f"LOF 检测异常点: {anomalies}")
| 方法 | 核心思想 | 优点 | 缺点 |
|---|---|---|---|
| 孤立森林 | 异常点更容易被孤立 | 高效,适合高维数据 | contamination 参数敏感 |
| LOF | 局部密度异常 | 处理密度不均匀数据 | 计算复杂度高,参数调优难 |
四、实战选择指南
说了这么多,到底怎么选?我总结了一个决策流程:
- 先看数据分布:画个直方图,如果近似正态分布,3-sigma 或 MAD 就够了。
- 再看时间特性:如果有明显的趋势或周期性,上移动平均或指数平滑。
- 最后看复杂度:如果数据维度高、模式复杂,再考虑孤立森林或 LOF。
我的经验法则:80% 的体量异常可以用移动平均 + 3-sigma 组合搞定。剩下 20% 的复杂场景,才需要机器学习。别一上来就上模型,先问问自己:简单的统计方法试过了吗?
嗯,这一章的内容就到这里。记住:异常检测不是越复杂越好,而是越合适越好。你手头的监控数据,用哪种方法最合适?不妨先跑一遍统计方法看看效果。