准备知识33 章节

软件工程中的单元测试，在 AI 数据工程里演化为了数据与代码契约测试 (Data & Model Contracts)。这一章我们将学习如何将混杂脚本重构为模块化函数（load.py, features.py, train.py），并使用 pytest 编写 5 大防御性契约测试。

初学者直觉：汽车生产线的质检卡规 (Quality Inspection)

把数据与模型代码比作汽车发动机的生产车间：

模块化重构：将所有零部件独立拆分为 load_data()（进货卡）、build_features()（打磨卡）、train_model()（组装卡）。
pytest 契约测试：车间里的自动卡规与质检员。每制作完一个零部件，卡规会自动测量“尺寸是否完全吻合”（Shape 匹配）、“钢材是否有气孔”（无缺失值）、“零部件数量是否守恒”（样本数切分守恒）。

1. 模块化重构与 5 大契约测试

1.1 模块化解耦规范

将单体 Notebook 代码重构拆分为 src/ 下的三大独立功能模块：

src/load.py：仅负责数据读取、校验与类型转换。
src/features.py：仅负责特征衍生、清洗与 ColumnTransformer 构建。
src/train.py：仅负责模型训练与评估指标计算。

1.2 5 大有意义的数据与模型契约测试

测试不是去测试第三方库本身（如测试 np.sin 是否正确），而是测试业务数据与代码行为的边界：

契约 1：Schema 与列名契约（必须包含所有指定特征列）。
契约 2：无非法缺失值契约（预处理输出的矩阵缺失值计数必须为 0）。
契约 3：切分数量守恒契约（ $N_{\text{train}} + N_{\text{test}} = N_{\text{total}}$ ）。
契约 4：矩阵 Shape 维度匹配契约（特征矩阵行数与标签 $y$ 的长度必须相等）。
契约 5：模型预测概率范围契约（输出概率必须严密落于 $[0, 1]$ 闭区间）。

2. 手算演练

手算练习：数据切分与维度守恒契约断言手算

假设原始读入数据集总行数 $N_{\text{raw}} = 250$ ：执行 train_test_split(..., test_size=0.2) 进行切分：

期望测试集与训练集行数计算：

N_{\text{test}} = \lfloor 250 \times 0.2 \rfloor = 50

N_{\text{train}} = 250 - 50 = 200

手算断言守恒等式：

N_{\text{train}} + N_{\text{test}} = 200 + 50 = 250 = N_{\text{raw}}

测试代码断言条件： assert len(X_train) + len(X_test) == len(X_raw) 手算推导成立！若结果不相等，说明在切分过程中有数据被无故丢弃或泄漏复写。

3. 动手实战：编写完整的 pytest 自动化契约测试脚本

下面的代码演示如何使用 pytest 的 fixture 脚手架编写 5 大契约测试：

pytest 5 大工程契约自动化测试代码

import pytest

import numpy as np

import pandas as pd

from sklearn.model_selection import train_test_split

from sklearn.linear_model import LogisticRegression

# 1. 模拟生成真实业务数据

def get_sample_data():

np.random.seed(42)

df = pd.DataFrame({

'age': np.random.randint(18, 65, 100),

'score': np.random.uniform(50, 100, 100),

'target': np.random.choice([0, 1], 100)

})

return df

# 2. 编写 5 大契约测试函数

def test_schema_contract():

"""契约 1: Schema 与关键列名必须存在"""

df = get_sample_data()

required_columns = {'age', 'score', 'target'}

assert required_columns.issubset(df.columns), "❌ 数据集中缺少必须的特征列！"

def test_no_missing_contract():

"""契约 2: 预处理后核心列不得含有 NaN"""

df = get_sample_data()

assert df[['age', 'score']].isnull().sum().sum() == 0, "❌ 特征列中存在未处理的缺失值 NaN！"

def test_split_conservation_contract():

"""契约 3: 切分样本数量守恒契约"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)

assert len(X_tr) + len(X_te) == len(X), "❌ 切分后训练集与测试集行数和不等于原始总行数！"

def test_shape_alignment_contract():

"""契约 4: 特征矩阵行数与标签 y 长度必须绝对匹配"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

assert X.shape[0] == len(y), "❌ 特征矩阵 X 的行数与标签 y 的长度不一致！"

def test_prediction_probability_contract():

