Transformer 预测模型，预测代码，可以直接替换数据。python代码，pytorch框架，有encoder decoder

Transformer 预测模型，预测代码，可以直接替换数据。python代码，pytorch框架，有encoder decoder。
多变量输入，单变量输出

2.informer预测 代码 模型
1.适合股票预测，风电预测等各类预测。
2.PyTorch框架实现 。
3.多输入单输出。
4.数据从excel/csv文件中读取，更换简单
且自带数据集，可自己替换数据集。
两个二选一哈，默认transformer。
Informer 时间序列预测

importpandas as pdimportnumpy as np from sklearn.preprocessingimportStandardScaler from sklearn.model_selectionimporttrain_test_splitimporttorch from torch.utils.dataimportDataset, DataLoaderimporttorch.nn as nnimporttorch.nn.functional as Fimportmatplotlib.pyplot as plt# 设置随机种子以保证可重复性torch.manual_seed(42)np.random.seed(42)# 数据加载与预处理def load_data(file_path, features, target, window_size, horizon): data=pd.read_csv(file_path,parse_dates=['timestamp'],index_col='timestamp')X=data[features].values y=data[target].values scaler_X=StandardScaler()scaler_y=StandardScaler()X_scaled=scaler_X.fit_transform(X)y_scaled=scaler_y.fit_transform(y.reshape(-1,1)).flatten()def create_sliding_windows(data, target, window_size, horizon): Xs, ys=[],[]foriinrange(len(data)- window_size - horizon +1):v=data[i:(i + window_size)]labels=target[(i + window_size):(i + window_size + horizon)]Xs.append(v)ys.append(labels)returnnp.array(Xs), np.array(ys)X_windows, y_windows=create_sliding_windows(X_scaled, y_scaled, window_size, horizon)X_train, X_test, y_train, y_test=train_test_split(X_windows, y_windows,test_size=0.2,random_state=42)class TimeSeriesDataset(Dataset): def __init__(self, X, y): self.X=torch.tensor(X,dtype=torch.float32)self.y=torch.tensor(y,dtype=torch.float32)def __len__(self):returnlen(self.X)def __getitem__(self, idx):returnself.X[idx], self.y[idx]train_dataset=TimeSeriesDataset(X_train, y_train)test_dataset=TimeSeriesDataset(X_test, y_test)train_loader=DataLoader(train_dataset,batch_size=32,shuffle=True)test_loader=DataLoader(test_dataset,batch_size=32,shuffle=False)returntrain_loader, test_loader, scaler_y# 定义模型组件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()*(-np.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): x=x + self.pe[:x.size(0), :]returnx class DataEmbedding(nn.Module): def __init__(self, c_in, d_model,dropout=0.1): super(DataEmbedding, self).__init__()self.value_embedding=nn.Linear(c_in, d_model)self.position_embedding=PositionalEncoding(d_model)self.dropout=nn.Dropout(p=dropout)def forward(self, x): x=self.value_embedding(x)+ self.position_embedding(x)returnself.dropout(x)class InformerModel(nn.Module): def __init__(self, enc_in, dec_in, c_out, seq_len, pred_len,factor=5,d_model=512,n_heads=8,e_layers=3,d_layers=2,d_ff=512,dropout=0.1,activation='gelu'): super(InformerModel, self).__init__()self.enc_embedding=DataEmbedding(enc_in, d_model, dropout)self.dec_embedding=DataEmbedding(dec_in, d_model, dropout)self.encoder=nn.TransformerEncoder(encoder_layer=nn.TransformerEncoderLayer(d_model=d_model,nhead=n_heads,dim_feedforward=d_ff,dropout=dropout),num_layers=e_layers)self.decoder=nn.TransformerDecoder(decoder_layer=nn.TransformerDecoderLayer(d_model=d_model,nhead=n_heads,dim_feedforward=d_ff,dropout=dropout),num_layers=d_layers)self.projection=nn.Linear(d_model, c_out)def forward(self, src, tgt): src=self.enc_embedding(src)tgt=self.dec_embedding(tgt)memory=self.encoder(src)output=self.decoder(tgt, memory)output=self.projection(output)returnoutput# 训练和评估模型def train_and_evaluate(model, optimizer, criterion, train_loader, test_loader,num_epochs=50):forepochinrange(num_epochs): model.train()running_loss=0.0forinputs, labelsintrain_loader: inputs=inputs.permute(1,0,2).to(device)# (seq_len, batch_size, input_dim)targets=labels.permute(1,0).unsqueeze(-1).to(device)# (seq_len, batch_size, output_dim)optimizer.zero_grad()outputs=model(inputs, targets[:-1])loss=criterion(outputs.view(-1), targets[1:].view(-1))loss.backward()optimizer.step()running_loss+=loss.item()print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {running_loss/len(train_loader):.4f}')# 评估模型model.eval()total_loss=0.0predictions=[]actuals=[]with torch.no_grad():forinputs, labelsintest_loader: inputs=inputs.permute(1,0,2).to(device)# (seq_len, batch_size, input_dim)targets=labels.permute(1,0).unsqueeze(-1).to(device)# (seq_len, batch_size, output_dim)outputs=model(inputs, targets[:-1])loss=criterion(outputs.view(-1), targets[1:].view(-1))total_loss+=loss.item()predictions.extend(outputs.cpu().numpy().squeeze())actuals.extend(targets[1:].cpu().numpy().squeeze())avg_loss=total_loss / len(test_loader)# 计算评价指标predictions=np.array(predictions)actuals=np.array(actuals)mse=np.mean((predictions-actuals)**2)rmse=np.sqrt(mse)mae=np.mean(np.abs(predictions-actuals))mape=np.mean(np.abs((predictions-actuals)/actuals))*100r2=1-((predictions-actuals)**2).sum()/((actuals-np.mean(actuals))**2).sum()print(f'Test Loss: {avg_loss:.4f}')print(f'R² Score: {r2:.4f}')print(f'RMSE: {rmse:.4f}')print(f'MAE: {mae:.4f}')print(f'MAPE: {mape:.4f}')# 保存结果到文本文件results=f"Test Loss: {avg_loss:.4f}\n"results+=f"R² Score: {r2:.4f}\n"results+=f"RMSE: {rmse:.4f}\n"results+=f"MAE: {mae:.4f}\n"results+=f"MAPE: {mape:.4f}\n"with open('results.txt','w')as f: f.write(results)# 可视化结果plt.figure(figsize=(12,6))plt.plot(actuals,label='Actual')plt.plot(predictions,label='Predicted')plt.title('Actual vs Predicted')plt.xlabel('Time Steps')plt.ylabel('Value')plt.legend()plt.savefig('prediction_vs_actual.png')plt.show()# 主函数if__name__=="__main__":# 参数设置file_path='data.csv'features=['feature1','feature2']target='target'window_size=10horizon=1# 单步预测，如果需要多步预测，设置更大的值input_dim=len(features)output_dim=horizon# 设备选择device=torch.device("cuda"iftorch.cuda.is_available()else"cpu")# 加载数据train_loader, test_loader, scaler_y=load_data(file_path, features, target, window_size, horizon)# 初始化模型model=InformerModel(enc_in=input_dim,dec_in=output_dim,c_out=output_dim,seq_len=window_size,pred_len=horizon,d_model=512,n_heads=8,e_layers=3,d_layers=2,d_ff=512,dropout=0.1,activation='gelu').to(device)criterion=nn.MSELoss()optimizer=torch.optim.Adam(model.parameters(),lr=0.001)# 训练和评估模型train_and_evaluate(model, optimizer, criterion, train_loader, test_loader)

