如何用 Python 和依赖图检测公开数据集中的隐藏目标泄漏

不久前，我把美国疾控中心（CDC）一份公开数据集中的五列数据喂给一个机器学习模型，让它预测同一文件里的第六列。模型的 R² 高达 0.998，对真实世界的模型来说，这几乎就是能拿到的最接近完美的分数。

这个分数看上去像是一次成功，但模型对现实世界几乎一无所知。第六列本来就是 CDC 用另外五列算出来的，所以模型只是逆向解出了 CDC 的计算公式而已。

数据科学家把这种问题称为目标泄漏（target leakage）：只要你喂给模型的输入已经以某种形式包含了答案，它就会出现。

这类泄漏在公开数据中很容易藏身，因为公开数据里相当大一部分本就是用其他公开数据算出来的。某个政府指数可能由一组调查列汇总而成，另一个指数又可能建立在第一个指数之上。这些计算方式，各机构都会在方法论文档（PDF）里解释清楚，但数据目录却很少把它们存成计算机能够核验的形式。

在本教程中，你将亲手编写这样一份记录，然后构建一个能读取它的小型 Python 工具。这个工具的用法类似于包管理器里的依赖检查器：你告诉它想预测什么、打算使用哪些列，凡是与目标存在推导关系的列——无论是目标由它推导而来，还是它由目标推导而来——它都会拒绝使用。

读完这篇教程，你将学会如何：

• 用最新的 CDC 数据和 scikit-learn 复现一次真实的数据泄漏

• 用一个名为 manifest 的小型 YAML 文件记录数据集是如何构成的

• 用广度优先搜索和深度优先搜索遍历这张图

• 打造一个检查工具，让它在遇到拼写错误、损坏的文件或空输入时大声报错

• 借助 GitHub Actions 在每次推送时自动运行检查

目录

• 准备工作

• 关键术语的通俗解释

• 第 1 步：亲眼看看数据泄漏

• 第 2 步：理解公开数据为何会泄漏

• 第 3 步：借鉴包管理器的思路

• 第 4 步：编写依赖清单

• 第 5 步：构建检查器

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

• 第 7 步：让检查器在遇到坏输入时大声报错

• 第 8 步：在 CI 中自动运行检查

• 第 9 步：吸取我的教训

• 用完整工具更进一步

• 结语

准备工作

要跟着操作，你需要准备：

• Python 3.10 或更高版本

• 对 pandas DataFrame 有基本了解

• 一个可以运行命令的终端

• 约 7 MB 的可用磁盘空间，用来存放 CDC 数据文件

先新建一个项目文件夹，并在其中创建虚拟环境，这样这些库就不会和你系统里的其他东西混在一起。然后安装本教程用到的三个库：

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

在 Windows 上，请用 .venv\Scripts\activate 代替 source 那一行命令；文中凡是出现 python3 的地方，请一律输入 python。本教程的所有命令都要在 leak-tutorial 文件夹内运行。

你在这篇教程里构建的每一个文件，都已收录在配套仓库 derives-from-tutorial 中，卡住时可以拿自己的成果与它对照。

完整版工具托管在一个公开的 GitHub 仓库里，文末会附上链接。

关键术语，用大白话讲清楚

本教程会反复用到下面这五个词：

• 目标变量（Target）：你希望模型预测的那一列。

• 协变量（Covariate）：喂给模型、帮它预测目标变量的数据列（很多人把它们叫做特征）。

• R²（R-squared，决定系数）：一个分数，用来衡量模型预测值与真实值的吻合程度。1.0 表示完全吻合，接近 0 则说明模型几乎解释不了什么。

• 人口普查区（Census tract）：美国的一种小型统计区域，通常约有 4,000 人，大致相当于一个街区的规模。

• 交叉验证（Cross-validation）：一种公平的模型测试方法：把数据分成五份，用四份训练、剩下一份测试，然后轮换，直到每一份都当过一次测试集。

第 1 步：亲眼看看这个泄漏

美国疾病控制与预防中心（CDC）发布 社会脆弱性指数（Social Vulnerability Index，简称 SVI）。应急规划人员借助这个指数，找出在洪水、热浪或疫情暴发时可能需要额外援助的社区。

CDC 是分层构建 SVI 的。起点是美国人口普查局组织的大型调查——美国社区调查（ACS）中的 16 个数据列，每一列都是一个百分比，比如贫困人口占比，或者没有汽车的家庭占比。

CDC 把这 16 列归入四大主题，并在每个主题内对所有人口普查区排名，再把四个主题排名合成一个总排名，也就是 RPL_THEMES。

CDC 构建 SVI 的流程：16 列 ACS 调查数据汇入四个主题排名，四个主题排名再汇入唯一一个总排名。图中每个黄色方框都由它下方的方框计算得出。

本教程里有个关键细节：CDC 把原始的 ACS 数据列和最终的排名一并打包在同一个 CSV 文件中发布。这样一来，我们就能轻松地同时拿到两者，放进同一个模型里。

下载加州数据文件：

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

我在这里使用 curl，是因为 macOS 上的某些 Python 安装在直接下载文件时，无法验证网站的安全证书。

现在创建一个名为 leak_demo.py 的文件：

"""leak_demo.py: predict a published index from the columns it was built from."""
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.model_selection import KFold, cross_val_score

# CDC marks missing values as -999, so turn those into proper blanks.
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}")


