https://mofanpy.com/tutorials/machine-learning/ML-practice/build-car-classifier-from-scratch1

汽车状态分类器练习

模型改为PyTorch实现
参考：https://blog.csdn.net/t18438605018/article/details/123563036

数据预处理

# 下载汽车数据
import pandas as pd
from urllib.request import urlretrieve

def load_data(download=True):
    # download data from : http://archive.ics.uci.edu/ml/datasets/Car+Evaluation
    if download:
        data_path, _ = urlretrieve("http://archive.ics.uci.edu/ml/machine-learning-databases/car/car.data", "car.csv")
        print("Downloaded to car.csv")

    # use pandas to view the data structure
    col_names = ["buying", "maint", "doors", "persons", "lug_boot", "safety", "class"]
    data = pd.read_csv("car.csv", names=col_names)
    return data

load_data(download=True)

Downloaded to car.csv

1728 rows × 7 columns

data = load_data(download=True)
# print(data.head)
for name in data.keys():
    print(name, data[name].unique())

Downloaded to car.csv
buying ['vhigh' 'high' 'med' 'low']
maint ['vhigh' 'high' 'med' 'low']
doors ['2' '3' '4' '5more']
persons ['2' '4' 'more']
lug_boot ['small' 'med' 'big']
safety ['low' 'med' 'high']
class ['unacc' 'acc' 'vgood' 'good']

# onehot预处理
def convert2onehot(data):
    # covert data to onehot representation
    return pd.get_dummies(data, prefix=data.columns)

new_data = convert2onehot(data)
new_data.to_csv("car_onehot.csv", index=False)
new_data

image.png

搭建模型

import numpy as np
import pandas as pd
import torch
import torch.utils.data as Data
import matplotlib.pyplot as plt
from IPython import display

# 打乱数据的顺序, 然后将训练和测试数据以 7/3 比例分开
# prepare training data
new_data = pd.read_csv("car_onehot.csv")
new_data = new_data.values.astype(np.float32)       # change to numpy array and float32
np.random.shuffle(new_data)
sep = int(0.7*len(new_data))
train_data = new_data[:sep]                         # training data (70%)
test_data = new_data[sep:]                          # test data (30%)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using {device} device")

# 输入数据转换为Tensor
torch_train_data = torch.tensor(train_data)
torch_test_data  = torch.tensor(test_data)

# 设置输入输出变量
x_train = torch_train_data[:, :21]
y_train = torch_train_data[:, 21:]
x_test = torch_test_data[:, :21]
y_test = torch_test_data[:, 21:]

# 搭建网络
net = torch.nn.Sequential(
    torch.nn.Linear(21,128),  # 隐藏层1
    torch.nn.ReLU(),
    torch.nn.Linear(128,128), # 隐藏层2
    torch.nn.ReLU(),
    torch.nn.Linear(128,4),   # 输出层
    #torch.nn.Softmax(dim=1)  # CrossEntropyLoss本身就会对输出取softmax，所以无需这一层
).to(device)
net

Using cuda device

Sequential(
  (0): Linear(in_features=21, out_features=128, bias=True)
  (1): ReLU()
  (2): Linear(in_features=128, out_features=128, bias=True)
  (3): ReLU()
  (4): Linear(in_features=128, out_features=4, bias=True)
)

# 定义优化器和损失函数
# opt       = torch.optim.Adam(net.parameters(), lr=0.2, betas=(0.9, 0.99)) 
opt       = torch.optim.SGD(net.parameters(), lr=0.2) 
loss_func = torch.nn.CrossEntropyLoss().to(device)

train_dataset = Data.TensorDataset(x_train, y_train)
test_dataset  = Data.TensorDataset(x_test, y_test)

# 利用DataLoader批训练
BATCH_SIZE = 32    # 每次训练数据数量
EPOCH      = 100  # 总共训练的轮次

train_loader  = Data.DataLoader(
    dataset = train_dataset,
    batch_size = BATCH_SIZE,
    shuffle = True,
    num_workers = 5
)

