1 案例介绍

Jones创办了一家二手手机公司，在经营过程中遇到了一个核心难题：如何准确估算二手手机的合理售价。为了解决这个问题，他系统性地收集了多家手机销售公司的历史交易数据。该数据集包含了二手手机的各项性能指标，以及对应的最终成交价格区间。价格被划分为0、1、2、3四个等级，分别代表从低到高的价格范围，这使得问题本质上成为一个多分类任务，而非传统的回归预测问题。

数据集包含以下20个特征字段：

特征名称	含义说明	单位
battery_power	电池总容量	毫安时（mAh）
blue	是否支持蓝牙	布尔值
clock_speed	处理器主频	GHz
dual_sim	是否支持双卡	布尔值
fc	前置摄像头像素	百万像素
four_g	是否支持4G网络	布尔值
int_memory	内置存储空间	GB
m_dep	手机厚度	cm
mobile_wt	手机重量	g
n_cores	CPU核心数	个
pc	后置主摄像头像素	百万像素
px_height	屏幕像素高度	像素
px_width	屏幕像素宽度	像素
ram	运行内存大小	MB
sc_h	屏幕高度	cm
sc_w	屏幕宽度	cm
talk_time	最长通话时长	小时
three_g	是否支持3G网络	布尔值
touch_screen	是否配备触控屏	布尔值
wifi	是否支持WiFi	布尔值
price_range	价格区间（目标变量）	0/1/2/3

我们的目标是帮助Jones挖掘出手机各项硬件配置与售价区间之间的内在关联规律。借助深度学习技术，构建一个全连接神经网络模型，通过学习历史数据中的模式，实现对二手手机价格区间的自动、准确预测。

整个实现流程遵循标准的机器学习项目规范，分为四个核心步骤：

• 数据准备：加载并预处理数据，划分训练集与验证集
• 模型构建：设计适合多分类任务的全连接神经网络架构
• 模型训练：配置损失函数、优化器，执行迭代训练并引入早停机制
• 模型评估：在验证集上测试模型性能，评估分类准确率

---

2 构建数据集

原始数据集共包含2000条样本记录，按照机器学习的最佳实践，我们将其划分为训练集和验证集：其中1600条（80%）用于模型训练，400条（20%）用于模型验证和性能评估。数据集划分采用sklearn.model_selection.train_test_split函数，并通过固定随机种子（random_state=88）确保结果的可重复性。

在数据预处理阶段，我们需要完成以下关键操作：

1. 特征与标签分离：使用pandas读取CSV文件，将前20列作为特征矩阵x，最后一列price_range作为目标标签y
2. 数据类型转换：特征值统一转换为np.float32类型以适配神经网络计算；标签值转换为np.int64类型用于交叉熵损失函数
3. Tensor封装：利用torch.from_numpy()将NumPy数组转换为PyTorch张量，并构建TensorDataset对象，便于后续与DataLoader配合使用

核心实现代码如下：

def create_dataset():
    # 使用pandas读取数据
    data = pd.read_csv(DATA_CSV)

    # 特征值和目标值分离
    x, y = data.iloc[:, :-1], data.iloc[:, -1]
    # 类型转换：特征值 → float32，目标值 → int64
    x = x.astype(np.float32)
    y = y.astype(np.int64)

    # 数据集划分（训练集:验证集 = 8:2）
    x_train, x_valid, y_train, y_valid = train_test_split(x, y, train_size=TRAIN_SIZE, random_state=RANDOM_STATE)

    # 转换为Tensor并移入到指定设备（CPU/GPU）
    x_train_tensor = torch.from_numpy(x_train.values).to(DEVICE)
    y_train_tensor = torch.tensor(y_train.values).to(DEVICE)
    x_valid_tensor = torch.from_numpy(x_valid.values).to(DEVICE)
    y_valid_tensor = torch.tensor(y_valid.values).to(DEVICE)

    # 构建TensorDataset数据集对象
    train_dataset = TensorDataset(x_train_tensor, y_train_tensor)
    valid_dataset = TensorDataset(x_valid_tensor, y_valid_tensor)

    # 返回：训练集、验证集、特征维度数、类别总数
    return train_dataset, valid_dataset, x_train.shape[1], len(np.unique(y))

