目标泄漏检测：用依赖清单和 Python 检查器守住特征边界

如果你把一份公开数据集的若干列喂给模型，去预测同一份文件里的另一列，有时会得到高得离谱的分数。R² 达到 0.99 以上，看起来像突破，实际上模型只是把发布机构当初的计算公式重新推导了一遍。这类问题叫目标泄漏（target leakage）：你选作输入的列，本身已经以某种形式包含了答案。公开数据尤其容易踩坑，因为大量公开指标是由其他公开指标算出来的——一个政府指数可能由若干调查列合成，第二个指数又建立在第一个指数之上。发布机构通常只在方法学 PDF 里用自然语言描述这套配方，数据目录很少把它存成机器可校验的形式。

这篇教程要做的，就是把那份配方自己写下来，再写一个小工具去读它。工具的逻辑类似包管理器里的依赖检查：你告诉它要预测什么、打算用哪些列，它拒绝任何位于目标推导路径上（上游或下游）的列。读完你会掌握：用真实公开数据复现一次泄漏、用一份 YAML 清单描述数据集是怎么被构造出来的、用广度优先搜索和深度优先搜索遍历这张图、让检查器在拼写错误和空输入时大声失败，以及把检查接进持续集成。

准备工作

开始之前需要准备几样东西：

• Python 3.10 或更新版本
• 对 pandas 的 DataFrame 有基本概念
• 一个能执行命令的终端
• 约 7 MB 空闲磁盘空间，用于存放示例数据文件

建一个独立的项目目录，并在里面创建虚拟环境，避免这些库污染系统环境。然后安装本教程用到的三个库：

mkdir leak-tutorial
cd leak-tutorial
python3 -m venv .venv
source .venv/bin/activate
pip install pandas scikit-learn pyyaml

Windows 上把激活那行换成 .venv\Scripts\activate，并且把后文所有 python3 换成 python。后续所有命令都在 leak-tutorial 目录内执行。

几个术语先讲清楚

• 目标（target）：你想让模型预测的那一列。
• 协变量（covariate）：你喂给模型、帮助它预测目标的列，很多人叫它特征。
• R²：衡量模型预测值与真实值接近程度的分数。1.0 表示完全吻合，接近 0 表示模型几乎没解释出什么。
• 普查区（census tract）：美国的一种小地理单元，通常容纳约 4000 人，大致相当于一个社区。
• 交叉验证：一种更公平的模型评估方式。把数据切成五份，用四份训练、一份测试，轮换到每一份都当过测试集为止。

操作步骤

第一步：亲手复现一次泄漏

美国疾病控制与预防中心（CDC）发布社会脆弱性指数（SVI），应急规划人员用它来定位在洪水、热浪或疾病暴发中可能需要额外援助的社区。SVI 是分层构造的：起点是美国社区调查（ACS）的 16 个列，每列都是一个百分比，比如生活在贫困中的人口占比、没有车辆的家庭占比。CDC 把这 16 列分成四个主题，在每个主题内对每个普查区排名，再把四个主题排名合成一个总排名 RPL_THEMES。

关键细节在于：CDC 把原始的 ACS 列和算好的排名放在同一个 CSV 文件里发布。这让人很容易把两者一起抓进模型。下载加州的数据文件：

curl -L -o California.csv https://svi.cdc.gov/Documents/Data/2022/csv/states/California.csv

这里用 curl 而不是 Python 直接下载，是因为某些 macOS 上的 Python 安装无法验证该网站的证书。接着创建 leak_demo.py：

"""leak_demo.py: 用构造目标的那些列去预测一个已发布的指数。"""
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.model_selection import KFold, cross_val_score

# CDC 用 -999 标记缺失值，先把它换成真正的空值。
df = pd.read_csv("California.csv", low_memory=False).replace(-999, float("nan"))

def score(inputs, target):
    data = df[inputs + [target]].dropna()
    model = HistGradientBoostingRegressor(random_state=0)
    folds = KFold(n_splits=5, shuffle=True, random_state=0)
    r2 = cross_val_score(model, data[inputs], data[target],
                         cv=folds, scoring="r2").mean()
    print(f"{target:<11} from {len(inputs)} column(s) "
          f"tracts={len(data)} R2 = {r2:.3f}")