"""契约 5: 模型输出概率值必须落入 [0.0, 1.0] 区间"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

model = LogisticRegression().fit(X, y)

probs = model.predict_proba(X)[:, 1]

assert np.all((probs >= 0.0) & (probs <= 1.0)), "❌ 模型输出概率超出了 [0, 1] 合法闭区间！"

# 运行测试集 (演示触发)

if __name__ == "__main__":

print("================ 开始执行 pytest 5 大契约测试 ================")

test_schema_contract()

test_no_missing_contract()

test_split_conservation_contract()

test_shape_alignment_contract()

test_prediction_probability_contract()

print("✅ 所有 5 大工程与数据质量契约测试 100% 顺利通过！")

import pytest
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression

# 1. 模拟生成真实业务数据
def get_sample_data():
    np.random.seed(42)
    df = pd.DataFrame({
        'age': np.random.randint(18, 65, 100),
        'score': np.random.uniform(50, 100, 100),
        'target': np.random.choice([0, 1], 100)
    })
    return df

# 2. 编写 5 大契约测试函数
def test_schema_contract():
    """契约 1: Schema 与关键列名必须存在"""
    df = get_sample_data()
    required_columns = {'age', 'score', 'target'}
    assert required_columns.issubset(df.columns), "❌ 数据集中缺少必须的特征列！"

def test_no_missing_contract():
    """契约 2: 预处理后核心列不得含有 NaN"""
    df = get_sample_data()
    assert df[['age', 'score']].isnull().sum().sum() == 0, "❌ 特征列中存在未处理的缺失值 NaN！"

def test_split_conservation_contract():
    """契约 3: 切分样本数量守恒契约"""
    df = get_sample_data()
    X = df[['age', 'score']]
    y = df['target']
    X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)
    assert len(X_tr) + len(X_te) == len(X), "❌ 切分后训练集与测试集行数和不等于原始总行数！"

def test_shape_alignment_contract():
    """契约 4: 特征矩阵行数与标签 y 长度必须绝对匹配"""
    df = get_sample_data()
    X = df[['age', 'score']]
    y = df['target']
    assert X.shape[0] == len(y), "❌ 特征矩阵 X 的行数与标签 y 的长度不一致！"

def test_prediction_probability_contract():
    """契约 5: 模型输出概率值必须落入 [0.0, 1.0] 区间"""
    df = get_sample_data()
    X = df[['age', 'score']]
    y = df['target']
    model = LogisticRegression().fit(X, y)
    probs = model.predict_proba(X)[:, 1]
    assert np.all((probs >= 0.0) & (probs <= 1.0)), "❌ 模型输出概率超出了 [0, 1] 合法闭区间！"

# 运行测试集 (演示触发)
if __name__ == "__main__":
    print("================ 开始执行 pytest 5 大契约测试 ================")
    test_schema_contract()
    test_no_missing_contract()
    test_split_conservation_contract()
    test_shape_alignment_contract()
    test_prediction_probability_contract()
    print("✅ 所有 5 大工程与数据质量契约测试 100% 顺利通过！")

4. 常见错误与避坑指南

现象	先问什么	处理方式
测试代码里写满了对 Scikit-Learn 底层算法的测试	是在测试第三方库还是测试自己的数据边界	严禁测试 `LogisticRegression` 的收敛数学公式，测试应该集中于数据 schema、Shape 匹配与切分契约
运行 `pytest` 时找不到 `src` 下的模块	是否缺少 `__init__.py` 或环境变量	在 `src/` 目录下创建空白 `__init__.py`，并在根目录运行 `pytest`
断言浮点数相等时使用 `assert a == b` 导致失败	是否存在浮点精度误差	使用 `pytest.approx` 或 `np.testing.assert_allclose(a, b, rtol=1e-5)` 断言浮点数

检查清单 (Checklist)

能说明将单体 Notebook 代码拆分为 load.py, features.py, train.py 模块化文件的优势。
能解释并编写数据契约测试中的 Schema 校验与无缺失值断言。
能手算切分数量守恒与 Shape 维度匹配契约条件。
能使用 pytest 编写 5 大有意义的数据与模型契约测试。
能说明为什么测试的重点应当是数据边界与代码契约，而非第三方库本身。

自测题

在机器学习工程中，为什么说“测试第三方库算法本身的正确性”是低效的测试？测试应该关注什么？
解释：在对模型预测概率做断言时，为什么应该断言 0.0 <= prob <= 1.0 而不是断言 prob in [0, 1]？
如果上游数据库将 gender 字段的枚举值从 ['M', 'F'] 突然变成了 [1, 2]，哪个契约测试会第一时间报警？
【代码阅读题】在上一节代码中，test_split_conservation_contract() 如果发生断言失败，说明代码中存在什么潜在漏洞？

点击查看参考答案

因为 Scikit-Learn、PyTorch 等第三方成熟库自身已有极完善的单元测试；我们自己的测试应当关注自身业务数据的 Schema 契约、预处理后的 Shape 匹配、缺失值消除与切分数量守恒。
因为概率值是连续的浮点数（如 0.85），在 $[0.0, 1.0]$ 闭区间内；而 prob in [0, 1] 表示集合离散判断（只允许概率恰好等于整数 0 或 1），会导致浮点概率预测报出误诊错误。
test_schema_contract() 或特征转换模块的契约测试会第一时间报错，因为类别值的改变会破坏后续 One-Hot 编码或数值转换的契约条件。
说明在切分数据时存在严重的索引丢失（如无故删除了行）、数据过滤逻辑位置错乱、或者是切分逻辑产生了非预期的重复复写。

下一步

进入下一个专项 03. 结构化实验记录 runs/ 与误分类错误分析，学习在 runs/ 目录中自动化记录 Git Commit、数据哈希、超参数与指标，并手写误分类错误分析。

参考资料

pytest documentation: Getting Started：pytest 官方使用指南。
scikit-learn: Common Pitfalls and Testing：代码边界与陷阱。
Python logging HOWTO：Python 日志与调试规范。

本页目录

从一个问题开始

1. 模块化重构与 5 大契约测试

1.1 模块化解耦规范

1.2 5 大有意义的数据与模型契约测试

2. 手算演练

手算练习：数据切分与维度守恒契约断言手算

3. 动手实战：编写完整的 pytest 自动化契约测试脚本

4. 常见错误与避坑指南

检查清单 (Checklist)

自测题

下一步

参考资料

核心概念术语表

\text{pytest.fixture}

提供可复用测试上下文的数据脚手架

\text{assert condition}

数据契约断言

关联推荐笔记

查看全部

01. 规范仓库结构、YAML 配置与随机性控制

工程习惯 50 分钟

03. 结构化实验记录 runs/ 与误分类错误分析

实验跟踪 60 分钟

从一个问题开始

初学者直觉：汽车生产线的质检卡规 (Quality Inspection)

把数据与模型代码比作汽车发动机的生产车间：

模块化重构：将所有零部件独立拆分为 load_data()（进货卡）、build_features()（打磨卡）、train_model()（组装卡）。
pytest 契约测试：车间里的自动卡规与质检员。每制作完一个零部件，卡规会自动测量“尺寸是否完全吻合”（Shape 匹配）、“钢材是否有气孔”（无缺失值）、“零部件数量是否守恒”（样本数切分守恒）。

1. 模块化重构与 5 大契约测试

1.1 模块化解耦规范

将单体 Notebook 代码重构拆分为 src/ 下的三大独立功能模块：

src/load.py：仅负责数据读取、校验与类型转换。
src/features.py：仅负责特征衍生、清洗与 ColumnTransformer 构建。
src/train.py：仅负责模型训练与评估指标计算。

1.2 5 大有意义的数据与模型契约测试

测试不是去测试第三方库本身（如测试 np.sin 是否正确），而是测试业务数据与代码行为的边界：

契约 1：Schema 与列名契约（必须包含所有指定特征列）。
契约 2：无非法缺失值契约（预处理输出的矩阵缺失值计数必须为 0）。
契约 3：切分数量守恒契约（ $N_{\text{train}} + N_{\text{test}} = N_{\text{total}}$ ）。
契约 4：矩阵 Shape 维度匹配契约（特征矩阵行数与标签 $y$ 的长度必须相等）。
契约 5：模型预测概率范围契约（输出概率必须严密落于 $[0, 1]$ 闭区间）。

2. 手算演练

手算练习：数据切分与维度守恒契约断言手算

假设原始读入数据集总行数 $N_{\text{raw}} = 250$ ：执行 train_test_split(..., test_size=0.2) 进行切分：

期望测试集与训练集行数计算：

N_{\text{test}} = \lfloor 250 \times 0.2 \rfloor = 50

N_{\text{train}} = 250 - 50 = 200

手算断言守恒等式：

N_{\text{train}} + N_{\text{test}} = 200 + 50 = 250 = N_{\text{raw}}

测试代码断言条件： assert len(X_train) + len(X_test) == len(X_raw) 手算推导成立！若结果不相等，说明在切分过程中有数据被无故丢弃或泄漏复写。

3. 动手实战：编写完整的 pytest 自动化契约测试脚本

下面的代码演示如何使用 pytest 的 fixture 脚手架编写 5 大契约测试：

pytest 5 大工程契约自动化测试代码

import pytest

import numpy as np

import pandas as pd

from sklearn.model_selection import train_test_split

from sklearn.linear_model import LogisticRegression

# 1. 模拟生成真实业务数据

def get_sample_data():

np.random.seed(42)

df = pd.DataFrame({

'age': np.random.randint(18, 65, 100),

'score': np.random.uniform(50, 100, 100),

'target': np.random.choice([0, 1], 100)

})

return df

# 2. 编写 5 大契约测试函数

def test_schema_contract():

"""契约 1: Schema 与关键列名必须存在"""

df = get_sample_data()

required_columns = {'age', 'score', 'target'}

assert required_columns.issubset(df.columns), "❌ 数据集中缺少必须的特征列！"

def test_no_missing_contract():

"""契约 2: 预处理后核心列不得含有 NaN"""

df = get_sample_data()

assert df[['age', 'score']].isnull().sum().sum() == 0, "❌ 特征列中存在未处理的缺失值 NaN！"

def test_split_conservation_contract():

"""契约 3: 切分样本数量守恒契约"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)

