06 机器学习项目完整流程：从数据拆分、Pipeline 到验证与测试

前言

前五篇分别介绍了机器学习的基本概念、Python 环境、NumPy 数组、Pandas 表格数据，以及 scikit-learn 的统一接口。到这里，我们已经可以写出：

model.fit(X, y)

但一次完整的机器学习实验远不止这一行。拿到数据后立即训练、使用全部数据训练又在原数据上评分、拆分前对全部样本标准化、反复查看测试集结果、只关注分数却不检查数据，这些做法都可能让实验结论失真。代码能运行，也不等于我们能够说明模型为什么这样训练、分数如何得到、别人能否复查。

另一个容易遗漏的环节是基线模型。如果没有简单参照物，即使得到一个看起来不错的数字，也无法判断复杂模型是否真正提供了改善。

本文用代码生成的模拟房价数据，依次完成下面的流程：

明确问题
→ 准备数据
→ 检查数据
→ 拆分数据
→ 建立基线
→ 创建预处理和模型流程
→ 训练
→ 验证
→ 确定方案
→ 最终测试
→ 预测新样本
→ 记录实验信息

重点不是深入线性回归原理，而是把前几篇的知识连接成一次规范、可复查且尽量可复现的实验。

第一步：明确机器学习问题

算法不是项目的起点。选择模型前，至少要回答：希望预测什么，输入特征是什么，输出是连续数值还是类别，模型结果将如何使用，用什么指标判断改善，以及错误预测会造成什么影响。

本文的教学问题是：根据四项模拟房屋特征预测模拟房价。

• area：房屋面积；
• rooms：房间数量；
• building_age：房龄；
• distance_to_center：距离市中心的距离；
• price：需要预测的模拟房价。

price 是连续数值，因此这是回归任务。本文用“万元”帮助理解目标单位，但所有样本及其关系均由代码人为生成，不代表任何真实城市、地区或房地产市场规律，不能用于现实定价、购房或投资决策。一次简单教学实验也不足以证明模型能够投入真实业务。

评价模型时使用平均绝对误差 MAE 和决定系数 R²。MAE 直接反映预测与目标平均相差多少，R²则帮助观察模型相对平均值参照方案解释了多少目标变化。

问题定义还要写清楚预测发生的时间点。假设模型用于房屋信息刚录入时给出参考值，那么输入特征必须在这一时刻已经存在。交易完成后才出现的字段，即使与价格关系很强，也不能作为输入。把“什么时候预测”和“当时知道什么”写明，比盲目增加特征更重要。

指标也应与使用方式对应。MAE 为 10 万元是否可以接受，不能只由算法决定：面向课堂演示、内部初筛和正式定价，误差成本完全不同。本文只比较模型是否超过教学基线，不设置现实业务门槛。真实项目必须在训练前约定可接受误差，避免得到结果后再挑选最有利的解释。

第二步：准备并检查数据

示例固定随机种子，使用 NumPy 生成 500 条样本，再用 Pandas 组成 DataFrame。人为关系是面积和房间数增加时价格总体提高，房龄和距中心距离增加时价格总体降低，并加入随机噪声。噪声用于模拟“同样特征不一定得到完全相同结果”，但它仍只是教学设定。

训练前应检查数据行列数、字段名称、数据类型、缺失值、重复记录、非有限值、特征与标签范围，以及特征和标签样本数是否一致。Pandas 中常用的检查包括：

print(data.head())
print(data.shape)
print(data.dtypes)
print(data.isna().sum())
print(data.describe())

head() 帮助确认字段是否错位，shape 给出行列数，dtypes 检查数值有没有误读为字符串，isna() 统计缺失，describe() 则快速展示范围与分布。数据检查不是为了增加几行输出，而是要在错误数据进入模型前及时停下。本例还检查重复行及 NaN、正负无穷等非有限数值。

本次实际运行得到完整数据形状 (500, 5)，缺失值和重复数据数量均为 0，模拟房价范围为 73.01～356.13。这里只能说明代码生成的数据通过了预设的基础检查，不能据此推断真实房价范围。

