一、基于DNN的风电功率预测

1.1、背景

        在全球能源转型的浪潮中，风力发电因其清洁和可再生的特性而日益重要。然而，风力发电功率的波动性给电网的稳定运行和能源调度带来了挑战。准确预测风力发电机的功率输出，对于优化能源管理、提高电网可靠性以及促进风能的高效利用至关重要。传统的预测方法在应对风力发电固有的复杂性和非线性时存在局限，因此，利用深度学习等先进人工智能技术，从历史运行数据中学习并预测风功率，已成为一个重要的研究方向。

1.2、目标

        旨在探索如何利用风机运行的历史数据，特别是风速、发电机转速和叶片角度等关键参数，构建一个深度学习模型来预测风力发电机的功率输出。通过对数据进行预处理、选择合适的模型架构、进行有效的训练和验证，并在独立的测试集上评估模型的预测性能，最终实现对风力发电机功率的准确预测，并可视化展示预测结果。

1.3、数据展示

二、数据准备与预处理

class PowerData(Dataset):
    def __init__(self,path,input_len):
        super().__init__()
        # 从CSV文件读取数据
        self.data=pd.read_csv(path)
        # 定义输入序列的长度
        self.input_len=input_len
        # 将功率值限制在0到1500之间，处理异常值
        self.data['功率(kW)']=np.minimum(self.data['功率(kW)'],1500)
        # 初始化MinMaxScaler，用于将功率值归一化到[-1, 1]的范围
        self.scaler=MinMaxScaler(feature_range=(-1,1))
        # 对功率数据进行归一化处理并存储
        self.data['power_nomalized']=self.scaler.fit_transform(self.data['功率(kW)'].values.reshape(-1,1))

    # 返回数据集的长度，即可以生成多少个输入-输出对
    def __len__(self):
        return len(self.data)-self.input_len

    # 根据给定的索引返回一个样本
    def __getitem__(self, idx):
        # 定义输入序列的起始和结束索引
        start_idx=idx
        end_idx=idx+self.input_len
        # 获取输入特征序列
        feature=self.data['power_nomalized'].values[start_idx:end_idx]
        # 获取对应的目标值（下一个时间点的功率值）
        target=self.data['power_nomalized'].values[end_idx:end_idx+1]
        # 将特征和目标转换为PyTorch张量并返回
        return  torch.tensor(feature,dtype=torch.float32),torch.tensor(target,dtype=torch.float32)

# 创建PowerData数据集实例，指定数据文件路径和输入序列长度
power_dataset=PowerData('./A01.csv',input_len=6)

Scaler之前：(tensor([327.0000, 327.0000, 252.6000, 211.2000, 211.2000, 159.0000]), tensor([184.2000]))
Scaler之后：(tensor([-0.5640, -0.5640, -0.6632, -0.7184, -0.7184, -0.7880]), tensor([-0.7544]))

# 定义训练集、验证集和测试集的比例
train_ratio = 0.8
val_ratio = 0.1
test_ratio = 0.1
# 计算各个数据集的大小
train_size = int(train_ratio * len(power_dataset))
val_size = int(val_ratio * len(power_dataset))
test_size = len(power_dataset) - train_size - val_size # 确保测试集大小正确

# 创建训练集、验证集和测试集的Subset
train_dataset=Subset(power_dataset,list(range(len(power_dataset)))[:train_size])
val_dataset=Subset(power_dataset,list(range(len(power_dataset)))[train_size:train_size+val_size])
test_dataset=Subset(power_dataset,list(range(len(power_dataset)))[train_size+val_size:])

# 创建训练集、验证集和测试集的数据加载器
train_dataloader=DataLoader(train_dataset,batch_size=64,shuffle=True)
val_dataloader=DataLoader(val_dataset,batch_size=64,shuffle=False)
test_dataloader=DataLoader(test_dataset,batch_size=1,shuffle=False)

三、模型构建

class RNN(nn.Module):
    def __init__(self, input_size=1, hidden_size=128, output_size=1):
        """
        初始化
        :param input_size: 输入数据的特征维度大小，每个时间步输入的特征向量的维度
        :param hidden_size:
        :param output_size:
        """
        super(RNN, self).__init__()
        self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)
        self.linear2 = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        # print("----", x.shape)  # torch.Size([64, 20])
        x = x.unsqueeze(2)  # torch.Size([64, 20, 1])
        w, h_n = self.rnn(x)  # h_n是最后一个时间步的隐藏状态
        # print("h_n:", h_n.shape)  # (num_layers*num_directions, batch, hidden_size)->[1, 64, 128]
        # 例如num_layers=2， [2, 64, 128]
        x = (self.linear2(h_n[-1])).squeeze(1)  # self.linear2的输出张量[64, 1]

        # x = F.relu(self.linear1(x))
        # x = self.linear2(x)
        return x