# Theme 1 is built from exactly these five columns.
score(["EP_POV150", "EP_UNEMP", "EP_HBURD", "EP_NOHSDP", "EP_UNINSUR"],
      "RPL_THEME1")

# EP_NOINT ships in the same file, and CDC leaves it out of the index.
score(["EP_NOINT"], "RPL_THEMES")

下面看看这个脚本具体做了什么：

1. 脚本会加载 CSV 文件，并把 CDC 用来标记缺失值的 -999 转成空白——因为 CDC 就是用 -999 来表示缺失数据的。

2. score 函数会训练一个梯度提升模型（这是处理数值表格时既强大又流行的选择），并用五折交叉验证来衡量 R²。

3. 第一次调用使用与 CDC 构建主题 1 时完全相同的五个列来预测主题 1。

4. 第二次调用则用 EP_NOINT（即家中没有订阅宽带互联网的家庭占比）来预测总排名。CDC 把这一列放在同一个文件里，却没有把它纳入指数。

运行：

python3 leak_demo.py

下面是 leak_demo.py的输出结果：主题 1（Theme 1）用 CDC 当初构建它的五列来预测，得分为 R² = 0.998；总体排名则用 CDC 未纳入指数的一列来预测，得分为 0.384。

第一个得分是 0.998，说明模型几乎完美地复现了 CDC 的主题 1。CDC 的公式是一套固定的配方，而模型手里恰好备齐了每一种原料。

第二个得分是 0.384。EP_NOINT 在文件里就紧挨着指数，它的得分反映的只是两个相关指标之间普通关联的强弱。

现在不妨想象一篇论文，声称预测社会脆弱性的模型拿到了 R² = 0.998。这个数字看起来像重大突破，其实只能说明模型摸到了 CDC 的那套配方。

本文的得分来自 scikit-learn 1.8.0。从 1.3.2 到 1.9.0 的各个版本中，这些得分精确到小数点后三位都完全一致，所以你跑出来的结果应该能对上。

第 2 步：理解公开数据为何会泄漏

SVI 这个例子不难察觉，因为输入数据和指数同在一个文件里。现实中大多数案例要隐蔽得多，因为这条链条会横跨好几个机构。

下面是一条真实存在的链条，横跨三个机构。FEMA 的国家风险指数（National Risk Index，NRI）包含一个社会脆弱性得分。根据 FEMA 的技术文档（2025 年 12 月发布的 1.20 版），这个得分来自美国人口普查局的社区韧性估计（Community Resilience Estimates），而人口普查局又是基于 ACS 调查数据算出这些估计的。

这样一来，FEMA 的风险得分和 ACS 的某一列就可能落在同一条链条的两端，尽管它们出自不同机构、发布在不同网站上。

要想明白计算机为什么会漏掉这种关联，你需要先了解一个数字可能携带的两种来历：

• 来源（Provenance）回答“这个数字从哪里来？”这个问题。例如，某个值取自 California.csv，而这个文件又来自 svi.cdc.gov。

• 推导（Derivation）回答“这个数字是怎么算出来的？”这个问题。例如，主题 1 就是拿五个 ACS 列算出来的。

一个数字可以有两种历史。左边是来源（provenance），记录这个数值来自哪个文件、哪个网站；右边是推导（derivation），记录 Theme 1 是由哪五列 ACS 数据算出来的。

大多数数据目录都把来源记录得很好，Google 的 Data Commons 就是个好例子：它把来源定义为“一次导入的物理单元”，告诉你某个数字是从哪个文件进来的。而推导呢，通常只存在于写给人看的 PDF 方法论文档里。

于是，当自动化流水线寻找有用的协变量时，它会顺理成章地把那些用来构建目标变量的列也一并收集进来。流水线看到分数很高，便保留了这些列。

第 3 步：向包管理器借一个思路

软件开发者很久以前就解决过一个非常类似的问题。

当你运行 pip install requests 时，pip 会先读出 requests 依赖哪些包，再读出那些包各自又依赖什么，沿着树一层层往下走。正因为每一条依赖都被白纸黑字记录下来，pip 在真正安装任何东西之前，就能发现树中任何位置的问题。

下面的示意图把这棵树和数据领域的同类问题放在了一起。

一模一样的形状出现了两次。左边是 pip 的依赖树： urllib3 和 certifi 供给 requests，后者再供给 my-app。右边是数据版本： ACS.EP_UNEMP 输入到预测 FEMA_NRI.risk_score 的模型中，同时这一列还经由另外两个产品爬进那个风险评分。

公共数据同样需要这样的记录。在上图的右半部分，ACS.EP_UNEMP（失业率）作为协变量进入模型；与此同时，同一列还经由另外两个产品一路向上爬进 FEMA_NRI.risk_score——而它正是目标变量。这一列同时处在回环的两端。

在计算机科学里，这种图叫做图（graph）。每个方框是一个节点（node），每条箭头是一条边（edge）。在数据图中，顺着箭头走永远是向上走、离出发点越来越远，所以这样的图是一张有向无环图，简称 DAG。“无环”的意思是：箭头不会构成任何环路。

在本文中，所有箭头一律从原料指向用它做出的成品。

描述节点在图中的位置，有两个“家族”术语很好用：

• 节点的祖先（ancestor）是指沿着箭头往回走能到达的所有节点，无论隔多少层。ACS 各列都是 SVI 的祖先。

