数据漂移与统计检验:分布怎么比、20+ stattest 怎么组织

30 秒导读: 模型上线后,线上数据会慢慢"变味"——今天进来的特征分布,和当初训练/评估时的分布不再一样,这叫数据漂移(data drift)。Evidently 检测漂移的做法极其朴素:把"现在的分布"(current)和"基准分布"(reference)放一起比一比,差异超过阈值就报 drift。难点全在"怎么比":数值列、类别列、文本列各有各的比法。Evidently 把 20 多种"比法"做成一张可插拔的注册表——每个统计检验(statistical test,简称 stattest)都是一个统一签名的小函数,返回 (距离/差异值, 是否漂移)。本章讲清这个统计内核:注册表机制、几个代表性检验的公式直觉,以及上层的漂移 metric / preset 如何"选检验、逐列跑、汇总成漂移列占比"。

本章聚焦统计检验这一类计算的内核与选择逻辑。metric / preset 的通用引擎机制(Metric 类型、Context 缓存、Report→Snapshot)已在 02-metrics-engine.md 讲透,这里只补"漂移这类计算里,统计检验是怎么被选中、被调用、被聚合的"。

---

1. 这是什么(零基础也能懂)

1.1 一句话定义

漂移检测 = 分布对比 + 阈值判定。 你手里有两批同一列的数据:

• reference(基准/参照):通常是训练集、或某个"正常时期"的数据快照。
• current(当前):线上刚收到的新一批数据。

把这两批数据的分布画出来叠在一起,如果形状差得够远(超过某个阈值),就说这一列"漂了"。

1.2 为什么要管它

模型是在 reference 分布上学出来的。一旦线上 current 分布偏离得太多——用户结构变了、上游数据管道改了口径、季节性来了——模型的预测就可能悄悄失准,而准确率指标往往要等真实标签回流才看得到。漂移检测是不需要标签的早期预警:光看输入/输出的分布变化,就能提前拉响警报。

1.3 用起来什么样

Evidently 的招牌用法就是一个 DataDriftPreset,method="psi" 指定用 PSI 这种比法(真实示例见 README.md:162-168):

from evidently import Report
from evidently.presets import DataDriftPreset

report = Report([
    DataDriftPreset(method="psi")   # 所有列都用 PSI 检验
], include_tests="True")
my_eval = report.run(iris_frame.iloc[:60], iris_frame.iloc[60:])  # (current, reference)
my_eval

跑完你会得到:每一列的漂移分数 + 判定,以及一个总览——"多少列漂了 / 占比多少 / 整个数据集算不算漂移"。

1.4 一句话直觉

把 stattest 想成一台"分布差异计"。 你把两条分布喂进去,它吐出一个数(差异有多大)和一个红绿灯(超没超阈值)。Evidently 备了 20 多台不同原理的差异计,你按名字("psi"、"ks"、"jensenshannon"……)挑一台,或者让它按数据类型自动挑。

---

2. 顶层全景(它大概怎么转)

一次漂移检测,数据是这样流动的(从上到下,以单列为主线):

       preset / metric 层  (core, 新引擎)
   ┌───────────────────────────────────────────────┐
   │  DataDriftPreset  ──展开──▶  每列一个 ValueDrift │
   │                    └──────▶  DriftedColumnsCount │  ← 汇总"漂了几列"
   │  presets/drift.py            metrics/column_statistics.py │
   └───────────────────────────────────────────────┘
                    │ 调用(桥接到 legacy)
                    ▼
       单列漂移计算  get_one_column_drift()          ← legacy/calculations/data_drift.py
   ┌───────────────────────────────────────────────┐
   │  1. 选检验  get_stattest(ref, cur, 类型, 名字)   │
   │  2. 跑检验  drift_test_function(ref, cur, ...)   │
   │  3. 拿结果  (drift_score, drifted, threshold)    │
   └───────────────────────────────────────────────┘
                    │ get_stattest 返回一个
                    ▼
       stattest 注册表  registry.py
   ┌───────────────────────────────────────────────┐
   │  名字 "psi" ─▶ StatTest 对象 ─▶ 实现函数 _psi   │
   │  名字 "ks"  ─▶ StatTest 对象 ─▶ 实现函数 ...    │
   │  ...20+ 个,每个是一个 .py 文件,自注册进表      │
   └───────────────────────────────────────────────┘