---

3 构建网络模型

针对本项目的多分类任务特点，我们设计了一个三层全连接神经网络（多层感知机，MLP）。网络结构采用"逐层递减"的隐藏单元设计策略，从输入层到输出层逐步压缩特征维度，最终映射到4个类别输出。

模型架构详情如下：

网络层	输入维度	输出维度	激活函数	补充操作
第1层（线性层）	input_dim（20）	HIDDEN_DIM1（64）	ReLU	Dropout（0.3）
第2层（线性层）	HIDDEN_DIM1（64）	HIDDEN_DIM2（32）	ReLU	Dropout（0.3）
第3层（线性层）	HIDDEN_DIM2（32）	output_dim（4）	无（线性输出）	—

设计考量：

• 层级递减结构：64 → 32 → 4，逐层提取高层次语义特征
• Dropout正则化：在前两个隐藏层后分别添加Dropout层（丢弃率0.3），有效抑制过拟合
• ReLU激活函数：引入非线性变换能力，同时缓解梯度消失问题
• 输出层无激活：直接输出logits，配合CrossEntropyLoss使用（该损失函数内部集成了Softmax）

对应的PyTorch实现：

class PhonePriceModel(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(PhonePriceModel, self).__init__()
        # 第一层：输入 → 64维
        self.linear1 = nn.Linear(input_dim, HIDDEN_DIM1)
        self.dropout1 = nn.Dropout(DROPOUT_RATE)
        # 第二层：64 → 32维
        self.linear2 = nn.Linear(HIDDEN_DIM1, HIDDEN_DIM2)
        self.dropout2 = nn.Dropout(DROPOUT_RATE)
        # 第三层：32 → 4维（分类输出）
        self.linear3 = nn.Linear(HIDDEN_DIM2, output_dim)

    def forward(self, x):
        x = torch.relu(self.linear1(x))
        x = self.dropout1(x)
        x = torch.relu(self.linear2(x))
        x = self.dropout2(x)
        output = self.linear3(x)  # 输出logits
        return output

---

4 模型训练

训练环节是整个流程的核心，我们精心配置了多项优化策略来提升模型的收敛效果和泛化能力：

训练配置概览

配置项	设定值	说明
批量大小（Batch Size）	32	小批量梯度下降
学习率（Learning Rate）	1×10⁻⁴	Adam优化器推荐初始值
优化器	Adam	自适应矩估计，收敛快
权重衰减（L2正则化）	1×10⁻⁴	防止过拟合
损失函数	交叉熵损失	适配多分类任务
学习率调度策略	StepLR	每30个epoch衰减为0.1倍
最大训练轮数	1000	配合早停提前终止
早停耐心值	75	验证损失连续75轮不下降则停止

早停机制（EarlyStopping）

早停是一种有效的正则化手段，当验证集损失在连续patience个epoch内没有显著改善时，自动终止训练，避免过拟合并节省计算资源。实现逻辑如下：

class EarlyStopping:
    def __init__(self, patience=10, min_delta=1e-4):
        self.patience = patience
        self.min_delta = min_delta      # 最小改善阈值
        self.counter = 0
        self.best_loss = float('inf')
        self.early_stop = False

    def __call__(self, val_loss):
        if val_loss < self.best_loss - self.min_delta:
            self.best_loss = val_loss   # 验证损失有显著改善，重置计数器
            self.counter = 0
        else:
            self.counter += 1           # 无显著改善，累加计数
            if self.counter >= self.patience:
                self.early_stop = True  # 触发早停
        return self.early_stop

训练主循环

在每个epoch中，依次执行以下操作：

1. 训练阶段：模型设置为train()模式，前向传播计算损失，反向传播更新参数
2. 验证阶段：模型设置为eval()模式，关闭梯度计算，在验证集上计算损失和准确率
3. 学习率更新：调用scheduler.step()按策略调整学习率
4. 早停判断：根据验证损失决定是否提前终止训练
5. 进度输出：实时打印当前epoch的训练损失、验证损失、验证准确率及耗时

训练结束后，模型权重被保存到model/phone_best.pth文件中，便于后续加载使用。

def train(train_dataset, valid_dataset, input_dim, class_num):
    # 初始化模型、优化器、调度器、早停对象
    torch.manual_seed(SEED)
    if torch.cuda.is_available():
        torch.cuda.manual_seed(SEED)

    train_loader = DataLoader(train_dataset, shuffle=True, batch_size=BATCH_SIZE)
    valid_loader = DataLoader(valid_dataset, shuffle=False, batch_size=BATCH_SIZE)

    model = PhonePriceModel(input_dim, class_num).to(DEVICE)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE, weight_decay=WEIGHT_DECAY)
    scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=LR_STEP_SIZE, gamma=LR_GAMMA)
    early_stopping = EarlyStopping(patience=PATIENCE)

    train_losses, val_losses = [], []


    for epoch_idx in range(NUM_EPOCHS):
        # —— 训练阶段 ——
        model.train()
        for x, y in train_loader:
            output = model(x)
            loss = criterion(output, y)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        
        # —— 验证阶段 ——
        model.eval()
        val_loss = 0.0
        correct = 0

        with torch.no_grad():
            for x, y in valid_loader:
                output = model(x)
                loss = criterion(output, y)
                val_loss += loss.item()
                y_pred = torch.argmax(output, dim=1)
                correct += (y_pred == y).sum()

        avg_val_loss = val_loss / len(valid_loader)
        val_losses.append(avg_val_loss)
        val_acc = correct.item() / len(valid_dataset)

        # 学习率调度
        scheduler.step()

        # 打印结果
        print('epoch: %4s train_loss: %.4f val_loss: %.4f val_acc: %.4f time: %.2fs' %
              (epoch_idx + 1, avg_train_loss, avg_val_loss, val_acc, time.time() - start))

        # 早停检查
        if early_stopping(avg_val_loss):
            print(f"Early stopping at epoch {epoch_idx + 1}")
            break

    # 保存模型
    torch.save(model.state_dict(), 'model/phone_best.pth')
    return train_losses, val_losses