• 节点的后代（descendant）则是指沿着箭头往前走能到达的所有节点。SVI 就是 ACS 各列的后代。

如此一来，泄漏检查就简化成了一条规则：每个协变量都必须避开目标变量的祖先和后代。

第 4 步：编写依赖清单

所谓清单（manifest），就是一个文件，里面列出每个产品，以及各自的构建来源。这里选用 YAML 格式，因为它方便人们阅读和编辑。

下面是一个测量型产品的清单示例：

ACS.EP_UNEMP:
  label: Unemployment rate
  measurementBasis: measured
  derivesFrom: []

derivesFrom: [] 中的空列表表示这个产品没有任何父产品，因为它直接来自一次调查。

而下面这个产品，则是由另一个产品构建而来：

FEMA_NRI.social_vulnerability:
  label: FEMA National Risk Index, social vulnerability
  measurementBasis: composite
  derivesFrom:
    - {variable: CENSUS_CRE.social_vulnerability, relation: identity, confidence: documented}

derivesFrom 中的每一项对应图中的一条边，而每条边都包含三项信息：

• variable 存放父产品的名称，该名称必须与文件中其他位置定义的某个产品相对应。

• relation 描述父产品的使用方式。

• confidence 记录你对这条边有多大把握。

五种关系如下：

关系	含义
component	父产品是数学运算中的一个组成要素，比如求和式里的某个数
modelled_from	父产品是某个统计模型的输入
identity	该产品其实就是父产品本身，只是换了名字重新发布
poststratified_on	父产品提供了人口权重
denominator	父产品是分数的分母，比如“人均病例数”中的人口数

可信度分为以下三级：

可信度	含义
certain	计算公式已公开发表，或者输入和输出打包在同一个文件中一起发布
documented	机构在其方法论文档中明确写明了这层关系
inferred	文档强烈暗示了这层关系，因此只能视为初步结论

可信度字段比乍看起来更重要。一张充满猜测的血缘图，只会重新制造出它本想解决的问题，所以每条边都应该说明背后有多少证据支撑。

每个产品还有一个 measurementBasis 字段：

测量依据	含义
measured	直接清点或调查得来，例如人口普查的计数
modelled	由统计模型或机器学习模型生成
composite	用固定的算术运算从其他产品计算得出

这个字段记录了公开数据目录通常略去不谈的一点：一个数字究竟是数出来的，还是预测出来的。人口普查的计数和随机森林的预测放进电子表格里看不出任何差别，但二者是完全不同性质的证据。

现在创建 mini-manifest.yaml，内容如下。它是完整清单的一个精简片段，收录了美国真实数据基础设施中的 11 个产品；每个产品只保留了几条真实的边，好让文件保持简短。

# A small slice of derivation-manifest.yaml, used in the tutorial.
schema: derives-from/0.2

products:

  # ---- measured: counted or surveyed directly
  ACS.EP_POV150:
    label: Population below 150% of the poverty line
    measurementBasis: measured
    derivesFrom: []

  ACS.EP_UNEMP:
    label: Unemployment rate
    measurementBasis: measured
    derivesFrom: []

  ACS.EP_NOVEH:
    label: Households with zero vehicles
    measurementBasis: measured
    derivesFrom: []

  SAT.chirps_rainfall:
    label: CHIRPS satellite rainfall
    measurementBasis: measured
    derivesFrom: []

  NVSS.mortality:
    label: Death certificate records
    measurementBasis: measured
    derivesFrom: []

  # ---- built from other products
  CENSUS_CRE.social_vulnerability:
    label: Census Community Resilience Estimates, social vulnerability
    measurementBasis: modelled
    derivesFrom:
      - {variable: ACS.EP_POV150, relation: modelled_from, confidence: documented}
      - {variable: ACS.EP_UNEMP,  relation: modelled_from, confidence: documented}
      - {variable: ACS.EP_NOVEH,  relation: modelled_from, confidence: documented}

  FEMA_NRI.social_vulnerability:
    label: FEMA National Risk Index, social vulnerability
    measurementBasis: composite
    derivesFrom:
      - {variable: CENSUS_CRE.social_vulnerability, relation: identity, confidence: documented}

  HVRI.bric:
    label: Baseline Resilience Indicators for Communities
    measurementBasis: composite
    derivesFrom:
      - {variable: ACS.EP_UNEMP, relation: component, confidence: documented}
      - {variable: ACS.EP_NOVEH, relation: component, confidence: documented}

  FEMA_NRI.community_resilience:
    label: FEMA National Risk Index, community resilience
    measurementBasis: composite
    derivesFrom:
      - {variable: HVRI.bric, relation: identity, confidence: documented}

  FEMA_NRI.expected_annual_loss:
    label: FEMA National Risk Index, expected annual loss
    measurementBasis: modelled
    derivesFrom: []

  FEMA_NRI.risk_score:
    label: FEMA National Risk Index, overall risk score
    measurementBasis: composite
    derivesFrom:
      - {variable: FEMA_NRI.expected_annual_loss, relation: component, confidence: certain}
      - {variable: FEMA_NRI.social_vulnerability, relation: component, confidence: certain}
      - {variable: FEMA_NRI.community_resilience, relation: component, confidence: certain}

第 5 步：构建 Linter