各部件一句话职责:

部件	干什么	在哪个文件
DataDriftPreset	招牌入口:一次给所有列配好漂移检测,展开成"总览 + 逐列" metric	src/evidently/presets/drift.py:24
ValueDrift	单列漂移 metric(core 新引擎侧)	src/evidently/metrics/column_statistics.py:530
DriftedColumnsCount	汇总 metric:数几列漂了、占比多少	src/evidently/metrics/column_statistics.py:748
get_one_column_drift	单列漂移的真正计算:选检验→跑→出结果	src/evidently/legacy/calculations/data_drift.py:102
StatTest / register_stattest	检验的统一封装 + 注册机制	src/evidently/legacy/calculations/stattests/registry.py:33,129
各 *_stattest.py	每个检验一个文件,含公式 + 自注册	src/evidently/legacy/calculations/stattests/

注意"两套引擎"这条暗线: 招牌能力"分布漂移"主要落在 legacy 层(src/evidently/legacy/calculations/),core 的新 metric(ValueDrift、DriftedColumnsCount)其实是桥接器——把参数打包成 DataDriftOptions,再调 legacy 的 get_one_column_drift。所以本章的统计内核几乎全在 legacy/calculations/stattests/。

---

3. 核心机制一:stattest 注册表(可插拔的"分布差异计")

3.1 它要解决的小问题

有 20 多种检验,还允许用户塞自己写的。怎么让"选一个检验"这件事既能按名字("psi")、又能按对象、又能传自定义函数,还不把 if-else 写满一屏?答案:注册表 + 统一签名。

3.2 统一签名:每个检验长一个样

所有检验函数都遵守同一个类型(registry.py:22-23):

StatTestFuncReturns = Tuple[float, bool]                                  # (差异值, 是否漂移)
StatTestFuncType = Callable[[pd.Series, pd.Series, ColumnType, float],    # (ref, cur, 列类型, 阈值)
                            StatTestFuncReturns]

一句话:每个检验就是"吃两条数据 + 列类型 + 阈值,吐出 (差异值, 超没超阈值)"。 这个 (float, bool) 的约定是整章的地基——不管背后是 KS 检验的 p 值、还是 PSI 的距离,对外都被抹平成同一副面孔。

3.3 StatTest:给函数套一层元数据

裸函数不够——还得知道它叫什么、能用于哪些列类型、默认阈值多少。这就是 StatTest 这个 dataclass(registry.py:33-54):

@dataclasses.dataclass
class StatTest:
    name: str                              # "psi"
    display_name: str                      # "PSI"(展示用)
    allowed_feature_types: List[ColumnType]# 能用于数值/类别/文本
    default_threshold: float = 0.05        # 不传阈值时的默认

它的 __call__(registry.py:40-54)才是对外统一入口:补上默认阈值 → 找到对应引擎的实现 → 调用 → 把 (drift_score, drifted) 包成 StatTestResult(带上 actual_threshold,方便上层展示"用了哪个阈值")。

3.4 注册:每个检验文件自己"上架"

打开任意一个检验文件,结尾都是同一套两步(以 PSI 为例,psi.py:59-66):

psi_stat_test = StatTest(                      # 1. 造一个 StatTest 元数据对象
    name="psi",
    display_name="PSI",
    allowed_feature_types=[ColumnType.Categorical, ColumnType.Numerical],
    default_threshold=0.1,
)
register_stattest(psi_stat_test, _psi)         # 2. 把它和实现函数 _psi 注册进表

register_stattest(registry.py:129-134)往三张全局表里各填一格:

全局表	键 → 值	用途
_registered_stat_tests	名字 → {列类型: StatTest}	按名字/类型查检验
_impls	StatTest → {引擎: 实现}	按引擎找具体实现(留了多引擎扩展位)
_registered_stat_test_funcs	实现函数 → 名字	反查:传进来的裸函数是不是已注册的

