分享一下基于transformer的时间序列预测模型python代码，给大家，记得点赞哦
#!/usr/bin/env python
# coding: 帅帅的笔者import torch
import torch.nn as nn
import numpy as np
import pandas as pd
import time
import math
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler# Set random seeds for reproducibility
torch.manual_seed(0)
np.random.seed(0)# Hyperparameters
input_window = 10
output_window = 1
batch_size = 250
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
epochs = 100
lr = 0.00005# Load data
df = pd.read_csv("data1.csv", parse_dates=["value"], index_col=[0], encoding='gbk')
data = np.array(df['value']).reshape(-1, 1)# Normalize data
scaler = MinMaxScaler(feature_range=(-1, 1))
data_normalized = scaler.fit_transform(data)# Split the data into train and validation sets
train_ratio = 0.828
train_size = int(len(data) * train_ratio)
val_size = len(data) - train_size
train_data_normalized = data_normalized[:train_size]
val_data_normalized = data_normalized[train_size:]# Define the Transformer model
class PositionalEncoding(nn.Module):def __init__(self, d_model, max_len=5000):super(PositionalEncoding, self).__init__()pe = torch.zeros(max_len, d_model)position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))pe[:, 0::2] = torch.sin(position * div_term)pe[:, 1::2] = torch.cos(position * div_term)pe = pe.unsqueeze(0).transpose(0, 1)self.register_buffer('pe', pe)def forward(self, x):return x + self.pe[:x.size(0), :]class TransAm(nn.Module):def __init__(self, feature_size=250, num_layers=1, dropout=0.1):super(TransAm, self).__init__()self.model_type = 'Transformer'self.src_mask = Noneself.pos_encoder = PositionalEncoding(feature_size)self.encoder_layer = nn.TransformerEncoderLayer(d_model=feature_size, nhead=10, dropout=dropout)self.transformer_encoder = nn.TransformerEncoder(self.encoder_layer, num_layers=num_layers)self.decoder = nn.Linear(feature_size, 1)self.init_weights()def init_weights(self):initrange = 0.1self.decoder.bias.data.zero_()self.decoder.weight.data.uniform_(-initrange, initrange)def forward(self, src):if self.src_mask is None or self.src_mask.size(0) != len(src):device = src.devicemask = self._generate_square_subsequent_mask(len(src)).to(device)self.src_mask = masksrc = self.pos_encoder(src)output = self.transformer_encoder(src, self.src_mask)output = self.decoder(output)return outputdef _generate_square_subsequent_mask(self, sz):mask = (torch.triu(torch.ones(sz, sz)) == 1).transpose(0, 1)mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))return mask# Create the dataset for the model
def create_inout_sequences(data, input_window, output_window):inout_seq = []length = len(data)for i in range(length - input_window - output_window):train_seq = data[i:i+input_window]train_label = data[i+input_window:i+input_window+output_window]inout_seq.append((train_seq, train_label))return inout_seqtrain_data = create_inout_sequences(train_data_normalized, input_window, output_window)
val_data = create_inout_sequences(val_data_normalized, input_window, output_window)# Train the model
model = TransAm().to(device)
criterion = nn.MSELoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=lr)
scheduler = torch.optim.lr_scheduler.StepLR(optimizer, 1.0, gamma=0.95)def train(train_data):model.train()total_loss = 0.for i in range(0, len(train_data) - 1, batch_size):data, targets = torch.stack([torch.tensor(item[0], dtype=torch.float32) for item in train_data[i:i+batch_size]]).to(device), torch.stack([torch.tensor(item[1], dtype=torch.float32) for item in train_data[i:i+batch_size]]).to(device)optimizer.zero_grad()output = model(data)loss = criterion(output, targets)loss.backward()torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5)optimizer.step()total_loss += loss.item()return total_loss / len(train_data)def validate(val_data):model.eval()total_loss = 0.with torch.no_grad():for i in range(0, len(val_data) - 1, batch_size):data, targets = torch.stack([torch.tensor(item[0], dtype=torch.float32) for item in val_data[i:i+batch_size]]).to(device), torch.stack([torch.tensor(item[1], dtype=torch.float32) for item in val_data[i:i+batch_size]]).to(device)output = model(data)loss = criterion(output, targets)total_loss += loss.item()return total_loss / len(val_data)best_val_loss = float("inf")
best_model = Nonefor epoch in range(1, epochs + 1):epoch_start_time = time.time()train_loss = train(train_data)val_loss = validate(val_data)scheduler.step()if val_loss < best_val_loss:best_val_loss = val_lossbest_model = model# Predict and denormalize the data
def predict(model, dataset):model.eval()predictions = []actuals = []with torch.no_grad():for i in range(len(dataset)):data, target = dataset[i]data = torch.tensor(data, dtype=torch.float32).to(device)output = model(data.unsqueeze(0))prediction = output.squeeze().cpu().numpy()predictions.append(prediction)actuals.append(target)return np.array(predictions), np.array(actuals)predictions, actuals = predict(best_model, val_data)
print("Predictions shape:", predictions.shape)
print("Actuals shape:", actuals.shape)predictions_denorm = scaler.inverse_transform(predictions)
actuals_denorm = scaler.inverse_transform(actuals.flatten().reshape(-1, 1))# Plot the results
plt.plot(predictions_denorm, label='Predictions')
plt.plot(actuals_denorm, label='Actuals')
plt.legend(['Predictions', 'Actuals'])
plt.xlabel('Timestep')
plt.ylabel('High')
plt.legend()
plt.show()更多时间序列预测代码：时间序列预测算法全集合--深度学习