# 主题 1 正是由这五列构造出来的。
score(["EP_POV150", "EP_UNEMP", "EP_HBURD", "EP_NOHSDP", "EP_UNINSUR"],
      "RPL_THEME1")

# EP_NOINT 也在同一个文件里，但 CDC 没有把它放进指数。
score(["EP_NOINT"], "RPL_THEMES")

脚本做了三件事：加载 CSV 并把 CDC 的 -999 标记转成空值；score 函数训练一个梯度提升模型（处理数值表格的常用强模型），并用五折交叉验证计算 R²；第一次调用用 CDC 构造主题 1 时使用的那五列去预测主题 1，第二次调用用 EP_NOINT（缺少宽带订阅的家庭占比）去预测总排名，而 CDC 把这个列留在指数之外。

运行它：

python3 leak_demo.py

输出会显示：主题 1 用那五列预测，R² 约为 0.998；总排名用 EP_NOINT 预测，R² 约为 0.384。第一个分数意味着模型几乎完美地重建了 CDC 的主题 1——因为 CDC 的公式是固定配方，而模型拿到了全部原料。第二个分数则反映了两个相关指标之间普通程度的关联。想象一下，如果某篇论文报告用模型预测社会脆弱性得到 R² = 0.998，那看起来像重大突破，实际上只说明模型找到了 CDC 的配方。

第二步：理解公开数据为什么会泄漏

SVI 这个例子容易识破，因为输入和指数就在同一个文件里。真实情况往往更难，因为推导链会跨机构延伸。举一条真实的链条：FEMA 的国家风险指数（NRI）中包含一个社会脆弱性得分，该得分来自人口普查局的社区韧性估计，而人口普查局又用 ACS 调查数据构造这些估计。于是 FEMA 的风险得分和某个 ACS 列可以处在同一条链的两端，尽管它们来自不同机构、不同网站。

要理解计算机为什么会漏掉这种关系，需要区分一个数字的两种历史：

• 来源（provenance）回答「这个数字是从哪里来的」。例如某个值来自 California.csv，而该文件来自 svi.cdc.gov。
• 推导（derivation）回答「这个数字是由什么算出来的」。例如主题 1 由五个 ACS 列计算得到。

大多数数据目录把来源记录得很好，而推导通常只存在于写给人类看的方法学 PDF 里。于是当自动化流水线搜索有用的协变量时，它会愉快地收集那些目标本身由之构造出来的列，看到高分就把它们留下。

第三步：借用包管理器的思路

软件开发者早就解决过非常类似的问题。执行 pip install requests 时，pip 会读取 requests 依赖什么、那些包又依赖什么，一路向下展开依赖树。因为每个依赖都被写下来了，pip 能在安装任何东西之前发现树中任何位置的麻烦。

数据版本的问题形状完全一样。假设 ACS.EP_UNEMP（失业率）作为协变量进入模型，用来预测 FEMA_NRI.risk_score；而同一个列又通过另外两个产品爬升，最终也进入那个风险得分。这一列同时处在环的两端。

在计算机科学里，这种图叫有向无环图（DAG）：每个方框是一个节点，每个箭头是一条边；箭头始终从原料指向由它制成的产品，因此顺着箭头走只会不断向上远离起点，不会形成任何环。两个家族词汇用来描述图中的位置：

• 一个节点的祖先，是从它出发沿箭头反向走能到达的任何东西，距离不限。ACS 列是 SVI 的祖先。
• 一个节点的后代，是从它出发沿箭头正向走能到达的任何东西。SVI 是 ACS 列的后代。

于是泄漏检查就变成一条简单规则：每个协变量都必须避开目标的祖先和后代。

第四步：编写依赖清单

清单（manifest）是一个列出所有产品以及每个产品由什么构造而成的文件。这里用 YAML，因为人读起来、改起来都方便。一个直接测量得到的产品长这样：

ACS.EP_UNEMP:
  label: 失业率
  measurementBasis: measured
  derivesFrom: []

derivesFrom: [] 这个空列表记录了零个父节点，因为这个产品直接来自调查。而一个由其他产品构造出来的产品长这样：

FEMA_NRI.social_vulnerability:
  label: FEMA 国家风险指数，社会脆弱性
  measurementBasis: composite
  derivesFrom:
    - {variable: CENSUS_CRE.social_vulnerability, relation: identity, confidence: documented}

