Python实现PSO-KNN光伏功率预测：粒子群自动寻优K与P的完整工程实战 从数据清洗、时间特征、严格时序验证，到PSO参数搜索、KNN回归、误差诊断与可部署改进

Python实现PSO-KNN光伏功率预测：粒子群自动寻优K与P的完整工程实战

从数据清洗、时间特征、严格时序验证，到PSO参数搜索、KNN回归、误差诊断与可部署改进

Python 光伏功率预测 粒子群优化 PSO K近邻 KNN 机器学习 时间序列预测 新能源 智能电网

光伏功率受辐照度、组件温度、云层变化和设备状态共同影响，短时预测既要处理强非线性，也要避免时间序列数据泄漏。本文给出一套可复现的 PSO-KNN 光伏功率预测工程方案：先完成时间对齐、异常治理、周期编码与滞后特征构造，再按时间顺序划分训练/验证/测试集；随后以验证集 RMSE 为适应度，利用粒子群算法联合搜索 KNN 的近邻数 K 与闵可夫斯基参数 P，最终在独立测试集上用 MAE、RMSE、R² 和残差图完成评价。文章同时补充真实项目中最容易被忽略的泄漏风险、夜间样本、预测步长、未来气象特征和部署边界，并提供可直接改造成实际电站程序的完整代码框架。

先看结论：这个项目真正难的不是“把 PSO 和 KNN 拼起来”

如果只把 K 和 P 交给粒子群搜索，代码很快就能跑通；但一个能用于真实光伏场景的预测项目，核心在于三件事：第一，所有数据处理必须尊重时间因果关系；第二，特征必须对应预测时刻真实可获得的信息；第三，参数寻优、最终训练和测试评价必须相互隔离。任何一项做错，都可能得到“指标很好、上线失效”的结果。

图1 光伏功率与辐照度短时波动示意

晴天时功率曲线通常接近稳定单峰；多云条件下，云团移动会造成分钟级辐照突降与恢复。KNN 的优势正来自这种“局部相似性”：当前气象状态与历史某些时刻相似时，可直接借鉴那些样本的目标功率；PSO 则负责自动寻找更合适的邻域尺度与距离形状。

1. 项目目标与适用边界

目标是预测未来一个或多个采样间隔的交流侧有功功率。以 15 分钟采样为例，forecast_steps=1 对应未来 15 分钟；forecast_steps=2 对应未来 30 分钟。本文重点演示单步短时预测。对于更长预测时域，建议引入数值天气预报（NWP）或未来气象预报特征，而不是默认未来辐照度“已知”。

2. 为什么选择 PSO-KNN

KNN 回归没有传统意义上的参数训练过程，它把历史样本本身视为知识库。对待预测样本 x，先计算其与训练样本的距离，再选取最近的 K 个样本。采用距离加权时，可写成：ŷ = Σ(wᵢyᵢ) / Σwᵢ，其中 wᵢ 常取 1/(dᵢ+ε)。闵可夫斯基距离 d(x,z) = (Σ|xⱼ-zⱼ|ᵖ)^(1/p)，P=1 接近曼哈顿距离，P=2 为欧氏距离。

图2 KNN 在标准化特征空间中寻找局部相似样本

K 太小会对噪声和异常点敏感，K 太大会把局部天气突变平均掉；P 又直接改变“相似”的几何定义。因此本文把 K 与 P 作为二维粒子位置，由 PSO 以验证集 RMSE 为目标自动寻优。

3. 完整模型架构

图3 PSO-KNN 光伏功率预测完整工程链路

推荐把工程拆成八个可验证模块。每个模块都应留下中间结果：清洗后样本数、特征列表、时间切分边界、标准化器参数来源、每轮最优 RMSE、最终超参数、测试指标和预测明细。这样一旦结果异常，可以定位到具体环节，而不是只看到最终 R²。

4. 数据字段与质量治理

4.1 一个必须修正的常见问题：不要在全量数据上先 bfill

对时间序列而言，先在全量数据上执行 bfill（后向填充），再切训练集/测试集，可能把未来观测值传播到过去样本，形成隐蔽的数据泄漏。更稳妥的做法是：先按时间切分，再在每个阶段使用只依赖过去信息的填充策略；或者只对极短缺口做因果插值，并明确最大缺口长度。

data = data.sort_values("timestamp").drop_duplicates("timestamp")
# 推荐：优先使用只依赖过去的前向填充
data[numeric_columns] = data[numeric_columns].ffill(limit=4)
# 对仍缺失的记录，根据业务规则删除或单独建模
data = data.dropna(subset=numeric_columns)