这三张表加上 stattests/__init__.py:5-29 里"import 即注册"的副作用(一 import 这个包,20 多个检验就全部自注册好了),构成了整张注册表。

3.5 选检验:三种入口,一个出口

get_stattest(registry.py:163-173)是选检验的总闸。用户能传三种东西,它统一返回一个 StatTest:

stattest_func 传进来的是……
   None      ──▶ _get_default_stattest()   按数据形状自动选(见 §3.6)
   "psi" 字符串 ─▶ 查 _registered_stat_tests  按名字取
   函数/StatTest对象 ─▶ get_registered_stattest 直接用 / 反查名字 / 当自定义包一层

自定义函数这条路很妙(registry.py:181-188):如果传进来一个没注册过的可调用对象,它当场造一个 name="" 的匿名 StatTest、把函数包进实现里就能用——用户写个符合签名的函数就能即插即用,无需注册。

3.6 不选时的默认:按"数据形状"自动挑

传 None 时走 _get_default_stattest(registry.py:137-160)。它的分支逻辑很值得记——Evidently 的默认漂移策略就藏在这:

列类型	样本量	取值基数 n_values	默认检验
文本	>1000	—	abs_text_content_drift(绝对值文本漂移)
文本	≤1000	—	perc_text_content_drift(百分位文本漂移)
数值	≤1000	≤5(且>2)	卡方 chisquare;=2 时用 z 检验
数值	≤1000	>5	KS 检验 ks
数值	>1000	≤5	Jensen-Shannon 距离
数值	>1000	>5	Wasserstein 距离
类别	≤1000	—	卡方;=2 类时用 z 检验
类别	>1000	—	Jensen-Shannon 距离

直觉:小样本用"有 p 值的假设检验"(KS/卡方,能给显著性),大样本改用"距离度量"(JS/Wasserstein,p 值在大样本下会过度敏感,距离更稳)。 数值列取值太少(≤5)时,当成类别处理。

---

4. 核心机制二:逐个看几个代表性检验

20 多个检验分两大流派,判定方向正好相反:

• 假设检验流(返回 p 值):p 值 = "两批数据其实同分布"的概率。p 值越小越可疑,所以判定是 p_value <= threshold(默认 0.05)。代表:KS、卡方、z 检验。
• 距离/散度流(返回 距离值):直接量两条分布差多远。值越大越漂,判定是 value >= threshold。代表:PSI、Jensen-Shannon、Wasserstein、Hellinger。

下面每个只点公式直觉 + file:line,不整段贴。

4.1 分箱:距离流的公共前置

PSI / JS / Hellinger 这类"比两条分布"的检验,先要把连续数据切成箱、数出每箱占比,才能逐箱相减。这一步共用 get_binned_data(utils.py:16-60):

• 数值且取值 >20 个 → 用 numpy 的 histogram_bin_edges(..., bins="sturges") 按 Sturges 规则自动定箱边,数每箱频率占比。
• 否则(类别,或取值很少的数值)→ 直接按 value_counts() 当每个取值一个箱。
• feel_zeroes 补零技巧(utils.py:44-58):某个箱在一侧占比为 0,会让后面 log(0) 或除零爆炸。这里把 0 替换成一个极小正数(约 1e-4 或更小),保证 PSI/KL 这类含对数的公式数值稳定。

4.2 PSI —— 群体稳定性指数(招牌默认之一)

_psi(psi.py:37-56)。分箱后逐箱算,再求和:

PSI = Σ_bin (ref% - cur%) × ln(ref% / cur%)

直觉:每个箱贡献"占比差 × 占比比值的对数",占比差得越多、比值偏得越狠,贡献越大。 求和就是总的"分布偏移量"。默认 n_bins=30、阈值 0.1,psi_value >= threshold 判漂移(psi.py:56,64)。经验档:<0.1 稳定、0.1~0.2 中等、>0.2 明显漂移。

