表格数据预测与 PyTorch 训练

任务书：回归流程交付

prediction 表含样本编号、事件日期、三个数值特征、连续目标和预先固定的 split。本项目是回归而非分类，使用三个特征预测连续 target，交付训练历史、最终测试预测、评价指标和可加载的 CPU 模型文件。不要把 sample_id、event_date、split 或 target 混入输入，模型输入严格只有 x1、x2、x3。

分割已经固定为 train、validation、test。训练集用于拟合预处理和梯度更新，验证集用于选择轮次，测试集在模型选择完成后只评价一次。划分包含日期顺序的约束，不能重新随机混合后声称仍检验未来泛化。先确认样本键唯一、三部分非空且互不重叠，再开始转换。

起始代码与防止泄漏

x3 存在缺失。先从训练集计算每个特征的均值作为填充值，再填充训练集，计算填充后的训练均值和总体标准差。标准差为零的列用一作为缩放因子，避免除零。验证与测试只应用这些已拟合参数，不能各自计算均值，也不能从整张表先标准化再划分。

起始代码：

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
raw=load("prediction")
cols=["x1","x2","x3"]
train=raw.loc[raw["split"].eq("train")].copy()
validation=raw.loc[raw["split"].eq("validation")].copy()
test=raw.loc[raw["split"].eq("test")].copy()
# TODO: 仅用 train 拟合 fill/mean/scale，再把三个分割转换成 float32 Tensor。

若训练集一整列缺失，均值无法定义，应停止并调整特征合同，不从验证集借值。每一步检查有限数值和形状，输入为样本数乘三，目标为样本数乘一。

训练与验证逐步解释

网络用三维输入、十六个隐藏单元、ReLU 和一维输出。MSELoss 衡量连续预测误差，目标形状与预测完全相同，否则广播可能悄悄计算错误的二维损失。每批依次清空梯度、前向预测、计算损失、反向传播、检查梯度、更新参数。累计训练损失时按批样本数加权，最后除以总样本数，不能直接平均不等长批次的均值。

每轮验证先 model.eval，再 no_grad，以正确切换行为并禁用梯度记录。eval 和 no_grad 解决不同问题，两者都写清。验证损失最小时复制 state_dict 中的张量；只保存引用会让所谓最佳权重随后续更新变化。完成八十轮后加载最佳状态，才进行一次测试评价。

复现、基线和保存加载

固定 PyTorch 随机种子、DataLoader 的随机生成器、CPU 和线程数，开启确定性算法，记录实际库版本。同一环境可复现不等于跨所有硬件和版本逐比特相同，所以交付锁定环境与输出记录。均值预测基线只使用训练目标均值，测试 MSE 与基线比较帮助解释模型作用，但不把超过某个阈值作为唯一正确性标准。

保存文件包含最佳权重、特征顺序、填充均值、标准化均值和缩放参数。加载到 CPU 后重建相同网络，从保存的预处理参数重新转换测试输入，并要求预测与原模型精确一致。没有预处理信息的权重不能独立正确推理。只加载本课自己生成的文件，不把未知来源的序列化文件当作无风险数据。

验收标准与错误诊断

完整答案在运行脚本中。验收检查样本划分互不重叠、预处理只用训练、所有 Tensor 有限且为 float32、预测目标形状一致、每个参数梯度存在且有限、网络在 CPU 上运行。最终模型必须处于 eval 模式，测试只用于报告，保存加载同时恢复预处理和权重，roundtrip 预测完全一致。输出训练与验证历史、测试表和指标文件供复核。

若损失 NaN，先检查缺失填充、标准差、学习率和输入范围；若损失变化异常，检查广播、未清零梯度和模式切换。若测试明显优于验证也不能立即庆祝，应调查划分和泄漏。最终报告同时说明这是合成回归数据与小型网络，训练成功证明流程正确，并不保证真实岗位数据上的泛化或金融收益。

版本与复现约定

本课程 pandas 示例以 2.3.3 为验证基线，页面运行记录给出实际版本。pandas 3.x 默认使用 Copy-on-Write：从表中取出的对象发生修改时，不据此假定原表同步变化。跨版本都应使用一次完整的 loc 赋值表达式；需要独立工作表时显式 copy。阅读网上的旧示例时，先核对版本、输入类型和索引，再判断结果能否复现。

完整示例与实际输出

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import json
import torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader
torch.set_num_threads(1)
torch.manual_seed(20261013)
torch.use_deterministic_algorithms(True)
raw=load("prediction")
assert raw["sample_id"].is_unique
cols=["x1","x2","x3"]
train=raw.loc[raw["split"].eq("train")].copy()
val=raw.loc[raw["split"].eq("validation")].copy()
test=raw.loc[raw["split"].eq("test")].copy()
assert all(len(d)>0 for d in [train,val,test])
assert not set(train["sample_id"]) & set(test["sample_id"])
fill=train[cols].mean()
assert fill.notna().all()
filled=train[cols].fillna(fill)
mean=filled.mean();scale=filled.std(ddof=0).replace(0,1)
def tensorize(frame):
    x=((frame[cols].fillna(fill)-mean)/scale).to_numpy(dtype=np.float32)
    y=frame["target"].to_numpy(dtype=np.float32).reshape(-1,1)
    assert np.isfinite(x).all() and np.isfinite(y).all()
    return torch.from_numpy(x),torch.from_numpy(y)