模拟数据也不能“随便造一组数”。如果目标与特征完全无关，线性模型无法学到有效关系；如果没有噪声且目标严格由输入公式计算，任务又会简单得不自然。本例同时设置基础线性关系和随机噪声，是为了让基线比较、训练误差与未知样本预测都有可观察的意义。人为关系必须在文章和代码中公开，不能把它包装成采集到的真实规律。

数据检查还应与后续每一步连接。例如发现重复记录时，不是看到数量后继续训练，而要判断它们是合法的多次观测还是复制错误；发现异常范围时，也不能机械删除，而要回到字段来源确认。检查代码提供信号，处理决定仍需要任务背景。

第三步：拆分特征 X 和标签 y

特征列和目标列应明确分开：

feature_columns = [
"area",
"rooms",
"building_age",
"distance_to_center",
]
X = data[feature_columns]
y = data["price"]

X 是特征矩阵，形状遵循“样本数 × 特征数”，本例为 (500, 4)；y 是一维预测目标，形状为 (500,)。两者的样本数量必须一致，同一行特征对应同一行标签。

不能把 price 放入 X。否则模型训练时直接看到了要预测的答案，离线分数可能异常漂亮，但预测新房屋时根本没有真实价格可供输入。这种情况称为目标泄漏，是数据泄漏的一种。

第四步：训练集、验证集和测试集

完整数据需要按职责拆分，而不是全部用于训练和评分。

训练集

训练集用于学习预处理参数和模型参数。标准化器的均值与标准差、线性模型的系数，都只能从训练集获得。开发期间可以在训练集上反复拟合候选方案。

验证集

验证集用于比较基线和候选模型、选择方案或超参数，并观察模型对未参与拟合数据的泛化表现。开发过程可以查看验证结果，但看得越多，方案也越可能逐渐适应这一份验证集，因此仍需独立测试集做最后检查。

测试集

测试集只用于最终评估已经确定的方案，模拟模型面对真正未见数据的表现。它不应参与模型选择、参数调整、特征修改或预处理规则的确定，也不能反复试到得到满意分数。

本文使用约 70% 训练、15% 验证、15% 测试的比例。它只是常见示例，不是固定规则；数据规模、类别分布、时间顺序和同一对象的多条记录都会影响合理拆分方式。

代码执行两次 train_test_split。第一次从完整数据留下 70% 训练集和 30% 临时集；第二次把临时集等分为验证集与测试集：

x_train, x_temp, y_train, y_temp = train_test_split(
X,
y,
test_size=0.30,
random_state=42,
)
x_validation, x_test, y_validation, y_test = train_test_split(
x_temp,
y_temp,
test_size=0.50,
random_state=42,
)

500 条样本最终得到训练集 350 条、验证集 75 条、测试集 75 条。固定 random_state=42 让拆分更容易复查。

随机拆分并非所有任务都适用

本文的样本由同一段代码独立生成，因此随机拆分便于教学。真实数据可能具有时间顺序、用户分组或空间关联。例如用未来月份预测过去、让同一个人的多条记录分散到训练和测试，都会高估模型面对新时间或新对象时的能力。时间序列通常按时间先后拆分，同一对象的记录则可能需要按组拆分。

拆分后还应保存样本标识或拆分规则。只有随机种子而没有原始数据版本，数据增加或排序改变后仍可能得到不同集合。本例数据完全由固定代码生成，所以能够重建；真实项目更适合记录数据快照、查询条件或稳定的样本编号。

第五步：建立基线模型

Baseline Model（基线模型）是一个简单的参考方案。本例使用：

DummyRegressor(strategy="mean")

它不学习面积、房龄等特征之间的关系，只始终预测训练标签的平均值。它不是最终方案，却给复杂模型设定了最低参照标准。

如果一个看起来更专业的模型没有明显超过平均值基线，可能说明特征信息不足、模型方案不合适、数据存在问题，或者评估流程有误。模型名称复杂不能证明它更有效，必须通过同一验证集和同一指标实际比较。