assert len(X_tr) + len(X_te) == len(X), "❌ 切分后训练集与测试集行数和不等于原始总行数！"

def test_shape_alignment_contract():

"""契约 4: 特征矩阵行数与标签 y 长度必须绝对匹配"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

assert X.shape[0] == len(y), "❌ 特征矩阵 X 的行数与标签 y 的长度不一致！"

def test_prediction_probability_contract():

"""契约 5: 模型输出概率值必须落入 [0.0, 1.0] 区间"""

df = get_sample_data()

X = df[['age', 'score']]

y = df['target']

model = LogisticRegression().fit(X, y)

probs = model.predict_proba(X)[:, 1]

assert np.all((probs >= 0.0) & (probs <= 1.0)), "❌ 模型输出概率超出了 [0, 1] 合法闭区间！"

# 运行测试集 (演示触发)

if __name__ == "__main__":

print("================ 开始执行 pytest 5 大契约测试 ================")

test_schema_contract()

test_no_missing_contract()

test_split_conservation_contract()

test_shape_alignment_contract()

test_prediction_probability_contract()

print("✅ 所有 5 大工程与数据质量契约测试 100% 顺利通过！")

import pytest
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression

4. 常见错误与避坑指南

现象	先问什么	处理方式
测试代码里写满了对 Scikit-Learn 底层算法的测试	是在测试第三方库还是测试自己的数据边界	严禁测试 `LogisticRegression` 的收敛数学公式，测试应该集中于数据 schema、Shape 匹配与切分契约
运行 `pytest` 时找不到 `src` 下的模块	是否缺少 `__init__.py` 或环境变量	在 `src/` 目录下创建空白 `__init__.py`，并在根目录运行 `pytest`
断言浮点数相等时使用 `assert a == b` 导致失败	是否存在浮点精度误差	使用 `pytest.approx` 或 `np.testing.assert_allclose(a, b, rtol=1e-5)` 断言浮点数

检查清单 (Checklist)

能说明将单体 Notebook 代码拆分为 load.py, features.py, train.py 模块化文件的优势。
能解释并编写数据契约测试中的 Schema 校验与无缺失值断言。
能手算切分数量守恒与 Shape 维度匹配契约条件。
能使用 pytest 编写 5 大有意义的数据与模型契约测试。
能说明为什么测试的重点应当是数据边界与代码契约，而非第三方库本身。

自测题

在机器学习工程中，为什么说“测试第三方库算法本身的正确性”是低效的测试？测试应该关注什么？
解释：在对模型预测概率做断言时，为什么应该断言 0.0 <= prob <= 1.0 而不是断言 prob in [0, 1]？
如果上游数据库将 gender 字段的枚举值从 ['M', 'F'] 突然变成了 [1, 2]，哪个契约测试会第一时间报警？
【代码阅读题】在上一节代码中，test_split_conservation_contract() 如果发生断言失败，说明代码中存在什么潜在漏洞？

点击查看参考答案

因为 Scikit-Learn、PyTorch 等第三方成熟库自身已有极完善的单元测试；我们自己的测试应当关注自身业务数据的 Schema 契约、预处理后的 Shape 匹配、缺失值消除与切分数量守恒。
因为概率值是连续的浮点数（如 0.85），在 $[0.0, 1.0]$ 闭区间内；而 prob in [0, 1] 表示集合离散判断（只允许概率恰好等于整数 0 或 1），会导致浮点概率预测报出误诊错误。
test_schema_contract() 或特征转换模块的契约测试会第一时间报错，因为类别值的改变会破坏后续 One-Hot 编码或数值转换的契约条件。
说明在切分数据时存在严重的索引丢失（如无故删除了行）、数据过滤逻辑位置错乱、或者是切分逻辑产生了非预期的重复复写。

下一步

参考资料

pytest documentation: Getting Started：pytest 官方使用指南。
scikit-learn: Common Pitfalls and Testing：代码边界与陷阱。
Python logging HOWTO：Python 日志与调试规范。