第 32 章 · Z 分数异常检测
这一章要解决什么：用一个统计指标找出内阻异常的电芯记录。 打掉一个盲区：Z 分数用的均值和标准差，会被异常值自己带偏 —— 数据越脏，它越不灵。所以还要学一个稳健的替代方案。
本章你需要完成的代码：demos/ch32-zscore.py
32.1 什么是 Z 分数
Z 分数（Z-score，又叫标准分数） 的公式：
= 这个值 = 均值 = 标准差
含义：这个值离"平均水平"有几个标准差。
| z 值 | 含义 | 正态分布下排位 |
|---|---|---|
0 | 正好等于均值 | 第 50% |
+1 | 比均值高 1 个标准差 | 前 15.9% |
+2 | 比均值高 2 个标准差 | 前 2.3% |
+3 | 比均值高 3 个标准差 | 前 0.13% |
-3 | 低 3 个标准差 | 后 0.13% |
正态分布下，|z| > 3 的样本约占 0.27%（大约 370 个里 1 个）。
所以 |z| > 3 天然成为一个异常阈值 —— 它来自统计学，不是拍脑袋定的。
它还有一个身份：Z 分数就是第 21 章讲的标准化的结果。 StandardScaler 输出的就是 z 值 —— 同一个公式，一个用途是"标准化特征"，一个用途是"找异常"。
32.2 算一遍
d = pd.read_csv(DATA / "dirty.csv")
COL = "resistance_mohm"
x = d[COL]
m, s = x.mean(), x.std()
print(f"均值 = {m:.4f}") # 56.1841
print(f"标准差 = {s:.4f}") # 23.1703
d["z"] = (x - m) / s点开：std() 的 ddof 口径说明（一个容易踩的细节）
标准差有两个版本：
| 分母 | Pandas 默认 | NumPy 默认 | |
|---|---|---|---|
| 总体标准差 | n | ✅ ddof=0 | |
| 样本标准差 | n - 1 | ✅ ddof=1 |
实测差异（本书数据）：
print(x.std()) # 23.1703 ← Pandas 默认 ddof=1
print(np.std(x.to_numpy())) # 23.1693 ← NumPy 默认 ddof=0差别很小（数据量大时 n 和 n-1 几乎一样），但跨库切换时要留意。
本书的做法：统一用 Pandas 的 .std()（ddof=1）， 并在需要时说清楚。关键是整个流程保持一致 —— 训练和推理用同一个口径就没问题。
怎么避免混淆：
x.std(ddof=1) # 明确写出来，读者不用猜Z 分数的分布
print(d["z"].describe().round(4))count 13897.0000
mean -0.0000
std 1.0000
min -0.7498
25% -0.5008
50% -0.1417
75% 0.3183
max 21.8567注意两点：
std = 1.0000—— 这是定义性的：Z 分数的标准差必然是 1；max = 21.86—— 最大的那个值离均值 21.9 个标准差！远远超过 3。
min 只有 -0.75（没有负向异常）—— 因为注入的异常都是"内阻变大"。
32.3 |z| > 3 抓到什么
mask = d["z"].abs() > 3
print(f"|z| > 3 的记录：{int(mask.sum()):,d} 条（占 {mask.mean():.2%}）")|z| > 3 的记录：68 条（占 0.49%）
被判异常的样本，内阻范围：219.49 ~ 562.61 mΩ
正常样本的内阻上限 ：87.57 mΩ物理上内阻应该在 38~90 mΩ 之间（数据生成时的设计范围）。
|z| > 3 抓到的 68 条，内阻全在 219 mΩ 以上 —— 显然是异常的。
⚠ 但 0.49% 和理论值 0.27% 不一样：
理论值（正态分布）：0.27%
实测值 ：0.49%原因：数据不是正态分布（右偏 + 人工注入的极端值）。
★ 教训：
"|z| > 3 约等于 0.27%" 这个说法依赖"正态分布"这个前提。 真实数据很少正态，所以阈值要用实测比例来校准，不能照抄。
怎么看实测比例是否合理：
rate = mask.mean()
if rate < 0.001:
print("可能阈值太严，漏检了")
elif rate > 0.05:
print("异常比例偏高 —— 要么阈值太松，要么数据本身有问题")
else:
print("比例在常见范围内")这些异常落在哪些电芯上
print(d.loc[mask, "cell_id"].value_counts())真实输出（节选）：
C008 12
C007 9
C006 8
...涉及 12 支电芯 —— 均匀分布在全部电芯上。
这个判断很重要（第 26 章讲过）：
均匀分布 → 采集/录入环节的随机错误 → 修数据管道
集中少数 → 那几支真的有问题 → 去查产线本数据是"均匀" —— 所以结论是"数据采集/注入造成的随机错误"， 而不是"某几支电芯有问题"。
32.4 ★ 核心问题：标准差被异常值自己撑大了
clean = pd.read_csv(DATA / "cycle_raw.csv")
print(f"脏数据：均值 {m:.4f}，标准差 {s:.4f}") # 56.1841, 23.1703
print(f"干净数据：均值 {clean[COL].mean():.4f}，"
f"标准差 {clean[COL].std():.4f}") # 54.8383, 11.6132
print(f"→ 标准差被撑大约 {s / clean[COL].std():.1f} 倍！") # 2.0 倍标准差从 11.61 被撑到 23.17 —— 大了 2 倍。
点开：这就是 Z 分数的死穴（附一个受控实验）
逻辑链：
异常值变大
↓
均值和标准差都被拉大（异常值也参与计算！）
↓
阈值 (mean ± 3σ) 被撑宽
↓
本该被抓到的异常，反而"落进了正常范围"
↓
漏检数据越脏，Z 分数越不灵。
受控实验：污染比例越大，Z 分数越失灵
注意：在本数据集上，普通 Z 和稳健 Z 命中完全相同（都是 68 条） —— 因为异常只占 0.49%，污染还不够严重。
所以我们要人为把污染程度调大，才能看清差别：
rng_ = np.random.default_rng(7)
base_ok = clean[COL].to_numpy() # 干净数据作为基底
for rate in (0.005, 0.01, 0.02, 0.05, 0.10):
x2 = base_ok.copy()
k_ = int(len(x2) * rate)
idx_ = rng_.choice(len(x2), k_, replace=False)
x2[idx_] = rng_.uniform(200.0, 600.0, k_) # 注入极端值
# ...（完整代码见 demos/ch32-zscore.py）真实输出：
污染比例 均值 标准差 普通 Z 命中 中位数 MAD 稳健 Z 命中
0.5% 56.57 28.38 69 52.89 9.21 69
1.0% 58.37 38.90 139 52.99 9.28 139
2.0% 61.49 50.90 264 53.12 9.39 279
5.0% 72.14 80.44 512 53.81 9.85 698
10.0% 88.99 109.32 617 54.93 10.71 1396读表（每行"应该抓到的个数"= 13967 × 污染比例）：
| 污染比例 | 应该抓到 | 普通 Z | 稳健 Z | 结论 |
|---|---|---|---|---|
| 0.5% | 69 | 69 | 69 | ✅ 一致 |
| 1.0% | 139 | 139 | 139 | ✅ 一致 |
| 2.0% | 279 | 264 | 279 | ⚠️ 普通 Z 漏 15 个 |
| 5.0% | 698 | 512 | 698 | ❌ 普通 Z 漏 186 个 |
| 10.0% | 1396 | 617 | 1396 | ❌ 普通 Z 漏掉一半以上 |
看标准差那一列就明白了：
污染 0.5% → 10%：
标准差：28.38 → 109.32 （涨 3.9 倍）
MAD ： 9.21 → 10.71 （涨 16%）★ 这就是"稳健（robust）"的含义： 统计量对极端值的敏感程度低。
★ 实践规则：
| 异常比例 | 建议 |
|---|---|
| < 1% | 普通 Z 分数就够了 |
| 1% ~ 5% | 换中位数 + MAD |
| > 5% | 考虑"这根本不是异常，而是两种分布混在一起" —— 该去查数据来源，而不是继续调阈值 |
最后一条很重要：如果 10% 的数据都"异常"，那它可能不是异常， 而是另一个总体（比如两批不同工艺的电芯混在一起）。
32.5 稳健版本：中位数 + MAD
med = x.median()
mad = (x - med).abs().median() # 中位数绝对偏差
robust_z = (x - med) / (1.4826 * mad) # 1.4826 让它在正态下与标准差等价真实输出：
中位数 = 52.9000
MAD = 9.2200
稳健 Z 命中数 = 68 条（占 0.49%）
普通 Z 命中数 = 68 条
两者都命中 = 68 条
只有普通 Z 命中 = 0 条
只有稳健 Z 命中 = 0 条在本数据集上两者命中完全相同 —— 因为这里异常只占 0.49%（见上面的受控实验）。
点开：MAD 和那个 1.4826 是什么？
MAD（Median Absolute Deviation，中位数绝对偏差）：
MAD = median(|x - median(x)|)三步：
- 算中位数；
- 算每个值到中位数的距离；
- 取这些距离的中位数。
为什么它稳健：
x = [45, 52, 53, 61, 999]
中位数 = 53
距离 = [8, 1, 0, 8, 946]
距离的中位数（MAD）= 8 ← 那个 946 被排到了最后，不影响中位数
对比标准差：
均值 = 242，标准差 = 379 ← 被 999 彻底带偏那个 1.4826 是什么：
它是"让 MAD 在正态分布下等于标准差"的校正系数。
推导：正态分布下，MAD ≈ 0.6745σ，所以 σ ≈ MAD / 0.6745 = 1.4826 × MAD。
用途：让"稳健 Z 分数"和"普通 Z 分数"在干净的正态数据上给出相同的尺度， 这样阈值 3 在两个方法里含义一致。
验证（本数据）：
print(f"标准差 = {s:.4f}") # 23.1703（脏数据，被撑大）
print(f"1.4826×MAD = {1.4826 * mad:.4f}") # 13.6696（稳健估计）
print(f"干净数据 σ = {clean[COL].std():.4f}") # 11.6132（真值）注意：1.4826 × MAD = 13.67 比干净数据的标准差 11.61 大一些 —— 说明这份数据的分布不是正态（实际是右偏的）， 所以这个校正系数在这里只是近似。
★ 教训：MAD 方法也依赖"分布形状"的假设。 任何统计方法都有前提 —— 知道前提，才知道什么时候不该用。
Python 里的实现：
# 手写
med = x.median()
mad = (x - med).abs().median()
robust_z = (x - med) / (1.4826 * mad)
# 或者用 scipy（需要安装）
from scipy.stats import median_abs_deviation
mad = median_abs_deviation(x, scale="normal") # scale="normal" 自动乘 1.4826本书手写（不引入额外依赖），但知道有现成的库也不错。
32.6 ⚠ 发现异常只是第一步
这一节是第 34 章的引子。
print("""
发现异常只是第一步，怎么处理是业务决定：
· 容量为负 → 物理上不可能，一定是录入错误 → 可以删
· 内阻 560 → 可能是传感器故障，也可能是真有问题的电芯 → 必须人工核查
""")★ 两类完全不同的"异常"
| 类型 | 例子 | 特征 | 处理 |
|---|---|---|---|
| 物理上不可能 | 容量 = -0.5 Ah | 值域越界，不需要统计 | 可以直接删/修 |
| 统计上偏离 | 内阻 = 560 mΩ | 值合法但极端 | 必须核查 |
第二类为什么必须核查：
- 可能是传感器故障（读数错了）→ 数据要修
- 可能是真有问题的电芯（内阻真的高）→ 这是最有价值的样本，不能删！
这就是"删数据要慎重"的根本原因 —— 你删掉的可能正是你要找的东西。
★ 物理规则应该排在统计规则前面
# ① 物理规则：不需要统计，直接拦
rule_phys = d["capacity_ah"] < 0 # 容量不可能是负数
# ② 统计规则：值合法但偏离
rule_stat = d["resistance_mohm"] > r_hi # 超过 IQR 上须
# 两条取并集
mask = rule_phys | rule_stat为什么顺序重要：均值和标准差会被异常值带偏（本节讲的就是这件事）， 而物理规则不依赖分布 —— 它永远可靠。
第 34 章会完整实现"物理规则 + 统计规则"的双保险方案。
NaN 不会被判成异常（但要单独处理）
print(f"本列还有 {int(d[COL].isna().sum())} 个 NaN")NaN > 3 的结果是 False（第 22 章讲过），所以 NaN 既不在 mask 里、 也不在 ~mask 的"正常"里 —— 是个盲区。
print(f"总数: {len(d)}")
print(f"异常: {int(mask.sum())}")
print(f"正常: {int((~mask).sum())}")
print(f"异常 + 正常 = {int(mask.sum()) + int((~mask).sum())} ← 差了 NaN 的个数")处理方式：
# 分开处理两类问题
missing = d[COL].isna() # 缺失 → 第 30 章的流程
outlier = d["z"].abs() > 3 # 异常 → 本章 + 第 34 章的流程
normal = ~(missing | outlier) # 真正"正常"的
print(f"缺失 {missing.sum()}, 异常 {outlier.sum()}, 正常 {normal.sum()}")★ 这个"三类划分"（缺失/异常/正常）是数据清洗的标准视角。
32.7 别在训练集上算阈值然后套到测试集
train = d.sample(frac=0.7, random_state=42)
test = d.drop(train.index)
mu_tr, sd_tr = train[COL].mean(), train[COL].std()
print(f"用训练集阈值判测试集：{int((((test[COL] - mu_tr) / sd_tr).abs() > 3).sum())} 条")
print(f"用测试集自己的阈值 ："
f"{int((((test[COL] - test[COL].mean()) / test[COL].std()).abs() > 3).sum())} 条")真实输出：
用训练集的阈值判测试集：|z_test| > 3 命中 19 条
用测试集自己的阈值 ：|z_test| > 3 命中 19 条（本数据上两者恰好相同，但这不是重点。）
点开：为什么统计量属于"模型的一部分"？
一个核心认知：
均值、标准差、IQR 边界、min/max 缩放系数 —— 这些"从数据里学出来的数字"，都是模型的一部分。
为什么：
它们决定了"什么样的值算异常"。如果训练时用的是训练集的均值和标准差， 那上线时也必须用同一套 —— 否则"异常"的定义就变了。
一个具体的例子：
# 训练时：某列均值 50、标准差 10，阈值 = 50 ± 30 = [20, 80]
# 上线后数据整体漂移了：均值变成 70
# 如果用线上数据重算：阈值 = [40, 100] → 原本 85 不算异常了
# 如果用训练时的阈值：[20, 80] → 85 会被标为异常（这才是对的！）用训练集的统计量，能检测出"数据漂移"； 用线上自己算的统计量，会"自动适应"漂移 —— 从而失去监控能力。
正确做法：
# 训练阶段：算好并保存
scaler_stats = {"mean": train[COL].mean(), "std": train[COL].std()}
pd.Series(scaler_stats).to_csv("outputs/scaler_stats.csv")
# 推理阶段：加载训练时的统计量
stats = pd.read_csv("outputs/scaler_stats.csv", index_col=0)["0"]
z_prod = (prod_df[COL] - stats["mean"]) / stats["std"]在 sklearn 里这是自动的：
scaler = StandardScaler()
scaler.fit(X_train) # 记住 mean_ 和 scale_
X_test_s = scaler.transform(X_test) # 用训练集的统计量但很多人会写成：
X_test_s = StandardScaler().fit_transform(X_test) # ❌ 测试集自己算了一套后者叫"数据泄漏" —— 测试集的信息泄漏到了预处理里， 评估结果会偏乐观。
★ 检查清单
| 步骤 | 训练时 | 推理时 |
|---|---|---|
| 算均值/标准差 | ✅ fit | ❌ 不要重算 |
| 应用标准化 | ✅ transform | ✅ transform（用训练的值） |
| 算异常阈值 | ✅ 在训练集上 | ❌ 不要重算 |
| 填充缺失（中位数） | ✅ 训练集中位数 | ✅ 用训练集的中位数 |
| 分箱边界 | ✅ 训练集分位数 | ✅ 用训练集的边界 |
一句话：训练阶段"学"参数，推理阶段"用"参数。
第 39 章的体检报告是"离线分析"（不是预测任务）， 所以可以用全量数据的统计量 —— 但要说清楚这一点。
32.8 完整示例与输出
cd code
python demos/ch32-zscore.py真实输出（节选）：
=== ② 算一遍 ===
均值 = 56.1841 mΩ
标准差 = 23.1703
|z| > 3 的记录：68 条（占 0.49%）
⚠ 这个占比和理论值 0.27% 不一样 —— 因为数据不是正态分布
=== ④ ★ 核心问题：标准差被异常值自己撑大了 ===
脏数据：均值 56.1841，标准差 23.1703
干净数据：均值 54.8383，标准差 11.6132
→ 标准差被撑大约 2.0 倍！
=== ⑤b 受控实验：污染比例越大，Z 分数越失灵 ===
污染比例 均值 标准差 普通 Z 命中 稳健 Z 命中
0.5% 56.57 28.38 69 69
5.0% 72.14 80.44 512 698
10.0% 88.99 109.32 617 1396
看标准差那一列：污染从 0.5% 涨到 10%，标准差涨 3.9 倍，MAD 只涨 16%本章练习
练习 1：用 Z 分数找内阻异常
对 dirty.csv 的 resistance_mohm：
- 算均值、标准差，构造 z 列；
- 数出
|z| > 3的条数与占比； - 打印异常样本的内阻范围、正常样本的上限；
- 对比干净数据的标准差，解释"为什么 Z 分数会被削弱"。
参考答案
from pathlib import Path
import pandas as pd
pd.set_option("display.width", 170)
CODE_DIR = Path(__file__).resolve().parent.parent.parent
DATA = CODE_DIR / "data" / "battery"
d = pd.read_csv(DATA / "dirty.csv")
clean = pd.read_csv(DATA / "cycle_raw.csv")
COL = "resistance_mohm"
# ①② Z 分数
x = d[COL]
m, s = x.mean(), x.std()
d["z"] = (x - m) / s
mask = d["z"].abs() > 3
print(f"均值 {m:.4f}，标准差 {s:.4f}")
print(f"|z| > 3：{int(mask.sum())} 条（{mask.mean():.2%}）") # 68 条，0.49%
# ③ 范围
print(f"异常样本 {d.loc[mask, COL].min():.2f} ~ {d.loc[mask, COL].max():.2f} mΩ")
print(f"正常样本上限 {d.loc[~mask, COL].max():.2f} mΩ")
# 异常样本 219.49 ~ 562.61 mΩ
# 正常样本上限 87.57 mΩ
# ④ 对比
print(f"干净数据标准差 {clean[COL].std():.4f}") # 11.6132
print(f"→ 标准差被撑大约 {s / clean[COL].std():.1f} 倍") # 2.0 倍四个要点：
- 实测 0.49% ≠ 理论 0.27% —— 因为数据不是正态分布。 阈值要用实测比例校准。
- 异常样本内阻 219~563 mΩ，远高于正常上限 87.57 —— 抓得很准。
- 标准差被撑大 2 倍 —— 这就是 Z 分数的死穴（见受控实验）。
- 异常均匀分布在 12 支电芯上 —— 说明是采集环节的随机错误。
完整练习：exercises/ex32-01-zscore
练习 2：为什么数据越脏，Z 分数越不灵？
用受控实验验证：在干净数据上注入不同比例的极端值， 看"普通 Z"和"稳健 Z"各能抓到多少。
参考答案
import numpy as np
import pandas as pd
clean = pd.read_csv(DATA / "cycle_raw.csv")
COL = "resistance_mohm"
base = clean[COL].to_numpy()
rng = np.random.default_rng(7)
print(f"{'污染':<8}{'应抓':>7}{'标准差':>10}{'普通 Z':>9}{'MAD':>8}{'稳健 Z':>9}")
for rate in (0.005, 0.01, 0.02, 0.05, 0.10):
x = base.copy()
k = int(len(x) * rate)
idx = rng.choice(len(x), k, replace=False)
x[idx] = rng.uniform(200.0, 600.0, k) # 注入极端值
s = pd.Series(x)
m, sd = s.mean(), s.std()
med = s.median()
mad = (s - med).abs().median()
n_z = int((((s - m) / sd).abs() > 3).sum())
n_rz = int((((s - med) / (1.4826 * mad)).abs() > 3).sum())
print(f"{rate:>7.1%}{k:>7d}{sd:>10.2f}{n_z:>9d}{mad:>8.2f}{n_rz:>9d}")真实输出：
污染 应抓 标准差 普通 Z MAD 稳健 Z
0.5% 69 28.38 69 9.21 69
1.0% 139 38.90 139 9.28 139
2.0% 279 50.90 264 9.39 279
5.0% 698 80.44 512 9.85 698
10.0% 1396 109.32 617 10.71 1396结论：
| 污染比例 | 普通 Z 的漏检 | 稳健 Z |
|---|---|---|
| ≤ 1% | 0 | 0 |
| 2% | 15 | 0 |
| 5% | 186 | 0 |
| 10% | 779（漏掉一半以上） | 0 |
原因：
标准差被异常值抬高 → 阈值 mean ± 3σ 变宽 → 漏检
标准差：28.38 → 109.32（涨 3.9 倍）
MAD 几乎不动 → 阈值稳定
MAD：9.21 → 10.71（涨 16%）★ 实践规则
| 异常比例 | 用什么 |
|---|---|
| < 1% | 普通 Z 分数 |
| 1% ~ 5% | 中位数 + MAD |
| > 5% | 停下来想：这可能是"两种分布混在一起"，不是异常 |
最后一条最值得记住：如果 10% 的数据都"异常"， 那更可能的解释是数据本身来自两个不同的总体 （比如两批工艺不同的电芯混在一起）。
这时该做的事是"去查数据来源"，而不是"继续调异常阈值"。
练习 3：为什么阈值不能在测试集上算？
train = d.sample(frac=0.7, random_state=42)
test = d.drop(train.index)
# 做法一：用训练集的统计量
mu_tr, sd_tr = train[COL].mean(), train[COL].std()
n1 = ((((test[COL] - mu_tr) / sd_tr).abs() > 3)).sum()
# 做法二：用测试集自己的统计量
n2 = ((((test[COL] - test[COL].mean()) / test[COL].std()).abs() > 3)).sum()
print(n1, n2)要求：说出"为什么做法二有问题"，并举一个它造成实际错误的场景。
参考答案
两个数字在本数据上恰好相同（19 和 19），但这不是重点。
做法二的问题：它让测试集"自己定义什么是异常"。
一个会造成实际错误的场景：
# 训练时：某传感器内阻均值 50 mΩ、标准差 10 mΩ
# 阈值 = 50 + 3×10 = 80 mΩ
# 上线后：传感器开始漂移，读数整体上移 +20 mΩ
# 真实情况：所有电芯的内阻读数都虚高了 20 mΩ
# ❌ 做法二（用线上数据重算）：
# 线上均值变成 70、标准差还是 10
# 新阈值 = 70 + 3×10 = 100
# → 真实内阻 85 的电芯（其实已异常）被判为"正常"
# → 传感器漂移被"自动适应"掉了，监控完全失效！
# ✅ 做法一（用训练时的阈值）：
# 用 80 mΩ 的阈值去判线上数据
# → 大量样本超过 80 → 立刻报警"数据分布漂移"
# → 这正是我们想要的监控能力★ 核心认知
训练时"学"参数（均值、标准差、阈值、分箱边界）， 推理时"用"这些参数，绝不重算。
为什么：
| 用训练集的参数 | 用线上的参数 | |
|---|---|---|
| 检测数据漂移 | ✅ 能 | ❌ 不能（自动适应） |
| 评估结果 | 真实 | 虚高 |
| 与训练时的一致性 | ✅ 保证 | ❌ 破坏 |
在 sklearn 里的正确写法：
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train) # fit + transform
X_test_s = scaler.transform(X_test) # 只 transform！❌ 常见错误：
X_test_s = StandardScaler().fit_transform(X_test)
# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
# 测试集自己 fit 了一套统计量 → 数据泄漏★ 一个"泄漏自检"清单
| 步骤 | 检查 |
|---|---|
| 标准化 | fit 只在训练集上调用过吗？ |
| 填充缺失 | 中位数/均值是训练集的吗？ |
| 异常阈值 | 在训练集上算的吗？ |
| 分箱边界 | 训练集的分位数吗？ |
| 特征选择 | 用训练集选的还是全量？ |
| 目标编码 | 训练集的均值吗？ |
第 39 章的体检报告是"离线分析"（不是预测）， 所以用全量统计量是对的 —— 但要在方法学说明里写清楚：
"本报告为离线数据分析，统计量基于全量数据计算。 若用于在线监控，需将统计量固化为基线（baseline）， 用基线阈值检测线上漂移。"
面向 AI 开发：异常检测是"数据质量监控"的核心
在 AI 项目里，异常检测有三个层次的应用：
| 层次 | 目标 | 时机 |
|---|---|---|
| 数据质量 | 找出"不可能的值" | 数据入库时 |
| 数据漂移 | 发现"分布变了" | 每天/每小时 |
| 业务异常 | 找出"需要关注的对象" | 分析时 |
① 数据质量：入库时拦截
def validate_row(row, rules):
"""按物理规则校验一行 —— 不需要统计，最快。"""
errors = []
for col, (lo, hi) in rules.items():
v = row.get(col)
if v is None or pd.isna(v):
continue
if not (lo <= v <= hi):
errors.append(f"{col}={v} 超出 [{lo}, {hi}]")
return errors
RULES = {
"capacity_ah": (0.0, 3.0),
"soh": (0.0, 1.05),
"resistance_mohm": (10.0, 200.0),
"temp_rise_c": (0.0, 60.0),
"coulombic_eff": (0.9, 1.0),
}为什么物理规则优先：它不需要历史数据、不依赖分布、永远可靠。
② 数据漂移：Z 分数是基础工具
def drift_report(df_new, baseline):
"""对比线上数据与基线，报告漂移情况。"""
rows = []
for col in baseline:
m0, s0 = baseline[col]["mean"], baseline[col]["std"]
m1 = df_new[col].mean()
z = (m1 - m0) / s0 # 均值偏移了几个标准差
rows.append({
"column": col,
"baseline_mean": round(m0, 3),
"current_mean": round(m1, 3),
"z_shift": round(z, 2),
"alert": "⚠️" if abs(z) > 2 else "",
})
return pd.DataFrame(rows)z_shift 的含义：当前均值偏离基线多少个标准差。
z_shift | 判断 |
|---|---|
| < 1 | 正常波动 |
| 1 ~ 2 | 留意 |
| > 2 | 显著漂移，检查上游 |
③ 业务异常：需要"标记 + 核查"而不是"自动删除"
本章的 68 条异常，处理方式不应该是"删掉"：
# ❌ 直接删
d_clean = d[d["z"].abs() <= 3]
# ✅ 标记 + 导出核查清单
d["is_outlier"] = d["z"].abs() > 3
d[d["is_outlier"]].to_csv("outputs/outlier_cases.csv", index=False)为什么：这 68 条里可能有"真的有问题"的电芯 —— 那才是最值得深入分析的样本（第 34 章详讲）。
★ 三个层次的对比
| 数据质量 | 数据漂移 | 业务异常 | |
|---|---|---|---|
| 方法 | 物理规则 | Z 分数（对基线） | Z 分数 / IQR |
| 阈值来源 | 行业标准 | 训练时的统计量 | 业务决策 |
| 动作 | 拒收/修正 | 报警/排查 | 标记 + 人工核查 |
| 删除数据？ | 可以（值不可能） | 不 | 绝不 |
★ 一个完整的监控思路
def monitor(df_new, baseline_path="outputs/baseline.csv"):
"""一个最小可用的数据监控：物理规则 + 漂移检测。"""
baseline = pd.read_csv(baseline_path, index_col=0)
# ① 物理规则：立刻拦截
bad = 0
for col, (lo, hi) in RULES.items():
if col in df_new.columns:
n = int((~df_new[col].between(lo, hi)).sum())
if n:
print(f"[物理规则] {col}: {n} 个值越界")
bad += n
# ② 漂移检测：对比基线
for col in baseline.index:
if col not in df_new.columns:
continue
m0, s0 = baseline.loc[col, "mean"], baseline.loc[col, "std"]
z = (df_new[col].mean() - m0) / s0
if abs(z) > 2:
print(f"[漂移告警] {col}: 均值偏移 {z:+.2f} 个标准差")
return bad这 20 行代码是"数据监控"的最小实现 —— 它能在数据出问题的当天发现问题，而不是等到模型效果下降几周后。
★ 一句话总结本章的价值
Z 分数是"发现异常"的工具，不是"处理异常"的规则。发现之后怎么做，取决于业务，不取决于统计。