Linter 是一种工具，它会在问题造成危害之前先行读取内容并发出警告。Flake8 这类代码 linter 读取的是源代码，而本文的 linter 读取的则是你的 manifest 文件和协变量列表。

新建一个名为 mini_lint.py 的文件。接下来我们将分六个部分逐步搭建，最终完成的文件不会超过 200 行。

第 1 部分：加载 YAML 并拒绝重复键

"""mini_lint.py: refuse covariates that sit on a derivation path to or from the target."""
import argparse
import sys
from collections import deque
from itertools import combinations

import yaml

RANK = {"certain": 3, "documented": 2, "inferred": 1}
DETERMINISTIC = {"component", "identity", "denominator"}


def fail(message):
    """Exit code 2 means the manifest or the command itself is broken."""
    print(message, file=sys.stderr)
    sys.exit(2)


# ---------------------------------------------------------------- step 1
class StrictLoader(yaml.SafeLoader):
    """A YAML loader that stops on a repeated key."""


def refuse_duplicates(loader, node, deep=False):
    seen = {}
    for key_node, _ in node.value:
        key = loader.construct_object(key_node, deep=deep)
        line = key_node.start_mark.line + 1
        if key in seen:
            fail(f"manifest error: key {key!r} appears twice "
                 f"(line {seen[key]} and line {line})")
        seen[key] = line
    return loader.construct_mapping(node, deep=deep)


StrictLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, refuse_duplicates)

RANK 把置信度词语转换成数字，工具由此才能找出一条路由中最薄弱的边。DETERMINISTIC 则列出了那些纯靠算术运算得出的关系。

fail 会打印一条消息，然后以退出码 2 终止程序。教程后面你会明白，退出码 2 为什么必须和退出码 1 严格区分开。

加载器还要防备 YAML 一个隐蔽的行为：如果同一个块里某个键出现了两次，PyYAML 会不动声色地保留后一份，把前一份丢掉。落到清单文件上，这可能抹掉某个产品的全部边，于是工具看到零条路由，进而心安理得地清掉一个本该报泄漏的协变量。

每当 PyYAML 构建一个映射（也就是 Python 字典）时，refuse_duplicates 都会执行一次。它遍历所有键，记住每个键所在的行号，一旦发现键重复，就立刻让程序停下来。

第 2 部分：读取产品与边

# ---------------------------------------------------------------- step 2
def load(path):
    try:
        with open(path) as fh:
            products = yaml.load(fh, StrictLoader)["products"]
    except (OSError, yaml.YAMLError, KeyError, TypeError) as e:
        fail(f"manifest error: unable to read {path}: {e}")

    edges = {}
    for name, product in products.items():
        edges[name] = []
        for e in product.get("derivesFrom") or []:
            if RANK.get(e.get("confidence")) is None:
                fail(f"manifest error: {name} has an edge with "
                     f"confidence {e.get('confidence')!r}")
            edges[name].append((e["variable"], e["relation"], e["confidence"]))
    return products, edges

load 用严格模式的加载器打开文件。读取过程中一旦出错——比如路径无效或 YAML 损坏——就会调用 fail。

接着，它会构建一个名为 edges 的字典：每个产品名对应一个 (parent, relation, confidence) 元组列表。例如：

edges["FEMA_NRI.social_vulnerability"]
# [("CENSUS_CRE.social_vulnerability", "identity", "documented")]

这一步还会检查每个置信度值，确保它是三个合法取值之一。否则，像 documneted 这样的拼写错误就可能蒙混过关，在后续排序时引发问题。

第三部分：先校验清单，再信任它

# ---------------------------------------------------------------- step 3
def undefined_names(products, edges):
    mentioned = {parent for rows in edges.values() for parent, _, _ in rows}
    return sorted(mentioned - set(products))


def find_cycle(edges):
    state = {}

    def visit(node, stack):
        state[node] = "open"
        stack.append(node)
        for parent, _, _ in edges.get(node, []):
            if state.get(parent) == "open":
                return stack[stack.index(parent):] + [parent]
            if parent in state:
                continue
            cycle = visit(parent, stack)
            if cycle:
                return cycle
        stack.pop()
        state[node] = "closed"
        return None

    for node in edges:
        if node in state:
            continue
        cycle = visit(node, [])
        if cycle:
            return cycle
    return None

父级名称里一旦出现笔误，比如把 HVRI.bric 错写成 HVRI.brick，就会产生一条指向文件中根本不存在的产品的边。遍历到这个死胡同就会停下，而它后面的所有协变量看起来都安然无恙。

undefined_names 会收集每条边中提到的父级名称，再从已定义产品的集合里做差集。剩下多出来的那些，要么是笔误，要么就是你忘了登记的产品。

find_cycle 负责确认清单确实是一个有向无环图（DAG）。在真实数据中，产品不可能由它自己构建出来；而一旦出现环路，寻路逻辑就会永远原地打转。

该函数采用带两种标记的深度优先搜索。搜索进入一个节点时，先把它标记为 open；等该节点上方的所有内容都探索完毕，再标记为 closed。如果搜索途中再次遇到一个仍处于 open 状态的节点，说明它绕了一个圈——函数会把找到的整个环返回出来，让你看得一清二楚。

第 4 部分：遍历图