---

5 模型预测与评估

模型训练完成后，我们使用独立的验证集对模型性能进行最终评估。评估流程如下：

1. 模型加载：实例化PhonePriceModel，加载已保存的最佳模型权重（model/phone_best.pth）
2. 数据加载：将验证集封装为DataLoader，按批次进行推理
3. 预测计算：对每个批次的输入数据执行前向传播，得到输出logits
4. 类别判定：使用torch.argmax(output, dim=1)获取每个样本的预测类别（0~3）
5. 准确率统计：比较预测结果与真实标签，计算总体分类准确率

def test(valid_dataset, input_dim, class_num):
    # 加载模型和训练好的网络参数
    model = PhonePriceModel(input_dim, class_num)
    model.load_state_dict(torch.load('model/phone_best.pth', map_location=DEVICE))
    model.to(DEVICE)

    dataloader = DataLoader(valid_dataset, batch_size=BATCH_SIZE, shuffle=False)
    
    correct = 0
    for x, y in dataloader:
        output = model(x)
        y_pred = torch.argmax(output, dim=1)  # 预测类别
        correct += (y_pred == y).sum()        # 累计正确预测数

    acc = correct.item() / len(valid_dataset)
    print(f'验证集准确率: {acc:.5f}')

---

6 环境依赖与运行

在运行本程序之前，请确保已安装以下Python依赖库：

pip install torch torchvision torchaudio      # PyTorch（根据CUDA版本选择）
pip install scikit-learn                      # 数据集划分与评估
pip install matplotlib                        # 可视化（可选）
pip install numpy pandas                      # 数据处理

程序入口位于if __name__ == '__main__':代码块中，按顺序执行数据加载、模型训练和评估三个环节：

