Z-score原理与统计假设
基本思想与公式
在统计学中，Z-score表示观测值与样本均值之间的偏离程度，单位为<强>标准差。核心公式z = (x - μ) / σ中，μ是样本均值，σ是样本标准差。阈值选择通常依赖领域知识，常用取3或2作为异常判定的分界线。异常判定基于观测值的绝对Z-score是否超过预设阈值。
使用Z-score的原因在于它将不同量纲的数据统一成标准化分布，从而便于跨变量筛选和对比。正态分布假设常作为参考，但现实数据可能偏离正态分布，因此在实际场景中需要结合其他方法进行对照与验证。标准化的意义在于将偏离程度与数据的离散程度联系起来，便于阈值的可迁移使用。
# 基本的单变量Z-score异常检测
import numpy as np
import pandas as pddef detect_zscore(series, threshold=3.0):mean = series.mean()std = series.std(ddof=0)z = (series - mean) / stdanomaly_mask = z.abs() > thresholdanomalies = series[anomaly_mask]return anomalies, z, anomaly_mask
滚动/分组场景下的放大应用
在时间序列或分组数据中，简单使用全局均值容易被趋势和季节性所影响。滚动平均与分组统计可以让检测对局部分布更加敏感，从而识别出局部异常。窗口选择与阈值设定的结合决定了检测的时效性与准确性。
下面给出滚动窗口的Z-score检测示例，演示如何在分组数据中维护局部统计量以进行异常判定。分组场景、局部阈值等是关键要点。
# 滚动窗口Z-score检测示例（按分组分组）
import numpy as np
import pandas as pddef detect_zscore_grouped(df, group_col, value_col, window=60, threshold=3.0):df = df.copy()rolling_mean = df.groupby(group_col)[value_col].transform(lambda s: s.rolling(window, min_periods=1).mean())rolling_std = df.groupby(group_col)[value_col].transform(lambda s: s.rolling(window, min_periods=1).std(ddof=0))z = (df[value_col] - rolling_mean) / rolling_std.replace({0: np.nan})mask = z.abs() > thresholddf['z_group_anomaly'] = maskdf['z_group'] = zreturn df
IQR方法原理与鲁棒性
统计分位数与异常边界
IQR（四分位距）方法基于数据的分位数来界定“正常”区间。Q1、Q3分别是第25百分位与第75百分位，IQR = Q3 - Q1。常用的异常边界定义为lower = Q1 - 1.5*IQR与upper = Q3 + 1.5*IQR，落在边界之外的点被判定为异常。边界因子（如1.5）可根据场景调整。
与以均值为中心的检测相比，IQR方法对极端值具有更高的鲁棒性，尤其适用于非对称分布、包含少量离群点的数据集。鲁棒性使其成为初步异常筛选的常用工具。
# IQR 异常检测函数
import pandas as pddef detect_iqr(series, factor=1.5):q1 = series.quantile(0.25)q3 = series.quantile(0.75)iqr = q3 - q1lower = q1 - factor * iqrupper = q3 + factor * iqrmask = (series < lower) | (series > upper)return mask, lower, upper
分组/IQR的鲁棒扩展
当数据按组分布或存在显著分布差异时，对每组单独计算IQR边界能够更准确地识别各自组内的异常。实现思路是对每个分组计算自己的Q1、Q3和IQR，并据此判断边界。分组边界、组间可比性等是关键点。
# 按组计算 IQR 边界的示例
def detect_iqr_grouped(df, group_col, value_col, factor=1.5):import numpy as npdf = df.copy()masks = []for key, group in df.groupby(group_col):q1 = group[value_col].quantile(0.25)q3 = group[value_col].quantile(0.75)iqr = q3 - q1lower = q1 - factor * iqrupper = q3 + factor * iqrm = (group[value_col] < lower) | (group[value_col] > upper)masks.append(m)# 将结果映射回原始数据结构，简化示例，实际应用中需对齐索引return df.assign(iqr_group_anomaly=masks[-1])
实战案例：Python实现的实战案例
数据场景与目标
在本节中，我们选取一个模拟的传感器数据集，包含时间戳与温度读数，演示如何同时应用<Z-score和IQR两种方法进行异常检测。传感器数据的连续性、噪声水平与变动趋势都会影响检测结果，因此需要展示两种方法的对比与应用要点。
通过一个简化案例，我们可以看到不同阈值设置时两种方法的敏感性与鲁棒性差异，以及两者在实际数据管线中的组合使用方式。对比分析有助于理解方法的优缺点与适用边界。
# 由两种方法组成的实战示例
import numpy as np
import pandas as pdnp.random.seed(42)
n = 500
# 模拟温度数据，带有噪声
temp = 20 + np.random.normal(0, 0.8, n)
# 注入一些离群点
temp[50] = 28.5
temp[200] = 11.0df = pd.DataFrame({'time': pd.date_range('2024-01-01', periods=n, freq='T'),'temp': temp})# Z-score 检测
mean = df['temp'].mean()
std = df['temp'].std(ddof=0)
df['z'] = (df['temp'] - mean) / std
df['z_anomaly'] = df['z'].abs() > 3# IQR 检测
q1 = df['temp'].quantile(0.25)
q3 = df['temp'].quantile(0.75)
iqr = q3 - q1
lower = q1 - 1.5 * iqr
upper = q3 + 1.5 * iqr
df['iqr_anomaly'] = (df['temp'] < lower) | (df['temp'] > upper)# 显示结果的一部分
print(df.loc[df['z_anomaly'] | df['iqr_anomaly'], ['time', 'temp', 'z', 'z_anomaly', 'iqr_anomaly']].head())
# 结果解释（文本注释）
# 具有异常标记的行应当被上报用于后续处理：
# - z_anomaly 使用全局均值与标准差，易受分布形态影响
# - iqr_anomaly 使用分位数边界，通常对极端值更鲁棒
在数据管线中的嵌入要点
将异常检测嵌入到数据管线时，实时性、可扩展性和容错性是关键指标。需要考虑批处理与流处理的切换、缺失值处理、以及结果存储与告警等设计点。同时，边界可解释性有助于后续审计与模型改进。
对比分析与注意事项
两种方法的对比要点
Z-score与IQR各有优劣：速度快、实现简单的通常是Z-score；对分布假设不敏感、鲁棒性强的往往来自IQR。缺失值处理、趋势/季节性的影响，以及阈值与边界的可解释性，是选择时的重要考量。
在多变量场景下，单一指标往往不足以覆盖所有异常形态，因此常见的做法是结合两种方法，形成综合策略，提升检测的覆盖率与稳定性。并集/交集策略、以及后续人工复核都可纳入设计之中。
# 简单的并集检测示例
def detect_both(series):z = (series - series.mean()) / series.std(ddof=0)z_anom = z.abs() > 3q1 = series.quantile(0.25)q3 = series.quantile(0.75)iqr = q3 - q1lower = q1 - 1.5 * iqrupper = q3 + 1.5 * iqriqr_anom = (series < lower) | (series > upper)return z_anom | iqr_anom