# ---------------------------------------------------------------- step 4
def ancestors(edges, node):
    """Every product that `node` was built from, at any distance."""
    found = set()
    queue = deque([node])
    while queue:
        current = queue.popleft()
        for parent, _, _ in edges.get(current, []):
            if parent in found:
                continue
            found.add(parent)
            queue.append(parent)
    return found


def routes(edges, start, goal):
    """Every path from start up to goal. Safe because step 3 ruled out cycles."""
    found = []
    for parent, relation, confidence in edges.get(start, []):
        step = (parent, relation, confidence)
        if parent == goal:
            found.append([step])
        else:
            for rest in routes(edges, parent, goal):
                found.append([step] + rest)
    return found


def describe(start, route):
    chain = " -> ".join([start] + [parent for parent, _, _ in route])
    weakest = min(route, key=lambda step: RANK[step[2]])[2]
    arithmetic = all(rel in DETERMINISTIC for _, rel, _ in route)
    kind = "deterministic" if arithmetic else "statistical"
    return [chain, f"{kind}, weakest link: {weakest}"]

ancestors 采用的是广度优先搜索（BFS）。不妨想象售票窗口前的一条队伍：先把起始产品放入队列；每一轮取出队首的产品，查出它的父节点，再把尚未见过的父节点逐个追加到队尾。等队列清空时，found 集合里就已经装下了各个层级上的全部祖先。

found 集合还能避免搜索重复访问同一个产品。这一点很重要，因为许多产品共用相同的父节点。

ancestors 告诉你某个协变量是否位于上游，而 routes 告诉你它是如何抵达那里的。

routes 使用带递归的深度优先搜索。对 start 的每个父节点，它先检查该父节点是不是目标：如果是，这一步本身就是一条完整路径；如果不是，函数就调用自身，找出从该父节点到目标的所有路径，再把当前这一步放到每条路径的开头。

这个函数会返回所有路径，这是有意为之。我完整工具的早期版本只报告最短路径，于是报告里永远只显示跳数最少的那一条——哪怕更长的路径其实有着更扎实的证据支撑。

递归在这里之所以安全，是因为第 3 部分已经确认过这张图是 DAG。

describe 会把一条路径转成两行可读文本。第一行是名称链条；第二行说明这条路径属于 deterministic（每一步都是算术运算）还是 statistical（链条中至少有一个模型），并给出沿途最弱的置信级别——毕竟一环薄弱，整条链都不牢靠。

第 5 部分：审计

审计会在图中寻找三种结构：

每张小图展示一种结构及其对应的判定结果：箭头指向目标（FAIL）、箭头从目标指出（FAIL），以及两个协变量挂在同一个共享输入上（REVIEW）。

• 祖先：协变量直接或经由其他产品流入目标。这是最典型的数据泄露，因此工具会将其报告为错误。

• 后代：目标流入了协变量。用孩子来预测它的父节点，泄露程度同样严重，所以这种情况也是错误。

• 共同祖先：两个协变量来自同一个输入。这类问题相对温和，因为二者携带的信息存在重叠，所以工具只会发出警告，交由人工审核。

# ---------------------------------------------------------------- step 5
def audit(products, edges, target, covariates):
    findings = []

    unknown = [n for n in [target, *covariates] if products.get(n) is None]
    if unknown:
        return [("ERROR", f"unknown name: {n}", ["check the spelling"])
                for n in unknown]

    target_ancestors = ancestors(edges, target)
    for cov in covariates:
        if cov in target_ancestors:
            found = routes(edges, target, cov)
            details = [line for r in found for line in describe(target, r)]
            findings.append(("ERROR", f"{cov} is an ancestor of the target "
                                      f"({len(found)} route(s))", details))
        if target in ancestors(edges, cov):
            found = routes(edges, cov, target)
            details = [line for r in found for line in describe(cov, r)]
            findings.append(("ERROR", f"{cov} is a descendant of the target "
                                      f"({len(found)} route(s))", details))

    for a, b in combinations(covariates, 2):
        if a in ancestors(edges, b) or b in ancestors(edges, a):
            findings.append(("ERROR", f"{a} and {b}: one is built from the other", []))
        elif ancestors(edges, a) & ancestors(edges, b):
            shared = sorted(ancestors(edges, a) & ancestors(edges, b))
            findings.append(("WARN", f"{a} and {b} share ancestors", shared))
    return findings

审计从名称检查开始：如果某个协变量的名字拼错了，工具会立刻报错——凡是清单之外的名称，它一概不认识，硬说这个名字安全，那纯属猜测。

接下来，工具会先算出目标的全部祖先，再逐个检查每个协变量是否落在这个集合里。同时它也会算出每个协变量的祖先，看目标是否出现在其中——目标的后代就是这样被揪出来的。

最后，combinations（来自 itertools 模块）会生成所有协变量的两两配对。若一个协变量由另一个构建而来，报错误；若两者共享哪怕一个祖先，则报警告。

第 6 部分：判定结果与退出码

