简介这份基于LSTM的电力负荷预测Python源码面向电力系统研究人员与机器学习工程师利用历史负荷数据构建高精度预测模型涵盖数据清洗、归一化、训练/测试集划分、LSTM网络搭建、训练与MAPE评估的完整链路。资源包共350个文件其中约170个csv文件为电力负荷历史数据集14个py脚本对应数据预处理、模型定义与训练评估另有index/meta/data等模型检查点文件压缩包大小11.36MB便于快速部署。目前已有83人学习下载。内容预览中给出了不同训练步数下的MAPE结果如7500步对应0.0474、6000步对应0.0476可直接对比模型收敛效果辅助调整训练参数与网络结构。通过研读源码读者既能掌握时间序列预测的工程实现方法也能将这套流程迁移到其他负荷预测或相关时序任务中具备较高的实践参考价值。1. 基于 LSTM 的电力负荷预测 python 源码为什么简单的 ARIMA 会先翻车电力负荷预测的核心难点不是模型多高级而是数据背后的周期性、趋势和天气扰动三重因素叠加。我早先用 ARIMA 做过一轮短期负荷预测换季降温那几天误差能到 8% 以上后来换成 LSTM 神经网络同样数据上把 MAPE 压到了 4% 左右。这也是为什么「lstm 时间序列预测 python」的搜索量一直不低——大家需要的不是概念是一段能跑通的基于 LSTM 的电力负荷预测 python 源码。这篇文章把我生产里用过的流程拆开讲数据怎么切、模型怎么搭、指标怎么验证、坑在哪。入门者可以照着把第一个版本跑出来熟手也能在参数边界和滚动预测上少踩几个坑。2. 数据准备把原始负荷序列切成长短合适的样本2.1 为什么不能把原始序列直接丢给 LSTMLSTM 是监督学习模型需要输入 X 和标签 y 成对出现。对电力负荷来说y 一般是未来某个时刻的负荷X 则是这个时刻之前的连续负荷值。直接把整段序列丢给 LSTM 没有意义因为模型不知道该从哪里学映射关系。常见做法是构造滑动窗口用过去 lookback 个时间步预测下一步。lookback 的取值与数据粒度强相关。如果数据是 15 分钟一个点那么一天 96 点一周 672 点我一般先试 96 或 192如果是小时级数据一天 24 点一周 168 点窗口可以取 24 或 48。窗口太短模型看不到日周期性窗口太长训练成本上去了而且引入的早期噪声反而会干扰预测。先从一个周期长度的窗口开始调比盲目堆神经元更有效。2.2 数据清洗缺失值和异常 0 值先处理负荷数据从 SCADA 系统导出来常见问题有两个时间戳不连续以及某几个点为 0 或缺失。如果直接把 0 当正常值喂给模型LSTM 会记住一个假的低谷后续预测在低谷时刻出现系统性偏差。我一般先统一频率再把明显异常的 0 值置为缺失最后做填充。import pandas as pd import numpy as np df pd.read_csv(load.csv, parse_dates[timestamp], index_coltimestamp) # 15 分钟间隔统一重采样频率缺失值会变成 NaN df df.resample(15min).mean() # 负荷为 0 大多是传感器或通讯故障先转成 NaN df[load] df[load].replace(0, np.nan) # 缺失不多时用前向 后向填充顶多能顶掉 2 个点 df[load] df[load].ffill().bfill()这段代码的要点是先resample统一到固定频率保证后面生成的窗口里每个时间步间隔相等。如果原始数据时间戳不连续resample会把缺失时间点补成 NaN再用ffill().bfill()填上。replace(0, np.nan)要谨慎如果数据来自单个用户真实 0 负荷是有可能的但关口计量数据极少为 0可以按业务规则判断。缺失比例超过 5% 时前后向填充就不够用了需要用线性插值或前一个同日的负荷值补齐。2.3 归一化必须在划分训练集之后做归一化的直觉是让数值落在 0 到 1 之间LSTM 的梯度更稳。但顺序至关重要必须先按时间顺序切出训练集和测试集再用训练集统计量做fit_transform最后用同一套统计量transform测试集。如果先对全量数据归一化再切分训练阶段就看到了测试区间的最小值和最大值测试指标的乐观程度会在部署时原形毕露。from sklearn.preprocessing import MinMaxScaler train_len int(len(df) * 0.8) train_df df.iloc[:train_len].copy() test_df df.iloc[train_len:].copy() scaler MinMaxScaler(feature_range(0, 1)) train_df[load_scaled] scaler.fit_transform(train_df[[load]]) test_df[load_scaled] scaler.transform(test_df[[load]])关键是.copy()否则对切片做赋值会触发 pandas 的 SettingWithCopyWarning更重要的是后续修改切片不会写回原数据。fit_transform只发生在train_df上transform使用同一组 min/max保证测试集数值范围与训练一致。如果将来新增别的特征每个特征都要单独指定列不要直接把二维数组塞给scaler.fit_transform否则会把所有列当成一组分布来压。2.4 滑动窗口生成 X 和 y定义create_sequences函数把一维序列转成监督学习样本。函数遍历从lookback到末尾的每个位置截取前lookback个点作为 X当前位置作为 y。训练集和测试集分开生成测试集的前lookback个点没有对应的完整窗口直接丢弃。def create_sequences(data, lookback96): X, y [], [] for i in range(lookback, len(data)): X.append(data[i - lookback:i]) y.append(data[i]) return np.array(X), np.array(y) lookback 96 # 用过去一天 96 个点预测下一点 X_train, y_train create_sequences(train_df[load_scaled].values, lookback) X_test, y_test create_sequences(test_df[load_scaled].values, lookback) print(X_train.shape, y_train.shape, X_test.shape, y_test.shape) # 期望输出类似 (n, 96), (n,), (m, 96), (m,)如果你希望预测未来一小时4 个点可以把y.append(data[i])改成y.append(data[i:i4])得到多步标签。但要注意直接多步输出和递归多步预测是两种完全不同的做法后面第 6 章会展开。这个函数在数据量到几百万条时会有点慢可以用 numpy 的 stride_tricks 替代不过几万条负荷数据用纯 Python 循环完全能接受我一般不会过早优化。2.5 为什么训练集不能随机打乱验证集却可以时间序列样本之间不是独立的前一小时的气温趋势会影响后一小时。如果随机打乱训练集LSTM 会在没有因果关系的乱序片段里找规律结果就是训练损失不降或模型记住了一个虚假的整体分布。model.fit默认shuffleTrue吗这里有一个容易混淆的点fit的shuffle参数默认是 True但对批量训练而言它只是在每个 epoch 内打乱样本顺序不会跨样本生成时间上错乱的序列。也就是说样本本身仍是按时间顺序生成的打乱只是改变喂入顺序不会破坏每个样本内部的时序。验证集怎么取Keras 的validation_split0.1是从训练集尾部顺序切出来的不会乱序也不会看到测试集。它相当于把训练序列最后 10% 的样本留出来观察模型是否过拟合。如果这批验证样本和测试样本时间上相邻验证 loss 往往能更真实地反映测试效果这也符合负荷预测的场景。3. 模型搭建与训练Keras LSTM 源码怎么配参数3.1 为什么用 Keras 而不是 PyTorchPython 生态里实现 LSTM 的主流选择有两个TensorFlow/Keras 和 PyTorch。对于电力负荷预测这种小数据量回归任务我倾向用 Keras。原因是 Sequential API 足够表达负荷预测最常见的“滑窗输入 全连接输出”结构训练日志、模型保存和部署都简单。PyTorch 更适合做需要动态图控制的研究原型但负荷预测场景里动态控制很少。如果团队里已经统一用 PyTorch那也无妨本文的源码换成 PyTorch 的nn.LSTM也是同样思路。核心不在框架而在数据切分和评估方式。3.2 一个可以跑通的两层 LSTM 模型结构模型设计从两层 LSTM 开始第一层return_sequencesTrue输出完整时间步序列第二层return_sequencesFalse只输出最后一步接一个全连接层输出预测值。Dropout 放在 LSTM 层后面目的是让隐层输出在训练时随机失活缓解过拟合。from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout model Sequential([ LSTM(units64, return_sequencesTrue, input_shape(lookback, 1)), Dropout(0.2), LSTM(units32, return_sequencesFalse), Dropout(0.2), Dense(units1) ]) model.compile(optimizeradam, lossmse) print(model.summary())注意input_shape(lookback, 1)。这里的第二维 1 是特征数也就是只用过去负荷一个变量。如果要多特征输入第二维改成特征数量样本形状要对应为(样本数, lookback, 特征数)。units 的初始值 64 和 32 是我在几万条 15 分钟负荷数据上的常用起点数据量翻几倍可以加到 128 和 64数据量小则降到 32 和 16。更小的模型训练更快也更不容易把噪声背下来。3.3 epochs、batch_size、learning_rate 怎么设训练参数不必一上来就追求最优先把一条能预测的基线跑通再看曲线调。epochs 30 起步batch_size 取 128。validation_split 取 0.1在训练尾部保留一段验证数据。history model.fit( X_train, y_train, epochs30, batch_size128, validation_split0.1, verbose1 )训练过程中观察val_loss是否在 15 个 epoch 后继续下降。如果下降很缓可以加到 60 个 epoch如果val_loss在某个点反弹说明过拟合开始应该提高 Dropout 或减小 units。学习率默认由 Adam 控制通常不用改但遇到 loss 不下降时可以显式设置learning_rate1e-4再试。注意学习率不能调得太低否则 30 个 epoch 内几乎不动我一般先用默认的 0.001 跑通再看曲线决定。3.4 多特征输入的扩展方式很多做负荷预测的项目不仅有负荷历史还有温度、湿度、星期几、是否节假日。把这些额外特征与负荷值一起作为输入通常能显著提升节假日的预测效果。实现上需要把特征列拼成二维矩阵归一化时对数值列各自单独处理。from sklearn.preprocessing import MinMaxScaler # 假设已有 train_df 包含 load_scaled, temp_scaled, is_holiday 三列 feature_names [load_scaled, temp_scaled, is_holiday] scaler MinMaxScaler() train_features scaler.fit_transform(train_df[feature_names]) test_features scaler.transform(test_df[feature_names]) def create_multivariate_sequences(features, lookback96): X, y [], [] for i in range(lookback, len(features)): X.append(features[i - lookback:i]) y.append(features[i, 0]) # 仍然只预测负荷 return np.array(X), np.array(y) X_train, y_train create_multivariate_sequences(train_features, lookback) X_test, y_test create_multivariate_sequences(test_features, lookback)这里y取的是第一列负荷而不是全部特征。X的第三维包含所有特征。如果is_holiday本来就是 0/1 二值不需要用 MinMaxScaler 压保留原值即可但为了统一放进同一个 scaler 也不会出错。特征越多数据需求量越大别一上来就堆 10 个特征先加一个温度和一个节假日标记看验证集是否有可感知的提升。3.5 训练曲线判断过拟合和欠拟合的两种模式训练完不要只看最终 loss画一条曲线更直观。history里保存了每个 epoch 的训练 loss 和验证 loss。import matplotlib.pyplot as plt plt.plot(history.history[loss], labeltrain loss) plt.plot(history.history[val_loss], labelval loss) plt.xlabel(epoch) plt.ylabel(loss) plt.legend() plt.show()如果训练 loss 持续下降而 val loss 在第若干轮反弹是典型的过拟合加 Dropout 或减少 units。如果两者都高且下降缓慢是欠拟合优先加 epochs、加大 LSTM units而不是继续加 Dropout。如果 val loss 一直高于 train loss 且稳定在一个平台可能只是验证集与训练集分布不同这时对比验证集和训练集的负荷均值是否明显偏差而不是急着调模型。4. 评估与可视化怎么证明模型真的变准了4.1 反归一化把预测值还原成可解释的 MW 数模型的输出是 0 到 1 之间的缩放值直接拿去做业务报告没有意义。预测完成后必须用训练时同一个 scaler 把结果还原为实际功率。pred_scaled model.predict(X_test, verbose0) pred scaler.inverse_transform(pred_scaled) y_actual scaler.inverse_transform(y_test.reshape(-1, 1))注意y_test的形状是(m,)需要先reshape(-1, 1)再inverse_transform否则 sklearn 会报维度错。这是很容易卡住的细节点。如果做多特征输入反归一化时要带负荷列对应的索引或者只把负荷列单独取出再还原。4.2 三个指标一起看MAE、RMSE、MAPE评估负荷预测只看单一是片面的。MAE 反映平均偏差RMSE 对突发大误差敏感MAPE 是百分比误差方便与业务阈值对比。我把三个都打印出来每次调参后对比。from sklearn.metrics import mean_absolute_error, mean_squared_error mae mean_absolute_error(y_actual, pred) rmse np.sqrt(mean_squared_error(y_actual, pred)) mape np.mean(np.abs((y_actual - pred) / y_actual)) * 100 print(fMAE: {mae:.2f} MW, RMSE: {rmse:.2f} MW, MAPE: {mape:.2f}%)这里有几个边界要提醒。第一如果测试集包含夜间低谷时段负荷值很小MAPE 会被几个低负荷点拉到虚高这时要按负荷区间分桶统计或者换 SMAPE。第二RMSE 对峰值敏感如果数据里有个别极端尖峰RMSE 可能比 MAE 看起来大得多这不一定代表模型差要看误差集中在哪里。第三这三个指标只能评估点预测如果想评估区间或概率预测需要换 Pinball Loss。4.3 可视化把负荷曲线和预测曲线贴在一起看指标只能给一个数字但预测到底滞后还是超前要从曲线里看。把预测值和真实值画在同一个坐标系里用测试集前面一段时间展示能直观看到是否有相位偏移或低谷偏移。import matplotlib.pyplot as plt plt.figure(figsize(12, 4)) plt.plot(y_actual[:500], labelactual, linewidth2) plt.plot(pred[:500], labelpredicted, linewidth1, alpha0.8) plt.xlabel(sample index (15 min per point)) plt.ylabel(load (MW)) plt.legend() plt.show()为什么取前 500 个点而不是全部全部测试集可能有几千个点曲线叠在一起看不出细节。前 500 个点大约覆盖 5 天能明显看到日周期和预测跟随情况。画图后如果发现预测比真实值滞后一步说明模型更像是在复制上一步的值这时要检查窗口是否太短或者数据里是否缺少当时可用的外部特征。滞后是负荷预测最常见的失败模式指标上看 MAE 不高但业务上往往不可接受。4.4 误差按小时聚合定位偏差发生的时间段曲线的偏差往往有固定时段比如晚间高峰或午间低谷。把误差按一天中的小时分组计算每个小时的平均误差和标准差能快速定位模型在哪些时段最容易失效。error np.array(y_actual).flatten() - np.array(pred).flatten() # 如果 y_actual 从测试集开始记录先构造对应的小时刻度 from datetime import timedelta start_time test_index[lookback] # 测试集第一个 y 对应的时间 hours [(start_time timedelta(minutes15 * i)).hour for i in range(len(error))] df_err pd.DataFrame({hour: hours, error: error}) hourly_mae df_err.groupby(hour)[error].apply(lambda x: np.abs(x).mean()) hourly_mae.plot(kindbar, figsize(10, 3))这段代码把 15 分钟样本按小时归组画出每个小时的绝对误差均值。如果误差集中在晚间 1822 点说明模型对晚高峰的社会用电行为刻画不足可以尝试加入当天是否为工作日的特征或者把晚间时段的样本在训练时单独加权。相比只看整体 MAPE这个聚合能告诉你下一步该改数据还是改模型。5. 避坑与常见问题排查我的 LSTM 源码跑通后依然翻车的五个场景先说明一个背景我在多个负荷预测项目里踩过的坑大部分不在模型代码本身而在数据管道和评估方式。下面五条是出现频率最高的每条按现象、原因、解决来写。5.1 归一化顺序写反导致测试集指标虚高现象训练集 MAE 和测试集 MAE 都异常低但模型部署到新数据后误差立刻反弹。原因初版代码里先对全量数据做了MinMaxScaler.fit_transform再切训练和测试集。测试集的 min/max 已经参与了训练时的缩放相当于模型在训练过程中知道了未来数据的分布边界测试指标自然乐观。解决调整顺序先按时间切分再只在训练集上fit_transform测试集只transform。这个坑不影响训练曲线但直接影响你对模型真实能力的判断。5.2 随机打乱样本模型学到了跨越时间的记忆现象用train_test_split(shuffleTrue)切分后训练 loss 下降很快测试 loss 也很低但预测曲线完全不像负荷曲线像是把训练集里某些片段原样搬过来。原因随机打乱行破坏了时间顺序样本里的 X 和 y 来自不同时间片段。LSTM 没有真正学习负荷的时间演进而是用输入片段去“检索”训练集里相似片段对应的标签这等于一种记忆。解决时间序列必须按顺序切分保证训练集时间完全早于测试集。Keras 的model.fit内部默认会对 mini-batch 做随机排布但这不会破坏每个样本内部的时序关系可以保留默认要禁的是从数据源头打乱行的逻辑。5.3 单步预测结果不错滚动预测却彻底发散现象用model.predict(X_test)单步评估时 MAPE 很低但把预测值喂回模型做多步递推第几步之后曲线变成水平线或剧烈振荡。原因单步评估每一步都用真实值作输入模型从未接触过“自己产生”的预测输入误差在递归过程中不断累积最终偏离真实分布。解决把滚动预测当作模型能否上线的验收标准。只做单步评估的报告不完整至少要补上第 6 章的rolling_predict结果。如果发散严重可以训练时用多步输出或加入 teacher forcing让模型在训练阶段就看到带噪声的预测输入。5.4 loss 卡住不动或者直接 NaN现象训练到几个 epoch 后 loss 不再下降或第一个 epoch 就出现 NaN。原因最常见的是数据没洗干净有 NaN 被create_sequences带进 X其次是学习率偏大梯度过冲再就是数值范围没归一化负荷值如果是几万 MW损失函数值本身巨大Adam 也可能压不住。解决训练前打印X_train和y_train的np.isnan().sum()与min/max。数据确认干净后看学习率是否默认 0.001必要时在 compile 里显式指定learning_rate1e-4。如果还 NaN把模型层数临时减成一层验证代码路径。5.5 验证 loss 一直低于训练 loss怎么看都不对现象验证 loss 比训练 loss 低一大截甚至出现验证 MAPE 只有训练 MAPE 一半的怪象。原因validation_split0.1取的尾部样本可能恰好处于负荷平稳期比如连续几天天气稳定验证集难度明显低于训练集平均难度。另一种原因是训练 loss 包含 Dropout 带来的随机噪声验证时 Dropout 被关闭模型表现自然更稳定。解决不要把验证 loss 和训练 loss 直接对比而是用第 4 章的测试集指标作为最终判定。如果测试集时间覆盖了多个典型天气过程这组指标才真正可信。反之如果测试期恰好撞上一段异常高温指标也会难看这不代表模型退步而是评估区间选择的问题。6. 一种更可靠的验收方法滚动预测与直接多步输出先看滚动预测代码。它的思路是从历史最后一个窗口出发每一步用模型生成一个新值再把它接在窗口尾部、丢弃最早的旧值循环steps次。这比单步评估更接近真实使用方式因为实际部署时未来值是不可见的你只能用自己预测出来的值继续推。def rolling_predict(model, init_seq, steps, scaler): preds [] seq init_seq.reshape(1, lookback, 1) for _ in range(steps): p model.predict(seq, verbose0).flatten()[0] preds.append(p) seq np.concatenate([seq[:, 1:, :], p.reshape(1, 1, 1)], axis1) return scaler.inverse_transform(np.array(preds).reshape(-1, 1))init_seq取测试集真实负荷的最后lookback个点steps是预测步数。以 15 分钟数据为例预测一天就是 96 步。滚动预测结果会随步数增加而误差扩大如果第 24 步以后误差尚可说明模型具备短期递推能力如果十几步就发散再决定是改模型结构还是接受更短的预测周期。比滚动预测更稳的另一条路是直接多步输出。把模型最后的Dense(1)改成Dense(k)训练标签从单步值换成未来 k 步的序列。例如预测下一天 96 点就设定output_dim96。直接多步避免了误差累积但模型要一次输出 96 个数输出空间变大训练数据需求量也更高。我一般的套路是两种都做直接多步用在小步长场景预测 424 点滚动预测用于验证极限步长。最后补充一个我养成的习惯每次提交源码前不只跑一遍model.predict还会把rolling_predict的结果画出来与真实曲线对比。只要滚动预测在目标步长内不发散我才会把结果交给业务方。这个习惯替我挡掉过不少上线后的返工也希望帮到你。本文还有配套的精品资源点击获取