model = RNN().to(device)

四、模型训练与验证

# 定义损失函数为均方误差
cri=nn.MSELoss()
# 定义优化器为Adam，并设置学习率和权重衰减
optim=torch.optim.Adam(model.parameters(),lr=0.001,weight_decay=1e-5)

epochs = 100
for epoch in range(1, epochs + 1):
    model.train()
    total_loss = 0
    for batch_feature, batch_target in train_dataloader:
        batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)
        # 前向传播
        y_pred = model(batch_feature)
        # y_pred [bs, 1] -> [bs,]
        loss = cri(y_pred, batch_target.view(-1))
        # 反向传播更新
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        total_loss += loss.item()
    train_loss = total_loss / len(train_dataloader)
    # 验证集的损失
    model.eval()
    total_loss = 0
    with torch.no_grad():
        for batch_feature, batch_target in val_dataloader:
            batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)
            y_pred = model(batch_feature)
            total_loss += cri(y_pred, batch_target.view(-1)).item()
    val_loss = total_loss / len(val_dataloader)
    print(f"Epoch:[{epoch}/{epochs}], Train Loss: {train_loss:.4f}, Eval Loss: {val_loss:.4f}")

五、模型评估与结果可视化

# 计算测试结果
model.eval()
predict_list = []
target_list = []
with torch.no_grad():
    for batch_feature, batch_target in test_dataloader:
        batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)
        y_pred = model(batch_feature)
        predict_list.append(y_pred.item())
        target_list.append(batch_target.item())

# 将预测结果反归一化
predict_list = power_dataset.scaler.inverse_transform(np.array(predict_list).reshape(-1, 1))
target_list = power_dataset.scaler.inverse_transform(np.array(target_list).reshape(-1, 1))

plt.plot(target_list, label="True values")
plt.plot(predict_list, label="Predict values")
plt.xlabel("Time")
plt.ylabel("Power")
plt.legend()
plt.show()

 六、完整代码

import os  
import random  
import torch  
import torch.nn as nn  
import torch.nn.functional as F  

import matplotlib.pyplot as plt  
import numpy as np  
from torch.utils.data import Dataset, Subset, DataLoader  
import pandas as pd  
from sklearn.preprocessing import MinMaxScaler  


# 定义函数设置随机种子，保证每次训练结果一致  
def setup_seed(seed):  
    np.random.seed(seed)  # Numpy模块设置随机种子  
    random.seed(seed)  # Python自带的random模块  
    os.environ['PYTHONHASHSEED'] = str(seed)  # 设置环境变量  
    torch.manual_seed(seed)  # PyTorch CPU随机种子  
    if torch.cuda.is_available():  
        torch.cuda.manual_seed(seed)  # CUDA随机种子  
        torch.cuda.manual_seed_all(seed)  # 所有GPU  
        torch.backends.cudnn.benchmark = False  # 禁用CuDNN的快速算法（结果可复现）  
        torch.backends.cudnn.deterministic = True  # 使用确定性算法  

setup_seed(1)  # 设置随机种子  

# 判断是否有GPU可用  
if torch.cuda.is_available():  
    device = torch.device("cuda")  # 使用GPU  
    print("CUDA is available. Using GPU.")  
else:  
    device = torch.device("cpu")  # 用CPU  
    print("CUDA is not available. Using CPU.")  


# 定义数据集类，用于加载和处理电功率数据  
class PowerData(Dataset):  
    def __init__(self, csv_path, sequence_len):  
        self.sequence_len = sequence_len  
        # 读取CSV数据  
        self.data = pd.read_csv(csv_path)  
        # 限制“功率(kW)”最大值为1500  
        self.data["功率(kW)"] = np.minimum(self.data["功率(kW)"], 1500)  
        # 使用MinMaxScaler将功率缩放到[-1, 1]  
        self.scaler = MinMaxScaler(feature_range=(-1, 1))  
        # 将原始“功率(kW)”列归一化，存储为新列“power_normalized”  
        self.data["power_normalized"] = self.scaler.fit_transform(  
            self.data["功率(kW)"].values.reshape(-1, 1))  

    def __len__(self):  
        # 计算数据集长度（考虑序列长度）  
        return len(self.data) - self.sequence_len  

    def __getitem__(self, idx):  
        # 获取起点和终点的索引  
        start_idx = idx  
        end_idx = idx + self.sequence_len  
        # 取连续的“power_normalized”作为输入特征  
        feature = self.data["power_normalized"].values[start_idx:end_idx]  
        # 目标值为序列后面一个点的标准化值  
        target = self.data["power_normalized"].values[end_idx:end_idx + 1]  
        # 转换为tensor  
        return torch.tensor(feature, dtype=torch.float32), torch.tensor(target, dtype=torch.float32)  