基线也要与任务匹配。回归可以使用均值或中位数，分类可以使用多数类别或简单规则。好的基线不追求复杂，而要容易解释、容易复现，并能代表“不使用候选模型时我们至少能做到什么”。只有候选方案稳定超过它，继续投入复杂度才有依据。

第六步：数据预处理

四个特征的尺度并不相同：面积可能为几十到几百，房间数只有 1 到 6，房龄可到几十，距离又是另一组范围。本例使用 StandardScaler（标准化器）处理数值特征。它会从数据中学习每列均值和标准差，再把特征转换到可比较的尺度。

错误流程是先在完整数据上拟合标准化器，再拆分训练集和测试集。这样验证集、测试集的统计信息会提前进入预处理参数。正确顺序是先拆分，再只用训练集拟合标准化器，并使用同一套规则转换验证集和测试集。

Pipeline（流水线）把预处理和模型组合为一个整体：

Pipeline(
steps=[
("scaler", StandardScaler()),
("regressor", LinearRegression()),
]
)

调用 fit 时，Pipeline 只从传入的训练数据学习；调用 predict 时，它自动应用已经学到的标准化规则，再执行预测。这既减少训练与预测处理不一致，也降低手动操作引入数据泄漏的风险。不过 Pipeline 不能替我们发现业务含义错误的字段，数据边界仍需人为确认。

第七步：训练模型

现在建立两个候选方案：平均值基线，以及 StandardScaler → LinearRegression。训练时只允许传入训练特征与训练标签：

baseline.fit(x_train, y_train)
model.fit(x_train, y_train)

验证集和测试集都不能传给 fit。其中测试集暂时保持封闭；验证集只在下一步比较方案时使用。线性回归原理会在下一篇单独展开，本篇只把它作为结构清晰的候选模型。

第八步：在验证集比较方案

两个方案分别预测同一验证集：

baseline_predictions = baseline.predict(x_validation)
model_predictions = model.predict(x_validation)

MAE（Mean Absolute Error，平均绝对误差）是预测值与真实值绝对误差的平均值，单位与目标相同，越接近 0 越好。R²（决定系数）衡量模型相对“预测平均值”解释目标变化的程度，通常越接近 1 越好；0 表示没有明显优于平均值参照，也可能出现负数。

本次实际验证结果如下：

基线模型验证集 MAE: 50.3327
基线模型验证集 R²: -0.0213
线性回归验证集 MAE: 9.2399
线性回归验证集 R²: 0.9628

代码只在“线性回归 MAE 小于基线 MAE”时选择 Pipeline；否则会选择基线并打印明确警告，不能预先宣称复杂模型有效。本次实际结果满足条件，因此确定 StandardScaler + LinearRegression Pipeline。这只说明它在这份人为生成的数据和当前拆分中优于简单基线，不证明它适合现实房价任务。

第九步：确定方案后进行最终训练

确定方案后有两种常见做法：直接保留仅在训练集上拟合的模型去测试，或者把训练集和验证集合并，用相同方案重新训练，再评估测试集。本文采用第二种，让已经完成模型选择的验证数据也参与最终参数学习。

具体步骤是合并 x_train 与 x_validation、合并对应标签，创建一个全新的相同 Pipeline，再在合并后的开发集上调用 fit。测试集绝不能并入。实际最终开发集有 425 条样本，即 350 条训练样本加 75 条验证样本。

之所以新建 Pipeline，而不是继续在旧模型上随意调用训练，是为了清楚表达“确定方案后，从干净对象开始完成最终训练”，也避免对象内部状态带来理解混乱。

第十步：最终测试

最终模型只在最后预测测试集一次。本次实际结果为：

最终测试集 MAE: 9.4223
最终测试集 R²: 0.9634

测试指标是当前实验对未知数据性能的一次估计，而不是现实业务保证。不能因为测试结果不好就继续调整参数，也不能查看结果后修改特征，再把同一份测试集称为“从未参与开发”的最终测试集。如果测试结果已经影响了方案，就应准备新的独立测试数据，或者明确承认原测试集已经参与开发。

验证集和测试集结果接近，是一个积极信号，但样本来自同一段模拟代码，仍不能替代真实来源、不同时间和不同分布的数据验证。