4.3 KS 检验 —— 数值列的经典假设检验

_ks_stat_test(ks_stattest.py:36-50)。直接调 scipy 的 ks_2samp,取它的 p 值:

p_value = ks_2samp(reference_data, current_data)[1]
return p_value, p_value <= threshold        # p 越小越漂,默认阈值 0.05

直觉:KS 统计量 = 两条经验累积分布曲线(CDF)之间的最大纵向缝隙。 缝隙越大越不像同分布,p 值越小。只用于数值列(ks_stattest.py:56)。

4.4 卡方检验 —— 类别列的经典假设检验

_chi_stat_test(chisquare_stattest.py:37-47)。把类别的频数摆成"观测 vs 期望"对比:

• 期望 f_exp = reference 各类别频数 × k_norm(k_norm = current 总数 / reference 总数,把两批规模拉齐,chisquare_stattest.py:43-44)。
• 观测 f_obs = current 各类别频数。
• scipy.stats.chisquare(f_obs, f_exp) 出 p 值,p_value < threshold 判漂移。

直觉:current 里每个类别的实际数量,和"若无漂移、按 reference 比例应有的数量"差多少;总偏差换算成卡方 p 值。

4.5 三个距离度量:JS / Wasserstein / Hellinger

它们都归"距离流",value >= threshold 判漂,默认阈值都是 0.1,但量的东西不同:

检验	量什么	核心调用	文件:行
Jensen-Shannon 距离	两条概率分布的"对称化 KL 散度"再开方,∈[0,1]	scipy.spatial.distance.jensenshannon(ref%, cur%)	jensenshannon.py:38-60
Wasserstein(归一化)	"把一堆土从 ref 形状搬成 cur 形状"的最小搬运功,除以 ref 的标准差归一化	stats.wasserstein_distance(ref, cur) / max(std, 0.001)	wasserstein_distance_norm.py:37-52
Hellinger 距离	由 Bhattacharyya 系数 Σ√(p·q) 导出:√(1 − Σ√(p·q)),∈[0,1]	手写逐箱累加 √(p1·p2) 再 √(1−·)	hellinger_distance.py:39-90

三者的分工直觉: JS/Hellinger 是纯概率形状差(只看每箱占比、无量纲),对"分布长得像不像"敏感;Wasserstein 带取值尺度(搬运距离含数值大小),对"整体平移了多少"敏感——比如所有值都 +100,JS 可能变化不大,Wasserstein 会明显变大。归一化除以 std 是为了让阈值跨列可比。

4.6 文本漂移 —— 非表格数据怎么比

文本没有"数值分布"可分箱,Evidently 换了个巧劲:训练一个"域分类器"来区分 ref 和 cur,分得越开 = 漂得越狠(text_content_drift.py → calculate_text_drift_score,data_drift_utils.py:135-166)。

流程(data_drift_utils.py:105-119 的 roc_auc_domain_classifier):

1. 给 reference 文本贴标签 0、current 文本贴标签 1,合并。
2. 拆训练/测试集,训一个 TfidfVectorizer(TF-IDF 词向量) + SGDClassifier 的 sklearn Pipeline。
3. 看这个分类器在测试集上的 ROC-AUC。

判定直觉:如果 ref/cur 其实同源,分类器根本分不开,ROC-AUC ≈ 0.5(等于瞎猜);分得越开、AUC 越高 = 两批文本内容差异越大 = 漂移。

两个变体:

• perc_text_content_drift(百分位,bootstrap=True):不用固定阈值,而是跑随机分类器上千次取百分位当基线(roc_auc_random_classifier_percentile,data_drift_utils.py:122-132),AUC 超过这个"随机能达到的高位"才算漂——等于自带显著性校正。默认阈值 0.95(text_content_drift.py:14,22)。
• abs_text_content_drift(绝对,bootstrap=False):直接和固定阈值 0.55 比 AUC(text_content_drift_abs.py:14,21)。默认路由里,样本 >1000 用绝对(省掉上千次 bootstrap 的开销),≤1000 用百分位(registry.py:139-142)。

---