# ---------------------------------------------------------------- step 6
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", default="mini-manifest.yaml")
    parser.add_argument("--target", required=True)
    parser.add_argument("--covariates", nargs="+", required=True)
    args = parser.parse_args()

    products, edges = load(args.manifest)
    missing = undefined_names(products, edges)
    if missing:
        fail(f"manifest error: undefined names: {', '.join(missing)}")
    cycle = find_cycle(edges)
    if cycle:
        fail(f"manifest error: cycle: {' -> '.join(cycle)}")

    findings = audit(products, edges, args.target, args.covariates)
    severities = {severity for severity, _, _ in findings}
    if "ERROR" in severities:
        verdict = "FAIL"
    elif "WARN" in severities:
        verdict = "REVIEW"
    elif ancestors(edges, args.target):
        verdict = "PASS"
    else:
        verdict = "UNTRACED"

    basis = products.get(args.target, {}).get("measurementBasis", "unknown")
    print(f"target      {args.target}  [{basis}]")
    print(f"covariates  {', '.join(args.covariates)}")
    print(f"verdict     {verdict}\n")
    for severity, message, details in findings:
        print(f"  {severity:<5} {message}")
        for line in details:
            print(f"        {line}")

    sys.exit(1 if verdict == "FAIL" else 0)


if __name__ == "__main__":
    main()

main 会读取命令行参数、加载清单文件，先执行两项自检，再运行审计，并将审计发现归入以下四种判定之一：

判定	触发条件	退出码
FAIL	至少存在一个错误	1
REVIEW	仅有警告	0
PASS	未发现任何问题，且目标记录有祖先节点	0
UNTRACED	未发现任何问题，且目标没有任何已记录的祖先节点	0

清单文件损坏或命令格式有误时，程序会以退出码 2 结束。

PASS 和 UNTRACED 值得单独说明。PASS 表示工具沿着一条真实的家族树完整走了一遍，并确认所有协变量都在这棵树之外；UNTRACED 则表示清单中该目标的家族树是空的，遍历根本无从展开。把这种情况也叫作"通过"，未免抬举了这个工具，所以它有了自己的名字。

退出码同样关键。退出码 1 表示检查正常运行并发现了泄漏，退出码 2 则表示检查本身出了问题。CI 流水线必须把这两种情况区分开：发现泄漏，该调整的是协变量；清单损坏，该修复的是文件。

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

案例 1：FEMA 的风险评分

FEMA 的综合风险评分由预期年损失乘以一个社区风险因子得出。这个因子由社会脆弱性评分和社区韧性评分构成，而这两项评分又各自经由不同机构的处理，一路追溯到 ACS 调查数据的列。

FEMA_NRI.risk_score 及其全部上游来源。图中以蓝色和红色绘制的两条路线都始于同一个 ACS 失业率列：一条经由美国人口普查局的韧性估计数据，另一条经由 HVRI 的 BRIC 指数。

假设你想用贫困率和失业率来预测这个风险评分：

python3 mini_lint.py --target FEMA_NRI.risk_score --covariates ACS.EP_POV150 ACS.EP_UNEMP

判定结果为 FAIL。贫困率通过一条路径抵达目标，失业率则通过两条路径抵达，其中一条是统计性的，另一条是确定性的。

判定结果为 FAIL，退出码为 1。

ACS.EP_POV150 通过一条路径抵达目标，而 ACS.EP_UNEMP 通过两条路径抵达。

失业率的第一条路径经过人口普查局的模型，因此工具将其标记为统计性泄漏；第二条路径经过 HVRI 的 BRIC 指数，失业率在那里是直接的组成部分，所以被标记为确定性泄漏。

试着在一堆列名组成的表格里用肉眼追踪这两条路径，你很快就会明白为什么需要图结构。图的遍历能在不到一秒内把两条路径都找出来。

案例 2：后代节点

现在把方向反过来：假设你想预测人口普查局的分数，却把 FEMA 重新发布的同一份数据当作协变量：

python3 mini_lint.py --target CENSUS_CRE.social_vulnerability --covariates FEMA_NRI.social_vulnerability

与前面相反的情形，结果依然是 FAIL。FEMA 转载发布的普查评分，经由一条确定性路径成了目标的后代。

FEMA 的评分直接由普查评分计算而来，把它当作输入特征，等于把答案直接喂到模型手里。该工具将这种情形识别为“后代”并予以捕获。

案例 3：REVIEW、PASS 与 UNTRACED

下面是三次运行，结果应当没有泄漏，或至少接近于没有泄漏：

python3 mini_lint.py --target NVSS.mortality --covariates FEMA_NRI.social_vulnerability HVRI.bric
python3 mini_lint.py --target FEMA_NRI.risk_score --covariates SAT.chirps_rainfall
python3 mini_lint.py --target NVSS.mortality --covariates SAT.chirps_rainfall

三次清理运行的结果：共享两个 ACS 父字段的组合返回 REVIEW，卫星降雨量对照风险评分返回 PASS，而没有任何已登记父字段的目标返回 UNTRACED。

第一次运行返回 REVIEW，因为这两个协变量共享两个 ACS 父字段，向模型传递的信息存在重叠。

第二次运行返回 PASS，因为风险评分有可追溯的来源谱系，而卫星降雨量完全不在这个谱系之内。

第三次运行返回 UNTRACED，因为死亡证明记录是直接统计得来的数据，没有任何已登记的父字段。

这些安静的通过结果与报错同样重要。一个对每个输入都发出警报的检查器毫无用处，所以一套好的测试用例必须包含本应通过的情况。

第 7 步：让 Linter 在遇到坏输入时大声报错

安全工具靠清晰的失败来赢得信任。对泄漏检查器而言，最糟糕的结局是某项检查悄悄跳过了工作，却依然亮出绿色的 PASS。这种情况有三种常见的成因。