第十一步：预测一个新样本

预测新房屋时，字段名称、含义和顺序必须与训练特征一致。使用 DataFrame 能保留这些信息：

new_house = pd.DataFrame(
[
{
"area": 105.0,
"rooms": 3,
"building_age": 8.0,
"distance_to_center": 6.5,
}
],
columns=feature_columns,
)
predicted_price = final_model.predict(new_house)[0]

本次实际输出为 222.83 万元。这个数值只是根据人为公式生成的数据所做的流程演示，不代表该房屋在任何真实市场中的价格。

数据泄漏是什么

Data Leakage（数据泄漏）指模型训练或选择过程中，意外使用了预测时本不应该获得的信息。泄漏往往让离线指标异常漂亮，真正面对新数据时却明显下降。

1. 把标签放进特征

预测房价时，把 price 本身或由它直接计算的字段加入 X，等于把答案交给模型。

2. 在完整数据上拟合预处理器

拆分前使用全部样本计算均值、标准差或缺失值填充值，会让验证集和测试集的分布信息进入训练过程。

3. 使用测试集选择模型

在测试集上比较线性回归、决策树和随机森林，再选择分数最高者，测试集就已经承担了验证集的职责，不再是独立最终评估。

4. 拆分前使用全部标签完成特征选择

如果先根据完整数据中各特征与标签的关系决定保留哪些字段，测试标签已经间接影响了模型方案。特征选择也应放进只使用训练信息的流程。

5. 重复样本跨越数据集

同一房屋或同一对象的重复记录同时进入训练和测试，模型可能记住对象特征，而不是学到可泛化规律。拆分前要识别重复和分组关系。

6. 使用预测时不可能获得的信息

用交易完成后才产生的字段预测交易价格，即使字段不叫 price，仍属于时间或业务信息泄漏。判断泄漏不能只看列名，还要问“真实预测时是否已经知道这个值”。

欠拟合与过拟合

欠拟合表示模型过于简单或特征不足，训练集表现差，验证集也差；过拟合表示模型过度适应训练样本，训练表现很好，验证表现却明显下降。比较合理的状态是训练集和验证集均达到任务可接受水平，二者差距不过大，最终测试与验证结果也基本一致。

仅看一个验证分数不足以完整判断欠拟合或过拟合，还需要同时查看训练指标、不同拆分或交叉验证结果，并结合业务误差要求。本篇使用简单线性回归，重点是流程边界，不深入调整模型复杂度。

如何保证实验可复现

可复现并不是“我再运行一次大概也能工作”，而是让别人能够知道数据和结果怎样产生。至少应记录：

本例输出解释器路径和依赖版本，使用固定种子，并把数据生成、拆分和训练都写进脚本。固定随机种子仍不能保证所有硬件、操作系统和库版本产生逐位相同的浮点数，但能显著提高复查能力。若环境版本变化导致末位数值略有差异，应记录版本，而不是把某一次输出当成永远不变的常量。

可以为每次实验保留一条简洁记录：实验时间、数据版本、特征列表、拆分规则、基线指标、候选方案参数、验证结果、最终选择理由和测试结果。若某次方案没有超过基线，也应保留负结果，而不是只记录成功尝试。这样回头查看时，才能区分“代码改了什么”“数据变了什么”和“分数为什么变化”。

复现还要求减少未记录的手工步骤。例如在表格软件中临时删除几行、在 IDE 里手动改一个参数但没有保存，都会让脚本与实际实验不一致。尽可能让数据生成、清洗、拆分、训练和评估都由同一入口执行，并让异常检查失败时明确停止，实验记录才具有可信度。

完整实践代码

以下是 06_ml_workflow.py 的完整内容。测试集只在方案确定并完成最终训练后用于评估。

