使用LSTM模型进行时序数据预测

使用LSTM模型进行时序数据预测

引用

CSDN

1.

https://blog.csdn.net/qq_42035021/article/details/141132537

LSTM模型简介

LSTM（Long Short-Term Memory）是一种循环神经网络（RNN）的变体，主要用于解决序列数据处理中的长期依赖问题。LSTM通过引入记忆单元和门控机制，能够有效地捕捉序列中的长期依赖关系，因此在时序数据预测、自然语言处理等领域得到了广泛应用。

研究目标

本项目的目标是使用临近若干时刻的气象观测数据，预测未来3个时刻的能见度。具体来说，我们将使用温度、露点温度、相对湿度、饱和水汽压差和能见度等气象要素作为输入特征，预测未来的能见度变化。

准备工作

环境准备

本项目需要Python 3.x环境，并安装以下必要库：

• PyTorch
• NumPy
• Pandas
• Scikit-learn

数据准备

数据来源为全国气象站逐小时观测数据，包含以下要素：

• 温度（TEM）
• 露点温度（DPT）
• 相对湿度（RHU）
• 饱和水汽压差（VAP）
• 能见度（VIS）

每个时刻的观测数据存储在一个CSV文件中，文件中的行代表不同的站点观测值，列代表不同的要素属性。

数据预处理

读取数据

首先需要从文件中读取数据。代码如下：

import pandas as pd
import os
import numpy as np

indir = r'./filepath'    # 保存所有数据文件的文件夹
file_list = os.listdir(indir)    # 获得所有数据文件的文件名
feature_vars = ['TEM', 'DPT', 'RHU', 'VAP', 'VIS']
label_vars = 'VIS'

dataset = []
for f in range(len(file_list)):
    df = pd.read_csv(os.path.join(indir, f), na_values=[999999, 999107], usecols=['Station_Id_C', 'Lat', 'Lon', 'TEM', 'DPT', 'RHU', 'VAP', 'VIS'])
    df = df.dropna(axis=0, how='any').query('Lat>17 & Lat<29 & Lon>108 & Lon<123')    # 剔除所有存在空值的行，并提取目标空间范围内的数据
    dataset.append(df.loc[:, feature_vars].values)
dataset = np.array(dataset, dtype=np.float32)    # (times, stations, features)

创建样本集

什么是样本？

在时序预测中，一个样本通常包含两部分：特征数据和标签数据。特征数据用于预测目标，标签数据则是预测的目标值。例如，如果我们想用当前时刻的温度和气压预测风速，那么温度和气压组成特征数据，风速则是标签数据。

构造样本集

假设我们需要用最近5个时刻的温度和气压数据，去预测未来3个时刻的风速数据。在这种情况下，我们有三个时间维度：

• time_step：用于预测的时间窗口长度
• pred_step：预测的目标时间长度
• times：原始观测数据的时间长度

代码实现如下：

time_step = 5
pred_step = 3
samples = []
X = []
Y = []
for i in range(dataset.shape[0] - time_step - pred_step + 1):
    X.append(dataset[i:i+time_step, :, :-1])    # (time_step, stations, features)
    Y.append(dataset[i+time_step:i+time_step+pred_step, :, -1])    # (pred_step, stations)
X = np.array(X).transpose(0, 2, 1, 3)    # (new_times, stations, time_step, features)
Y = np.array(Y).transpose(0, 2, 1)    # (new_times, stations, pred_step)

samples_x = X.reshape(-1, time_step, features)    # 形状为(samples, time_step, features)
samples_y = Y.reshape(-1, pred_step)    # 形状为(samples, pred_step)

拆分样本集

将样本集拆分成训练集、验证集和测试集：

from sklearn.model_selection import train_test_split

train_x, tmp_x, train_y, tmp_y = train_test_split(data_x, data_y, train_size=0.6, random_state=42, shuffle=True)
val_x, test_x, val_y, test_y = train_test_split(tmp_x, tmp_y, train_size=0.5, random_state=42, shuffle=True)

