电力多源异构数据融合检测框架：SCADA、防火墙与IED协同分析

简介本资源面向具备Python基础与机器学习知识的电力系统安全研究者、研究生及工控安全工程师聚焦关键基础设施防护场景提供一套针对电力系统网络入侵检测的多传感器异构数据融合完整方案。内容涵盖数据预处理、多源特征提取、融合算法设计如加权平均、D-S证据理论或简单级联策略及轻量级分类模型构建等核心环节显著提升异常行为识别精度与响应鲁棒性。资源为单文件docx文档77KB结构清晰含原理说明、算法流程图、关键代码片段注释及复现指导便于读者理解融合逻辑并迁移至实际SCADA/EMS仿真环境测试。目前已有34人学习下载适合用于科研复现、课程设计参考或工业控制系统安全防护体系的技术验证。1. 电力监控系统里“看不见的刀”当SCADA日志、防火墙流表、IED遥信信号同时报警却没人能说清攻击到底从哪扇门进来你有没有遇到过这种场景某天凌晨三点调度中心大屏上跳红——变电站A的RTU通信中断、防火墙告警显示异常SSH爆破、而同一时刻智能电表集群上报的电压谐波畸变率突增127%。三路数据都指向“被攻破”但没人敢拍板这是真实入侵还是设备老化导致的级联误报抑或是恶意攻击者精心设计的“数据迷雾”——这正是当前电力工控系统最真实的困境传感器越来越多告警越来越密可决策延迟反而拉长了。这个项目干了一件很实在的事它不造新算法而是用一套轻量、可解释、端到端可复现的Python流水线把SCADA时序数据、网络流量特征、继电保护动作信号这三类典型异构源拧成一股能直接喂给分类器的“融合向量”。它不是替代SIEM而是让SIEM的原始输入本身具备更强的攻击语义表达力。适合正在写电力系统网络安全方向论文的研究生、刚接手变电站安全加固项目的工程师或者想把多源日志分析落地到实际PLC/RTU环境的开发人员。它解决的不是“能不能检测”而是“为什么这次告警更可信”。2. 架构拆解三层数据管道 双通道融合决策拒绝黑匣子式端到端训练这个项目没走深度学习端到端那一套比如直接扔进Transformer堆叠而是采用分层解耦可插拔设计核心就一张图数据预处理层 → 特征工程层 → 融合决策层。每一层都留了明确的接口和配置点改一个参数就能切算法换一个CSV就能跑新场景。下面逐层拆解。2.1 数据预处理层三类传感器的“方言翻译”电力系统里没有标准数据格式。SCADA点表是毫秒级时间戳浮点值防火墙日志是文本行IP五元组IED事件报文是IEC61850 GOOSE帧里的布尔量时间戳。项目用pandas做了三套独立清洗器关键不是“标准化”而是保留各自的时间语义粒度SCADA数据scada_data.csv每行含timestamp, device_id, voltage_a, current_b, freq, status_code。清洗重点是滑动窗口重采样——不是简单降频而是用resample(5S).mean().bfill()保证5秒内所有有效采样都被聚合缺失值用后向填充避免引入未来信息。网络流数据firewall_flows.csv字段为src_ip, dst_ip, proto, src_port, dst_port, bytes_sent, duration, flags, timestamp。清洗逻辑是会话聚合按src_ipdst_ipprotodst_port分组统计每5分钟内连接数、平均包长、SYN/FIN比例等12维特征输出flow_features_5min.csv。IED事件数据ied_events.csv格式为timestamp, device_id, event_type, value, quality_flag。清洗关键是事件对齐将离散事件如“断路器分闸”转换为5秒窗口内的布尔序列1该窗口发生事件再做二阶差分提取“事件簇密度”。提示所有清洗脚本都放在preprocess/目录下main_preprocess.py是总入口。它不强制要求三类数据时间完全对齐——而是以SCADA时间轴为基准用pd.merge_asof()做最近邻左连接容忍±2秒偏差。这是工业现场的真实约束硬对齐反而失真。2.2 特征工程层从原始信号到攻击指纹的升维操作预处理后的数据仍是“扁平”的。这一层要提炼出攻击行为特有的时空耦合特征。项目没用PCA或AutoEncoder这类无监督降维而是基于电力系统知识设计了4类手工特征特征类型具体实现物理意义代码位置时序动力学特征对SCADA电压序列计算Hurst指数、Lempel-Ziv复杂度、近似熵ApEn正常电网负载变化有自相似性DDoS攻击导致的通信拥塞会破坏此特性features/timeseries_features.py网络行为异常度基于防火墙流特征用Isolation Forest打分再与SCADA负载率做皮尔逊相关性校正单独看网络流量可能正常但若高流量时段恰好伴随低负载率则极可能是扫描行为features/network_anomaly.py事件关联强度统计IED事件在SCADA异常窗口如电压越限前后的发生概率比Odds Ratio“保护装置误动”常是APT横向移动的副产品其与主异常的时间偏移量是关键线索features/event_correlation.py跨源一致性指标计算三源数据在相同时间窗内的方差比var(SCADA)/var(Flow)/var(IED)攻击者很难同步伪造三类传感器的噪声分布一致性骤降是强入侵信号features/cross_source_consistency.py这些特征最终拼接成一个37维向量SCADA 15维 Flow 12维 IED 6维 Cross-source 4维存为features/fused_features.csv。注意所有特征计算都封装成FeatureExtractor类支持fit_transform()和transform()分离方便线上部署时只用transform()。2.3 融合决策层双通道投票机制保障鲁棒性融合不是简单拼接。项目采用双通道并行决策动态加权投票通道A模型驱动用XGBoost对37维特征做二分类0正常1入侵输出概率p_model通道B规则驱动硬编码3条电力安全规则例如“若SCADA电压谐波畸变率8% AND 防火墙SYN包占比95% AND IED事件簇密度5次/分钟则置信度0.95”。输出确定性标签label_rule最终融合结果不是取平均而是final_score 0.7 * p_model 0.3 * (label_rule * 0.95)权重0.7/0.3可调且label_rule仅在满足全部条件时才激活避免误触发。这种设计让系统既保留机器学习的泛化能力又不失电力领域专家经验的兜底保障。注意fusion/目录下的voting_fusion.py实现了该逻辑。它接受features/fused_features.csv和labels/true_labels.csv人工标注的入侵时段输出results/fusion_prediction.csv含每5分钟窗口的final_score、model_prob、rule_trigger三列便于后续阈值调优。3. 运行指南5分钟跑通全流程从数据加载到ROC曲线生成别被“多源”吓住——整个流程就是三个命令的事。项目结构极度精简无多余依赖Python 3.8即可。3.1 环境准备与数据准备先创建干净虚拟环境强烈建议避免与现有项目冲突python -m venv power_fusion_env source power_fusion_env/bin/activate # Linux/Mac # power_fusion_env\Scripts\activate # Windows pip install -r requirements.txtrequirements.txt只含6个核心包pandas1.5.3,numpy1.23.5,scikit-learn1.2.2,xgboost1.7.5,matplotlib3.7.1,seaborn0.12.2。没有PyTorch/TensorFlow纯传统ML栈部署成本极低。数据准备分两步将提供的sample_data/目录含scada_data.csv,firewall_flows.csv,ied_events.csv,true_labels.csv复制到项目根目录检查时间戳格式所有CSV的timestamp列必须是ISO格式2023-05-12T14:23:05.123若为Unix时间戳需在preprocess/config.py中修改TIMESTAMP_FORMAT unix3.2 执行三步流水线第一步数据清洗与对齐运行主清洗脚本生成对齐后的中间数据python preprocess/main_preprocess.py --input_dir sample_data/ --output_dir processed_data/成功后processed_data/下会出现scada_aligned.csv,flow_aggregated.csv,ied_boolean.csv三个文件。检查行数是否一致应均为2880行对应24小时×5分钟窗口。第二步特征提取调用特征工厂生成37维融合特征python features/feature_extractor.py --input_dir processed_data/ --output_dir features/ --window_sec 300关键参数--window_sec 300即5分钟窗口可按需调整如对实时性要求高可设为60。输出features/fused_features.csv共2880行×37列。第三步融合预测与评估运行融合主程序自动训练XGBoost、应用规则引擎、生成评估报告python fusion/main_fusion.py --features_path features/fused_features.csv \ --labels_path sample_data/true_labels.csv \ --output_dir results/ \ --model_weight 0.7输出results/目录包含fusion_prediction.csv每行含时间戳、模型概率、规则触发标志、融合分数roc_curve.pngROC曲线图AUC0.923见下图confusion_matrix.png混淆矩阵F1-score0.87补充说明main_fusion.py内部已实现5折交叉验证--model_weight参数会同时影响训练和测试阶段。若只想用预训练模型可加--load_model_path models/xgb_best.pkl跳过训练。3.3 快速验证用单条命令查看核心逻辑想立刻确认融合是否生效直接运行诊断脚本python utils/diagnose_fusion.py --feature_csv features/fused_features.csv \ --label_csv sample_data/true_labels.csv \ --window_idx 1440 # 查看中午12点那条记录输出类似[DIAGNOSTIC] Window 2023-05-12T12:00:00: - SCADA_Hurst: 0.28 (normal 0.5) → anomaly flag ON - Flow_SYN_ratio: 0.97 (attack threshold 0.95) → anomaly flag ON - IED_event_odds: 3.2 (baseline 1.0) → anomaly flag ON - Cross-variance_ratio: 0.41 (normal ~1.0) → anomaly flag ON → Rule engine TRIGGERED (all 4 flags active) → XGBoost prob: 0.89 → FINAL SCORE: 0.7*0.89 0.3*0.95 0.908 → CLASSIFIED AS INTRUSION这就是“多源协同”的具象化——单看任一源都可能犹豫但四路证据链闭合结论就变得不容置疑。4. 避坑指南我在现场调试时踩过的5个深坑现在全给你垫好这个项目看似简单但在真实变电站数据上跑通我花了整整两周。以下是血泪总结的5个高频翻车点每个都附带定位方法和修复命令4.1 现象preprocess/main_preprocess.py报错KeyError: timestamp原因原始CSV的列名含不可见空格或大小写不一致如Timestamp或 TIME_STAMP pandas.read_csv()默认不strip列名。解决在preprocess/utils.py的load_csv()函数里加清洗逻辑def load_csv(filepath): df pd.read_csv(filepath) # 新增清理列名空格并转小写 df.columns [col.strip().lower() for col in df.columns] return df然后确保所有CSV首行是纯英文列名无中文、空格、特殊符号。4.2 现象features/feature_extractor.py运行缓慢CPU占用100%卡死原因scipy.signal的welch功率谱计算在短序列上开销巨大而项目默认对每个5分钟窗口300点都做FFT。解决在timeseries_features.py中将welch替换为快速近似# 原始慢代码注释掉 # f, Pxx welch(series, fs1, nperseg64) # 替换为用numpy直接算周期图快10倍 def fast_periodogram(series): n len(series) freqs np.fft.rfftfreq(n, d1.0) # fs1Hz psd np.abs(np.fft.rfft(series))**2 / n return freqs, psd4.3 现象fusion/main_fusion.py训练XGBoost时ValueError: Input contains NaN原因某些SCADA设备在通信中断期间上报-999.0作为无效值而IsolationForest对NaN敏感导致特征向量出现NaN。解决在特征提取最后一步统一处理# 在 feature_extractor.py 的 save_features() 函数末尾添加 features_df features_df.fillna(features_df.median()) # 用中位数填充比均值更鲁棒 features_df.to_csv(output_path, indexFalse)4.4 现象ROC曲线AUC只有0.55模型像随机猜测原因true_labels.csv的标签格式错误。项目要求标签列名为label值为0或1但常见错误是用True/False字符串需转为1/0用normal/attack需映射时间戳未对齐标签时间比特征时间晚5分钟解决运行校验脚本python utils/validate_labels.py --features features/fused_features.csv \ --labels sample_data/true_labels.csv它会输出精确的对齐误差如“标签时间偏移283秒”并生成修正后的labels_fixed.csv。4.5 现象规则引擎永远不触发rule_trigger列全0原因规则阈值过于严苛。原代码中SYN_ratio 0.95在真实防火墙日志里极少出现正常扫描也仅0.8左右。解决打开fusion/rule_engine.py调整阈值# 原始太严 if syn_ratio 0.95 and harmonic_distortion 0.08 and ied_density 5: # 修改为更贴近实战 if syn_ratio 0.82 and harmonic_distortion 0.05 and ied_density 3:玄学经验在sample_data/里找3个已知攻击时段手动计算这三项值取最小值的0.9倍作为新阈值。5. 二次开发实战如何把这套融合框架迁移到你的变电站3个必改点1个后悔药拿到代码不是终点而是起点。我在某110kV变电站落地时发现原项目有3个必须修改的“水土不服”点以及1个保命用的“后悔药”机制。下面直接给可抄作业的方案。5.1 必改点1适配你的SCADA点表结构2分钟搞定原项目假设SCADA数据有固定字段voltage_a,current_b等但不同厂家点表千差万别。不要改清洗逻辑改映射配置在preprocess/config.py中新增字典# 根据你的RTU型号选择映射 SCADA_MAPPING { 南瑞NS3000: { voltage: Ua, # 原项目用voltage_a → 映射到Ua current: Ib, # 原项目用current_b → 映射到Ib freq: Freq, status: DevSta }, 许继XJ300: { voltage: Vol_A_Phase, current: Cur_B_Phase, freq: Sys_Freq, status: Run_Status } }然后在preprocess/scada_cleaner.py的clean_scada()函数开头加# 自动读取配置 vendor 南瑞NS3000 # 从命令行参数或config.ini读取 mapping SCADA_MAPPING[vendor] df df.rename(columns{ mapping[voltage]: voltage_a, mapping[current]: current_b, mapping[freq]: freq, mapping[status]: status_code })这样换一家RTU只需改两行配置不用碰核心代码。5.2 必改点2接入你的防火墙日志格式支持Suricata/Zeek/Snort原项目只支持CSV格式的简化流日志。要接真实防火墙需扩展解析器。以Suricata EVE JSON为例在preprocess/firewall_parser.py中新增import json def parse_suricata_eve(eve_file): 解析Suricata EVE日志提取关键网络特征 flows [] with open(eve_file) as f: for line in f: try: record json.loads(line.strip()) if record.get(event_type) flow: flows.append({ src_ip: record[src_ip], dst_ip: record[dest_ip], proto: record[proto], src_port: record[src_port], dst_port: record[dest_port], bytes_sent: record[flow][bytes_toclient] record[flow][bytes_toserver], duration: record[flow][end_time] - record[flow][start_time], timestamp: record[timestamp] }) except json.JSONDecodeError: continue # 跳过非JSON行 return pd.DataFrame(flows)再在main_preprocess.py里加判断if firewall_format suricata_eve: df_flow parse_suricata_eve(input_path) else: df_flow pd.read_csv(input_path)5.3 必改点3替换XGBoost为你的在线学习模型如River库原项目用XGBoost是为离线训练方便但变电站需要持续学习新攻击模式。换成river库的HoeffdingTreeClassifier支持单样本增量更新pip install river修改fusion/main_fusion.py# 原XGBoost训练注释掉 # model xgb.XGBClassifier() # model.fit(X_train, y_train) # 替换为River在线学习 from river import tree model tree.HoeffdingTreeClassifier( grace_period2000, # 每2000样本更新一次树结构 split_confidence1e-7 ) # 在预测循环中增量学习 for i, (x, y_true) in enumerate(zip(X_test, y_test)): y_pred model.predict_one(x) # 单样本预测 model.learn_one(x, y_true) # 单样本学习 # ... 记录结果这样模型部署后每收到一条新特征向量就自动进化无需停机重训。5.4 后悔药一键回滚到任意历史版本的融合策略现场调试时经常改着改着就把效果搞差了。项目内置了strategy_versioning.py每次运行main_fusion.py都会自动生成版本快照# 运行时自动保存 python fusion/main_fusion.py --save_version v2.1_attack_pattern_fix # 查看所有版本 python utils/list_versions.py # 回滚到指定版本复制配置文件模型 python utils/rollback_version.py --version v1.8_baseline快照内容包括config.yaml所有参数、models/xgb_best.pkl模型、results/roc_curve_v1.8.png评估图。从那以后我每次上线新策略都强制走一遍--save_version哪怕只是改了一个小数点。希望帮到你。本文还有配套的精品资源点击获取