"""一次包含训练、验证和最终测试的完整机器学习流程。"""
from future import annotations
import sys
from pathlib import Path
from typing import Any
RANDOM_STATE = 42
N_SAMPLES = 500
FEATURE_COLUMNS = [
"area",
"rooms",
"building_age",
"distance_to_center",
]
def configure_console_encoding() -> None:
"""让包含 R² 等字符的输出在常见 Windows 终端中可显示。"""
for stream in (sys.stdout, sys.stderr):
reconfigure = getattr(stream, "reconfigure", None)
if callable(reconfigure):
reconfigure(encoding="utf-8", errors="replace")
def check_environment() -> dict[str, Any]:
"""检查 Python 与直接依赖，并返回本例使用的对象。"""
if sys.version_info < (3, 8):
raise RuntimeError("需要 Python 3.8 或更高版本。")
try:
import numpy as np
import pandas as pd
import sklearn
from sklearn.dummy import DummyRegressor
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error, r2_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
except ImportError as exc:
missing_name = exc.name or "未知依赖"
raise RuntimeError(
f"缺少依赖 {missing_name!r}。请确认当前 Conda 环境已安装 "
"NumPy、Pandas 和 scikit-learn。"
) from exc

return {
"np": np,
"pd": pd,
"sklearn": sklearn,
"DummyRegressor": DummyRegressor,
"LinearRegression": LinearRegression,
"mean_absolute_error": mean_absolute_error,
"r2_score": r2_score,
"train_test_split": train_test_split,
"Pipeline": Pipeline,
"StandardScaler": StandardScaler,
}
def generate_dataset(deps: dict[str, Any], n_samples: int = N_SAMPLES) -> Any:
"""生成只用于教学的模拟房屋数据。"""
if n_samples < 400:
raise ValueError("教学数据至少需要 400 个样本。")
np = deps["np"]
pd = deps["pd"]
rng = np.random.default_rng(RANDOM_STATE)

area = rng.uniform(45.0, 220.0, n_samples)
rooms = np.clip(
np.rint(area / 38.0 + rng.normal(0.0, 0.65, n_samples)),
1,
6,
).astype(int)
building_age = rng.uniform(0.0, 45.0, n_samples)
distance_to_center = rng.uniform(0.5, 25.0, n_samples)
noise = rng.normal(0.0, 12.0, n_samples)

# 下面的关系由代码人为设定，只为演示完整流程，单位可理解为“万元”。
price = (
130.0
+ 0.90 * area
+ 8.0 * rooms
– 1.20 * building_age
– 2.50 * distance_to_center
+ noise
)

return pd.DataFrame(
{
"area": area,
"rooms": rooms,
"building_age": building_age,
"distance_to_center": distance_to_center,
"price": price,
}
)
def validate_dataset(data: Any, deps: dict[str, Any]) -> tuple[int, int]:
"""检查形状、类型、缺失、重复和非有限数值。"""
np = deps["np"]
pd = deps["pd"]
expected_columns = FEATURE_COLUMNS + ["price"]
if not isinstance(data, pd.DataFrame):
raise ValueError("数据必须是 Pandas DataFrame。")
if data.empty:
raise ValueError("数据为空，无法继续。")
if list(data.columns) != expected_columns:
raise ValueError(f"字段或顺序不符合预期：{expected_columns}")
if not all(pd.api.types.is_numeric_dtype(data[column]) for column in data.columns):
raise ValueError("本例所有字段都必须是数值类型。")

missing_count = int(data.isna().sum().sum())
duplicate_count = int(data.duplicated().sum())
if missing_count != 0:
raise ValueError(f"发现 {missing_count} 个缺失值。")
if duplicate_count != 0:
raise ValueError(f"发现 {duplicate_count} 条重复数据。")
if not np.isfinite(data.to_numpy(dtype=float)).all():
raise ValueError("数据中存在无穷大或非有限数值。")
if (data["area"] &lt;= 0).any() or (data["price"] &lt;= 0).any():
raise ValueError("面积和模拟房价应为正数。")

return missing_count, duplicate_count
def split_dataset(X: Any, y: Any, deps: dict[str, Any]) -> tuple[Any, …]:
"""用两次拆分得到约 70%/15%/15% 的训练、验证和测试集。"""
if X.shape[0] != len(y):
raise ValueError("X 与 y 的样本数量不一致。")
if list(X.columns) != FEATURE_COLUMNS:
raise ValueError("特征列名称或顺序不符合预期。")
train_test_split = deps["train_test_split"]
x_train, x_temp, y_train, y_temp = train_test_split(
X,
y,
test_size=0.30,
random_state=RANDOM_STATE,
)
x_validation, x_test, y_validation, y_test = train_test_split(
x_temp,
y_temp,
test_size=0.50,
random_state=RANDOM_STATE,
)

