简介：这份基于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 为什么不能把原始序列直接丢给 LSTM
LSTM 是监督学习模型，需要输入 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_col='timestamp') # 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, lookback=96): 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:i+4])，得到多步标签。但要注意，直接多步输出和递归多步预测是两种完全不同的做法，后面第 6 章会展开。这个函数在数据量到几百万条时会有点慢，可以用 numpy 的 stride_tricks 替代；不过几万条负荷数据用纯 Python 循环完全能接受，我一般不会过早优化。
2.5 为什么训练集不能随机打乱，验证集却可以
时间序列样本之间不是独立的，前一小时的气温趋势会影响后一小时。如果随机打乱训练集，LSTM 会在没有因果关系的乱序片段里找规律，结果就是训练损失不降或模型记住了一个虚假的整体分布。model.fit默认shuffle=True吗？这里有一个容易混淆的点：fit的shuffle参数默认是 True，但对批量训练而言，它只是在每个 epoch 内打乱样本顺序，不会跨样本生成时间上错乱的序列。也就是说，样本本身仍是按时间顺序生成的，打乱只是改变喂入顺序，不会破坏每个样本内部的时序。
验证集怎么取？Keras 的validation_split=0.1是从训练集尾部顺序切出来的，不会乱序，也不会看到测试集。它相当于把训练序列最后 10% 的样本留出来，观察模型是否过拟合。如果这批验证样本和测试样本时间上相邻，验证 loss 往往能更真实地反映测试效果，这也符合负荷预测的场景。
3. 模型搭建与训练：Keras LSTM 源码怎么配参数
3.1 为什么用 Keras 而不是 PyTorch
Python 生态里实现 LSTM 的主流选择有两个：TensorFlow/Keras 和 PyTorch。对于电力负荷预测这种小数据量回归任务，我倾向用 Keras。原因是 Sequential API 足够表达负荷预测最常见的“滑窗输入 + 全连接输出”结构，训练日志、模型保存和部署都简单。PyTorch 更适合做需要动态图控制的研究原型，但负荷预测场景里动态控制很少。如果团队里已经统一用 PyTorch，那也无妨，本文的源码换成 PyTorch 的nn.LSTM也是同样思路。核心不在框架，而在数据切分和评估方式。
3.2 一个可以跑通的两层 LSTM 模型结构
模型设计从两层 LSTM 开始：第一层return_sequences=True输出完整时间步序列，第二层return_sequences=False只输出最后一步，接一个全连接层输出预测值。Dropout 放在 LSTM 层后面，目的是让隐层输出在训练时随机失活，缓解过拟合。
from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout model = Sequential([ LSTM(units=64, return_sequences=True, input_shape=(lookback, 1)), Dropout(0.2), LSTM(units=32, return_sequences=False), Dropout(0.2), Dense(units=1) ]) model.compile(optimizer='adam', loss='mse') 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, epochs=30, batch_size=128, validation_split=0.1, verbose=1 )训练过程中，观察val_loss是否在 15 个 epoch 后继续下降。如果下降很缓，可以加到 60 个 epoch；如果val_loss在某个点反弹，说明过拟合开始，应该提高 Dropout 或减小 units。学习率默认由 Adam 控制，通常不用改，但遇到 loss 不下降时，可以显式设置learning_rate=1e-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, lookback=96): 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'], label='train loss') plt.plot(history.history['val_loss'], label='val 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, verbose=0) 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(f"MAE: {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], label='actual', linewidth=2) plt.plot(pred[:500], label='predicted', linewidth=1, alpha=0.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(minutes=15 * 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(kind='bar', figsize=(10, 3))这段代码把 15 分钟样本按小时归组，画出每个小时的绝对误差均值。如果误差集中在晚间 18～22 点，说明模型对晚高峰的社会用电行为刻画不足，可以尝试加入当天是否为工作日的特征，或者把晚间时段的样本在训练时单独加权。相比只看整体 MAPE，这个聚合能告诉你下一步该改数据还是改模型。
5. 避坑与常见问题排查：我的 LSTM 源码跑通后依然翻车的五个场景
先说明一个背景：我在多个负荷预测项目里踩过的坑，大部分不在模型代码本身，而在数据管道和评估方式。下面五条是出现频率最高的，每条按现象、原因、解决来写。
5.1 归一化顺序写反，导致测试集指标虚高
现象：训练集 MAE 和测试集 MAE 都异常低，但模型部署到新数据后误差立刻反弹。
原因：初版代码里先对全量数据做了MinMaxScaler.fit_transform，再切训练和测试集。测试集的 min/max 已经参与了训练时的缩放，相当于模型在训练过程中知道了未来数据的分布边界，测试指标自然乐观。
解决：调整顺序，先按时间切分，再只在训练集上fit_transform，测试集只transform。这个坑不影响训练曲线，但直接影响你对模型真实能力的判断。
5.2 随机打乱样本，模型学到了跨越时间的记忆
现象：用train_test_split(shuffle=True)切分后，训练 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_rate=1e-4。如果还 NaN，把模型层数临时减成一层，验证代码路径。
5.5 验证 loss 一直低于训练 loss，怎么看都不对
现象：验证 loss 比训练 loss 低一大截，甚至出现验证 MAPE 只有训练 MAPE 一半的怪象。
原因：validation_split=0.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, verbose=0).flatten()[0] preds.append(p) seq = np.concatenate([seq[:, 1:, :], p.reshape(1, 1, 1)], axis=1) return scaler.inverse_transform(np.array(preds).reshape(-1, 1))init_seq取测试集真实负荷的最后lookback个点，steps是预测步数。以 15 分钟数据为例，预测一天就是 96 步。滚动预测结果会随步数增加而误差扩大，如果第 24 步以后误差尚可，说明模型具备短期递推能力；如果十几步就发散，再决定是改模型结构还是接受更短的预测周期。
比滚动预测更稳的另一条路是直接多步输出。把模型最后的Dense(1)改成Dense(k)，训练标签从单步值换成未来 k 步的序列。例如预测下一天 96 点，就设定output_dim=96。直接多步避免了误差累积，但模型要一次输出 96 个数，输出空间变大，训练数据需求量也更高。我一般的套路是两种都做：直接多步用在小步长场景（预测 4～24 点），滚动预测用于验证极限步长。
最后补充一个我养成的习惯：每次提交源码前，不只跑一遍model.predict，还会把rolling_predict的结果画出来，与真实曲线对比。只要滚动预测在目标步长内不发散，我才会把结果交给业务方。这个习惯替我挡掉过不少上线后的返工，也希望帮到你。
本文还有配套的精品资源，点击获取