5. 时间特征与滞后特征

图4 多源特征共同定义相似运行状态

小时、月份具有周期性，直接使用 23 与 0 会让模型误认为二者相距很远，因此采用 sin/cos 双通道编码。短时预测还应加入 power_lag_1、power_lag_2 等滞后功率；若数据量充足，可进一步加入 1 小时滚动均值、辐照变化率、晴空指数、太阳高度角等物理增强特征。

data["hour_float"] = data["timestamp"].dt.hour + data["timestamp"].dt.minute / 60
data["hour_sin"] = np.sin(2*np.pi*data["hour_float"]/24)
data["hour_cos"] = np.cos(2*np.pi*data["hour_float"]/24)
data["power_lag_1"] = data["power_kw"].shift(1)
data["target_power_kw"] = data["power_kw"].shift(-forecast_steps)

6. 严格时间顺序验证：比模型本身更重要

图5 训练集、验证集、测试集按时间顺序隔离

训练集负责建立邻域库，验证集只负责选择 K/P，测试集只在所有选择完成后使用一次。StandardScaler 必须先在训练集 fit，再 transform 验证集和测试集。超参数确定后，可以用训练+验证数据重新拟合最终 scaler 与 KNN，但测试集始终不能参与任何统计量估计。

7. PSO 粒子群优化机制

图6 PSO 在 K-P 二维空间中的搜索示意

每个粒子位置 xᵢ=[K,P]，速度更新采用经典形式：vᵢ(t+1)=ωvᵢ(t)+c₁r₁(pbestᵢ-xᵢ)+c₂r₂(gbest-xᵢ)，随后 xᵢ(t+1)=xᵢ(t)+vᵢ(t+1)。其中 ω 控制惯性，c₁ 控制个体学习，c₂ 控制群体学习。K 是离散变量，因此评价前四舍五入并裁剪到合法整数范围；P 保持连续。

图7 PSO 验证集 RMSE 收敛过程（合成数据演示）

工程上不建议盲目增加粒子数和迭代次数。KNN 的一次评估需要在验证集上执行近邻查询，样本量较大时计算成本会迅速增加。可先用 15~30 个粒子、20~40 轮迭代建立基线，再根据收敛曲线决定是否扩大搜索预算。

8. 可直接运行的核心代码

下面给出整理后的核心实现。代码重点修正了三个问题：避免全量双向填充造成未来信息泄漏；明确训练/验证/测试职责；超参数确定后重新在训练+验证集上拟合最终预处理器与模型。

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsRegressor
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

RANDOM_SEED = 42
RATED_POWER_KW = 1000.0
FORECAST_STEPS = 1
PARTICLE_COUNT = 24
ITERATION_COUNT = 30
K_LOWER, K_UPPER = 2, 40
P_LOWER, P_UPPER = 1.0, 3.0
W, C1, C2 = 0.72, 1.45, 1.45
rng = np.random.default_rng(RANDOM_SEED)

data = pd.read_csv("pv_power_data.csv")
data["timestamp"] = pd.to_datetime(data["timestamp"], errors="coerce")
data = data.dropna(subset=["timestamp"]).sort_values("timestamp")
data = data.drop_duplicates("timestamp").reset_index(drop=True)

numeric_columns = [
"irradiance", "ambient_temp", "module_temp",
"humidity", "wind_speed", "power_kw"
]
data[numeric_columns] = data[numeric_columns].apply(pd.to_numeric, errors="coerce")

# 因果填充：只使用过去值；limit 应按采样周期和业务容忍度设置
data[numeric_columns] = data[numeric_columns].ffill(limit=4)
data["power_kw"] = data["power_kw"].clip(0, RATED_POWER_KW)
data["irradiance"] = data["irradiance"].clip(lower=0)
data["humidity"] = data["humidity"].clip(0, 100)
data = data.dropna(subset=numeric_columns).reset_index(drop=True)

hour_float = data["timestamp"].dt.hour + data["timestamp"].dt.minute / 60
data["hour_sin"] = np.sin(2*np.pi*hour_float/24)
data["hour_cos"] = np.cos(2*np.pi*hour_float/24)
month = data["timestamp"].dt.month
data["month_sin"] = np.sin(2*np.pi*month/12)
data["month_cos"] = np.cos(2*np.pi*month/12)
data["weekday"] = data["timestamp"].dt.weekday
data["power_lag_1"] = data["power_kw"].shift(1)
data["target_power_kw"] = data["power_kw"].shift(-FORECAST_STEPS)
data = data.dropna().reset_index(drop=True)