if __name__ == '__main__':
    print(f"使用设备: {DEVICE}")
    
    # 1. 加载数据
    train_dataset, valid_dataset, input_dim, class_num = create_dataset()
    print(f"输入特征数：{input_dim}，分类个数：{class_num}")
    
    # 2. 模型训练
    train(train_dataset, valid_dataset, input_dim, class_num)
    
    # 3. 模型评估
    test(valid_dataset, input_dim, class_num)

通过以上完整的四步流程，我们构建了一个端到端的二手手机价格区间预测系统，能够帮助Jones高效、准确地为回收的手机产品定价提供决策支持。

完整代码

from torch.utils.data import TensorDataset
from torch.utils.data import DataLoader
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import time

from torchsummary import summary

# ==================== 超参数配置 ====================
# 数据路径
DATA_CSV = r'data/手机价格预测.csv'

# 数据划分参数
TRAIN_SIZE = 0.8
RANDOM_STATE = 88
VALIDATION_SIZE = 0.2  # 验证集比例

# 训练参数（优化）
BATCH_SIZE = 32
LEARNING_RATE = 1e-4  # 学习率
NUM_EPOCHS = 1000  # 轮数
PATIENCE = 75  # 早停耐心值

# 模型参数（简化）
HIDDEN_DIM1 = 64  # 减小，避免过拟合
HIDDEN_DIM2 = 32  # 逐层递减

# 优化器选择
OPTIMIZER = 'adam'  # 使用Adam替代SGD
WEIGHT_DECAY = 1e-4  # L2正则化

# 学习率调度
LR_SCHEDULER = 'step'  # 使用学习率调度
LR_STEP_SIZE = 30  # 每30个epoch衰减
LR_GAMMA = 0.1  # 衰减因子

# Dropout（防止过拟合）
DROPOUT_RATE = 0.3

# 设备配置
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"

# 随机种子
SEED = 42


# ==================================================

# 构建数据集
def create_dataset():
    # 使用pandas读取数据
    data = pd.read_csv(DATA_CSV)

    # 特征值和目标值
    x, y = data.iloc[:, :-1], data.iloc[:, -1]
    # 类型转换：特征值，目标值
    x = x.astype(np.float32)
    y = y.astype(np.int64)

    # 数据集划分
    x_train, x_valid, y_train, y_valid = \
        train_test_split(x, y, train_size=TRAIN_SIZE, random_state=RANDOM_STATE)

    # 转换为tensor并移入到设备
    x_train_tensor = torch.from_numpy(x_train.values).to(DEVICE)
    y_train_tensor = torch.tensor(y_train.values).to(DEVICE)
    x_valid_tensor = torch.from_numpy(x_valid.values).to(DEVICE)
    y_valid_tensor = torch.tensor(y_valid.values).to(DEVICE)

    # 构建数据集
    train_dataset = TensorDataset(x_train_tensor, y_train_tensor)
    valid_dataset = TensorDataset(x_valid_tensor, y_valid_tensor)

    # 返回结果:训练集数据，验证集数据，样本特征维度和目标分类个数
    return train_dataset, valid_dataset, x_train.shape[1], len(np.unique(y))