test_loader  = Data.DataLoader(
    dataset = test_dataset,
    batch_size = BATCH_SIZE,
    shuffle = True,
    num_workers = 5
)

def show_img_acc_and_loss():
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8, 4))

    ax1.cla()
    ax1.plot(epoches, train_acc_his, 'r', label = 'train_acc')
    ax1.plot(epoches, test_acc_his, 'b--', label = 'test_acc')
    ax1.legend()
    ax1.set_xlabel('Epoches')
    ax1.set_ylabel('Accuracy')
    ax1.set_title('Model Accuracy')

    ax2.cla()
    ax2.plot(epoches, train_loss_his, 'r', label = 'train_loss')
    ax2.plot(epoches, test_loss_his, 'b--', label = 'test_loss')
    ax2.legend()
    ax2.set_xlabel('Epoches')
    ax2.set_ylabel('Loss')
    ax2.set_title('Model Loss')

    # plt.show()
    # plt.pause(0.5)
    # display.clear_output(wait=True) # 刷新图片

#show_img_acc_and_loss()   

def show_img_Ratio():
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8, 4)) 

    # 各个batch的数据合并
    im_tp1 = [b for a in train_pred_his for b in a]
    # 合并list语句，参考：
    # https://blog.csdn.net/cxj540947672/article/details/107337082?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_baidulandingword~default-0-107337082-blog-113962252.pc_relevant_3mothn_strategy_recovery&spm=1001.2101.3001.4242.1&utm_relevant_index=3
    im_tp1 = np.array(im_tp1)

    im_tt1 = [b for a in train_targ_his for b in a]
    # 合并list语句，参考：
    # https://blog.csdn.net/cxj540947672/article/details/107337082?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_baidulandingword~default-0-107337082-blog-113962252.pc_relevant_3mothn_strategy_recovery&spm=1001.2101.3001.4242.1&utm_relevant_index=3
    im_tt1 = np.array(im_tt1)

    im_tp2 = [b for a in test_pred_his for b in a]
    # 合并list语句，参考：
    # https://blog.csdn.net/cxj540947672/article/details/107337082?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_baidulandingword~default-0-107337082-blog-113962252.pc_relevant_3mothn_strategy_recovery&spm=1001.2101.3001.4242.1&utm_relevant_index=3
    im_tp2 = np.array(im_tp2)

    im_tt2 = [b for a in test_targ_his for b in a]
    # 合并list语句，参考：
    # https://blog.csdn.net/cxj540947672/article/details/107337082?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_baidulandingword~default-0-107337082-blog-113962252.pc_relevant_3mothn_strategy_recovery&spm=1001.2101.3001.4242.1&utm_relevant_index=3
    im_tt2 = np.array(im_tt2)

    for c in range(4):
        tp1 = ax1.bar(c+0.1, height=100*(im_tp1 == c).sum()/len(im_tp1), width=0.2, color='r')
        tt1 = ax1.bar(c-0.1, height=100*(im_tt1 == c).sum()/len(im_tp1), width=0.2, color='b')
        tp2 = ax2.bar(c+0.1, height=(im_tp2 == c).sum(), width=0.2, color='r')
        tt2 = ax2.bar(c-0.1, height=(im_tt2 == c).sum(), width=0.2, color='b')

    ax1.set_xticks(range(4), ["accepted", "good", "unaccepted", "very good"])
    ax1.legend(handles=[tp1, tt1], labels=["prediction", "target"])
    ax1.set_ylim(0, 100) 
    ax1.set_ylabel('Ratio (%)')
    ax1.set_title('Train Datasets')

    ax2.set_xticks(range(4), ["accepted", "good", "unaccepted", "very good"])
    ax2.legend(handles=[tp2, tt2], labels=["prediction", "target"])
    ax2.set_ylim(0, 400) 
    ax2.set_ylabel('Ratio (%)')
    ax2.set_title('Test Datasets')

    # plt.show()
    # plt.pause(0.5)
    # display.clear_output(wait=True) # 刷新图片

#show_img_Ratio()

# 训练并测试模型