feature_columns = [
"irradiance", "ambient_temp", "module_temp", "humidity", "wind_speed",
"hour_sin", "hour_cos", "month_sin", "month_cos",
"weekday", "power_lag_1"
]
X = data[feature_columns].to_numpy(float)
y = data["target_power_kw"].to_numpy(float)

n = len(X)
train_end = int(n * 0.70)
valid_end = int(n * 0.85)

X_train, y_train = X[:train_end], y[:train_end]
X_valid, y_valid = X[train_end:valid_end], y[train_end:valid_end]
X_test, y_test = X[valid_end:], y[valid_end:]

scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_valid_s = scaler.transform(X_valid)

def fitness(position):
k = int(np.clip(np.rint(position[0]), K_LOWER,
min(K_UPPER, len(X_train_s))))
p = float(np.clip(position[1], P_LOWER, P_UPPER))
model = KNeighborsRegressor(
n_neighbors=k, weights="distance",
metric="minkowski", p=p, n_jobs=-1
)
model.fit(X_train_s, y_train)
pred = model.predict(X_valid_s)
return float(np.sqrt(mean_squared_error(y_valid, pred)))

lower = np.array([K_LOWER, P_LOWER], dtype=float)
upper = np.array([K_UPPER, P_UPPER], dtype=float)
positions = rng.uniform(lower, upper, size=(PARTICLE_COUNT, 2))
velocities = rng.uniform(-0.5, 0.5, size=(PARTICLE_COUNT, 2))
pbest_pos = positions.copy()
pbest_score = np.full(PARTICLE_COUNT, np.inf)
gbest_pos = positions[0].copy()
gbest_score = np.inf

for i in range(PARTICLE_COUNT):
s = fitness(positions[i])
pbest_score[i] = s
if s < gbest_score:
gbest_score, gbest_pos = s, positions[i].copy()

for _ in range(ITERATION_COUNT):
r1 = rng.random((PARTICLE_COUNT, 2))
r2 = rng.random((PARTICLE_COUNT, 2))
velocities = (
W * velocities
+ C1 * r1 * (pbest_pos - positions)
+ C2 * r2 * (gbest_pos - positions)
)
positions = np.clip(positions + velocities, lower, upper)

for i in range(PARTICLE_COUNT):
s = fitness(positions[i])
if s < pbest_score[i]:
pbest_score[i] = s
pbest_pos[i] = positions[i].copy()
if s < gbest_score:
gbest_score = s
gbest_pos = positions[i].copy()

best_k = int(np.clip(np.rint(gbest_pos[0]), K_LOWER, K_UPPER))
best_p = float(np.clip(gbest_pos[1], P_LOWER, P_UPPER))

# 最终模型：只合并训练集+验证集，测试集仍保持完全独立
final_scaler = StandardScaler()
X_train_valid_s = final_scaler.fit_transform(X[:valid_end])
X_test_s = final_scaler.transform(X_test)

final_model = KNeighborsRegressor(
n_neighbors=best_k, weights="distance",
metric="minkowski", p=best_p, n_jobs=-1
)
final_model.fit(X_train_valid_s, y[:valid_end])

test_pred = np.clip(final_model.predict(X_test_s), 0, RATED_POWER_KW)
mae = mean_absolute_error(y_test, test_pred)
rmse = np.sqrt(mean_squared_error(y_test, test_pred))
r2 = r2_score(y_test, test_pred)

result = pd.DataFrame({
"timestamp": data["timestamp"].iloc[valid_end:].to_numpy(),
"actual_power_kw": y_test,
"predicted_power_kw": test_pred,
"error_kw": test_pred - y_test
})
result.to_csv("pso_knn_prediction_result.csv",
index=False, encoding="utf_8_sig")

print("best_k =", best_k)
print("best_p =", round(best_p, 4))
print("validation_rmse =", round(gbest_score, 4))
print("test_mae =", round(mae, 4))
print("test_rmse =", round(rmse, 4))
print("test_r2 =", round(r2, 4))

9. 可复现实验结果：先用合成数据验证流程，再替换真实电站数据

为了避免虚构真实电站指标，下面的数值来自本文脚本生成的可复现合成数据，仅用于验证程序链路是否正确，不代表任何实际电站性能。真实项目应使用自己的历史数据重新运行，并保留测试时段、额定容量、采样周期和天气分布说明。

图8 独立测试集真实功率与预测功率（合成数据演示）

