Python 现代开发学习笔记

(十四)先分组,再用 NumPy 拟合预处理

1. 字段顺序、类型与形状

FEATURES 同时服务于读取、元数据和预测检查。按 minutes、sessions 排列,240 行得到 x.shape=(240,2),目标保持 y.shape=(240,1)。沿用上一章的 float32,让 NumPy 数组与模型参数类型相同。

# 项目文件:model_project/src/modeldelivery/data.py
import csv
from dataclasses import dataclass
import hashlib
from pathlib import Path

import numpy as np

FEATURES = ["minutes", "sessions"]


@dataclass
class Prepared:
    x: np.ndarray
    y: np.ndarray
    splits: dict
    scaler: dict
    source_hash: str


def prepare(path, seed=7):
    path = Path(path)
    with path.open(encoding="utf-8", newline="") as source:
        rows = list(csv.DictReader(source))
    if len(rows) < 10:
        raise ValueError("训练数据不足十条,请检查输入文件")
    try:
        # 按固定列序构造二维数组,让训练与推理解释相同的特征。
        x = np.asarray(
            [[float(row[name]) for name in FEATURES] for row in rows], dtype=np.float32
        )
        y = np.asarray([[float(row["pages"])] for row in rows], dtype=np.float32)
    except (KeyError, TypeError, ValueError) as error:
        raise ValueError("CSV 需要 minutes、sessions、pages 三列数值") from error
    if not np.isfinite(x).all() or not np.isfinite(y).all() or (x <= 0).any():
        raise ValueError("输入需要有限数值,分钟数和次数应大于零")

np.isfinite() 检查每个元素是不是普通有限数值。CSV 缺列、出现无法转换的文本或无穷数时,准备数据阶段就指出问题;没有开始发布版本。hashlib.sha256 对原 CSV 文件的字节计算摘要,记录这次训练究竟读取了哪份输入。

2. 训练组决定缩放,其他组沿用

# 续写同一文件:model_project/src/modeldelivery/data.py
    # NumPy 先决定原始样本分组,任何统计量都在划分之后计算。
    order = np.random.default_rng(seed).permutation(len(x))
    a, b = int(len(x) * 0.6), int(len(x) * 0.8)
    splits = dict(zip(["train", "val", "test"], [order[:a], order[a:b], order[b:]]))
    train_x, train_y = x[splits["train"]], y[splits["train"]]
    scaler = {
        "x_mean": train_x.mean(0, keepdims=True),
        "x_scale": np.maximum(train_x.std(0, keepdims=True), 1e-6),
        "y_mean": train_y.mean(0, keepdims=True),
        "y_scale": np.maximum(train_y.std(0, keepdims=True), 1e-6),
    }
    return Prepared(
        ((x - scaler["x_mean"]) / scaler["x_scale"]).astype(np.float32),
        ((y - scaler["y_mean"]) / scaler["y_scale"]).astype(np.float32),
        splits,
        scaler,
        hashlib.sha256(path.read_bytes()).hexdigest(),
    )

先用 default_rng(seed).permutation() 得到不重复的下标顺序,再分成 144、48、48 条。只有 train_x 和 train_y 参与均值与标准差计算;验证和测试组使用相同变换。keepdims=True 保留 (1,2) 或 (1,1),减法和除法沿行广播。某列标准差很小时取 1e-6,避免除以零。

与上一章对照时注意实际的分组方法:这里先由 NumPy 决定下标,再交给 PyTorch。保存 splits.json 后,可以直接检查每组包含哪些原始行。测试还会把验证组分钟数改为很大的值,再核对训练统计量没有改变。

先随机排列 240 个下标,再切成 144、48、48,最后只用训练部分拟合四组缩放数组

笔记正文与全部题解 · Ctrl / ⌘ K