样本归一化

为了消除不同特征之间数值范围差异的影响，需要对数据进行归一化处理：

from sklearn.preprocessing import MinMaxScaler

scale_x = MinMaxScaler()
train_x = scale_x.fit_transform(train_x.reshape(-1, len(feature_vars)).reshape(-1, time_step, len(feature_vars))
val_x = scale_x.transform(val_x.reshape(-1, len(feature_vars))).reshape(-1, time_step, len(feature_vars))
test_x = scale_x.transform(test_x.reshape(-1, len(feature_vars))).reshape(-1, time_step, len(feature_vars))

scale_y = MinMaxScaler()
train_y = scale_y.fit_transform(train_y)
val_y = scale_y.transform(val_y)
test_y = scale_y.transform(test_y)

模型训练

定义LSTM模型

使用PyTorch定义LSTM模型：

import torch
import torch.nn as nn
import torch.optim as optim

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers=1, dropout_rate=0.2):
        super(LSTMModel, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.num_layers = num_layers
        self.dropout_rate = dropout_rate
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.dropout = nn.Dropout(dropout_rate)
        self.fc1 = nn.Linear(hidden_size, output_size)
        self.relu = nn.ReLU()

    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
        out, _ = self.lstm(x, (h0, c0))
        out = self.dropout(out[:, -1, :])
        out = self.fc1(out)
        out = self.relu(out)
        return out

模型训练

定义数据加载器和训练函数：

def set_loader(x, y, batch_size):
    tensor_x = torch.from_numpy(x)
    tensor_y = torch.from_numpy(y)
    loader = DataLoader(TensorDataset(tensor_x, tensor_y), batch_size=batch_size, shuffle=True)
    return loader

def lstm_train(model, epochs, train_loader, val_loader, learning_rate=0.01, plot_loss=False):
    loss_function = nn.MSELoss()
    optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    train_losses = []
    val_losses = []

    for epoch in range(epochs):
        model.train()
        train_loss = 0.0
        for x_batch, y_batch in train_loader:
            optimizer.zero_grad()
            outputs = model(x_batch)
            loss = loss_function(outputs, y_batch)
            loss.backward()
            optimizer.step()
            train_loss += loss.item() * x_batch.size(0)
        train_loss /= len(train_loader.dataset)
        train_losses.append(train_loss)

        model.eval()
        val_loss = 0.0
        with torch.no_grad():
            for x, y in val_loader:
                outputs = model(x)
                loss = loss_function(outputs, y)
                val_loss += loss.item() * x.size(0)
        val_loss /= len(val_loader.dataset)
        val_losses.append(val_loss)
        print(f'Epoch [{epoch+1}/{epochs}], Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}')

    if plot_loss:
        fig = plt.figure(figsize=[8,6])
        ax = fig.add_subplot(111)
        ax.plot(train_losses, 'b', label='train_losses')
        ax.plot(val_losses, 'r', label='val_losses')
        ax.legend()
        ax.set_title(f'Epochs:{epochs}  learning_rate:{learning_rate}')
        plt.show()

    return model

模型预测

使用训练好的模型进行预测：

hidden_size = 16
input_size = len(feature_vars)
output_size = pred_time
batch_size = 16
epochs = 500

train_loader = set_loader(train_x, train_y, batch_size)
val_loader = set_loader(val_x, val_y, batch_size)

model = LSTMModel(input_size, hidden_size, output_size, num_layers=1, dropout_rate=0.3)
model = lstm_train(model, epochs, train_loader, val_loader, learning_rate=0.001, plot_loss=False)

model.eval()
with torch.no_grad():
    out = model(torch.from_numpy(test_x))
pred = scale_y.inverse_transform(out)

结果分析

模型预测结果如下图所示：

虽然原数据的变化范围较大，导致预测结果在极值处表现较差，但模型基本能够捕捉到能见度的变化趋势。