图9 残差诊断（合成数据演示）

评价时不要只看 R²。MAE 更直观地表示平均偏差，RMSE 会放大少数严重误差，残差图则能帮助判断模型是否在高功率区、爬坡段或突发云遮阶段存在系统性偏差。若误差随功率增大而明显扩散，可考虑按额定功率归一化，或分天气类型建立子模型。

10. 真实项目中最容易踩的 8 个坑

未来气象泄漏

预测未来 1 小时却直接使用未来 1 小时的实测辐照度，会得到不可上线的“理想指标”。若未来气象未知，应使用当前观测、滞后量或 NWP 预报。

全量标准化

在切分前对全量 X 执行 fit_transform，会让测试期均值和方差进入训练流程。

双向填充

bfill 会用未来值补过去缺口，时间序列尤其危险。

随机切分

train_test_split(shuffle=True) 会把相邻时刻打散，使训练集“看见”测试期附近状态。

夜间样本占比过高

大量零功率夜间样本会让整体 RMSE 看起来更漂亮。建议同时报告白天/有效辐照时段指标。

只调 K 不管特征

距离模型对特征空间极其敏感。错误特征、冗余特征和量纲问题，往往比 K 的影响更大。

只报告单次指标

至少保留按月份、天气、功率区间的分层误差，必要时做滚动回测。

忽略计算复杂度

KNN 预测阶段需要查询历史样本。数据规模很大时，应考虑 KDTree/BallTree、样本压缩、近似近邻或分季节建库。

11. 从“能跑”升级到“可部署”的增强方案

12. 工程目录建议

pso_knn_pv/
├─ data/
│ └─ pv_power_data.csv
├─ outputs/
│ ├─ prediction_result.csv
│ └─ figures/
├─ src/
│ ├─ preprocess.py
│ ├─ features.py
│ ├─ pso_optimizer.py
│ ├─ train.py
│ └─ evaluate.py
├─ config.py
├─ requirements.txt
└─ main.py

将数据治理、特征工程、优化器和评价模块拆开后，后续替换算法（如 PSO-SVR、PSO-XGBoost、LSTM）时，不必重写整个项目。生产环境还应保存 scaler、feature_columns、best_k、best_p、训练数据时间范围和额定容量等元数据，确保离线训练与在线推理一致。

13. 结果解释：什么时候 PSO-KNN 值得用

PSO-KNN 适合中小规模数据、局部相似规律明显、需要快速建立强基线的光伏预测任务。它的优势是结构透明、训练成本低、容易解释“参考了哪些历史时刻”；局限是预测阶段计算量随样本数增长，对特征尺度和异常样本敏感，且面对从未出现过的极端天气时外推能力有限。因此，它更适合作为可靠基线、边缘侧轻量模型或集成学习中的一个成员，而不是在所有数据规模和预测时域上替代深度时序模型。

14. 一份可直接执行的实验检查清单

• 时间戳已排序、去重，采样间隔已核验；
• 缺失值处理不使用未来信息；
• 目标 shift 与预测步长完全一致；
• 所有输入特征在预测时刻真实可获得；
• 训练/验证/测试按时间隔离；
• StandardScaler 只在允许的历史数据上 fit；
• PSO 只看验证集，不看测试集；
• 最终指标同时报告 MAE、RMSE、R²，并查看残差；
• 夜间与白天指标至少做一次分层核验；
• 保存最优参数、随机种子、数据时间范围和结果文件。

15. 总结

PSO-KNN 的价值不在于把两个算法简单叠加，而在于建立一套“局部相似建模 + 自动超参数搜索 + 严格时序验证”的完整方法。KNN 负责从历史数据中寻找与当前状态最接近的运行片段，PSO 负责自动确定邻域规模和距离形状；数据治理、时间特征、滞后变量和防泄漏设计，则决定这套方法能否从实验代码走向真实电站。

当你把本文示例替换为真实电站数据时，建议先保持模型不变，优先检查数据质量和时间因果关系；随后再逐步加入天气分型、滚动回测、NWP、物理特征和不确定性区间。这样得到的每一次指标提升，都更容易解释、复现和验证。

附录：运行环境

建议环境：Python 3.10+；NumPy、pandas、scikit-learn、matplotlib。KNeighborsRegressor、StandardScaler 与误差指标均来自 scikit-learn。实际部署前请锁定依赖版本，并在新版本升级后重新执行回归测试。

说明：文中的实验曲线与指标为固定随机种子生成的合成数据演示，用于验证流程与代码结构；不冒充真实光伏电站测试结果。