total_after_split = len(x_train) + len(x_validation) + len(x_test)
if total_after_split != len(X):
raise ValueError("拆分后的样本总数与原数据不一致。")

return x_train, x_validation, x_test, y_train, y_validation, y_test
def build_baseline(deps: dict[str, Any]) -> Any:
"""创建始终预测训练标签平均值的基线模型。"""
return deps"DummyRegressor"
def build_pipeline(deps: dict[str, Any]) -> Any:
"""创建标准化与线性回归组成的 Pipeline。"""
return deps["Pipeline"](
steps=[
("scaler", deps"StandardScaler"),
("regressor", deps"LinearRegression"),
]
)
def evaluate_regression(
model: Any,
X: Any,
y: Any,
deps: dict[str, Any],
) -> tuple[float, float]:
"""返回平均绝对误差 MAE 和决定系数 R²。"""
predictions = model.predict(X)
mae = float(deps["mean_absolute_error"](y, predictions))
r2 = float(deps["r2_score"](y, predictions))
return mae, r2
def prepare_final_training_data(
x_train: Any,
x_validation: Any,
y_train: Any,
y_validation: Any,
deps: dict[str, Any],
) -> tuple[Any, Any]:
"""合并训练集和验证集，但明确不包含测试集。"""
pd = deps["pd"]
x_development = pd.concat([x_train, x_validation], axis=0).reset_index(drop=True)
y_development = pd.concat([y_train, y_validation], axis=0).reset_index(drop=True)
if len(x_development) != len(y_development):
raise ValueError("最终开发集的 X 与 y 样本数量不一致。")
return x_development, y_development
def predict_new_sample(final_model: Any, deps: dict[str, Any]) -> float:
"""预测一条字段名称和顺序均与训练数据一致的新样本。"""
pd = deps["pd"]
new_house = pd.DataFrame(
[
{
"area": 105.0,
"rooms": 3,
"building_age": 8.0,
"distance_to_center": 6.5,
}
],
columns=FEATURE_COLUMNS,
)
return float(final_model.predict(new_house)[0])
def main() -> None:
configure_console_encoding()
deps = check_environment()
np = deps["np"]
pd = deps["pd"]
print("=== 环境信息 ===")
print(f"Python: {sys.version.split()[0]}")
print(f"NumPy: {np.__version__}")
print(f"Pandas: {pd.__version__}")
print(f"Scikit-learn: {deps['sklearn'].__version__}")
print(f"Interpreter: {Path(sys.executable).resolve()}")

data = generate_dataset(deps)
missing_count, duplicate_count = validate_dataset(data, deps)
X = data[FEATURE_COLUMNS].copy()
y = data["price"].copy()

print("\\n=== 数据检查 ===")
print(f"完整数据形状: {data.shape}")
print(f"特征矩阵形状: {X.shape}")
print(f"标签形状: {y.shape}")
print(f"特征名称: {FEATURE_COLUMNS}")
print(f"缺失值数量: {missing_count}")
print(f"重复数据数量: {duplicate_count}")
print(f"特征值全部有限: {np.isfinite(X.to_numpy(dtype=float)).all()}")
print(f"模拟房价范围: {y.min():.2f} ～ {y.max():.2f}")

(
x_train,
x_validation,
x_test,
y_train,
y_validation,
y_test,
) = split_dataset(X, y, deps)

print("\\n=== 数据拆分 ===")
print(f"训练集样本数: {len(x_train)}")
print(f"验证集样本数: {len(x_validation)}")
print(f"测试集样本数: {len(x_test)}")

baseline = build_baseline(deps)
model = build_pipeline(deps)
baseline.fit(x_train, y_train)
model.fit(x_train, y_train)