5. 核心机制三:检验如何被 metric / preset 调用、聚合

前两节讲"单个检验怎么算"。这节讲上层引擎怎么选它、逐列跑它、汇总成一句结论。

5.1 单列:get_one_column_drift 才是真正的调用点

不管从哪个 metric 进来,单列漂移最终都汇到 get_one_column_drift(data_drift.py:102)。它的核心就两行(data_drift.py:173-174):

drift_test_function = get_stattest(reference_column, current_column, column_type, stattest)  # 选
drift_result = drift_test_function(reference_column, current_column, column_type, threshold) # 跑

先 get_stattest 选出一个 StatTest(用户没指定就走 §3.6 的自动默认),再调用它拿到 StatTestResult。 之后从 drift_result.drift_score / .drifted / .actual_threshold 取三件套(data_drift.py:327-329),连同分布画图数据一起打包成 ColumnDataDriftMetrics。

5.2 core 侧的 ValueDrift 只是桥接器

新引擎的单列 metric ValueDrift,其计算(ValueDriftCalculation.calculate,column_statistics.py:564-575)干的事就是:把 method / threshold 打包成 DataDriftOptions,然后调 legacy 的 get_one_column_drift。core 不重写统计,只做参数转发——再次印证"招牌漂移能力在 legacy"。

参数怎么落到具体某列?靠 DataDriftOptions.get_feature_stattest_func(options/data_drift.py:143)和 get_threshold(:105):按优先级 按列指定 > 按类型(cat/num/text) > 全局 method > None(自动) 解析出该列该用哪个检验、哪个阈值。

5.3 汇总:从"每列一个 bool"到"数据集漂没漂"

DriftedColumnsCount(column_statistics.py:748)负责总览。聚合逻辑在 get_dataset_drift(data_drift.py:403-405):

number_of_drifted_columns = sum([1 if drift.drift_detected else 0 for _, drift in drift_metrics.items()])
share_drifted_columns = number_of_drifted_columns / len(drift_metrics)
dataset_drift = bool(share_drifted_columns >= drift_share)   # 漂移列占比超过 drift_share(默认 0.5)

一句话:逐列拿到 drifted 布尔 → 数有几个 True → 除以列数得占比 → 占比过半(可调 drift_share)就宣布"数据集漂移"。 这就是从"20+ 检验的 (float, bool)"一路收敛到一句业务结论的全链路。

5.4 preset 把这一切一次配好

DataDriftPreset.generate_metrics(presets/drift.py:98-140)把上面两块拼起来:构造一份 DataDriftOptions,然后生成 一个 DriftedColumnsCount(总览) + 每列一个 ValueDrift(明细)。method="psi" 这种参数就是在这里被塞进 options、再往下透传到每列的 get_stattest。

---

6. 巧妙之处(可借鉴的技术)

• (float, bool) 统一签名抹平异构算法。 20 多种数学原理天差地别的检验(p 值 / 散度 / 分类器 AUC),对外全长成"吃两条 Series 吐 (值, 漂不漂)"。上层逻辑完全不必知道背后是 KS 还是 PSI——这是策略模式的教科书用法。见 registry.py:22-23。
• import 即注册。 每个检验文件末尾一句 register_stattest(...),stattests/__init__.py 一次 import 全部,注册表自动装满。加一个新检验 = 加一个文件,零改动中心代码。见 stattests/__init__.py:5-29。
• 未注册的自定义函数即插即用。 传进来的裸函数若不在注册表里,当场包成匿名 StatTest 就能跑——用户写个符合签名的函数就能用,不必先注册。见 registry.py:181-188。
• 默认策略按"样本量 × 基数"分档。 小样本用带 p 值的假设检验、大样本改用距离度量,避开了"p 值在大样本下过度敏感"的经典陷阱。见 registry.py:137-160。
• 文本漂移转化为"能不能被分类器区分开"。 把没有数值分布的文本,巧妙地归约成一个二分类 ROC-AUC 问题,还用随机分类器的 bootstrap 百分位自带显著性校正。见 data_drift_utils.py:135-166。
• 补零护栏。 分箱后把占比为 0 的箱替换成极小正数,专门给 PSI/KL 这类含 log/除法的公式兜底,避免 inf/nan。见 utils.py:44-58。