# 构建网络模型
class PhonePriceModel(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(PhonePriceModel, self).__init__()
        # 1. 第一层: 输入为维度为 input_dim, 输出维度为: HIDDEN_DIM1
        self.linear1 = nn.Linear(input_dim, HIDDEN_DIM1)
        self.dropout1 = nn.Dropout(DROPOUT_RATE)
        # 2. 第二层: 输入为维度为 HIDDEN_DIM1, 输出维度为: HIDDEN_DIM2
        self.linear2 = nn.Linear(HIDDEN_DIM1, HIDDEN_DIM2)
        self.dropout2 = nn.Dropout(DROPOUT_RATE)
        # 3. 第三层: 输入为维度为 HIDDEN_DIM2, 输出维度为: output_dim
        self.linear3 = nn.Linear(HIDDEN_DIM2, output_dim)

    def forward(self, x):
        # 前向传播过程
        x = torch.relu(self.linear1(x))
        x = self.dropout1(x)
        x = torch.relu(self.linear2(x))
        x = self.dropout2(x)
        output = self.linear3(x)
        return output


class EarlyStopping:
    def __init__(self, patience=10, min_delta=1e-4):
        self.patience = patience
        self.min_delta = min_delta
        self.counter = 0
        self.best_loss = float('inf')
        self.early_stop = False

    def __call__(self, val_loss):
        if val_loss < self.best_loss - self.min_delta:
            self.best_loss = val_loss
            self.counter = 0
        else:
            self.counter += 1
            if self.counter >= self.patience:
                self.early_stop = True
        return self.early_stop


# 模型训练过程
def train(train_dataset, valid_dataset, input_dim, class_num):
    torch.manual_seed(SEED)
    if torch.cuda.is_available():
        torch.cuda.manual_seed(SEED)

    train_loader = DataLoader(train_dataset, shuffle=True, batch_size=BATCH_SIZE)
    valid_loader = DataLoader(valid_dataset, shuffle=False, batch_size=BATCH_SIZE)

    model = PhonePriceModel(input_dim, class_num).to(DEVICE)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE, weight_decay=WEIGHT_DECAY)
    scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=LR_STEP_SIZE, gamma=LR_GAMMA)
    early_stopping = EarlyStopping(patience=PATIENCE)

    train_losses, val_losses = [], []

    for epoch_idx in range(NUM_EPOCHS):
        # 训练阶段
        model.train()
        train_loss = 0.0
        start = time.time()

        for x, y in train_loader:
            output = model(x)
            loss = criterion(output, y)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            train_loss += loss.item()

        avg_train_loss = train_loss / len(train_loader)
        train_losses.append(avg_train_loss)

        # 验证阶段
        model.eval()
        val_loss = 0.0
        correct = 0

        with torch.no_grad():
            for x, y in valid_loader:
                output = model(x)
                loss = criterion(output, y)
                val_loss += loss.item()
                y_pred = torch.argmax(output, dim=1)
                correct += (y_pred == y).sum()

        avg_val_loss = val_loss / len(valid_loader)
        val_losses.append(avg_val_loss)
        val_acc = correct.item() / len(valid_dataset)

        # 学习率调度
        scheduler.step()

        # 打印结果
        print('epoch: %4s train_loss: %.4f val_loss: %.4f val_acc: %.4f time: %.2fs' %
              (epoch_idx + 1, avg_train_loss, avg_val_loss, val_acc, time.time() - start))

        # 早停检查
        if early_stopping(avg_val_loss):
            print(f"Early stopping at epoch {epoch_idx + 1}")
            break

    # 保存模型
    torch.save(model.state_dict(), 'model/phone_best.pth')
    return train_losses, val_losses


# 模型预测结果评估
def test(valid_dataset, input_dim, class_num):
    # 加载模型和训练好的网络参数
    model = PhonePriceModel(input_dim, class_num)
    model.load_state_dict(torch.load('model/phone.pth', map_location=DEVICE))
    model.to(DEVICE)

    # 构建加载器
    dataloader = DataLoader(valid_dataset, batch_size=BATCH_SIZE, shuffle=False)

    # 评估测试集
    correct = 0

    # 遍历测试集中的数据
    for x, y in dataloader:
        # 数据已经在GPU上
        # 将其送入网络中
        output = model(x)
        # 获取类别结果
        y_pred = torch.argmax(output, dim=1)
        # 获取预测正确的个数
        correct += (y_pred == y).sum()

    # 求预测精度
    print('Acc: %.5f' % (correct.item() / len(valid_dataset)))


if __name__ == '__main__':
    print(f"使用设备: {DEVICE}")

    # 获取数据
    train_dataset, valid_dataset, input_dim, class_num = create_dataset()
    print("输入特征数：", input_dim)
    print("分类个数：", class_num)

    # 模型实例化
    model = PhonePriceModel(input_dim, class_num)
    # summary(model, input_size=(input_dim,), batch_size=BATCH_SIZE, device=DEVICE)

    # 模型训练
    train(train_dataset, valid_dataset, input_dim, class_num)

    # 模型预测结果
    test(valid_dataset, input_dim, class_num)