多源数据融合驱动的电力系统网络入侵检测Python实现

简介面向电力系统网络安全研究与工业控制系统安全方向的技术资料聚焦网络入侵场景下的多源数据融合检测方法适合具备Python基础、熟悉机器学习与数据处理的研究生及科研人员参考。内容围绕异构传感器数据的预处理、特征提取、融合算法设计与入侵检测模型构建展开可用于早期预警、异常行为识别与融合策略复现优化。压缩包共1个docx文件约77KB便于集中阅读方法说明与代码实现思路也适合在仿真数据或实际数据集上对照测试、调优。已有35人学习可作为关键基础设施安全监控中多传感器协同检测的实践参考帮助读者理解多源数据协同检测机制并进一步迁移到电力系统安全防护研究。1. 一条没人敢拍板的告警电力系统网络入侵检测为什么要做多源融合凌晨两点监控后台弹出一条告警某台交换机的端口出现高频连接请求。值班的人看一眼物理侧的量测电压电流纹丝不动SCADA 上没有任何异常操作记录。这到底是扫描器在试探还是某个业务系统在做批量同步只看一路数据谁也不敢定性。这就是多传感器融合在电力系统网络入侵检测里最实际的价值网络流量探针看到的是连接行为异常主机日志看到的是进程和账号在干什么SCADA/量测通道看到的是物理侧有没有跟着动。三路数据单独看都只够给出一半结论按时间轴对齐后交给一个融合模型才可能把误报压下去、把漏报捞回来。这份 Python 代码实现的正是这条链路——多源数据融合、特征提取、融合判别一份能在普通工作站上跑通的工程骨架适合做工业控制安全、电力监控安全方向的研究生和工程侧开发拿去做二次开发。2. 项目骨架拆解从采集端到告警落库的六层链路与文件分工拿到一份检测类代码我第一件事不是看模型而是看数据从哪儿进、从哪儿出。模型可以换数据流一旦定歪了后面全是白工。这份代码的结构比较克制没有堆花活目录划分基本按数据流位置来切而不是按算法名来切这一点是对的。2.1 目录结构与各文件职责展开压缩包后主干目录大致是这样的形态不同人整理时文件名会有出入但职责分层是一致的power_ids/ ├── config/ │ └── settings.yaml # 路径、时间容忍度、窗口长度、阈值等全部外置 ├── data/ │ ├── raw/ # 原样落盘的网络流量、主机日志、量测数据 │ └── processed/ # 对齐后的宽表、特征矩阵可复现的中间产物 ├── src/ │ ├── io_utils.py # 读文件、字段归一、时间戳统一成 datetime │ ├── align.py # 多源时间对齐核心是 merge_asof │ ├── features.py # 滑动窗口特征、频域特征、行为特征 │ ├── fuse.py # 特征级融合与决策级融合两套实现 │ ├── model.py # 训练、交叉验证、模型持久化 │ ├── evaluate.py # 混淆矩阵、误报率、告警时延 │ └── run_pipeline.py # 唯一入口按六层顺序串起来 ├── notebooks/ │ └── eda.ipynb # 探索性分析不参与生产链路 ├── requirements.txt └── README.mdio_utils.py干的是最枯燥但最容易翻车的活把不同来源的时间戳统一成同一时区、同一精度的datetime64把字段名里的空格、大小写、单位差异抹平。很多人图省事直接在align.py里顺手做等到要接入第四路数据时就会发现改一处崩三处。我一般会把所有字段名映射写成一张显式的字典放在配置里宁可多写二十行也不要在业务代码里做隐式猜测。run_pipeline.py只负责串联不写任何算法逻辑。这一点值得学它的存在让你能一条命令重跑整条链路也让你在复现别人结果时有个确定的起点。做法上通常是把每一步的中间结果落成 parquet 或 pickle加一个--stage参数可以只重跑从特征提取开始这一段省掉每次几分钟的数据读取。2.2 技术栈选择和依赖清单选型上这份代码走的是轻量可解释路线pandas numpy 做对齐和特征scipy 做频域变换scikit-learn 做分类与评估imbalanced-learn 处理正负样本极不平衡matplotlib/seaborn 出图PyYAML 读配置。没有引入深度学习和分布式框架这不是能力问题是场景问题。电力监控网络里的攻击样本本来就少一个变电站一年可能攒不出上千条真实攻击记录深度学习在这种样本量下基本是过拟合训练器。而且检测系统上线后要能解释为什么这条被判成攻击运维才敢处置一个树模型能吐出特征重要度一个五层网络吐不出来。所以在这类项目里GBDT、随机森林这类模型反而是更合适的选择。依赖清单不用写死版本号锁定大版本区间即可避免在别人的机器上因为小版本差异装不上numpy pandas scipy scikit-learn imbalanced-learn matplotlib seaborn pyyaml pyarrowpyarrow是我额外建议装的中间产物用 parquet 存比 CSV 快一个数量级体积也小得多。整条链路在 8G 内存的机器上跑几十万条记录没有压力真要上到千万级再考虑把对齐那层换成按天分片处理而不是一上来就上集群。2.3 六层链路每一层的输入、输出和失败点整条链路可以拆成下面六层每层都有明确的输入输出出问题时能快速定位是哪一层断的层输入输出典型失败点采集接入原始 CSV / 日志 / 量测点表统一字段名的 DataFrame时间戳格式不统一时间对齐多路 DataFrame一张以基准时间为索引的宽表时区、精度不一致导致全空清洗宽表去重、补缺、异常值截断后缺失值比例过高被静默丢弃特征提取宽表特征矩阵 标签窗口跨过标签边界造成泄漏融合多源特征/多模型输出融合特征或融合概率权重拍脑袋定判别告警融合分数0/1 判定 告警事件阈值用默认 0.5采集接入层的核心是不改原始文件所有转换都在内存里做data/raw永远保持只读。这听起来是老生常谈但真出过事故有人在原始目录上做了一次就地清洗后来想回滚发现原始数据已经没了后悔药没地方买。时间对齐层是整个项目的技术难点也是最有价值的部分第 3 章会展开。这里先记一句对齐的基准一定要选采样最密的那一路用稀疏的那一路当基准等于把高频信息全部扔掉。清洗层要注意的是不要静默丢数据。缺失值超过一定比例的记录应该被标记出来单独看而不是dropna()一删了事——被删掉的那批记录里很可能恰好就是攻击导致的采集中断。特征提取层要防的是标签泄漏。滑动窗口如果跨过了攻击结束的时间点窗口里的统计量就会混入攻击后的信息模型在训练集上表现极好上线后一塌糊涂。常见做法是在标签变化点前后各留一段隔离带隔离带内的窗口直接丢弃。融合层和判别层是策略问题不是代码问题第 4 章细讲。这里只强调一点融合权重和判定阈值必须是配置项不能硬编码在函数里否则每换一个场景就要改代码重新打包。3. 预处理与特征工程把三种采样率的数据对齐到同一张宽表多源融合项目里八成的功夫花在这一章。模型再花哨喂进去的是错位的数据出来的就是玄学结果。3.1 三类数据源的字段画像与时间对齐电力监控场景下能拿到的三类数据采样节奏差别很大数据源典型采样率关键字段异常时的表征网络流量探针秒级或按流汇总五元组、包数、字节数、连接时长、失败连接数端口扫描、突发连接、长连接心跳缺失主机与设备日志事件触发不规则账号、登录结果、进程名、配置变更异常登录、提权、配置被改量测/SCADA 通道毫秒级到秒级电压、电流、开关状态、遥控指令量测跳变、遥测遥信不一致三类数据的共同点是都有时间戳难点是精度和时区不一致。对齐的常见做法是以采样最密的那一路为基准用 pandas 的merge_asof做就近匹配并且用一个显式的容忍度参数卡死偏差范围。容忍度一般取最慢那一路采样周期的二分之一超过就留空而不是硬凑一个值过去。import pandas as pd def align_sources(frames, ts_colts, tolerance2s): frames: dict键为数据源名值为带时间戳列的 DataFrame ts_col: 统一后的时间戳列名已转成 datetime64 tolerance: 允许的最大时间偏差超过则匹配为空 返回: 以采样最密的一路为基准的宽表 # 基准选行数最多的那一路避免用稀疏源上采样造出假数据 base_name max(frames, keylambda k: len(frames[k])) base frames[base_name].sort_values(ts_col).reset_index(dropTrue) base[_base_src] base_name for name, df in frames.items(): if name base_name: continue right df.sort_values(ts_col).copy() # direction 用 nearest日志是事件触发可能略早于现象出现 base pd.merge_asof( base, right, onts_col, directionnearest, tolerancepd.Timedelta(tolerance), suffixes(, f_{name}), ) return base这段逻辑有三个参数值得说清。tolerance决定了多宽松地认为这两条记录是同一时刻发生的设太大会把几分钟前的日志贴到当前流量上造成因果倒置设太小宽表里会大面积 NaN模型学不到跨源关系。directionnearest相比默认的backward更适合事件型日志因为日志写入往往比实际行为晚几十毫秒只往前找会系统性地错配。suffixes必须显式给否则两路数据里同名字段比如都有status会在无提示的情况下被覆盖这类问题排查起来极其难受。对齐完还要检查一件事每路数据的匹配率。匹配率低于 60% 的数据源要么是时间戳单位错了秒当成毫秒要么是时区没统一。这两个原因占了不对齐问题的绝大多数。3.2 特征提取时域统计、频域能量、行为窗口对齐后的宽表是逐条的但入侵检测真正关心的是一段时间里发生了什么所以要切滑动窗口。窗口长度按业务定网络侧的扫描行为通常持续十几秒物理侧的异常量测可能只持续几百毫秒一个窗口装不下两种节奏通常要同时保留长短两套窗口再在对齐层合并。下面的函数对每个数值列产出时域统计和频域能量占比import numpy as np import pandas as pd def window_features(df, value_cols, window10s, step5s): df: 对齐后的宽表含 ts 列 value_cols: 需要抽特征的数值列 window: 窗口长度应覆盖一次完整攻击行为 step: 滑动步长一般取 window 的一半保证重叠但不冗余 out [] for start in pd.date_range(df[ts].min(), df[ts].max(), freqstep): seg df[(df[ts] start) (df[ts] start pd.Timedelta(window))] if len(seg) 3: # 样本太少的窗口直接跳过避免统计量失真 continue row {ts: start, win_n: len(seg)} for c in value_cols: x seg[c].astype(float).to_numpy() row[f{c}_mean] x.mean() row[f{c}_std] x.std(ddof0) row[f{c}_p95] np.percentile(x, 95) # 高频能量占比突发扫描、短时脉冲类异常在这里最敏感 spec np.abs(np.fft.rfft(x - x.mean())) half len(spec) // 2 row[f{c}_hf_ratio] spec[half:].sum() / (spec.sum() 1e-9) out.append(row) return pd.DataFrame(out).sort_values(ts).reset_index(dropTrue)std(ddof0)用总体标准差是为了避免小窗口下样本标准差被放大导致特征分布在不同窗口长度间漂移。p95比max更抗单点毛刺单条采集异常不会让整个窗口的特征爆炸。hf_ratio的计算里除以spec.sum() 1e-9是防止全零窗口除零这个1e-9看着不起眼却是我见过最多人漏掉的一行。窗口标签的取法也要小心。一个窗口只要和攻击区间有交集就标 1是最简单的做法但会让攻击区间前后各扩散出一个窗口宽度的假阳性模型学到的是攻击附近就算攻击。我一般会要求窗口内攻击时长占比超过 30% 才标正样本占比在 0 到 30% 之间的窗口直接丢弃。这样正样本会少一些但边界干净得多。3.3 缺失值、标准化和不平衡采样的正确顺序这三步的顺序错了评估结果会好看得离谱。正确顺序是先填补再标准化最后在训练折内部做重采样。用 imbalanced-learn 的 Pipeline 可以把这个顺序固化成代码不给人犯错的余地from sklearn.impute import SimpleImputer from sklearn.preprocessing import StandardScaler from imblearn.over_sampling import SMOTE from imblearn.pipeline import Pipeline as ImbPipeline def build_pipeline(clf, k_neighbors5): clf: 任意实现了 fit/predict_proba 的分类器 k_neighbors: SMOTE 的近邻数样本极少时要调小否则报错 注意必须用 imblearn 的 Pipelinesklearn 的 Pipeline 不会在 fit 时重采样 return ImbPipeline([ (imputer, SimpleImputer(strategymedian)), # 中位数比均值抗异常值 (scaler, StandardScaler()), # 树模型可省线性/NN 必留 (smote, SMOTE(random_state42, k_neighborsk_neighbors)), (clf, clf), ])这里的关键细节是必须使用imblearn.pipeline.Pipeline。sklearn 自带的 Pipeline 只在fit时调用各步的fit_transformpredict时会跳过重采样步骤但它的交叉验证不会自动把重采样限制在训练折内——如果你手动在交叉验证前对整个数据集做 SMOTE验证集里就会混入由训练样本合成出来的假样本评估结果虚高十几个点是常事这就是典型的黑匣子式自欺欺人。k_neighbors在少数类样本少于 6 条时会直接报错这时要么改成RandomOverSampler要么换成基于代价的class_weightbalanced别硬撑。4. 融合层与检测模型特征级拼接和决策级加权两条路线怎么选融合发生在哪一层决定了整个系统的可解释性和可维护性。这份代码两条路线都留了接口实际用哪条要看你手上的数据长什么样。4.1 特征级融合横向拼接后交给一个强分类器特征级融合也叫早期融合把三路数据抽出的特征横向拼成一个大矩阵直接喂给一个分类器。优点是一次训练得到一个模型特征之间的交互关系能被模型自动学到缺点是任何一路数据缺失整行特征就退化了而且特征维度上去以后重要度解释会变模糊。from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.ensemble import HistGradientBoostingClassifier from sklearn.impute import SimpleImputer def build_feature_fusion(net_cols, phy_cols, log_cols): 按数据源分组做预处理避免不同量纲的列被同一套缩放参数污染 net_cols / phy_cols / log_cols: 各路源数据抽出的特征列名 def one_group(cols): return Pipeline([ (imp, SimpleImputer(strategymedian)), (sc, StandardScaler()), ]) pre ColumnTransformer([ (net, one_group(net_cols), net_cols), (phy, one_group(phy_cols), phy_cols), (log, one_group(log_cols), log_cols), ], remainderdrop) return Pipeline([ (pre, pre), (clf, HistGradientBoostingClassifier( max_depth6, learning_rate0.05, max_iter300, class_weightbalanced, )), ])分组做ColumnTransformer而不是一股脑StandardScaler是因为网络侧特征包数、字节数和物理侧特征电压、电流量纲差好几个数量级混在一起缩放会把小量纲特征压成噪声。max_depth6和learning_rate0.05是这类中小样本表格数据的稳妥起点深度再往上加训练集 AUC 会继续涨验证集开始掉就是过拟合的信号。class_weightbalanced和外面的 SMOTE 二者留一个就行同时用会让少数类被重复加权两次。4.2 决策级融合三路模型各自打分再按代价加权决策级融合晚期融合让每一路数据训练一个独立的子模型各自输出是攻击的概率再按权重合成一个最终分数。它对单路数据缺失更宽容——某一路没有数据就把它的权重临时降到 0其他两路照常工作。代价是要训练和维护多个模型。import numpy as np def weighted_fusion(prob_list, weights, eps1e-6): prob_list: 各子模型输出的攻击概率长度一致 weights: 与 prob_list 同长度的权重按各路在验证集上的可靠性定 返回: 融合后的攻击概率 w np.asarray(weights, dtypefloat) w w / (w.sum() eps) # 归一化避免权重和不为 1 stacked np.vstack(prob_list) # shape: (n_src, n_samples) return stacked.T w def weight_by_cost(models, X_val_list, y_val, cost_fp1.0, cost_fn10.0): 按误报/漏报代价反推权重代价高的模型权重低 scores [] for m, X in zip(models, X_val_list): p m.predict_proba(X)[:, 1] pred (p 0.5).astype(int) cost cost_fp * ((pred 1) (y_val 0)).sum() \ cost_fn * ((pred 0) (y_val 1)).sum() scores.append(1.0 / (cost 1.0)) # 代价越低权重越大 return scores权重不能拍脑袋按 0.5/0.3/0.2 分。电力场景里漏报的代价远高于误报一次漏掉的入侵可能意味着一条馈线被远程操作所以cost_fn一般设成cost_fp的 5 到 20 倍用验证集上的实际代价反推权重更站得住脚。1.0 / (cost 1.0)是个简单的反比映射加了 1 是防止某一路代价为 0 时权重无限大。这套做法不优雅但足够稳而且改代价系数就能调整系统性格不用动模型代码。两条路线的选择建议数据完整率高、样本量够每一类都有几百条以上、追求单模型解释性的走特征级数据源经常缺一路、或者各路数据到达时间差异大需要异步判定的走决策级。实际工程里我见过不少项目两条都用线上跑决策级保证可用性离线用特征级做归因分析。4.3 评估口径误报率、漏报率、告警时延怎么算入侵检测的评估不能只看准确率。正负样本 1:500 的数据集上模型全判负样本也能拿到 99.8% 的准确率这个数字毫无意义。真正要盯的是三个数指标含义电力场景下的期望召回率Recall真实攻击里被抓到的比例越高越好低于 0.85 基本不能用误报率FPR正常流量被判成攻击的比例单站点每天误报控制在个位数告警时延攻击开始到产出告警的时间差小于一个窗口长度加推理耗时from sklearn.metrics import confusion_matrix def report(y_true, y_score, threshold): 按给定阈值算三个核心指标threshold 由业务代价决定不是默认 0.5 y_pred (y_score threshold).astype(int) tn, fp, fn, tp confusion_matrix(y_true, y_pred, labels[0, 1]).ravel() return { recall: tp / (tp fn 1e-9), fpr: fp / (fp tn 1e-9), precision: tp / (tp fp 1e-9), alerts_per_day: fp, # 具体到每天弹多少条运维才看得懂 } def sweep_threshold(y_true, y_score, thresholds): 扫一遍阈值挑出满足误报上限的最高召回点 rows [] for t in thresholds: r report(y_true, y_score, t) rows.append({threshold: t, **r}) return rows把误报换算成每天多少条告警比报一个 0.003 的 FPR 有用得多因为运维关心的就是这个数。阈值扫描的做法是先定死每天误报不超过 N 条这条红线再在满足红线的阈值里选召回最高的那个。这个顺序不能反先追召回再压误报最后往往压不下来——告警炸屏之后运维会直接把这一路告警静音检测系统等于没上线。5. 排查手记跑不起来、指标虚高、告警炸屏的六个真实原因5.1 环境与数据接入类问题现象一merge_asof执行成功但整张宽表全是 NaN。原因基本是两个左右两张表的时间戳一个是datetime64[ns]一个是字符串或者时区一个带08:00一个不带。merge_asof不会报错只会静默匹配失败。解决方法是读文件时就统一pd.to_datetime(..., utcTrue)再.dt.tz_convert(Asia/Shanghai)全链路只用一个时区永远不在业务代码里转。我曾经在这上面耗掉大半天从数据的字段名查到内存最后发现是时区。现象二读一个几百兆的 CSV 直接 MemoryError。原因是pd.read_csv默认会把整张表读进内存而且字符串列会被推成 object 类型内存占用翻好几倍。解决方法是显式指定dtype和usecols把用不到的大字段直接排除或者用chunksize分块处理每块抽完特征就丢弃中间结果落 parquet。别指望加内存条加完你还是会写同样的代码。现象三SMOTE 在训练时报n_neighbors相关的错误。原因是少数类样本数少于k_neighbors 1。电力场景下某些攻击类型可能只有三五条记录这很常见。解决方法是按类别的样本量动态调k_neighbors或者直接用class_weightbalanced替代重采样。硬把k_neighbors设成 1 生成出来的样本基本是原样本的复制对模型没好处。5.2 评估与线上表现类问题现象四离线评估 F1 有 0.96上线后基本什么都抓不到。最常见的元凶是随机切分。时间序列数据用train_test_split随机打乱等于让模型在训练时看到了未来的样本同一个攻击事件被拆到训练集和测试集两边模型只要记住这次事件的特征就能预测出来。解决方法是严格按时间切分前 70% 做训练、中间 15% 做验证、最后 15% 做测试而且切分点要落在攻击事件之间的空档里不能从一次攻击中间切开。现象五准确率 0.999混淆矩阵里正类一个都没预测出来。原因就是样本极度不平衡加上用了默认阈值 0.5。模型学到了全判负样本这个局部最优。解决办法不是换模型是换评估口径和阈值策略先算召回率再用 4.3 节的方法按误报上限扫阈值或者用 PR 曲线下的面积替代 ROC-AUC 作为模型选择依据。在 1:500 的比例下ROC-AUC 会给出一个相当乐观的假象。现象六刚上线两天告警面板一天几百条。除了阈值问题还有一个容易被忽略的原因是基线没对齐。训练集里正常流量的分布和上线环境里的分布可能完全不同——测试环境的流量强度、设备型号、业务时段都变了模型把没见过的正常判成了攻击。解决办法分两步上线前先用影子模式跑一到两周只记录不告警收集线上真实分布然后做一次增量校准把线上正常样本加入训练集重训或者对特征做分位数量纲归一化减少分布漂移的影响。6. 二次开发换数据源、加一路传感器、改判定阈值的三步改法这份代码最值得拿来改的地方是它的分层足够干净改动可以限制在很小的范围内。6.1 换数据源和加一路传感器只动配置层如果只是把网络流量从按流汇总换成包级特征或者新增一路来自继电保护装置的操作日志理论上不需要动fuse.py和model.py。要做的三件事是在settings.yaml里注册新的数据源名、采样周期、字段映射在features.py里给它指定一套窗口参数把新特征列名加进build_feature_fusion的分组列表。只要新数据有一个时间戳列 一组数值列 数据源标识这三件套对齐层直接复用。6.2 验证改动的回放法改完别急着看指标先用回放法验证。做法是在一段已知正常的历史数据里人为插入几段有标注的攻击片段可以从公开数据里裁剪也可以按攻击手法合成然后让整条链路从头跑到尾看三件事插入的攻击片段是否全部被检出、检出的时间点比插入点晚多少、误报有没有因为新增特征而上升。这个方法比单纯看测试集指标靠谱因为它验证的是端到端的行为而不是模型在固定测试集上的分数。我习惯把每次改动的回放结果存成一张对比表改了什么、召回涨了还是跌了、误报变成多少一眼能看出来。顺带提一句资源包里除了这份 Python 实现还并排放着一批同方向的 Matlab/Python 算例做电力系统优化、负荷预测、信号处理的都有接口风格不太一样但数据准备和评估那套思路是相通的可以对照着看融合逻辑该怎么拆。我现在每次动特征工程都会强制自己先跑一遍完整回放再看指标——因为吃过一次亏某个改完之后验证集涨了三个点的提交上线第一周把一整条线路的告警刷爆了最后查出来是新加的一个频域特征在夜间低流量时段数值异常。从那以后我给自己立了个规矩任何特征改动必须先在回放数据上过一遍白天的和夜间的两个时段。希望这篇拆解能帮到你少走几步弯路。本文还有配套的精品资源点击获取