线性回归

一维线性回归

最小二乘法，偏导数为0

import torch
from torch.autograd import Variable
import matplotlib.pyplot as plt
import numpy as np
import torch.nn as nn
import torch.optim as optim

x_train = np.array([[3.3], [4.4], [5.5], [6.71], [6.93], [4.168], [9.779], 						[6.182], [7.59], [2.167], [7.042],
                    [10.791], [5.313], [7.997], [3.1]], dtype=np.float32)
y_train = np.array([[1.7], [2.76], [2.09], [3.19], [1.694], [1.573], 						[3.366], [2.596], [2.53], [1.221],
                    [2.827], [3.465], [1.65], [2.904],										[1.3]],dtype=np.float32)

x_train = torch.from_numpy(x_train)
y_train = torch.from_numpy(y_train)

class LinearRegression(nn.Module):
    def __init__(self):
        super(LinearRegression, self).__init__()
        self.linear = nn.Linear(1, 1)

    def forward(self, x):
        assert isinstance(x, object)
        out = self.linear(x)
        return out
if torch.cuda.is_available():
    model = LinearRegression().cuda()
else:
    model = LinearRegression()
    
    
# 定义损失函数和优化函数，使用均方误差作为优化函数，使用梯度下降进行优化

criterion = nn.MSELoss()
optimizer = optim.SGD(model.parameters(), lr=0.01)

# 开始训练模型
num_epochs = 1000
for epoch in range(num_epochs):
    inputs = Variable(x_train)
    target = Variable(y_train)

    # forward
    out = model(inputs)
    loss = criterion(out, target)
    # backward
    optimizer.zero_grad() # 归零梯度 
    loss.backward()
    optimizer.step()

    if (epoch+1) % 20 == 0:
        print('Epoch[{}/{}],loss:{:.6f}'.format(epoch+1, num_epochs, loss.item()))

if __name__ == '__main__':
    model.eval()
    predict = model(Variable(x_train))
    predict = predict.data.numpy()
    plt.plot(x_train.numpy(), y_train.numpy(), 'ro', label='Original data')
    plt.plot(x_train.numpy(), predict, label='Fitting Line')
    plt.show()

多维线性回归

import torch
import numpy as np
import matplotlib.pyplot as plt
from torch import nn

# -------------------------------------数据准备--------------------------------------
# 目标权重和偏置
w = torch.FloatTensor([2.0, 3.0, 4.0]).unsqueeze(1)
b = torch.FloatTensor([0.5])


# 一次生成32个数据
def create_data(batch_size=32):
    random = torch.randn(batch_size)
    random = random.unsqueeze(1)  # 添加一个维度
    # 纵向连接tensor
    x = torch.cat([random ** i for i in range(1, 4)], 1)  # b/x/^2/x^3
    # 矩阵乘法
    y = x.mm(w) + b[0]  # mm表示矩阵相乘，mul为对应元素相乘
    if torch.cuda.is_available():
        return x.cuda(), y.cuda()
    return x, y


# -------------------------------------自定义模型--------------------------------------
class PloyRegression(nn.Module):
    def __init__(self):
        super(PloyRegression, self).__init__()
        self.ploy = nn.Linear(3, 1)  # 输入3维(分别表示x/x^2/x^3)，输出1维

    def forward(self, x):
        out = self.ploy(x)
        return out


model = PloyRegression()
if torch.cuda.is_available():
    model = model.cuda()

# ------------------------损失函数、优化器的选择----------------------------
criterion = torch.nn.MSELoss()
optimizer = torch.optim.SGD(model.parameters(), lr=1e-3)

# ------------------------开始训练----------------------------
# 使用均方误差，随机梯度下降
epoch = 0
while True:
    # 创建数据
    batch_x, batch_y = create_data()  # 一次生成32个数据
    # 前向传播
    output = model(batch_x)
    # 损失计算
    loss = criterion(output, batch_y)
    # 获取损失值
    loss_value = loss.data.cpu().numpy()
    # 梯度置零
    optimizer.zero_grad()
    # 反向传播
    loss.backward()
    # 更新参数
    optimizer.step()

    epoch += 1
    # 损失函数小于一定的值才会退出来
    if loss_value < 1e-3:
        break
    # 每100步打印一次损失
    if (epoch + 1) % 100 == 0:
        print("Epoch{}, loss:{:.6f}".format(epoch + 1, loss_value))