derivesFrom 里的每个条目就是图中的一条边，每条边携带三个事实：

• variable：父产品的名称，必须与清单中定义的某个产品匹配。
• relation：父产品与子产品之间的关系类型，例如 identity 表示直接沿用，其他常见取值还包括聚合、加权、排名等。
• confidence：这条边的可信程度，例如 documented 表示有文档依据。

把 SVI 的构造写进清单时，主题排名和总排名都要作为产品出现，它们的 derivesFrom 分别指向那 16 个 ACS 列和四个主题排名。这样，当有人把 EP_UNEMP 当作协变量、把 RPL_THEMES 当作目标时，检查器就能顺着边发现前者是后者的祖先。

第五步：构建检查器

检查器需要做四件事：读入清单、构建邻接表、从目标出发做双向遍历、对每个协变量判定是否落在禁区内。核心遍历用两种搜索实现：

from collections import deque

def ancestors(graph, node):
    """沿反向边做广度优先搜索，收集所有祖先。"""
    seen, queue = set(), deque([node])
    while queue:
        current = queue.popleft()
        for parent in graph.parents.get(current, []):
            if parent not in seen:
                seen.add(parent)
                queue.append(parent)
    return seen

def descendants(graph, node):
    """沿正向边做深度优先搜索，收集所有后代。"""
    seen, stack = set(), [node]
    while stack:
        current = stack.pop()
        for child in graph.children.get(current, []):
            if child not in seen:
                seen.add(child)
                stack.append(child)
    return seen

广度优先用队列，适合从目标向上逐层展开祖先；深度优先用栈，适合顺着推导链一路向下。两者都维护一个 seen 集合，避免在共享子图上重复访问。得到两个集合后，禁区就是它们的并集：

def check(graph, target, covariates):
    forbidden = ancestors(graph, target) | descendants(graph, target)
    offenders = [c for c in covariates if c in forbidden]
    if offenders:
        raise LeakageError(
            f"以下协变量位于目标 {target} 的推导路径上：{offenders}"
        )
    return True

注意规则是双向的：既拒绝目标的祖先（输入里含有构造目标的原料），也拒绝目标的后代（输入里含有由目标算出来的东西）。后者同样常见，比如用某个指数去预测它派生出的子指标。

第六步：在真实案例上运行检查器

把 SVI 的清单加载进来，先跑一个应当通过的组合：用 EP_NOINT 预测 RPL_THEMES。因为 CDC 没有把 EP_NOINT 放进指数，它既不是目标的祖先也不是后代，检查通过。再跑一个应当被拦下的组合：用那五个 ACS 列预测 RPL_THEME1。检查器会报出全部五个违规列。

跨机构链条同样能测。在清单里补上 FEMA NRI 的社会脆弱性得分及其指向人口普查局社区韧性估计的边，再补上后者指向 ACS 列的边。此时用 ACS.EP_UNEMP 预测 FEMA_NRI.risk_score 会被拦下，尽管两者来自不同机构、不同网站——这正是纯靠来源信息无法发现的情形。

第七步：让检查器在坏输入上大声失败

一个安静的检查器比没有检查器更危险，因为它会给出虚假的安全感。需要显式处理的坏输入至少有三类：

• 拼写错误：协变量或目标名在清单里不存在。此时应报错并给出最接近的候选名，而不是默默跳过。
• 文件损坏：YAML 解析失败、derivesFrom 引用了未定义的产品、边缺少必填字段。应在加载阶段就抛出异常，并指出出错的行。
• 空输入：协变量列表为空，或目标为空。空协变量列表会让检查永远通过，必须显式拒绝。

实现上，加载清单时先做一遍完整性校验：遍历所有产品的 derivesFrom，确认每个 variable 都能在清单中找到定义，否则抛出带产品名的异常。检查入口处先断言目标非空、协变量列表非空。拼写建议可以用标准库的 difflib.get_close_matches 生成。

第八步：在持续集成里自动运行

把检查接进 GitHub Actions，让每次推送都跑一遍。工作流大致包含这些步骤：检出代码、安装 Python、安装依赖、运行检查脚本。检查脚本读取仓库中的清单文件和一个描述「目标 + 协变量」的配置文件，任何违规都以非零退出码结束，从而让流水线失败。