train_acc_his  = []
train_loss_his = []
test_acc_his   = []
test_loss_his  = []
epoches        = []

#fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8, 4))

for epoch in range (EPOCH):
    # print ('Epoch: ', (int(epoch)+1))

    #______________训练模型

    net.train() # 训练模式

    # 记录各个steps的历史值
    loss_his = [] 
    acc_his  = []
    train_pred_his = []
    train_targ_his = []

    for step, (b_x, b_y) in enumerate(train_loader):
        b_x, b_y = b_x.to(device), b_y.to(device)
        output = net(b_x)  # 调用网络
        loss   = loss_func(output, b_y) # 计算损失
        opt.zero_grad()    # 清空梯度
        loss.backward()    # 反向传播
        opt.step()         # 更新参数

        loss_his.append(loss.data.cpu().cpu().numpy()) # 记录历史损失
        # 计算准确度
        prediction = torch.max(output, 1)[1] # 取softmax输出最大值作为预测值
        # 第一个1表示返回这一行中最大值，第二个1表示只返回索引值，参考：
        # https://blog.csdn.net/weixin_43635550/article/details/100534904
        pred = prediction.data.cpu().numpy()
        target = torch.max(b_y, 1)[1]
        targ = target.data.cpu().numpy()
        acc = (pred==targ).sum() / b_y.shape[0] # 这一个batch的accuracy
        acc_his.append(acc)
        train_pred_his.append(pred)
        train_targ_his.append(targ)
        # print('step: ', step)

    train_acc  = sum(acc_his)/len(acc_his)   # 当前epoch的训练准确度
    train_loss = sum(loss_his)/len(loss_his) # 当前epoch的损失值

    #______________测试模型

    net.eval() # 测试模式

    # 记录各个steps的历史值
    loss_his = [] 
    acc_his  = []   
    test_pred_his = []
    test_targ_his = []

    for step, (b_x, b_y) in enumerate(test_loader):
        b_x, b_y = b_x.to(device), b_y.to(device)
        output = net(b_x)  # 调用网络
        loss   = loss_func(output, b_y) # 计算损失

        loss_his.append(loss.data.cpu().numpy()) # 记录历史损失
        # 计算准确度
        prediction = torch.max(output, 1)[1] # 取softmax输出最大值作为预测值
        # 第一个1表示返回这一行中最大值，第二个1表示只返回索引值，参考：
        # https://blog.csdn.net/weixin_43635550/article/details/100534904
        pred = prediction.cpu().numpy()
        target = torch.max(b_y, 1)[1]
        targ = target.cpu().numpy()
        acc = (pred==targ).sum() / b_y.shape[0] # 这一个batch的accuracy
        acc_his.append(acc)
        test_pred_his.append(pred)
        test_targ_his.append(targ)
        # print('step: ', step)

    test_acc  = sum(acc_his)/len(acc_his)   # 当前epoch的训练准确度
    test_loss = sum(loss_his)/len(loss_his) # 当前epoch的损失值

    #______________打印训练和测试结果
    if (int(epoch)+1) % 5 == 0:
        epoches.append((int(epoch)+1))

        print("Epoch:  %i" % (int(epoch)+1),"| Train Accuracy: %.2f" % train_acc, "| Train Loss: %.2f" % train_loss)
        train_acc_his.append(train_acc)      # 记录各个epoch的历史值
        train_loss_his.append(train_loss)

        print("Epoch:  %i" % (int(epoch)+1),"| Test Accuracy: %.2f" % test_acc, "| Test Loss: %.2f" % test_loss)
        test_acc_his.append(test_acc)      # 记录各个epoch的历史值
        test_loss_his.append(test_loss)

        # —————————————结果可视化
        show_img_acc_and_loss()
        show_img_Ratio()
        plt.pause(0.5)
        display.clear_output(wait=True) # 刷新图片

Epoch:  100 | Train Accuracy: 1.00 | Train Loss: 0.00
Epoch:  100 | Test Accuracy: 0.99 | Test Loss: 0.02

output_14_1.png

output_14_2.png