当前位置：首页 > news >正文

用Python和LSTM搞定风电功率预测：从数据清洗到区间预测的完整实战（附2018年数据集）

news 2026/5/24 5:50:49

风电功率预测实战：基于Python与LSTM的完整实现指南

风电作为清洁能源的重要组成部分，其功率预测对电网调度和能源管理至关重要。本文将带您从零开始构建一个完整的LSTM风电功率预测系统，涵盖数据预处理、模型构建、训练优化到结果可视化的全流程。不同于简单的理论介绍，我们更注重实战操作和代码细节，确保您能够亲手复现整个项目。

1. 环境准备与数据加载

在开始之前，我们需要配置好Python环境并安装必要的库。推荐使用Anaconda创建虚拟环境以避免依赖冲突：

conda create -n wind_power python=3.8 conda activate wind_power pip install tensorflow pandas numpy matplotlib scikit-learn

数据集采用2018年6-8月的风电功率记录，采样间隔为15分钟。原始数据通常以CSV格式存储，我们可以使用Pandas轻松加载：

import pandas as pd # 加载原始数据 data = pd.read_csv('wind_power_2018.csv', parse_dates=['timestamp'], index_col='timestamp') print(data.head())

典型的风电数据集包含以下字段：

timestamp: 时间戳（15分钟间隔）
wind_speed: 风速（m/s）
wind_direction: 风向（度）
temperature: 温度（℃）
power: 风电功率（kW）

注意：实际数据中可能存在缺失值或异常值，我们需要在下一步进行清洗和处理。

2. 数据预处理与特征工程

高质量的数据预处理是模型成功的关键。我们需要依次完成以下步骤：

2.1 缺失值与异常值处理

# 检查缺失值 print(data.isnull().sum()) # 简单填充（可根据实际情况选择更复杂的方法） data.fillna(method='ffill', inplace=True) # 移除明显异常值（功率不可能为负） data = data[data['power'] >= 0]

2.2 数据标准化与差分

时序数据通常需要进行标准化和差分以消除量纲和趋势影响：

from sklearn.preprocessing import MinMaxScaler # 初始化标准化器 scaler = MinMaxScaler(feature_range=(0, 1)) # 对特征进行标准化 scaled_features = scaler.fit_transform(data.drop('power', axis=1)) scaled_power = scaler.fit_transform(data[['power']]) # 一阶差分 diff_power = data['power'].diff().dropna()

2.3 构建监督学习数据集

LSTM需要将时序数据转换为监督学习格式。我们定义函数将数据转换为"过去N个时间点预测未来M个时间点"的形式：

def create_dataset(data, look_back=20, look_forward=1): X, y = [], [] for i in range(len(data)-look_back-look_forward): X.append(data[i:(i+look_back)]) y.append(data[(i+look_back):(i+look_back+look_forward)]) return np.array(X), np.array(y)

3. LSTM模型构建与训练

3.1 网络架构设计

我们构建一个双层LSTM网络，使用分位数回归作为损失函数：

from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense from tensorflow.keras.optimizers import Adam def build_quantile_lstm(input_shape, quantiles=[0.1, 0.5, 0.9]): model = Sequential() model.add(LSTM(64, return_sequences=True, input_shape=input_shape)) model.add(LSTM(64)) # 为每个分位数添加输出层 outputs = [] for q in quantiles: outputs.append(Dense(1)(model.output)) quantile_model = Model(inputs=model.input, outputs=outputs) quantile_model.compile(optimizer=Adam(learning_rate=0.001), loss=lambda y_true, y_pred: quantile_loss(y_true, y_pred, q=quantiles)) return quantile_model

3.2 分位数损失函数实现

分位数回归需要自定义损失函数：

def quantile_loss(y_true, y_pred, q): e = y_true - y_pred return tf.reduce_mean(tf.maximum(q*e, (q-1)*e), axis=-1)

3.3 模型训练与验证

将数据划分为训练集和测试集，并开始训练：

# 划分训练测试集 train_size = int(len(X) * 0.8) X_train, X_test = X[:train_size], X[train_size:] y_train, y_test = y[:train_size], y[train_size:] # 构建并训练模型 model = build_quantile_lstm(input_shape=(X_train.shape[1], X_train.shape[2])) history = model.fit(X_train, [y_train]*3, # 三个输出对应三个分位数 epochs=30, batch_size=64, validation_data=(X_test, [y_test]*3))

