电力系统网络入侵检测中的多传感器融合Python实现

简介围绕电力系统网络入侵检测中的多传感器融合方法这份文档面向具备Python编程基础、从事电力系统安全或工业控制系统安全研究的科研人员与研究生帮助其理解如何整合来自不同传感器的异构数据提升异常行为识别精度与响应效率。内容覆盖数据预处理、特征提取、融合算法设计与入侵检测模型构建等关键环节可用于潜在入侵的早期预警、算法复现以及融合策略优化等场景。压缩包共1个文件为docx文档大小约77KB结构紧凑便于查阅与整理。已有35人学习适合作为多源数据协同检测的入门参考与课题实践素材读者可据此梳理检测流程、把握分类模型设计要点并在仿真环境或实际数据中进一步测试与调优。1. 一路告警压不住的时候多传感器融合在电力系统网络入侵检测里到底做什么电力监控场景里有个老毛病只看流量探针一次规约异常可能只是主站正常的遥控下发只看主机日志一次集中登录可能只是运维在做例行巡检。告警刷屏、值班员麻木、真事件埋在噪声里这是不少现场的血泪经验。多传感器融合落到电力系统的网络入侵检测上说白了就是把流量统计、主机日志、设备事件、安全设备告警这几路异构数据按同一根时间轴对齐再在特征层或决策层合成一个判断。这份 Python 代码实现给的是能跑的骨架数据接入、时基对齐、特征拼接、决策级加权、评估与阈值调节一条链走通。适合有 Python 基础、又要在工控安全或电力监控方向做落地的研究生和工程师也适合拿它当自己项目的融合基线。2. 把四路数据拧到一根时间轴上重采样、缺失标记与标签对齐融合模型效果差十次里有六次不是算法问题是数据没对齐。流量是按秒来的主机日志是事件触发的保护装置的动作记录可能几分钟才有一条安全设备告警还带着自己的时区配置。直接按行 concat等于把不同时刻发生的事强行说成同时发生模型学到的全是噪声。这一章把「对齐」这件事拆成三步先认清楚各路数据的时间口径再定重采样网格最后处理缺失和标签窗口。2.1 四类数据源与它们各自的时间口径在电力监控网络里能拿到的融合输入通常是这四类字段名各家不同但语义差不多数据源典型字段原始时间粒度时间口径坑流量统计五元组、包数、字节数、目的端口秒级或会话级采集器落盘时间与抓包时间不一致主机/设备日志时间戳、来源 IP、消息文本、级别事件触发稀疏设备本地时钟未同步跨时区电力业务事件遥测遥信变位、保护动作、遥控记录毫秒到分钟时标可能来自装置内部晶振安全设备告警规则 ID、源目 IP、严重级别事件触发告警时间常是「检测完成时间」而非「发生时间」真正麻烦的是第四类。很多安全设备的告警时间戳记的是规则匹配完成的时刻一条会话持续几十秒告警可能落在会话结束之后。拿这个时间去对齐流量窗口特征和标签就错位了。常见做法是优先用会话开始时间拿不到就退回采集时间并在特征表里单独留一列标记这个来源用的是哪种时间口径。2.2 统一时基为什么选 10 秒网格重采样网格的粒度没有标准答案但有取舍。1 秒网格能保留突发细节代价是稀疏日志源会产生大量空窗口特征矩阵稀疏到模型学不动60 秒网格平滑了噪声又容易把一次持续十几秒的扫描行为摊薄成「几乎没变化」。我一般会先用 10 秒网格做基线理由是它大致覆盖一次 TCP 会话的生存周期也接近多数日志源的聚合上报周期。真要调就在 5 秒和 30 秒之间做一次对比看验证集上的 PR-AUC 变化别拍脑袋。import pandas as pd import numpy as np WIN 10s # 统一时间网格决定后面所有特征的观测粒度 def load_flow(path: str) - pd.DataFrame: 流量统计会话级记录聚合成固定窗口 df pd.read_csv(path, parse_dates[ts]) # 关键一步统一转 UTC。混着本地时解析跨夏令时必翻车 df[ts] pd.to_datetime(df[ts], utcTrue) df df.set_index(ts).sort_index() out df.resample(WIN).agg( flow_pkts(pkts, sum), # 窗口内总包数 flow_bytes(bytes, sum), # 窗口内总字节数 flow_conn_cnt(src_ip, count), # 会话条数 flow_dst_port_nuniq(dst_port, nunique), # 独立目的端口数扫描行为的强特征 ) return out def load_syslog(path: str) - pd.DataFrame: 主机日志把消息文本压成计数特征 df pd.read_csv(path, parse_dates[ts]) df[ts] pd.to_datetime(df[ts], utcTrue) # 用正则粗筛失败/拒绝类关键字比上 NLP 模型省事也够用 df[is_fail] df[msg].str.contains( rfailed|denied|invalid, caseFalse, regexTrue ).astype(int) df df.set_index(ts).sort_index() out df.resample(WIN).agg( log_cnt(msg, count), log_fail_cnt(is_fail, sum), log_src_nuniq(src_ip, nunique), ) return out代码里resample(WIN)是核心它把不规则的原始记录压到等间隔网格上。sum和count的选择有讲究包数、字节数用sum会话数和独立 IP 数用count或nunique混用会让特征的物理含义变味。utcTrue这个参数不要省本地时区加上夏令时会让某些窗口凭空多出或少掉一小时的数据。如果你的数据源时标是带偏移的字符串先统一转 UTC 再做任何聚合顺序不能反。2.3 缺失不是零缺失标记与「距上次事件时间」稀疏源重采样之后会出现大量 NaN。直接fillna(0)是最常见的做法也是最容易埋雷的做法。深夜两点没有登录失败日志和「这个窗口日志采集器掉线了」在模型眼里都会变成log_fail_cnt0含义完全不同。我的处理方式是两个都保留数值填 0同时给每个来源加一列*_seen表示本窗口是否真有观测。再补一类时间特征就是「距上次事件的时间差」。对保护动作、安全告警这种低频源这个特征比计数本身更有信息量因为它保留了事件之间的间隔结构。def build_feature_table(parts: dict, label: pd.DataFrame) - pd.DataFrame: parts: {flow: df, log: df, alarm: df}统一 join 到同一网格 start min(df.index.min() for df in parts.values()) end max(df.index.max() for df in parts.values()) idx pd.date_range(startstart, endend, freqWIN, tzUTC) X pd.DataFrame(indexidx) for name, df in parts.items(): sub df.reindex(idx) # 观测标记本窗口该来源是否有真实数据 X[f{name}_seen] sub.notna().any(axis1).astype(int) # 距上次事件的秒数第一次出现前用一个大常数占位 has_evt sub.notna().any(axis1) last_seen pd.Series( np.where(has_evt, idx.view(int64) // 10**9, np.nan), indexidx ).ffill() X[f{name}_since_last_s] ( idx.view(int64) // 10**9 - last_seen ).fillna(86400).clip(upper86400) # 数值列填 0但 seen 列已经保留了「这是填充值」的信息 X X.join(sub.fillna(0), howleft) # 标签按窗口右边界对齐详见 2.4 X X.join(label, howleft) return Xsince_last_s用clip(upper86400)是为了防止极端值把标准化拉偏。seen列在特征重要性里往往排得很靠前不是因为它多有业务含义而是因为它替模型区分了「真零」和「假零」。2.4 标签窗口与特征窗口必须错开这是最隐蔽的一个坑。假设标签定义是「窗口 [t, t10s) 内是否发生入侵」而特征用的是同一个窗口 [t, t10s) 的统计量。攻击发生时流量必然突增这个突增同时进了特征和标签模型离线能刷到 0.99上线直接崩。正确做法是把特征窗口整体前移特征取 [t-W, t)标签取 (t, tH]中间留一段缓冲两边不重叠。def make_label(event_ts: pd.Series, idx: pd.DatetimeIndex, horizon: str 10s) - pd.DataFrame: 事件时间戳 - 窗口标签只在 (t, thorizon] 内标记为 1 y pd.Series(0, indexidx, dtypeint) for t in event_ts: # 找事件之后的第一个网格点避免把事件当刻算进来 pos idx.searchsorted(t, sideright) if pos len(idx): y.iloc[pos] 1 return y.to_frame(label) # 训练时再把特征整体 shift 一个窗口保证 X 只看过去 X_train X.shift(1).dropna() y_train X[label].loc[X_train.index]shift(1)这一步别嫌麻烦它相当于在时间轴上给特征留了一格的「反应时间」。如果你用的是更大的窗口比如 60 秒特征配 10 秒标签那就该 shift 6 格换算关系要自己盯住。标签事件的粒度如果比网格细searchsorted用sideright保证事件不会被算到自己所在的窗口里。3. 特征级还是决策级融合层次的选择与可跑的代码对齐做完下一个问题是在哪一层融合。这不是玄学取决于各路数据能不能拼到同一张表里、以及子模型之间的误差是不是相关的。选错层次后面的调参全是白费。3.1 三个层次的边界层次做法适用条件典型代价数据级原始信号直接合成同构传感器、采样率一致电力多源场景几乎不可用特征级各路特征拼成宽表单模型训练时间对齐可靠、特征量可控高维稀疏易受缺失影响决策级每路单独建模概率再加权子模型误差不相关、某路常缺失需要额外验证集定权重电力系统里我不建议碰数据级采样率差着两三个数量级硬合成只会引入伪影。特征级是默认起点实现简单、可解释性好只要对齐做扎实效果通常够用。决策级的价值在于鲁棒性某一路探针掉线时其他路的模型还能给出分数不会整体失效。判断要不要上决策级看一个指标就行——各路数据在你的数据集里的缺失率如果某一类源缺失超过 20%特征级模型会被这路的缺失标记带偏这时候拆成决策级更稳。3.2 特征级融合拼接之后的标准化与降维特征级融合的坑主要在量纲。包数动辄上万日志条数常常是个位数树模型不在乎这个线性模型和神经网络会在包数那一列上打转。标准化必须做降维看情况特征数在几十列这个量级PCA 往往帮不上忙反而丢信息上百列且高度共线时保留 95% 方差是个合理起点。from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler from sklearn.ensemble import RandomForestClassifier def build_feature_level_model(n_estimators: int 300, max_depth: int None): 特征级融合所有源拼一张表单模型出分 return Pipeline([ (scaler, StandardScaler()), # 树模型可省但留着方便换线性模型 (clf, RandomForestClassifier( n_estimatorsn_estimators, max_depthmax_depth, class_weightbalanced, # 入侵样本少先压住多数类 n_jobs-1, random_state42, )), ]) model build_feature_level_model() model.fit(X_train, y_train) score model.predict_proba(X_test)[:, 1] # 取「攻击」类概率class_weightbalanced是按类别频率反比给权重参数不用手调但对极端不平衡正样本低于 1%效果有限那时候要靠采样或者代价敏感第 4 章再说。predict_proba[:, 1]取的是正类概率后面决策级融合和阈值扫描都依赖这个连续分数不要提前用predict二值化。3.3 决策级融合加权概率与软投票决策级的做法是每路数据单独训一个子模型输出概率后加权平均。前提是子模型真的只看到自己那一路数据如果三个子模型都喂了同一张宽表加权平均只是把同一个模型的方差平均掉收益有限。import numpy as np def late_fusion(proba: dict, weights: dict) - np.ndarray: proba: {flow: p1, log: p2, alarm: p3}均为攻击类概率 keys list(proba) w np.array([weights[k] for k in keys], dtypefloat) w w / w.sum() # 归一化防止权重和不为 1 导致分数越界 stack np.vstack([proba[k] for k in keys]) return (stack * w[:, None]).sum(axis0) def soft_vote(proba: dict, thr: float 0.5) - np.ndarray: 等权软投票作为加权融合的对照基线 stack np.vstack(list(proba.values())) return (stack.mean(axis0) thr).astype(int)w / w.sum()这一行别省。三个权重写 0.5、0.3、0.3加权和可能超过 1分数就不是概率了后面阈值扫描全乱。软投票的对照很有必要如果加权融合打不过等权说明权重估计本身在过拟合退回去用等权更稳。3.4 权重怎么定才不至于过拟合权重不是拍脑袋给的。三种定法按可靠性排一是在验证集上按子模型各自的 AP 归一化简单但只看单模型质量二是用验证集做小网格搜索权重组合数不多时可行三是把三个子模型的输出概率当特征训一个逻辑回归做堆叠这本质上是在学权重。我一般先用第一种出一版基线再用第三种验证有没有提升两者差异很小就保留第一种参数少、好解释。from sklearn.linear_model import LogisticRegression def stack_fusion(train_proba: dict, y_train, test_proba: dict): 堆叠把子模型概率当特征学一组融合权重 keys list(train_proba) Xtr np.vstack([train_proba[k] for k in keys]).T Xte np.vstack([test_proba[k] for k in keys]).T lr LogisticRegression(max_iter1000) lr.fit(Xtr, y_train) return lr.predict_proba(Xte)[:, 1], dict(zip(keys, lr.coef_[0]))堆叠的系数能直接读出来当权重参考但要注意它是在验证集上训的最终评估还得留一段完全没碰过的时间段。系数出现负数也别慌说明那一路在当前数据上起到了反向指示作用通常是该路数据质量有问题值得回头查采集链路。4. 融合模型的训练与评估类别不平衡下的可信指标入侵检测的数据集天然不平衡一个月的流量里真正的攻击窗口可能只有几十个。这个比例下准确率是废指标99% 的准确率可能意味着模型把所有样本都判成正常。评估体系搭不对融合做得再花哨也证明不了价值。4.1 采样与代价敏感怎么选三种手段过采样、欠采样、代价敏感。欠采样会丢掉大量正常样本的时间结构对时序场景伤害大一般不单独用。过采样里 SMOTE 这类插值方法在时序特征上要谨慎它会在两个不相邻的时间点之间造样本物理上不成立。我的默认组合是class_weightbalanced加阈值调节只有在正样本极少比如少于 50 个时才考虑对正样本做随机过采样而且只在训练折内做绝不能在全量数据上做再切分。4.2 指标看 PR-AUC 和固定误报率下的召回ROC-AUC 在极端不平衡时会虚高因为负样本太多假阳性率的计算被稀释。PR-AUC平均精度对正类更敏感是主指标。工程上还要看一个数在每天允许 N 条误报的前提下能拿到多少召回。这个数直接对应值班员的工作量比任何曲线都实在。from sklearn.metrics import average_precision_score, precision_recall_curve def report(y_true, score, max_fp_per_day: int 20, n_days: int 30): ap average_precision_score(y_true, score) prec, rec, thr precision_recall_curve(y_true, score) # precision_recall_curve 的 thresholds 比 prec/rec 少一个索引要减一 allowed_fp max_fp_per_day * n_days n_neg int((y_true 0).sum()) max_fpr allowed_fp / max(n_neg, 1) fpr 1 - prec[:-1] * 0 (1 - prec[:-1]) # 占位见下方说明 ok (1 - prec[:-1]) max_fpr best_rec float(rec[:-1][ok].max()) if ok.any() else 0.0 return {pr_auc: round(ap, 4), recall_at_fp_budget: round(best_rec, 4)}上面那段fpr是占位写法实际算假阳性率要用混淆矩阵别照抄。更稳的写法是遍历阈值直接数 FPdef recall_at_budget(y_true, score, max_fp: int): best 0.0 for t in np.unique(score): pred (score t).astype(int) fp int(((pred 1) (y_true 0)).sum()) tp int(((pred 1) (y_true 1)).sum()) fn int(((pred 0) (y_true 1)).sum()) if fp max_fp: best max(best, tp / max(tp fn, 1)) return bestmax_fp是按你的评估周期换算的30 天数据、每天允许 20 条就是 600。这个参数一变结论可能完全反过来所以报告指标时一定要把预算写清楚别只扔一个 PR-AUC 出去。4.3 按时间切分不要随机切时序数据随机切分会让未来信息泄漏到训练集分数虚高这是最典型的翻车点。切分要按时间顺序前 70% 训练、后 30% 测试中间可以留一段做验证调阈值。更严谨的做法是做滚动窗口验证用第 1 到 3 周训练、第 4 周测试再往后滚一周看指标稳不稳。def time_split(X, y, train_ratio: float 0.7): cut int(len(X) * train_ratio) return X.iloc[:cut], X.iloc[cut:], y.iloc[:cut], y.iloc[cut:] def rolling_eval(X, y, n_folds: int 4): 滚动验证每折用前面的数据训紧邻的一段测 size len(X) // (n_folds 1) scores [] for i in range(1, n_folds 1): tr_end size * i te_end size * (i 1) model build_feature_level_model() model.fit(X.iloc[:tr_end], y.iloc[:tr_end]) p model.predict_proba(X.iloc[tr_end:te_end])[:, 1] scores.append(average_precision_score(y.iloc[tr_end:te_end], p)) return scores滚动验证的分数通常比单次切分低这是正常的也更接近上线后的真实表现。如果滚动各折之间方差很大说明模型对时间段敏感可能是某几周的攻击类型和别的周差异太大这本身就是要写进结论的发现。5. 避坑排查五个把融合结果做废的细节下面这几条都是我在联调里真实撞过的现象看着都像「模型不行」查下去全是数据或流程问题。每条按现象、原因、解决三段写照着排查能省不少时间。5.1 融合后指标反而低于单源现象单独用流量特征 PR-AUC 有 0.72拼上日志和告警之后掉到 0.61。原因日志源的时间戳带着本地时区偏移转 UTC 时只对流量做了处理日志整体错位了几小时拼进来的是另一段时间的噪声。解决所有来源在入库前统一pd.to_datetime(..., utcTrue)并在特征表里留一列source_tz_offset记录原始偏移量。上线前跑一次对齐自检任取一个已知事件时刻看四路数据在该时刻前后是否有对应的观测对不上就是时区或时钟问题。5.2 离线 0.99上线一塌糊涂现象交叉验证分数极高部署后几乎全判正常或全判攻击。原因标签窗口和特征窗口重叠攻击发生时的流量突增同时进了特征和标签或者标准化、填充这类预处理在切分之前对全量数据做了测试集统计量泄漏进训练。解决特征窗口整体前移一个窗口用shift实现所有 fit 类操作放进 Pipeline只在训练折上 fit。检查方法很土但有效把标签打乱重训一遍如果分数还是很高说明有泄漏。5.3 深夜时段的误报特别多现象白天的检测基本正常凌晨误报集中爆发。原因凌晨业务流量本来就低缺失填充用 0 之后正常窗口的特征和攻击窗口长得很像模型把「低流量」学成了「异常」。解决加*_seen观测标记列让模型能区分「真零」和「没采到」同时对流量类特征做分时段标准化或者直接把小时作为周期特征喂进去。排查时按小时分组统计误报率一眼就能定位。5.4 决策级权重在测试集上表现很好换一批数据就崩现象加权融合的权重是在全量验证集上网格搜出来的当时提升明显下一批数据上反而不如等权。原因权重搜索用到的验证集和最终评估集重叠或者验证集太小搜出来的权重只是在拟合那几十个正样本。解决权重只用训练期内的一段验证数据定评估集完全隔离样本量不足时优先用等权软投票或者用逻辑回归堆叠并加 L2 正则。权重系数的绝对值不该出现某一项接近 1、其余接近 0 的情况那说明融合退化成单源了。5.5 某一路探针掉线整个系统静默失效现象采集链路中断后融合模型不再报任何告警值班员以为是「网络变干净了」。原因缺失填充把整路数据变成 0模型输出稳定在低分区同时没有任何监控覆盖「输入特征是否还在更新」这件事。解决在推理侧加输入健康检查统计每个来源最近 N 个窗口的seen比例低于阈值就降级到可用子集的模型并单独上报告警。这一条属于工程护栏不是模型能力但没有它融合系统的黑匣子属性会让人很难受。5.6 一个可复用的上线前自检脚本把上面几条揉成一个函数每次换数据集先跑一遍能挡掉大部分低级错误。def sanity_check(X: pd.DataFrame, y: pd.Series, meta: dict) - dict: report {} # 1) 时间轴是否单调、有无重复 report[is_sorted] X.index.is_monotonic_increasing report[dup_ts] int(X.index.duplicated().sum()) # 2) 各来源观测率低于 0.5 要考虑降级 for c in [c for c in X.columns if c.endswith(_seen)]: report[fseen_rate_{c}] round(float(X[c].mean()), 3) # 3) 标签比例太极端要回头查标签构造 report[pos_rate] round(float(y.mean()), 4) # 4) 特征与标签的即时相关性高得离谱大概率是泄漏 corr X.corrwith(y).abs().sort_values(ascendingFalse) report[top_corr] corr.head(3).round(3).to_dict() report[tz] str(X.index.tzinfo) return reporttop_corr里如果出现 0.9 以上的特征先别高兴回头查这个特征是不是在标签窗口内算出来的。seen_rate低于 0.5 的来源要么补采集要么在融合时降权别硬拼。6. 上线前的回放验证与阈值迟滞两个能少加班的技巧模型训完不等于能用。我见过的失败上线一半死在阈值抖上分数在 0.5 附近来回跳同一个攻击事件连续触发十几条告警值班员直接把这套系统静音了。回放验证加迟滞判断是我现在必做的两步。6.1 用历史数据做影子回放别拿测试集分数直接当上线依据。把过去一个月的原始数据按时间顺序灌进推理链路输出分数流和当时的真实事件记录比对。重点看三件事一是每天触发的告警条数分布有没有某天爆量二是同一事件是否被重复触发三是采集中断的那几个小时系统表现是否合理。回放脚本不需要多复杂把特征构造和推理封成一个predict_window(df)函数按窗口循环调用就行。def replay(feature_df: pd.DataFrame, model, step: int 1): 按时间顺序逐窗推理模拟线上分数流 scores [] for i in range(0, len(feature_df), step): row feature_df.iloc[[i]] scores.append(float(model.predict_proba(row)[:, 1][0])) return pd.Series(scores, indexfeature_df.index[::step])6.2 阈值加迟滞压住分数抖动单阈值判定在边界附近必然抖。工程上的做法是双阈值迟滞分数超过高位才置为告警要降到低位以下才解除中间区间保持上一状态。参数含义取值建议high触发告警的分数按误报预算扫出来的阈值再上浮 0.05low解除告警的分数high 减去 0.15 到 0.25min_hold最短保持窗口数3 到 6 个窗口覆盖一次会话的持续时长class Hysteresis: def __init__(self, high: float 0.8, low: float 0.55, min_hold: int 3): self.high, self.low, self.min_hold high, low, min_hold self.state, self.hold 0, 0 def update(self, score: float) - int: if self.state 0 and score self.high: self.state, self.hold 1, 0 # 越过高位进入告警态 elif self.state 1: self.hold 1 if score self.low and self.hold self.min_hold: self.state 0 # 低位持续足够久才解除 return self.statemin_hold这个参数别设太小设成 1 等于没做迟滞。high和low的间距决定灵敏度间距太窄照样抖太宽会漏掉短时攻击我一般在回放数据上把间距从 0.1 扫到 0.3看「同一事件重复告警次数」这个指标落到多少。还有一个习惯值得养成把融合分数当排序用而不是只当判定用。值班员先看分数最高的前 20 条比看一个二值告警列表信息量大得多尤其在阈值还没调准的初期。资源包里的 Python 脚本按数据接入、特征构造、融合、评估拆成了独立模块样例数据的字段和文中列名对得上照着第 2 章的列名改就能接自己的数据。从那以后我每接一个融合场景都强制先把时区对齐、标签窗口重叠、按时间切分的基线分数这三件事跑一遍再谈模型选型——顺序反了后面全是白工。希望帮到你。本文还有配套的精品资源点击获取