发电机安全裕度预测数据集

import os import numpy as np import pandas as pd from tqdm import tqdm import torch from torch import nn import torch.utils.data as data import torch.nn.functional as F from torch import tensor import torch.utils.data as Data import math from matplotlib import pyplot from sklearn.model_selection import train_test_split import matplotlib.pyplot as plt import warnings warnings.filterwarnings("ignore") import torch import torch.nn as nn import math from datetime import datetime, timedelta plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示中文标签 plt.rcParams['axes.unicode_minus'] = False # 用来正常显示负号 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") data = pd.read_csv("data(2).csv", encoding = 'utf-8') # 1 3 7 是 预测列 data_x=data[['发电机2','发电机3']].values data_y=data['安全裕度最小值'].values print(len(data_y)) # 四个数据划分为一组 用前三个预测后一个 data_4_x=[] data_4_y=[] for i in range(0,len(data_y)-4,4): data_4_x.append(data_x[i:i+3]) data_4_y.append(data_y[i+4]) print(len(data_4_x),len(data_4_y)) x_train,x_test,y_train,y_test = train_test_split(np.array(data_4_x),np.array(data_4_y),test_size=0.2) class DataSet(Data.Dataset): def __init__(self, data_inputs, data_targets): self.inputs = torch.LongTensor(data_inputs) self.label = torch.LongTensor(data_targets) def __getitem__(self, index): return self.inputs[index], self.label[index] def __len__(self): return len(self.inputs) Batch_Size=8 DataSet = DataSet(np.array(x_train), list(y_train)) train_size = int(len(x_train) * 0.8) test_size = len(y_train) - train_size train_dataset, test_dataset = torch.utils.data.random_split(DataSet, [train_size, test_size]) TrainDataLoader = Data.DataLoader(train_dataset, batch_size=Batch_Size, shuffle=True, drop_last=True) TestDataLoader = Data.DataLoader(test_dataset, batch_size=Batch_Size, shuffle=True, drop_last=True) 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) # pe.requires_grad = False self.register_buffer('pe', pe) def forward(self, x: torch.Tensor): chunk = x.chunk(x.size(-1), dim=2) out = torch.Tensor([]).to(x.device) for i in range(len(chunk)): out = torch.cat((out, chunk[i] + self.pe[:chunk[i].size(0), ...]), dim=2) return out def transformer_generate_tgt_mask(length, device): mask = torch.tril(torch.ones(length, length, device=device)) == 1 mask = ( mask.float() .masked_fill(mask == 0, float("-inf")) .masked_fill(mask == 1, float(0.0)) ) return mask class Transformer(nn.Module): """标准的Transformer编码器-解码器结构""" def __init__(self, n_encoder_inputs, n_decoder_inputs,Sequence_length,d_model=512, dropout=0.1, num_layer=8): """ 初始化 :param n_encoder_inputs: 输入数据的特征维度 :param n_decoder_inputs: 编码器输入的特征维度，其实等于编码器输出的特征维度 :param d_model: 词嵌入特征维度 :param dropout: dropout :param num_layer: Transformer块的个数 Sequence_length: transformer 输入数据 序列的长度 """ super(Transformer, self).__init__() self.input_pos_embedding = torch.nn.Embedding(5000, embedding_dim=d_model) self.target_pos_embedding = torch.nn.Embedding(5000, embedding_dim=d_model) encoder_layer = torch.nn.TransformerEncoderLayer(d_model=d_model, nhead=8, dropout=dropout, dim_feedforward=4 * d_model) decoder_layer = torch.nn.TransformerDecoderLayer(d_model=d_model, nhead=8, dropout=dropout, dim_feedforward=4 * d_model) self.encoder = torch.nn.TransformerEncoder(encoder_layer, num_layers=2) self.decoder = torch.nn.TransformerDecoder(decoder_layer, num_layers=4) self.input_projection = torch.nn.Linear(n_encoder_inputs, d_model) self.output_projection = torch.nn.Linear(n_decoder_inputs, d_model) self.linear = torch.nn.Linear(d_model, 1) self.ziji_add_linear=torch.nn.Linear(Sequence_length, 1) def encode_in(self, src): src_start = self.input_projection(src).permute(1, 0, 2) in_sequence_len, batch_size = src_start.size(0), src_start.size(1) pos_encoder = (torch.arange(0, in_sequence_len, device=src.device).unsqueeze(0).repeat(batch_size, 1)) pos_encoder = self.input_pos_embedding(pos_encoder).permute(1, 0, 2) src = src_start + pos_encoder src = self.encoder(src) + src_start return src def decode_out(self, tgt, memory): tgt_start = self.output_projection(tgt).permute(1, 0, 2) out_sequence_len, batch_size = tgt_start.size(0), tgt_start.size(1) pos_decoder = (torch.arange(0, out_sequence_len, device=tgt.device).unsqueeze(0).repeat(batch_size, 1)) pos_decoder = self.target_pos_embedding(pos_decoder).permute(1, 0, 2) tgt = tgt_start + pos_decoder tgt_mask = transformer_generate_tgt_mask(out_sequence_len, tgt.device) out = self.decoder(tgt=tgt, memory=memory, tgt_mask=tgt_mask) + tgt_start out = out.permute(1, 0, 2) # [batch_size, seq_len, d_model] out = self.linear(out) return out def forward(self, src, target_in): src = self.encode_in(src) out = self.decode_out(tgt=target_in, memory=src) # print("out.shape:",out.shape)# torch.Size([batch, 3, 1]) # 原本代码中的输出 # 上边的这个输入可以用于很多任务的输出 可以根据任务进行自由的变换 # 下面是自己修改的 # 使用全连接变成 [batch,1] 构成了基于transformer的回归单值预测 out=out.squeeze(2) out=self.ziji_add_linear(out) return out model = Transformer(n_encoder_inputs=2, n_decoder_inputs=2,Sequence_length=3).to(device) # # 3 表示Sequence_length transformer 输入数据 序列的长度 def test(): with torch.no_grad(): val_epoch_loss=[] # for i in range(0, len(x_test),batch):# batch是 1 测试用1测试就行 for index, (inputs, targets) in enumerate(TrainDataLoader): # inputs = x_test[i:i+batch] # targets = y_test[i:i+batch] # if len(inputs) == batch: # 最后一个batch可能不足长度 舍弃 inputs = torch.tensor(inputs).to(device) targets = torch.tensor(targets).to(device) inputs = inputs.float() targets = targets.float() tgt_in = torch.rand((Batch_Size, 3, 2)) outputs = model(inputs,tgt_in) loss = criterion(outputs.float(), targets.float()) val_epoch_loss.append(loss.item()) return np.mean(val_epoch_loss) epochs = 50 optimizer = torch.optim.Adam(model.parameters(), lr=0.01) criterion = torch.nn.MSELoss().to(device) val_loss = [] train_loss = [] best_test_loss = 10000000 for epoch in tqdm(range(epochs)): train_epoch_loss = [] # for i in range(0, len(x_train),batch):# batch是 1 for inde