---

7. 边界与局限(诚实)

• 漂移 ≠ 模型变差。 检验只回答"分布变了没",不回答"变了要不要紧"。分布漂了模型也可能照样准,反之亦然——漂移是预警信号,不是结论。
• 阈值是经验值,得按业务调。 PSI 的 0.1、KS 的 0.05 都是默认约定,不同数据规模/业务容忍度下需要重设(默认档见各文件 default_threshold)。
• 距离流没有 p 值。 JS/Wasserstein/Hellinger/PSI 返回的是距离,不带统计显著性;"超阈值"是工程判定而非假设检验的 α 水平。
• 文本漂移较重、且有随机性。 要训练分类器(perc 变体还要上千次 bootstrap),比表格检验慢;虽然固定了 random_state,但本质仍是采样近似。见 data_drift_utils.py:105-132。
• 本章不穷举全部 20+ 检验。 Anderson-Darling、Cramér-von Mises、Epps-Singleton、Fisher exact、G-test、KL 散度、Mann-Whitney U、MMD、T-test、TVD、Energy distance 等都在同目录、同套注册机制下,可按 §3 的模式自行按 file:line 阅读。

---

8. 代码地图(导航索引)

主题	文件路径	符号名
检验统一签名 & StatTest 封装	src/evidently/legacy/calculations/stattests/registry.py	StatTestFuncType、StatTest、StatTestResult
注册机制(三张全局表)	src/evidently/legacy/calculations/stattests/registry.py	register_stattest、_registered_stat_tests、_impls
选检验入口 + 自动默认	src/evidently/legacy/calculations/stattests/registry.py	get_stattest、_get_default_stattest、get_registered_stattest
分箱与补零	src/evidently/legacy/calculations/stattests/utils.py	get_binned_data
PSI	src/evidently/legacy/calculations/stattests/psi.py	_psi、psi_stat_test
KS 检验	src/evidently/legacy/calculations/stattests/ks_stattest.py	_ks_stat_test、ks_stat_test
卡方	src/evidently/legacy/calculations/stattests/chisquare_stattest.py	_chi_stat_test、chi_stat_test
Jensen-Shannon	src/evidently/legacy/calculations/stattests/jensenshannon.py	_jensenshannon
Wasserstein(归一化)	src/evidently/legacy/calculations/stattests/wasserstein_distance_norm.py	_wasserstein_distance_norm
Hellinger	src/evidently/legacy/calculations/stattests/hellinger_distance.py	_hellinger_distance
文本漂移(域分类器)	src/evidently/legacy/utils/data_drift_utils.py	calculate_text_drift_score、roc_auc_domain_classifier
文本漂移两变体	src/evidently/legacy/calculations/stattests/text_content_drift.py / text_content_drift_abs.py	perc_text_content_drift_stat_test、abs_text_content_drift_stat_test
全部检验的注册清单	src/evidently/legacy/calculations/stattests/__init__.py	__all__
单列漂移真正计算	src/evidently/legacy/calculations/data_drift.py	get_one_column_drift
数据集级聚合	src/evidently/legacy/calculations/data_drift.py	get_dataset_drift
core 单列漂移 metric(桥接)	src/evidently/metrics/column_statistics.py	ValueDrift、ValueDriftCalculation
core 汇总 metric	src/evidently/metrics/column_statistics.py	DriftedColumnsCount、DriftedColumnCalculation
招牌 preset	src/evidently/presets/drift.py	DataDriftPreset
每列选检验/阈值	src/evidently/legacy/options/data_drift.py	get_feature_stattest_func、get_threshold

---

同组导航: index.md(Evidently 总览) · 01-data-model.md(数据模型与 Descriptor) · 02-metrics-engine.md(度量与运行引擎) · 03-llm-as-judge.md(LLM 即评委) · 05-observability-and-ops.md(从评估到监控)