陷阱 1：名称拼写错误

想试试的话，把 mini-manifest.yaml 复制为 typo-manifest.yaml，然后将 FEMA_NRI.community_resilience 条目里的 HVRI.bric 改成 HVRI.brick。接着运行下面两条命令：

python3 mini_lint.py --target FEMA_NRI.risk_score --covariates ACS.EP_POV15
python3 mini_lint.py --manifest typo-manifest.yaml --target FEMA_NRI.risk_score --covariates ACS.EP_UNEMP

两个拼写错误，两种退出码。拼错的协变量以退出码 1 报 FAIL，而清单里拼错的父节点则以退出码 2 触发清单错误。

拼错的协变量（ACS.EP_POV15）会以退出码 1 报 FAIL；清单里拼错的父节点（HVRI.brick）则会触发退出码 2 的清单错误。这两种情况都会在任何遍历开始之前终止整个运行。

陷阱 2：重复的 YAML 键

再复制一份清单，命名为 sneaky-manifest.yaml，然后在文件最末尾的 FEMA_NRI.risk_score 块内添加一行：

    derivesFrom: []

现在 risk score 中出现了两个 derivesFrom 键。新建一个 peek.py，看看原生 PyYAML 对此会作何处理：

import yaml

with open("sneaky-manifest.yaml") as fh:
    doc = yaml.safe_load(fh)

print(doc["products"]["FEMA_NRI.risk_score"]["derivesFrom"])

现在运行 peek.py，然后对同一个文件运行 linter：

python3 peek.py
python3 mini_lint.py --manifest sneaky-manifest.yaml --target FEMA_NRI.risk_score --covariates ACS.EP_UNEMP

同一个重复键的两种呈现。普通的 yaml.safe_load 打印出空列表，对此毫无提示；严格加载器则会点名重复的键、给出两处行号，并以退出码 2 结束。

普通的 yaml.safe_load 返回空列表，因为 PyYAML 保留了第二个键，丢掉了全部三条真实边，而且对此一言不发。基于这种加载器构建的 linter 会一条路径都查不到，让所有会泄漏的协变量统统漏网。

严格加载器则以退出码 2 中止运行，并准确点出两处行号。

陷阱 3：空的协变量列表

第三个陷阱很容易被忽视。在 CI 中，协变量列表可能来自某个文件或 shell 变量。一旦文件为空，或者变量名拼错了，命令最终拿到的协变量数量就是零。

我完整工具的早期版本会照单全收，照样打印 PASS——对一项跳过了全部工作的检查来说，这无异于开出了健康无恙的证明。

python3 mini_lint.py --target FEMA_NRI.risk_score --covariates

现在，空的 --covariates 列表会在 argparse 处直接中止运行，任何遍历都还没开始，退出码为 2。

在 mini_lint.py 里，nargs="+" 告诉 argparse：--covariates 至少要接一个值；required=True 则把这个参数本身设为必填。现在只要传入空列表，整条流水线就会立刻亮红灯，退出码为 2。

第 8 步：在 CI 中自动运行检查

CI（持续集成）会在你每次推送代码时替你执行检查。下面这个 GitHub Actions 工作流，会在每次 push 和每个 pull request 时运行 lint 检查。把它保存为 .github/workflows/lineage.yml，放进一个根目录下有 mini_lint.py 和 mini-manifest.yaml 的仓库：

name: lineage-check

on: [push, pull_request]

jobs:
  lint-lineage:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v5
      - uses: actions/setup-python@v6
        with:
          python-version: "3.12"
      - run: pip install pyyaml
      - name: Check covariates against the target's lineage
        run: |
          python3 mini_lint.py \
            --target FEMA_NRI.risk_score \
            --covariates $(cat features.txt)

将协变量名称保存在仓库根目录下的 features.txt 文件里，每行一个：

SAT.chirps_rainfall

$(cat features.txt) 这一部分会把那些名字拼接进命令。有了这份文件，任务顺利通过，状态变绿。之后如果有人加入 ACS.EP_UNEMP 这类泄漏协变量，任务就会以退出码 1 结束，状态变红。要是有人不小心清空了 features.txt，argparse 会以退出码 2 退出，任务同样变红。

退出码	含义	CI 结果
0	检查运行完毕，返回 PASS、REVIEW 或 UNTRACED	绿色
1	检查运行完毕，发现了泄漏	红色
2	清单损坏，或命令格式有误	红色

REVIEW 和 UNTRACED 同样以退出码 0 结束。如果你希望流水线在这两种情况下也停下来，可以修改 main 的最后一行，让 PASS 以外的任何判定都以退出码 1 结束。

配套仓库运行的正是这套工作流，你可以在仓库的 Actions 标签页中查看运行结果。

第 9 步：吸取我的教训

构建完整清单让我体会到：linter 的可靠程度，完全取决于它读取的那张图。我在 SVI 上错了两次，而且方向恰好相反，但根源是同一个习惯——只信文件里的列，却忽视了列背后的方法学。

错误一：SVI 加州文件里有 24 个名字以 EP_ 开头的列，但 CDC 只把其中 16 个纳入指数排名。我的第一版清单把 24 列全部算上，还把 EP_NOINT（宽带订阅数）记成了指数的组成部分。这一列确实在文件里，只是 CDC 排名时并未使用它，于是我把这条边删掉了。