xtrain,ytrain=tensorize(train)
xval,yval=tensorize(val)
xtest,ytest=tensorize(test)
assert xtrain.shape==(len(train),3) and ytrain.shape==(len(train),1)
generator=torch.Generator().manual_seed(20261013)
loader=DataLoader(TensorDataset(xtrain,ytrain),batch_size=32,shuffle=True,generator=generator)
def build_model():
    return nn.Sequential(nn.Linear(3,16),nn.ReLU(),nn.Linear(16,1))
model=build_model().cpu()
optimizer=torch.optim.Adam(model.parameters(),lr=0.01)
loss_fn=nn.MSELoss()
best_loss=float("inf");best_state=None;best_epoch=0;history=[]
for epoch in range(80):
    model.train()
    total=0.0;n=0
    for xb,yb in loader:
        optimizer.zero_grad(set_to_none=True)
        prediction=model(xb)
        assert prediction.shape==yb.shape
        loss=loss_fn(prediction,yb)
        assert torch.isfinite(loss)
        loss.backward()
        assert all(p.grad is not None and torch.isfinite(p.grad).all() for p in model.parameters())
        optimizer.step()
        total+=loss.item()*len(xb);n+=len(xb)
    model.eval()
    with torch.no_grad():v=loss_fn(model(xval),yval).item()
    assert np.isfinite(v)
    history.append({"epoch":epoch+1,"train_mse":total/n,"validation_mse":v})
    if v<best_loss:
        best_loss=v;best_epoch=epoch+1
        best_state={k:value.detach().clone() for k,value in model.state_dict().items()}
assert best_state is not None
model.load_state_dict(best_state)
model.eval()
# 模型选择完成后只进行一次测试集评价。
with torch.no_grad():
    predicted=model(xtest)
    test_mse=loss_fn(predicted,ytest).item()
    baseline_mse=loss_fn(torch.full_like(ytest,ytrain.mean().item()),ytest).item()
checkpoint={"state_dict":best_state,"columns":cols,"fill":fill.tolist(),"mean":mean.tolist(),"scale":scale.tolist(),"best_epoch":best_epoch}
path=OUT/"project_model.pt"
torch.save(checkpoint,path)
saved=torch.load(path,map_location="cpu",weights_only=True)
restored=build_model()
restored.load_state_dict(saved["state_dict"])
restored.eval()
# 保存的预处理参数也参与重新推理，而不是仅检验网络权重。
f=pd.Series(saved["fill"],index=saved["columns"])
m=pd.Series(saved["mean"],index=saved["columns"])
s=pd.Series(saved["scale"],index=saved["columns"])
reloaded_x=torch.from_numpy(((test[saved["columns"]].fillna(f)-m)/s).to_numpy(dtype=np.float32))
with torch.no_grad():reloaded_prediction=restored(reloaded_x)
torch.testing.assert_close(predicted,reloaded_prediction,rtol=0,atol=0)
assert not restored.training and predicted.device.type=="cpu"
pd.DataFrame(history).to_csv(OUT/"project_model_history.csv",index=False)
pd.DataFrame({"sample_id":test["sample_id"].to_numpy(),"target":ytest.numpy().ravel(),"prediction":predicted.numpy().ravel()}).to_csv(OUT/"project_model_predictions.csv",index=False)
metrics={"torch_version":torch.__version__,"pandas_version":pd.__version__,"best_epoch":best_epoch,"validation_mse":best_loss,"test_mse":test_mse,"baseline_mse":baseline_mse,"roundtrip":"exact"}
(OUT/"project_model_metrics.json").write_text(json.dumps(metrics,indent=2),encoding="utf-8")
print(json.dumps(metrics,indent=2))
print("形状、梯度、CPU 推理及保存加载一致性已通过")

实际运行通过CPU Python execution.

{
  "torch_version": "2.9.1+cpu",
  "pandas_version": "2.3.3",
  "best_epoch": 76,
  "validation_mse": 0.027654949575662613,
  "test_mse": 0.021679464727640152,
  "baseline_mse": 2.3331830501556396,
  "roundtrip": "exact"
}
形状、梯度、CPU 推理及保存加载一致性已通过

五道自测：先回答，再核对

两项动手任务

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import json
import torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader
torch.set_num_threads(1)
torch.manual_seed(20261013)
torch.use_deterministic_algorithms(True)
raw=load("prediction")
assert raw["sample_id"].is_unique
cols=["x1","x2","x3"]
train=raw.loc[raw["split"].eq("train")].copy()
val=raw.loc[raw["split"].eq("validation")].copy()
test=raw.loc[raw["split"].eq("test")].copy()
assert all(len(d)>0 for d in [train,val,test])
assert not set(train["sample_id"]) & set(test["sample_id"])
fill=train[cols].mean()
assert fill.notna().all()
filled=train[cols].fillna(fill)
mean=filled.mean();scale=filled.std(ddof=0).replace(0,1)
def tensorize(frame):
    x=((frame[cols].fillna(fill)-mean)/scale).to_numpy(dtype=np.float32)
    y=frame["target"].to_numpy(dtype=np.float32).reshape(-1,1)
    assert np.isfinite(x).all() and np.isfinite(y).all()
    return torch.from_numpy(x),torch.from_numpy(y)