当然可以！下面我将为你提供两个版本的代码：一个是基于Transformer Encoder-Decoder架构的时间序列预测模型，另一个是Informer模型。这两个模型都适合多变量输入和单变量输出，并且可以从Excel/CSV文件中读取数据。

1. Transformer Encoder-Decoder 模型

项目介绍

• 模型架构:

  • Encoder: 处理输入时间序列数据。
  • Decoder: 生成未来的预测值。

• 功能:

  • 多变量输入，单变量输出。
  • 数据从Excel/CSV文件中读取，更换简单。
  • 自带数据集，可自己替换数据集。
  • 评价指标: R², RMSE, MAE, MAPE
  • 结果保存到文本中，可以后续处理。

代码实现

2. Informer 预测模型

项目介绍

• 模型架构:

  • 基于Informer论文中的架构，结合了自注意力机制和局部特征提取。

• 功能:

  • 多变量输入，单变量输出。
  • 数据从Excel/CSV文件中读取，更换简单。
  • 自带数据集，可自己替换数据集。
  • 评价指标: R², RMSE, MAE, MAPE
  • 结果保存到文本中，可以后续处理。

代码实现

如何使用这些代码

1. 准备数据：

   • 确保你的数据集格式正确（例如CSV文件），并且包含特征列和目标列。
   • 示例数据data.csv的结构如下：

     timestamp,feature1,feature2,target 2023-01-01 00:00:00,0.1,0.2,3.4 2023-01-01 00:01:00,0.5,0.6,7.8 ...