sequence_len = 20  # 序列长度为20  
power_dataset = PowerData("./A01.csv", sequence_len)  

# 按比例划分训练集、验证集、测试集  
train_ratio = 0.8  
val_ratio = 0.1  
test_ratio = 0.1  
train_size = int(train_ratio * len(power_dataset))  
val_size = int(val_ratio * len(power_dataset))  
test_size = int(test_ratio * len(power_dataset))  
# 生成索引列表  
indices = list(range(len(power_dataset)))  
# 划分子集  
train_dataset = Subset(power_dataset, indices[:train_size])  
val_dataset = Subset(power_dataset, indices[train_size:train_size + val_size])  
test_dataset = Subset(power_dataset, indices[train_size + val_size:])  

# 创建数据加载器（DataLoader）  
train_dataloader = DataLoader(train_dataset, batch_size=64, shuffle=True)  
val_dataloader = DataLoader(val_dataset, batch_size=64, shuffle=False)  
test_dataloader = DataLoader(test_dataset, batch_size=1, shuffle=False)  


# 定义RNN模型  
class RNN(nn.Module):  
    def __init__(self, input_size=1, hidden_size=128, output_size=1):  
        super(RNN, self).__init__()  
        # RNN层  
        self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)  
        # 全连接层，将隐藏状态映射到输出  
        self.linear2 = nn.Linear(hidden_size, output_size)  

    def forward(self, x):  
        # 输入x的形状：(batch, 序列长度)  
        x = x.unsqueeze(2)  # 增加特征维度，使形状变为：(batch, 序列长度, 1)  
        w, h_n = self.rnn(x)  # 返回全部时间步的输出和最后一个时间步的隐藏状态  
        # 取最后一个隐藏状态，经过线性层映射  
        x = (self.linear2(h_n[-1])).squeeze(1)  # 输出形状：(batch, 1)  
        return x  

# 实例化模型，并转到设备（GPU或CPU）  
model = RNN().to(device)  
# 定义损失函数（均方误差）  
cri = nn.MSELoss()  
# 定义优化器（Adam）  
optimizer = torch.optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-5)  

# 训练模型，迭代100轮  
epochs = 100  
for epoch in range(1, epochs + 1):  
    # 训练阶段  
    model.train()  
    total_loss = 0  
    for batch_feature, batch_target in train_dataloader:  
        batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)  
        # 前向传播  
        y_pred = model(batch_feature)  
        # 计算损失（注意：batch_target形状是(batch_size, 1)，需要转换为一维）  
        loss = cri(y_pred, batch_target.view(-1))  
        # 梯度清零  
        optimizer.zero_grad()  
        # 反向传播  
        loss.backward()  
        # 更新参数  
        optimizer.step()  

        total_loss += loss.item()  
    # 计算当前轮的平均训练损失  
    train_loss = total_loss / len(train_dataloader)  

    # 验证集  
    model.eval()  
    total_loss = 0  
    with torch.no_grad():  
        for batch_feature, batch_target in val_dataloader:  
            batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)  
            y_pred = model(batch_feature)  
            total_loss += cri(y_pred, batch_target.view(-1)).item()  
    val_loss = total_loss / len(val_dataloader)  

    print(f"Epoch:[{epoch}/{epochs}], Train Loss: {train_loss:.4f}, Eval Loss: {val_loss:.4f}")  

# 测试模型  
model.eval()  
predict_list = []  
target_list = []  
with torch.no_grad():  
    for batch_feature, batch_target in test_dataloader:  
        batch_feature, batch_target = batch_feature.to(device), batch_target.to(device)  
        y_pred = model(batch_feature)  
        # 获取预测值和真实值（单个样本）  
        predict_list.append(y_pred.item())  
        target_list.append(batch_target.item())  

# 反归一化还原到原始功率  
predict_list = power_dataset.scaler.inverse_transform(np.array(predict_list).reshape(-1, 1))  
target_list = power_dataset.scaler.inverse_transform(np.array(target_list).reshape(-1, 1))  

# 绘图显示预测与真实值  
plt.plot(target_list, label="True values")  
plt.plot(predict_list, label="Predict values")  
plt.xlabel("Time")  
plt.ylabel("Power")  
plt.legend()  
plt.show()