xtrain,ytrain=tensorize(train)
xval,yval=tensorize(val)
xtest,ytest=tensorize(test)
assert xtrain.shape==(len(train),3) and ytrain.shape==(len(train),1)
generator=torch.Generator().manual_seed(20261013)
loader=DataLoader(TensorDataset(xtrain,ytrain),batch_size=32,shuffle=True,generator=generator)
def build_model():
    return nn.Sequential(nn.Linear(3,16),nn.ReLU(),nn.Linear(16,1))
model=build_model().cpu()
optimizer=torch.optim.Adam(model.parameters(),lr=0.01)
loss_fn=nn.MSELoss()
best_loss=float("inf");best_state=None;best_epoch=0;history=[]
for epoch in range(80):
    model.train()
    total=0.0;n=0
    for xb,yb in loader:
        optimizer.zero_grad(set_to_none=True)
        prediction=model(xb)
        assert prediction.shape==yb.shape
        loss=loss_fn(prediction,yb)
        assert torch.isfinite(loss)
        loss.backward()
        assert all(p.grad is not None and torch.isfinite(p.grad).all() for p in model.parameters())
        optimizer.step()
        total+=loss.item()*len(xb);n+=len(xb)
    model.eval()
    with torch.no_grad():v=loss_fn(model(xval),yval).item()
    assert np.isfinite(v)
    history.append({"epoch":epoch+1,"train_mse":total/n,"validation_mse":v})
    if v<best_loss:
        best_loss=v;best_epoch=epoch+1
        best_state={k:value.detach().clone() for k,value in model.state_dict().items()}
assert best_state is not None
model.load_state_dict(best_state)
model.eval()
# 模型选择完成后只进行一次测试集评价。
with torch.no_grad():
    predicted=model(xtest)
    test_mse=loss_fn(predicted,ytest).item()
    baseline_mse=loss_fn(torch.full_like(ytest,ytrain.mean().item()),ytest).item()
checkpoint={"state_dict":best_state,"columns":cols,"fill":fill.tolist(),"mean":mean.tolist(),"scale":scale.tolist(),"best_epoch":best_epoch}
path=OUT/"project_model.pt"
torch.save(checkpoint,path)
saved=torch.load(path,map_location="cpu",weights_only=True)
restored=build_model()
restored.load_state_dict(saved["state_dict"])
restored.eval()
# 保存的预处理参数也参与重新推理，而不是仅检验网络权重。
f=pd.Series(saved["fill"],index=saved["columns"])
m=pd.Series(saved["mean"],index=saved["columns"])
s=pd.Series(saved["scale"],index=saved["columns"])
reloaded_x=torch.from_numpy(((test[saved["columns"]].fillna(f)-m)/s).to_numpy(dtype=np.float32))
with torch.no_grad():reloaded_prediction=restored(reloaded_x)
torch.testing.assert_close(predicted,reloaded_prediction,rtol=0,atol=0)
assert not restored.training and predicted.device.type=="cpu"
pd.DataFrame(history).to_csv(OUT/"project_model_history.csv",index=False)
pd.DataFrame({"sample_id":test["sample_id"].to_numpy(),"target":ytest.numpy().ravel(),"prediction":predicted.numpy().ravel()}).to_csv(OUT/"project_model_predictions.csv",index=False)
metrics={"torch_version":torch.__version__,"pandas_version":pd.__version__,"best_epoch":best_epoch,"validation_mse":best_loss,"test_mse":test_mse,"baseline_mse":baseline_mse,"roundtrip":"exact"}
(OUT/"project_model_metrics.json").write_text(json.dumps(metrics,indent=2),encoding="utf-8")
print(json.dumps(metrics,indent=2))
print("形状、梯度、CPU 推理及保存加载一致性已通过")

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
train=pd.DataFrame({"x":[1.,3.,None]})
val=pd.DataFrame({"x":[1000.]})
fill=train.mean()
fitted=train.fillna(fill)
mean=fitted.mean();scale=fitted.std(ddof=0).replace(0,1)
before=(float(fill["x"]),float(mean["x"]),float(scale["x"]))
val.loc[:,"x"]=-1000.
after=(float(train.mean()["x"]),float(train.fillna(fill).mean()["x"]),float(train.fillna(fill).std(ddof=0)["x"]))
assert before==after
print("训练预处理参数不受验证值改变:",before)

官方资料与掌握状态

• PyTorch 完整训练入门
• 保存与加载
• 可复现性

完成两项任务后，按独立完成程度记录。状态会尽量保存在当前浏览器，可导出备份。