简介：本资源面向具备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_ip+dst_ip+proto+dst_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类手工特征：

这些特征最终拼接成一个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.txt

requirements.txt只含6个核心包：pandas==1.5.3,numpy==1.23.5,scikit-learn==1.2.2,xgboost==1.7.5,matplotlib==3.7.1,seaborn==0.12.2。没有PyTorch/TensorFlow，纯传统ML栈，部署成本极低。

数据准备分两步：

1. 将提供的sample_data/目录（含scada_data.csv,firewall_flows.csv,ied_events.csv,true_labels.csv）复制到项目根目录
2. 检查时间戳格式：所有CSV的timestamp列必须是ISO格式（2023-05-12T14:23:05.123），若为Unix时间戳需在preprocess/config.py中修改TIMESTAMP_FORMAT = 'unix'

3.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.png：ROC曲线图（AUC=0.923，见下图）
• confusion_matrix.png：混淆矩阵（F1-score=0.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, fs=1, nperseg=64) # 替换为：用numpy直接算周期图（快10倍） def fast_periodogram(series): n = len(series) freqs = np.fft.rfftfreq(n, d=1.0) # fs=1Hz psd = np.abs(np.fft.rfft(series))**2 / n return freqs, psd

4.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, index=False)

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_period=2000, # 每2000样本更新一次树结构 split_confidence=1e-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，哪怕只是改了一个小数点。希望帮到你。

本文还有配套的精品资源，点击获取