baseline_mae, baseline_r2 = evaluate_regression(
baseline, x_validation, y_validation, deps
)
model_mae, model_r2 = evaluate_regression(model, x_validation, y_validation, deps)

print("\\n=== 验证集比较 ===")
print(f"基线模型验证集 MAE: {baseline_mae:.4f}")
print(f"基线模型验证集 R²: {baseline_r2:.4f}")
print(f"线性回归验证集 MAE: {model_mae:.4f}")
print(f"线性回归验证集 R²: {model_r2:.4f}")

if model_mae &lt; baseline_mae:
selected_name = "StandardScaler + LinearRegression Pipeline"
final_model = build_pipeline(deps)
else:
selected_name = "DummyRegressor 基线模型"
final_model = build_baseline(deps)
print("警告: 线性回归的验证集 MAE 未优于基线，不能宣称复杂方案有效。")
print(f"选择的模型方案: {selected_name}")

x_development, y_development = prepare_final_training_data(
x_train,
x_validation,
y_train,
y_validation,
deps,
)
final_model.fit(x_development, y_development)
test_mae, test_r2 = evaluate_regression(final_model, x_test, y_test, deps)
predicted_price = predict_new_sample(final_model, deps)

print("\\n=== 最终训练与测试 ===")
print(f"最终开发集样本数: {len(x_development)}")
print(f"最终测试集 MAE: {test_mae:.4f}")
print(f"最终测试集 R²: {test_r2:.4f}")
print(f"新房屋预测价格: {predicted_price:.2f} 万元（仅教学演示）")
if name == "main":
try:
main()
except (RuntimeError, ValueError) as exc:
print(f"程序无法继续：{exc}", file=sys.stderr)
sys.exit(1)
except Exception as exc: # 给初学者保留清楚的错误入口
print(f"出现未预期错误：{type(exc).name}: {exc}", file=sys.stderr)
sys.exit(1)

在项目根目录运行：

python 06_ml_workflow.py

本文在现有 Conda 环境中实际运行，程序完整输出如下：

=== 环境信息 ===
Python: 3.11.4
NumPy: 1.24.3
Pandas: 1.5.3
Scikit-learn: 1.3.0
Interpreter: C:\\Users\\32981\\anaconda3\\python.exe
=== 数据检查 ===
完整数据形状: (500, 5)
特征矩阵形状: (500, 4)
标签形状: (500,)
特征名称: ['area', 'rooms', 'building_age', 'distance_to_center']
缺失值数量: 0
重复数据数量: 0
特征值全部有限: True
模拟房价范围: 73.01 ～ 356.13
=== 数据拆分 ===
训练集样本数: 350
验证集样本数: 75
测试集样本数: 75
=== 验证集比较 ===
基线模型验证集 MAE: 50.3327
基线模型验证集 R²: -0.0213
线性回归验证集 MAE: 9.2399
线性回归验证集 R²: 0.9628
选择的模型方案: StandardScaler + LinearRegression Pipeline
=== 最终训练与测试 ===
最终开发集样本数: 425
最终测试集 MAE: 9.4223
最终测试集 R²: 0.9634
新房屋预测价格: 222.83 万元（仅教学演示）

小结：让训练与评估边界清晰

一次可信的机器学习实验，核心不是尽快调用 fit，而是先明确问题和指标，检查数据边界，拆分训练、验证和测试职责，建立简单基线，只从训练数据学习预处理与模型参数，再用验证集选择方案，最后一次性评估测试集。

Pipeline 能帮助保持预处理一致，基线能告诉我们复杂方案是否真的改善，独立测试集则保护最终评估不被开发过程污染。实验还需要记录随机种子、环境版本、特征顺序、数据方法和模型参数，才能让结果被复查。

本例的高分来自一份按线性关系人为生成的教学数据，不能外推到现实房价。真正项目还要面对数据来源、时间变化、业务成本、隐私和部署监控等问题。

下一篇预告

下一篇是第 7 篇《线性回归：预测连续数值》。我们会在本篇完整流程的基础上，进一步理解线性回归如何表示特征与连续目标之间的关系，以及系数、误差和评估指标应当怎样解释。