错误二：随后我又矫枉过正，把那八个未参与排名的列统统标成了随指数一同发布的安全旁观者。这八列中有七列与种族和族裔相关。核对原始数值时我发现，这七列相加正好等于 E_MINRTY，而且在全部 9,109 个普查区上，最大偏差为零。

EP_MINRTY 是主题 3 的唯一输入，主题 3 再汇入总指数。

为什么这七列并非旁观者：它们相加恰好等于 EP_MINRTY，它向上跳一层喂给主题 3，主题 3 再向上跳一层喂给总体指数。 EP_NOINT位于虚线框内，与它们打包在同一个文件里发布，却始终置身指数之外。

可见，这七列正是该指数向上两跳的祖先节点。我自己的 linter 就曾把它们放行为 SVI 目标的安全协变量——而这类误放行，恰恰是这套工具要防范的问题。

各产品的交叉验证 R²，由旁边列出的那些列预测得出。凡是基于自身输入列预测，得分都不低于 0.987，而 EP_NOINT只是随指数一同发布，仅达到 0.384。

图表把差距摆得明明白白：那七列种族与族裔数据足以把主题 3 重建到 R² = 0.992，而 EP_NOINT 单独去预测总体指数，只能达到 0.384。

如今在把任何一列标记为安全之前，我都会遵守更严格的规矩：先读方法论文档，再到数据中实测这层关系。

完整版清单用一个名为 coPublishedNonInputs 的字段记录这一结果：它列出所有与指数同文件发布、却完全不参与指数计算的列。就 SVI 而言，如今在册的只有 EP_NOINT 一项。

用完整版工具走得更远

本教程中的迷你 linter 只覆盖核心思路。完整项目还额外包含：

• 一份覆盖美国与全球数据的清单，收录 60 个产品、75 条派生边，包括 CDC PLACES、FEMA 国家风险指数、WorldPop、AlphaEarth 卫星嵌入，以及 WFP 的 HungerMap LIVE

• 每个带派生边的产品都配有一份书面证据说明；一旦早先的论断站不住脚，还会附上 correction 字段

• 一个 --graph 模式，可打印整张派生图

• 内置的八套真实审计案例

• reproduce_svi.py 脚本，可将本文中每个 R² 数值与 CDC 在线文件逐一核对

• 一个锁定版本的 Dockerfile，确保所有数值都能精确复现

想上手试试：

git clone https://github.com/Adeniyikayodee/dependency_manifest.git
cd dependency_manifest
python3 lint_lineage.py
python3 lint_lineage.py --graph
python3 reproduce_svi.py

完整工具在完整清单上的运行结果：60 个产品、75 条派生边，八项审计结果为 5 项 FAIL、1 项 REVIEW、1 项 PASS 和 1 项 UNTRACED。

这份清单并不讳言自身的局限：全球官方综合指数估计有 400 多个，它只覆盖了其中 60 个，而且仍有四条边标记为inferred。第一次审计这份文件就发现了六个错误，其中四个出在我原本已标注为certain或documented的边上。

最有资格撰写这类记录的是各机构自己，因为它们早就以 PDF 文档描述过自己的编制方法。只要在公共数据模式中新增两个字段——derivesFrom和measurementBasis——每个数据生产者就都有了存放这些已知信息的地方。

如果你平时就和公共数据打交道，不妨出一份力：补充自己熟悉的产品，或者对照原始文件，核对那些标记为inferred的边。

结语

公共数据中的目标泄漏，就藏在各机构编制指数的“配方”里。模型只要重新发现其中一条配方，分数就能接近满分——但这样的分数对真实世界几乎说明不了什么。

在本教程中，你完成了以下工作：

• 仅凭 CDC 自带的五个输入列，就以 R² = 0.998 复现了它的 Theme 1

• 把溯源（数字从何而来）与派生（数字由什么计算而来）区分开来

• 编写了一份 YAML 清单，为每条派生边记录关系类型和置信水平

• 构建了一个 linter：用广度优先搜索查找祖先，用深度优先搜索枚举每条路径

• 让 linter 在遇到拼写错误、重复的 YAML 键、循环依赖或空协变量列表时立即报错退出

• 把这项检查接入 GitHub Actions，并通过清晰的退出码反馈结果

在信任公共数据上的高分之前，先问问自己：目标变量是用什么构建出来的？把答案写下来之后，只需几行 Python，就能在每次训练模型时自动完成核查。

本教程所用代码存放在 derives-from-tutorial 中，完整的工具、清单文件和复现脚本都可以在主仓库 dependency_manifest 中找到。该项目已存档至 Zenodo，DOI 为 10.5281/zenodo.22274757，你可以基于 CC0 许可协议自由使用。

参考资料

• CDC/ATSDR 社会脆弱性指数：https://svi.cdc.gov/

• FEMA 国家风险指数技术文档 v1.20（2025 年 12 月）：https://www.fema.gov/sites/default/files/documents/fema_national-risk-index_technical-documentation.pdf

• 美国人口普查局社区韧性估计：https://www.census.gov/programs-surveys/community-resilience-estimates.html

• CDC PLACES 方法论，发表于《慢性疾病预防》（Preventing Chronic Disease），2022 年：https://www.cdc.gov/pcd/issues/2022/21_0459.htm

• Data Commons 数据模型：https://docs.datacommons.org/data_model.html