综合实战：一条完整的数据管道

本章收尾。把前八节的内容串成一个真实项目：从原始 CSV 到可用于训练的数据集。

---

任务

给定一份学生成绩原始数据（有缺失、有重复、有脏值），要求产出：

1. 一份清洗后的数据
2. 一份统计报告
3. 一张可视化图
4. 转成 PyTorch 能直接用的张量

---

完整代码

"""成绩数据处理管道：清洗 → 统计 → 可视化 → 张量。"""
import argparse, logging
from pathlib import Path

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import torch

log = logging.getLogger(__name__)
SUBJECTS = ["语文", "数学", "英语"]


def load(path: Path) -> pd.DataFrame:
    """读 CSV，自动兼容 utf-8 和 gbk 两种编码。"""
    if not path.exists():
        raise FileNotFoundError(f"找不到文件：{path}")
    try:
        return pd.read_csv(path, encoding="utf-8")
    except UnicodeDecodeError:
        log.warning("utf-8 解码失败，改用 gbk")
        return pd.read_csv(path, encoding="gbk")


def clean(df: pd.DataFrame) -> pd.DataFrame:
    """去重、转类型、处理缺失和越界值。"""
    before = len(df)
    df = df.drop_duplicates().copy()
    log.info("去重：%d → %d 行", before, len(df))

    for col in SUBJECTS:
        # 脏值（"缺考"、"-"）会变成 NaN，而不是让程序崩掉
        df[col] = pd.to_numeric(df[col], errors="coerce")
        # 越界分数视为录入错误，一并当缺失处理
        bad = ~df[col].between(0, 100)
        if bad.sum():
            log.warning("%s 有 %d 个越界分数，按缺失处理", col, bad.sum())
            df.loc[bad, col] = np.nan

    # 三科全缺的行没有分析价值，直接丢弃
    df = df.dropna(subset=SUBJECTS, how="all")
    # 部分缺失用该科均值补，并记录补了多少
    for col in SUBJECTS:
        n = df[col].isna().sum()
        if n:
            log.info("%s 补了 %d 个均值", col, n)
            df[col] = df[col].fillna(df[col].mean())
    return df


def report(df: pd.DataFrame) -> pd.DataFrame:
    """按班级汇总各科表现。"""
    df["总分"] = df[SUBJECTS].sum(axis=1)
    stats = df.groupby("班级")[SUBJECTS].agg(["mean", "max", "min"]).round(1)
    log.info("共 %d 个班级", df["班级"].nunique())
    return stats


def plot(df: pd.DataFrame, out: Path) -> None:
    fig, axes = plt.subplots(1, 2, figsize=(12, 4))
    axes[0].hist(df["总分"], bins=20, edgecolor="black")
    axes[0].set_title("Total score distribution")
    df.groupby("班级")["总分"].mean().plot(kind="bar", ax=axes[1])
    axes[1].set_title("Average by class")
    fig.tight_layout()
    fig.savefig(out, dpi=120)
    log.info("图已保存：%s", out)


def to_tensor(df: pd.DataFrame) -> torch.Tensor:
    """标准化后转成张量，可直接喂给模型。"""
    x = df[SUBJECTS].to_numpy(dtype=np.float32)
    x = (x - x.mean(axis=0)) / x.std(axis=0)     # 按列标准化
    return torch.from_numpy(x)


def main():
    p = argparse.ArgumentParser(description="成绩数据处理管道")
    p.add_argument("--input", type=Path, required=True)
    p.add_argument("--outdir", type=Path, default=Path("out"))
    args = p.parse_args()

    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s [%(levelname)s] %(message)s")
    args.outdir.mkdir(parents=True, exist_ok=True)

    df = clean(load(args.input))
    stats = report(df)

    df.to_csv(args.outdir / "cleaned.csv", index=False, encoding="utf-8-sig")
    stats.to_csv(args.outdir / "report.csv", encoding="utf-8-sig")
    plot(df, args.outdir / "chart.png")

    t = to_tensor(df)
    log.info("张量形状 %s，均值 %.4f", tuple(t.shape), t.mean().item())
    torch.save(t, args.outdir / "features.pt")


if __name__ == "__main__":
    main()

python pipeline.py --input scores.csv --outdir out

---

这段代码用到了本章的每一节

用到的技术	出自
虚拟环境、requirements	第 1 节
pandas 读取与选择	第 2 节
去重、缺失、groupby	第 3 节
try/except、主动 raise	第 4 节
生成器思想（大文件可换 chunksize）	第 5 节
NumPy → 张量、标准化	第 6 节
函数拆分、类型标注、文档字符串	第 7 节
argparse、logging、pathlib	第 8 节

---

三个值得注意的工程细节

1. encoding="utf-8-sig"：导出 CSV 时用它，Excel 打开中文才不乱码（sig 是 BOM 标记）。
2. 每一步都记录了「改了多少」——去重了几行、补了几个均值。静默的数据清洗是最危险的，你必须知道数据被动了什么。
3. 越界分数没有直接删，而是当缺失处理——保留了这一行的其他科成绩，信息损失更小。

---

本章小结

这一章把 Python 从「能写脚本」推到了「能做数据工程」：

环境隔离 → 数据清洗 → 健壮性 → 内存效率 → 张量 → 可读可复用

下一站：编程路线上，你已经具备读懂深度学习代码的全部前置能力；AI 路线上，接下来是机器学习基础和《动手学深度学习》导读——那里的每一段代码，你现在都能读懂了。

练习：自己造一份带脏值的 CSV（故意加入重复行、缺考、超过 100 的分数），跑通上面的管道，检查日志里报告的清洗数量对不对。