4. 结果分析与可视化

4.1 预测区间可视化

我们可以绘制不同时间步长的预测区间：

import matplotlib.pyplot as plt def plot_prediction_intervals(actual, pred_low, pred_median, pred_high, title): plt.figure(figsize=(12, 6)) plt.plot(actual, label='Actual Power', color='blue') plt.plot(pred_median, label='Median Prediction', color='green') plt.fill_between(range(len(actual)), pred_low.flatten(), pred_high.flatten(), color='gray', alpha=0.3, label='80% Prediction Interval') plt.title(title) plt.xlabel('Time Steps') plt.ylabel('Normalized Power') plt.legend() plt.show() # 获取测试集预测结果 pred_low, pred_median, pred_high = model.predict(X_test) plot_prediction_intervals(y_test, pred_low, pred_median, pred_high, '30-min Ahead Prediction')

4.2 评估指标计算

计算不同时间步长的预测误差：

from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score def evaluate_predictions(y_true, y_pred, step): mae = mean_absolute_error(y_true, y_pred) rmse = np.sqrt(mean_squared_error(y_true, y_pred)) r2 = r2_score(y_true, y_pred) print(f'{step}min预测指标:') print(f'MAE: {mae:.4f}') print(f'RMSE: {rmse:.4f}') print(f'R2: {r2:.4f}\n') return mae, rmse, r2 # 评估不同时间步长 time_steps = [30, 60, 90, 120, 150] metrics = [] for step in time_steps: # 这里需要根据实际预测步长调整输入数据 mae, rmse, r2 = evaluate_predictions(y_test, pred_median, step) metrics.append((step, mae, rmse, r2))

4.3 概率密度可视化

对于特定时间点，我们可以绘制预测功率的概率密度分布：

import seaborn as sns def plot_probability_density(actual, samples, time_point): plt.figure(figsize=(10, 6)) sns.kdeplot(samples, label='Predicted Distribution', fill=True) plt.axvline(x=actual, color='r', linestyle='--', label='Actual Value') plt.title(f'Probability Density at {time_point}') plt.xlabel('Normalized Power') plt.ylabel('Density') plt.legend() plt.show() # 示例：选择测试集中特定时间点 sample_point = 100 # 示例索引 plot_probability_density(y_test[sample_point], np.random.normal(loc=pred_median[sample_point], scale=(pred_high[sample_point]-pred_low[sample_point])/2, size=1000), data.index[train_size + sample_point])

5. 模型优化与生产部署

5.1 超参数调优

可以通过网格搜索寻找最佳超参数组合：

from sklearn.model_selection import GridSearchCV from tensorflow.keras.wrappers.scikit_learn import KerasRegressor # 创建Keras模型包装器 def create_model(units=64, learning_rate=0.001): model = build_quantile_lstm(input_shape=(X_train.shape[1], X_train.shape[2])) model.compile(optimizer=Adam(learning_rate=learning_rate), loss=lambda y_true, y_pred: quantile_loss(y_true, y_pred, q=[0.1, 0.5, 0.9])) return model # 定义参数网格 param_grid = { 'units': [32, 64, 128], 'learning_rate': [0.01, 0.001, 0.0001], 'batch_size': [32, 64, 128] } # 执行网格搜索 grid = GridSearchCV(estimator=KerasRegressor(build_fn=create_model, epochs=20), param_grid=param_grid, cv=3) grid_result = grid.fit(X_train, y_train)

5.2 模型保存与加载

训练好的模型可以保存为HDF5格式以便后续使用：

model.save('wind_power_lstm.h5') # 加载模型 from tensorflow.keras.models import load_model loaded_model = load_model('wind_power_lstm.h5', custom_objects={'quantile_loss': quantile_loss})

5.3 实时预测API示例

将模型部署为REST API供生产环境调用：

from flask import Flask, request, jsonify import numpy as np app = Flask(__name__) model = load_model('wind_power_lstm.h5') @app.route('/predict', methods=['POST']) def predict(): data = request.json['data'] data = np.array(data).reshape(1, -1, 1) # 根据实际输入形状调整 pred_low, pred_median, pred_high = model.predict(data) return jsonify({ 'prediction': pred_median.tolist(), 'lower_bound': pred_low.tolist(), 'upper_bound': pred_high.tolist() }) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)