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

生成测试数据

# 长期趋势
def trend(time, slope=0):
    return slope * time

# 季节趋势
def seasonal_pattern(season_time):
    return np.where(season_time < 0.4,
                    np.cos(season_time * 2 * np.pi),
                    1 / np.exp(3 * season_time))
def seasonality(time, period, amplitude=1, phase=0):
    season_time = ((time + phase) % period) / period
    return amplitude * seasonal_pattern(season_time)

# 噪声
def noise(time, noise_level=1):
    return np.random.randn(len(time)) * noise_level

X = torch.arange(1, 1001)
# Y = 0.7 * X + 100 + torch.randn(X.size())
Y = trend(X, 0.3) + seasonality(X, period=365, amplitude=30) + noise(X, 15) + 200
X.shape, Y.shape

(torch.Size([1000]), torch.Size([1000]))

plt.plot(X.numpy(), Y.numpy());

对测试数据进行处理

# 模型的数据的类型需要是32位浮点型
X = X.type(torch.float32)
Y = Y.type(torch.float32)
X.dtype, Y.dtype

(torch.float32, torch.float32)

# 模型的数据需要进行归一化或者标准化，下面是归一化
X = (X - X.min()) / (X.max() - X.min())
Y = (Y - Y.min()) / (Y.max() - Y.min())
plt.plot(X.numpy(), Y.numpy());

定义模型和模型参数

# 线性模型只有两个参数斜率k，和偏置b
# 线性模型的方程为y = k * x + b
k = nn.Parameter(torch.rand(1, dtype=torch.float32))
b = nn.Parameter(torch.rand(1, dtype=torch.float32))
# 下面输出中的requires_grad=True 表示该参数需要计算梯度
# 梯度用于在反向传播中对参数进行优化，优化方法即梯度下降
k, b 

(Parameter containing:
 tensor([0.6231], requires_grad=True),
 Parameter containing:
 tensor([0.0044], requires_grad=True))

def linear_model(x):
    return k * x + b

梯度下降优化参数

# 可以通过改变学习率lr和epoch_num学习各自的用途

# 定义优化器，用于更新模型的参数，即传入的k和b
optimizer = torch.optim.SGD([k, b], lr=0.01)
# 损失函数，模型优化的目的
loss_func = nn.MSELoss()

# 每个epoch表示把全部的数据过一遍
epoch_num = 2000
for epoch in range(epoch_num):
    # 获取模型预测结果
    y_pred = linear_model(X)
    # 计算损失值
    loss = loss_func(y_pred, Y)
    # 将梯度设为0
    optimizer.zero_grad()
    # 反向传播，计算梯度
    loss.backward()
    # 执行梯度下降，优化参数
    optimizer.step()
k, b

(Parameter containing:
 tensor([0.8825], requires_grad=True),
 Parameter containing:
 tensor([0.0419], requires_grad=True))

# detach()函数用于将参数设置为不需要梯度
k2 = k.detach().numpy()[0]
b2 = b.detach().numpy()[0]

plt.plot(X, Y);
plt.plot(X, k2 * X + b2);

优化模型

class LinearModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.k = nn.Parameter(torch.rand(1, dtype=torch.float32))
        self.b = nn.Parameter(torch.rand(1, dtype=torch.float32))

    def forward(self, x):
        return self.k * x + self.b

model = LinearModel()
# 定义优化器，用于更新模型的参数，即传入的k和b
optimizer = torch.optim.SGD([k, b], lr=0.01)
# 损失函数，模型优化的目的
loss_func = nn.MSELoss()

epoch_num = 2000
for epoch in range(epoch_num):
    y_pred = model(X)
    loss = loss_func(y_pred, Y)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
k, b

(Parameter containing:
 tensor([0.8825], requires_grad=True),
 Parameter containing:
 tensor([0.0419], requires_grad=True))

k2 = k.detach().numpy()[0]
b2 = b.detach().numpy()[0]

plt.plot(X, Y);
plt.plot(X, k2 * X + b2);

随机梯度下降

# 前面执行梯度下降时，我们是一次将全部的数据都传入模型
# 但在实际应用中，可能会由于数据太大，没法全部传入模型
# 因此，可以一次传入一部分数据，这便是随机梯度下降
# 随机梯度下降的核心是，梯度是期望。期望可使用小规模的样本近似估计。

model = LinearModel()
# 定义优化器，用于更新模型的参数，即传入的k和b
optimizer = torch.optim.SGD([k, b], lr=0.01)
# 损失函数，模型优化的目的
loss_func = nn.MSELoss()

# 每个epoch表示把全部的数据过一遍
epoch_num = 2000 
# iter_step表示在一个epoch内抽取几个小规模样本
iter_step = 10
# batch_size表示小规模样本的大小
batch_size = 100
for epoch in range(epoch_num):
    for i in range(iter_step):
        random_samples = torch.randint(X.size()[0], (batch_size, ))
        X_i, Y_i = X[random_samples], Y[random_samples]
        y_pred = model(X_i)
        loss = loss_func(y_pred, Y_i)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
k, b

(Parameter containing:
 tensor([0.8825], requires_grad=True),
 Parameter containing:
 tensor([0.0419], requires_grad=True))

k2 = k.detach().numpy()[0]
b2 = b.detach().numpy()[0]

plt.plot(X, Y);
plt.plot(X, k2 * X + b2);