# -------------------------------------测试--------------------------------------
model.eval()  # 开启验证模式

# 构造数据
x_train = np.array([[i] for i in range(20)], dtype=np.float32)
x_train = torch.from_numpy(x_train)

x = torch.cat([x_train ** i for i in range(1, 4)], 1)
y = x.mm(w) + b
# 绘制数据点
plt.plot(x_train.numpy(), y.numpy(), 'ro')
# 提取拟合参数
w_get = model.ploy.weight.data.T
b_get = model.ploy.bias.data
print('w:{},b:{}'.format(w_get.cpu().numpy(), b_get.cpu().numpy()))
# 计算预测值
Y_get = x.mm(w_get.cpu()) + b_get.cpu()
plt.plot(x_train.numpy(), Y_get.numpy(), '-')
plt.show()

分类问题

二分类算法———Logistic

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

# 设置随机种子
seed_value = 2024
np.random.seed(seed_value)


# Sigmoid激活函数
def sigmoid(z):
    return 1 / (1 + np.exp(-z))


# 定义逻辑回归算法
class LogisticRegression:
    def __init__(self, learning_rate=0.003, iterations=100):
        self.learning_rate = learning_rate  # 学习率
        self.iterations = iterations  # 迭代次数

    def fit(self, X, y):
        # 初始化参数
        self.weights = np.random.randn(X.shape[1])
        self.bias = 0

        # 梯度下降
        for i in range(self.iterations):
            # 计算sigmoid函数的预测值, y_hat = w * x + b
            y_hat = sigmoid(np.dot(X, self.weights) + self.bias)

            # 计算损失函数
            loss = (-1 / len(X)) * np.sum(y * np.log(y_hat) + (1 - y) * np.log(1 - y_hat))

            # 计算梯度
            dw = (1 / len(X)) * np.dot(X.T, (y_hat - y))
            db = (1 / len(X)) * np.sum(y_hat - y)

            # 更新参数
            self.weights -= self.learning_rate * dw
            self.bias -= self.learning_rate * db

            # 打印损失函数值
            if i % 10 == 0:
                print(f"Loss after iteration {i}: {loss}")

    # 预测
    def predict(self, X):
        y_hat = sigmoid(np.dot(X, self.weights) + self.bias)
        y_hat[y_hat >= 0.5] = 1
        y_hat[y_hat < 0.5] = 0
        return y_hat

    # 精度
    def score(self, y_pred, y):
        accuracy = (y_pred == y).sum() / len(y)
        return accuracy


# 导入数据
iris = load_iris()
X = iris.data[:, :2]
y = (iris.target != 0) * 1

# 划分训练集、测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.15, random_state=seed_value)

# 训练模型
model = LogisticRegression(learning_rate=0.03, iterations=1000)
model.fit(X_train, y_train)

# 结果
y_train_pred = model.predict(X_train)
y_test_pred = model.predict(X_test)

score_train = model.score(y_train_pred, y_train)
score_test = model.score(y_test_pred, y_test)

print('训练集Accuracy: ', score_train)
print('测试集Accuracy: ', score_test)

# 可视化决策边界
x1_min, x1_max = X[:, 0].min() - 0.5, X[:, 0].max() + 0.5
x2_min, x2_max = X[:, 1].min() - 0.5, X[:, 1].max() + 0.5
xx1, xx2 = np.meshgrid(np.linspace(x1_min, x1_max, 100), np.linspace(x2_min, x2_max, 100))
Z = model.predict(np.c_[xx1.ravel(), xx2.ravel()])
Z = Z.reshape(xx1.shape)
plt.contourf(xx1, xx2, Z, cmap=plt.cm.Spectral)
plt.scatter(X[:, 0], X[:, 1], c=y, cmap=plt.cm.Spectral)
plt.xlabel("Sepal length")
plt.ylabel("Sepal width")
plt.show()