2. 替换数据路径：

   • 在load_data函数中，将'data.csv'替换为你的数据文件路径。

   file_path='your_data_file.csv'

3. 调整窗口大小和预测步数：

   • 根据你的需求调整window_size和horizon参数。

   window_size=20# 更改窗口大小horizon=5# 更改预测步数

4. 运行代码：

   • 将上述代码复制到你的Python脚本中，并运行该脚本。
   • 确保你已经安装了所需的库：

     pipinstallpandas numpy scikit-learn torch matplotlib

示例：使用自定义数据集

假设你有一个新的数据集new_data.csv，其内容如下：

timestamp,feature1,feature2,target 2023-01-01 00:00:00,0.1,0.2,3.4 2023-01-01 00:01:00,0.5,0.6,7.8 ...

你可以按照以下步骤进行替换：

1. 修改数据路径：

   file_path='new_data.csv'

2. 调整窗口大小和预测步数：

   window_size=20# 更改窗口大小horizon=5# 更改预测步数

3. 运行完整的代码：

   • 将所有代码整合到一个Python脚本中，并运行该脚本。

注释说明

代码中包含了详细的注释，帮助你理解每个部分的功能。以下是关键部分的注释：

• 数据加载与预处理：

  • load_data: 加载数据并进行标准化，创建滑动窗口。

• 模型组件：

  • PositionalEncoding: 添加位置编码，帮助模型理解序列的位置信息。
  • DataEmbedding: 结合值嵌入和位置嵌入。
  • TransformerModel: 实现Transformer Encoder-Decoder架构。
  • InformerModel: 实现Informer架构，结合自注意力机制和局部特征提取。

• 训练和评估模型：

  • train_and_evaluate: 训练模型并计算评价指标，保存结果并可视化。

结果

运行代码后，你将得到以下结果：

1. 控制台输出：

   • 每个epoch的损失值。
   • 测试集上的评价指标（R², RMSE, MAE, MAPE）。

2. 文件输出：

   • results.txt: 包含测试集上的评价指标。

3. 图像输出：

   • prediction_vs_actual.png: 实际值与预测值的对比图。

希望这些详细的信息和代码能够帮助你顺利实施和优化你的项目。如果你有任何进一步的问题或需要更多帮助，请随时提问！

默认选择 Transformer Encoder-Decoder 模型

默认情况下，我们将使用Transformer Encoder-Decoder模型。如果你更倾向于使用Informer模型，只需切换到相应的代码块即可。

运行默认的 Transformer Encoder-Decoder 模型

确保你已经安装了所需的库：

pipinstallpandas numpy scikit-learn torch matplotlib

然后运行以下命令来执行Transformer Encoder-Decoder模型：

python transformer_encoder_decoder.py

其中transformer_encoder_decoder.py是包含上述Transformer Encoder-Decoder模型代码的文件。

运行 Informer 模型

同样地，确保你已经安装了所需的库，然后运行以下命令来执行Informer模型：

python informer.py

其中informer.py是包含上述Informer模型代码的文件。

希望这些详细的指导和代码示例能帮助你成功实现和优化你的时间序列预测项目。如果有任何问题或需要进一步的帮助，请随时告诉我！