name: leakage-check
on: [push, pull_request]
jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install pyyaml
      - run: python3 check_leakage.py --manifest manifest.yaml --spec model_spec.yaml

这样，当有人往特征列表里加了一个位于目标推导路径上的列时，PR 会直接变红，而不是等到模型上线后才发现分数虚高。

一个完整示例

下面把前面的片段串成一个可运行的最小流程。假设目录里有 manifest.yaml 和 model_spec.yaml 两个文件。

manifest.yaml：

ACS.EP_UNEMP:
  label: 失业率
  measurementBasis: measured
  derivesFrom: []

ACS.EP_POV150:
  label: 贫困率
  measurementBasis: measured
  derivesFrom: []

CDC_SVI.RPL_THEME1:
  label: SVI 主题 1 排名
  measurementBasis: composite
  derivesFrom:
    - {variable: ACS.EP_UNEMP, relation: weighted, confidence: documented}
    - {variable: ACS.EP_POV150, relation: weighted, confidence: documented}

CDC_SVI.RPL_THEMES:
  label: SVI 总排名
  measurementBasis: composite
  derivesFrom:
    - {variable: CDC_SVI.RPL_THEME1, relation: weighted, confidence: documented}

model_spec.yaml：

target: CDC_SVI.RPL_THEMES
covariates:
  - ACS.EP_UNEMP
  - ACS.EP_POV150

检查脚本：

import sys
import yaml
from collections import deque

class LeakageError(Exception):
    pass

def load(path):
    with open(path, encoding="utf-8") as fh:
        raw = yaml.safe_load(fh)
    if not isinstance(raw, dict) or not raw:
        raise ValueError(f"{path} 为空或格式不正确")
    parents, children = {}, {}
    for name, entry in raw.items():
        parents.setdefault(name, [])
        children.setdefault(name, [])
        for edge in (entry or {}).get("derivesFrom") or []:
            parent = edge.get("variable")
            if parent not in raw:
                raise ValueError(f"{name} 引用了未定义的产品 {parent}")
            parents[name].append(parent)
            children.setdefault(parent, []).append(name)
    return parents, children

def walk(start, table):
    seen, queue = set(), deque([start])
    while queue:
        current = queue.popleft()
        for nxt in table.get(current, []):
            if nxt not in seen:
                seen.add(nxt)
                queue.append(nxt)
    return seen

def main():
    parents, children = load("manifest.yaml")
    with open("model_spec.yaml", encoding="utf-8") as fh:
        spec = yaml.safe_load(fh)
    target = spec.get("target")
    covariates = spec.get("covariates") or []
    if not target:
        raise ValueError("未指定目标列")
    if not covariates:
        raise ValueError("协变量列表为空，检查将永远通过")
    forbidden = walk(target, parents) | walk(target, children)
    offenders = [c for c in covariates if c in forbidden]
    if offenders:
        raise LeakageError(f"协变量 {offenders} 位于目标 {target} 的推导路径上")
    print(f"检查通过：{len(covariates)} 个协变量均不在 {target} 的推导路径上")

if __name__ == "__main__":
    try:
        main()
    except (LeakageError, ValueError) as exc:
        print(f"检查失败：{exc}", file=sys.stderr)
        sys.exit(1)

运行 python3 check_leakage.py，因为 ACS.EP_UNEMP 和 ACS.EP_POV150 都是 CDC_SVI.RPL_THEMES 的祖先，脚本会以非零退出码结束并打印违规列。把 model_spec.yaml 里的协变量换成一个不在推导链上的列，检查就会通过。

注意事项

• 清单的准确性完全取决于写它的人。如果某条推导边没有被记录，检查器就查不出来。清单需要随着数据产品的更新而维护。
• 边的 confidence 字段只是标注，不参与判定。低置信度的边同样会触发拦截，这是有意为之：宁可误报，也不要漏报。
• 检查器只处理显式写进清单的关系。跨机构链条需要把中间产品也补进清单，否则链条会断在中间。
• 空协变量列表必须显式拒绝，否则检查会静默通过，给出虚假的安全感。
• 示例中使用的库版本、数据文件大小、具体分数会随版本和数据集更新而变化，实际数值请以官方当前发布的信息为准。
• 把检查接进 CI 时，清单文件和模型配置都应纳入版本控制，这样每次特征变更都能被追溯。