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

简介面向电力系统网络安全方向的科研人员与研究生这份Python代码实现聚焦多传感器融合下的网络入侵检测将来自不同传感器、协议与业务节点的异构数据经预处理、特征提取与融合算法处理后输入入侵检测模型以提升异常行为识别精度与响应效率可用于关键基础设施安全监控的早期预警研究。压缩包共1个文件为docx文档约77KB集中呈现方法思路与代码实现说明便于按模块对照理解数据融合逻辑与分类模型设计。已有35人学习适合具备Python与机器学习基础、希望复现并优化多源数据协同检测策略的读者参考也可作为工业控制系统与物联网安全场景的实践素材。1. 电力系统网络入侵检测为什么要走多传感器融合这条路值班室的屏幕上跳出一条告警某台工控主机在凌晨两点向外发起了一批异常连接。单看流量探针的 NetFlow 记录像是例行的心跳包重传单看主机侧的命令日志只是几条权限内的查询单看保护装置的量测电压电流纹丝不动。三路数据各自都不足以定性可把时间轴对齐、放在同一个 60 秒窗口里看结论瞬间反转——连接目标、命令序列、量测扰动三者同时出现偏移这是一次有预谋的扫描加探测。多传感器融合在电力系统网络入侵检测里的价值就在这儿它不指望某一路信号够强而是靠多源数据融合把弱证据叠成强证据。这套 Python 代码实现干的就是这件事从异构数据对齐、特征提取、融合层设计到入侵检测模型训练评估一条链路走通。它适合有 Python 基础、做电力系统安全或工控安全方向的开发者与研究生想复现一套能改、能跑、能对比基线的融合框架。2. 项目结构与技术选型从目录布局看数据怎么流拿到一个算法包先别看模型先看目录。目录结构往往比任何说明文档都诚实它直接暴露了作者对数据流的理解哪些是入口、哪些是中间产物、哪些是可以替换的插件位。这套代码的组织方式属于典型的轻量科研工程——没有上框架、没有 ORM、没有服务化全部围绕「一次实验跑完出指标」这个目标来铺。2.1 目录结构与各模块职责清理掉缓存和临时文件后有效结构大致如下power-ids-fusion/ ├── config/ │ └── settings.yaml # 路径、窗口长度、融合权重、模型超参 ├── data/ │ ├── raw/ # 三路原始数据netflow / hostlog / pmu │ └── processed/ # 对齐、清洗后的中间宽表 ├── src/ │ ├── loader.py # 读文件、统一表头、打来源标记 │ ├── preprocess.py # 时间窗对齐、缺失值、标准化、划分 │ ├── extract.py # 统计特征、PCA 降维 │ ├── fusion.py # 特征级加权拼接、软投票、D-S 证据合成 │ ├── models.py # 单源基线与融合模型的训练封装 │ ├── evaluate.py # 指标计算、混淆矩阵、对比表输出 │ └── pipeline.py # 串起全流程的编排入口 ├── scripts/ │ └── run_pipeline.py # 命令行入口接 argparse ├── requirements.txt └── README.md模块职责主要函数loader.py异构表头归一、时间戳解析、来源打标load_source()preprocess.py按实体时间窗聚合、对齐成宽表、缺失填补、缩放build_window_features()、split_and_scale()extract.py滚动统计特征、PCAadd_rolling_stats()、fit_pca()fusion.py三种融合策略feature_level_fusion()、soft_voting()、ds_combine()models.py基线与融合模型训练train_baseline()、train_fusion()evaluate.py统一口径评估report_all()这套划分的关键在fusion.py被单独拎出来。很多同类代码把融合逻辑硬塞在训练函数里改一种融合策略就得动模型代码改一次跑一次最后谁也不敢动。把融合拆成独立模块之后融合策略和分类器变成两个正交维度可以自由组合成「特征级 随机森林」「决策级 D-S XGBoost」这样的矩阵做消融。我在做对比实验时最看重的就是这种可正交拆解的结构它决定了你后面能不能在论文或报告里讲清楚提升到底来自融合还是来自换模型。2.2 依赖清单与运行环境依赖不多都是数据处理和机器学习的老面孔版本上不追求最新追求能复现# requirements.txt 的核心部分 python3.9 # 3.8~3.11 均可3.12 上部分科学计算包编译会折腾 numpy1.24.3 pandas2.0.3 scikit-learn1.3.0 imbalanced-learn0.11.0 # SMOTE 等过采样处理攻击样本稀疏 pyyaml6.0 # 读 settings.yaml matplotlib3.7.2 # 画混淆矩阵与融合前后对比 joblib1.3.2 # 模型持久化比 pickle 对 numpy 数组更友好安装照常走一条命令python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt这里有个容易被忽略的点imbalanced-learn和scikit-learn的版本是强耦合的。0.11 之后的imbalanced-learn要求scikit-learn 1.2如果本机已经装了旧版 sklearnpip 会静默降级imbalanced-learn然后在SMOTE里抛一个看着毫不相干的AttributeError。常见做法是装完之后打印一次版本号对一下别等跑到一半才排查。提示不要在全局 Python 环境里装这套依赖。工控安全方向的同学本机往往还跑着别的仿真工具科学计算包版本打架是最典型的玄学故障来源。2.3 数据契约三路传感器的字段约定与时间对齐基准融合系统的成败八成在数据接入这一层就决定了。这套代码假设三路数据分别是网络流量侧连接级记录含源地址、目的地址、协议、持续时长、上下行字节数、时间戳粒度通常是秒级。主机日志侧工控主机上的操作事件含事件时间、主机标识、操作类型、执行用户粒度不均匀闲时几小时一条、忙时一秒几条。物理量测侧保护或测量装置上送的量测量含采样时间、设备编号、电压、电流粒度可能是毫秒到秒级。三者字段名、时间精度、采样频率全不一样。所以代码里第一件事不是建模是定契约统一时间戳字段名timestamp、统一实体标识字段名src、统一来源标记source。后续所有聚合、对齐、融合都只认这三个内部字段原始字段名在loader里一次性映射掉。对齐基准选择 60 秒窗口是因为电力系统网络入侵的典型行为端口扫描、暴力登录、异常指令下发在秒级到分钟级尺度上才呈现出可辨识的模式窗口太短噪声占主导窗口太长又会把不同事件糊在一起。这个值放在settings.yaml里后续做敏感性分析时可以按 30s/60s/300s 各跑一轮。3. 数据接入与预处理把三路异构数据拉齐到同一条时间轴预处理占了整套代码将近一半的代码量这不是过度设计。真实场景下你拿到的三份数据时间戳单位都可能不一样——一份是 Unix 秒、一份是带时区的字符串、一份是设备本地时间。不做统一就直接拼表得到的是一张行数对不上、索引全 NaN 的废墟。3.1 多源数据加载与时间窗对齐先看加载层。核心动作是三次映射列名映射、时间解析、来源打标。# src/loader.py import pandas as pd # 三路异构表头 - 内部统一字段新增数据源只需要在这里加一条映射 COL_MAPS { netflow: {src_ip: src, dst_ip: dst, proto: proto, duration: dur, fwd_bytes: fwd_bytes, bwd_bytes: bwd_bytes, ts: timestamp}, hostlog: {event_time: timestamp, host: src, cmd: cmd, uid: uid}, pmu: {sample_time: timestamp, dev_id: src, u1: volt, i1: curr}, } def load_source(path: str, source_name: str, ts_unit: str s) - pd.DataFrame: 读单路数据并归一化表头。 path : 原始文件路径 source_name : 来源标记写进 source 列融合与消融实验都靠它 ts_unit : 时间戳单位s 表示 Unix 秒ms 表示毫秒字符串时间传 None df pd.read_csv(path) df df.rename(columnsCOL_MAPS[source_name]) df[source] source_name # 时间解析数值型按单位换算字符串型直接 to_datetime if ts_unit is None: df[timestamp] pd.to_datetime(df[timestamp], errorscoerce) else: df[timestamp] pd.to_datetime(df[timestamp], unitts_unit, errorscoerce) # 解析失败的行直接丢留着会在后续 resample 里变成幽灵分组 return df.dropna(subset[timestamp]).sort_values(timestamp).reset_index(dropTrue)ts_unit参数看着不起眼实际上是最容易翻车的地方。pd.to_datetime带units时如果传进来的是毫秒值解析结果会跑到 1970 年附近然后你在后面按时间窗聚合时得到一个空表报错信息却是key not found非常迷惑。所以我在加载完之后会顺手 assert 一下时间范围落在合理区间内宁可早报错。加载完是聚合对齐。思路是每路数据按「实体 时间窗」做聚合得到一个以(src, window_start)为索引、以统计量为列的小表再把三张小表按键横向拼起来。# src/preprocess.py import pandas as pd def build_window_features(df: pd.DataFrame, window: str 60s) - pd.DataFrame: 按 源实体 时间窗 聚合输出透视宽表。 window : pandas 的 offset 别名60s/5min 均可 # 不同来源的聚合口径不同先按 source 分流 out {} for name, sub in df.groupby(source): if name netflow: agg {dur: [sum, max], fwd_bytes: sum, bwd_bytes: sum} elif name hostlog: # 日志侧没有数值量用条数当强度特征 sub sub.assign(cnt1.0) agg {cnt: [sum, max]} else: # pmu agg {volt: [mean, std], curr: [mean, std]} g (sub.set_index(timestamp) .groupby(src) # 同一台设备内部做时间窗 .resample(window) .agg(agg)) g.columns [f{name}_{a}_{b} for a, b in g.columns] out[name] g # 三路数据在外层 concat索引不一致的地方补 NaN交给填补环节处理 wide pd.concat(out.values(), axis1) return wide.sort_index()这段的核心是groupby(src).resample(window)的组合。注意顺序先按实体分组再在组内做时间重采样这样每台设备、每个源地址都拥有自己独立且连续的时间轴不会出现 A 设备的数据被聚合到 B 设备窗口里的情况。用反了先 resample 再 groupby就会得到一张实体被打散的表而且不报错是最阴的那种 bug。对齐之后三路数据在时间上依然是错开的——网络侧每分钟都有数据主机日志可能连续几个小时空窗。pd.concat(axis1)按索引对齐空窗处自动补 NaN这是符合预期的行为接下来由填补环节决定这些 NaN 代表这段确实没有事件还是采集缺失。3.2 缺失值填补与标准化顺序错了指标就废了缺失值处理要先想清楚语义。网络侧的空窗通常是这段时间真的没有连接填 0 是对的物理量测的空窗往往是采集设备离线填 0 会造出电压为 0 的假告警这类应该用前后插值再补一个缺失标记列。# src/preprocess.py import numpy as np import pandas as pd def impute_features(wide: pd.DataFrame) - pd.DataFrame: 按列语义分流填补并保留缺失指示列。 out wide.copy() for col in out.columns: miss out[col].isna() if not miss.any(): continue # 保留缺失标记模型能学到这个窗口这路数据缺了本身就是信号 out[f{col}__isna] miss.astype(np.int8) if col.startswith(pmu_): # 量测量用前后线性插值首尾仍缺的再补中位数 out[col] out[col].interpolate(limit_directionboth) out[col] out[col].fillna(out[col].median()) else: # 流量与日志侧空窗代表无事件补 0 out[col] out[col].fillna(0.0) return out那个__isna标记列值得单独说一句。工业场景里采集链路本身中断往往就是攻击的前兆——先打掉监控再动手。把这个缺失模式显式喂给模型比指望插值算法凭空猜出正确数值靠谱得多。填补完才是划分和缩放顺序绝不能反# src/preprocess.py from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler def split_and_scale(X, y, test_size: float 0.3, seed: int 42): 先划分再缩放避免测试集统计量泄漏进训练过程。 test_size : 测试集比例样本不平衡时要确保测试集里攻击样本绝对数够 X_tr, X_te, y_tr, y_te train_test_split( X, y, test_sizetest_size, stratifyy, random_stateseed) scaler StandardScaler().fit(X_tr) # 只在训练集上 fit return (scaler.transform(X_tr), scaler.transform(X_te), y_tr, y_te, scaler)stratifyy保证了训练集和测试集的正负样本比例一致样本量小的时候不加这个参数很可能出现测试集里攻击样本只有个位数的情况评估出来的 F1 波动大得没法看。而fit只在训练集上做是防止信息泄漏的基本纪律——用了全量数据算均值和方差测试集的分布信息就提前漏进了训练指标会虚高好几个点上线之后原形毕露。3.3 特征提取滚动统计与 PCA 降维怎么配合窗口聚合只给了你一堆静态统计量但入侵行为的判别信息很多藏在「变化」里连接数突然翻倍、字节数从平稳变得忽高忽低。加一层滚动统计能把这部分时序信息显式提出来。# src/extract.py import pandas as pd def add_rolling_stats(wide: pd.DataFrame, cols: list, k: int 5) - pd.DataFrame: 对关键列加 k 窗口滚动均值与差分捕捉变化趋势。 cols : 需要加滚动特征的列名一般挑流量字节数和日志条数 k : 滚动窗口长度以时间窗个数计5 表示看最近 5 个窗口 out wide.copy() for c in cols: if c not in out.columns: continue out[f{c}__roll{k}] out[c].rolling(k, min_periods1).mean() out[f{c}__diff] out[c].diff().fillna(0.0) return outmin_periods1是必须的否则序列开头 4 个窗口全是 NaN被后面的填补逻辑当缺失处理等于人为在数据头部造了一段异常。diff()的首行同样补 0。然后是 PCA。这里的定位要摆正PCA 在这套代码里不是主力特征提取手段而是压缩与去冗余。多源融合之后特征维度很快破百其中大量是强相关比如fwd_bytes_sum和bwd_bytes_sum在正常业务下高度同步直接喂给树模型虽然不影响结果但会让模型解释和特征重要性排序变得很难看。# src/extract.py from sklearn.decomposition import PCA def fit_pca(X_tr, X_te, var_ratio: float 0.95): 保留累计解释方差 95% 的主成分只对训练集 fit。 var_ratio : 累计方差贡献率阈值0.95 在多数教程里是默认起点 pca PCA(n_componentsvar_ratio, svd_solverfull, random_state42) return pca.fit_transform(X_tr), pca.transform(X_te), pcasvd_solverfull在特征数几百这个量级完全够用别急着上随机化版本随机化求解会引入额外的随机性做对比实验时会干扰你判断提升到底来自哪里。另外提醒一句PCA 之后的主成分不再有物理含义第一主成分对应电压均值这种解释基本都是事后附会写报告时别硬编。4. 融合层实现特征级拼接、决策级投票与证据合成融合层是这套代码真正的差异化部分。项目里给了三条路线从浅到深分别是特征级融合、决策级软投票、D-S 证据理论合成。三条路线在工程上的定位完全不同特征级最容易落地也最容易过拟合软投票最省事但需要各路模型概率可比D-S 理论表达力最强但对冲突处理敏感。4.1 特征级融合加权拼接与它的边界特征级融合最直白——把三路特征拼成一个大向量交给同一个分类器。# src/fusion.py import numpy as np def feature_level_fusion(blocks: dict, weights: dict None) - np.ndarray: 把各路特征块按权重放大后横向拼接。 blocks : {来源名: ndarray(n_samples, n_features)} weights : {来源名: float}不给则全部为 1.0 weights weights or {} parts [] for name, blk in blocks.items(): w float(weights.get(name, 1.0)) # 加权而非简单拼接物理量测这类低噪数据可以给更高话语权 parts.append(blk * w) return np.hstack(parts)weights这个设计有争议但我觉得合理。电力系统里三路数据的信噪比天差地别物理量测采样规整、噪声可控主机日志稀疏且格式不统一流量侧受业务波动影响大。等权拼接等于默认三者可信度相同实际并不成立。加权就是把这层先验塞进去。不过要提醒边界weights千万别在测试集上调。我见过不止一次有人拿测试集指标反推出一组最优权重写进报告说是自适应加权融合实际就是过拟合。正确做法是划分出验证集或者干脆用固定先验权重比如按各路单源模型的验证集 F1 归一化把测试集彻底留干净。4.2 决策级融合软投票与 D-S 证据合成决策级融合的思路是每路数据先各自训练一个分类器再在概率层面合成。好处是各路模型可以完全不同——流量侧用 XGBoost、日志侧用逻辑回归、量测侧用一个小 MLP互不干扰。软投票实现很轻# src/fusion.py import numpy as np def soft_voting(probs: dict, weights: dict None) - np.ndarray: 加权软投票返回攻击类的融合概率。 probs : {来源名: ndarray(n_samples, 2)}第 1 列必须是攻击类概率 weights : {来源名: float}默认等权 weights weights or {k: 1.0 for k in probs} num sum(weights[k] * probs[k][:, 1] for k in probs) den sum(weights.values()) return num / den注意概率列的顺序必须统一。sklearn 的predict_proba列顺序跟着classes_走字符标签按字典序排attack会排在normal前面跟你直觉相反。稳妥做法是在训练后显式记下clf.classes_取概率时按标签名索引别硬编码[:, 1]。D-S 证据理论要复杂一档。它把每个传感器的输出看成对「识别框架」的一组基本概率分配BPA用 Dempster 合成规则把多组 BPA 合成一组。这套代码里识别框架取{attack, normal}另外保留一个全集{attack, normal}表示不确定。# src/fusion.py from itertools import product def prob_to_bpa(p_attack: float, alpha: float 0.9) - dict: 把单源攻击概率转成基本概率分配。 alpha : 对该传感器的信任度0.9 表示留 10% 质量给不确定 return { frozenset({attack}): alpha * p_attack, frozenset({normal}): alpha * (1.0 - p_attack), frozenset({attack, normal}): 1.0 - alpha, # 不确定部分 } def ds_combine(bpas: list, eps: float 1e-9) - dict: Dempster 合成规则支持任意个 BPA 依次合成。 result bpas[0] for bpa in bpas[1:]: new, conflict {}, 0.0 for a, ma in result.items(): for b, mb in bpa.items(): inter a b if not inter: # 交集为空计入冲突质量 K conflict ma * mb continue new[inter] new.get(inter, 0.0) ma * mb norm 1.0 - conflict if norm eps: # 完全冲突合成规则失效退化为均匀分布而不是硬判 return {frozenset({attack}): 0.5, frozenset({normal}): 0.5} result {k: v / norm for k, v in new.items()} return resultalpha这个参数是整套融合逻辑里最值得调的一个。它表达的是我有多信这个传感器。物理侧数据质量高可以给 0.95日志侧噪声大给 0.7剩下 30% 质量留给不确定集——这样即使某一路给出了极端错误的判断在合成时也会因为不确定质量的存在而被稀释不至于一票否决。conflict的处理也很关键。当两路传感器给出完全相反的证据一个说攻击概率接近 1一个说接近 01 - conflict会趋近于 0直接除会炸。代码里退化成均匀分布等于告诉下游这单我判不了交给人看。这个保守策略在安全场景里远比硬判一个结果要好——宁可多一条待复核的告警也别漏掉一次真实入侵。决策时比较两个单元素焦元的质量# src/fusion.py def ds_predict(p_attack_list: list, alphas: list, thr: float 0.5) - int: 多源 BPA 合成后给出 0/1 判定。 bpas [prob_to_bpa(p, a) for p, a in zip(p_attack_list, alphas)] fused ds_combine(bpas) m_attack fused.get(frozenset({attack}), 0.0) return int(m_attack thr)阈值thr别死守 0.5。安全运营的实际诉求通常是高召回、可容忍一定误报把阈值压到 0.35 左右能明显提召回代价是多几条待确认工单——这个换算在电力系统里通常是划算的。4.3 训练与评估口径不统一就比不出高低模型封装得很薄主要作用是把超参收进配置、把随机种子固定住# src/models.py from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import f1_score, recall_score, roc_auc_score def train_baseline(X_tr, y_tr, X_te, y_te, seed: int 42): 单源基线模型用于和融合结果对照。 clf RandomForestClassifier( n_estimators300, # 树数量300 在该数据规模下收益已趋平 max_depthNone, class_weightbalanced, # 双重保险与 SMOTE 二选一即可 n_jobs-1, random_stateseed) clf.fit(X_tr, y_tr) prob clf.predict_proba(X_te)[:, list(clf.classes_).index(attack)] pred (prob 0.5).astype(int) return { f1: f1_score(y_te, pred), recall: recall_score(y_te, pred), auc: roc_auc_score(y_te, prob), model: clf, }评估这块我建议固定四件事否则不同融合策略之间的对比毫无意义同一个训练/测试划分、同一个随机种子、同一个判定阈值、同一套指标定义。项目里evaluate.py输出的对比表大致长这样策略召回率F1AUC说明仅流量侧0.780.740.86单源基线仅日志侧0.610.580.71事件稀疏样本少仅量测侧0.550.600.68对网络攻击不敏感特征级融合0.890.850.93维度膨胀需 PCA决策级软投票0.870.840.92各路模型可异构D-S 合成0.910.870.94冲突样本可人工复核这张表能说明的东西比自己脑补多得多。单源量测侧 AUC 只有 0.68看着很差但它在融合里贡献不小——因为它捕捉的是网络手段造成的物理侧微扰这类样本流量侧和日志侧都看不出来。融合提升的本质是互补不是简单叠加。5. 避坑与排查跑通这套代码最常见的几个翻车点融合类项目的坑有个共同特征不报错指标还挺好看但结果是假的。下面这几条都是我在类似项目里真实踩过或者帮人排查过的按现象、原因、解决三步给。5.1 数据层面的坑坑一时间戳单位不统一对齐后得到一张空表。现象是build_window_features跑完行数为 0或者行数正常但所有物理量测列全是 NaN。原因通常是某一路的时间戳是毫秒而ts_unit传了s解析出的日期落在 1970 年跟另外两路差了半个世纪concat时索引完全不相交。解决办法是在load_source加一条断言把时间范围打出来assert df[timestamp].dt.year.min() 2000。更稳妥的做法是在settings.yaml里为每个来源单独配置ts_unit别用全局默认值。坑二先 resample 再 groupby实体被打散。现象是聚合结果的索引里出现了本该属于不同设备的特征混在一起比如某台根本不发日志的设备它的窗口里却有了日志条数。原因是df.resample(window).agg(...)之后再做groupby(src)此时src列已经被聚合掉了剩下的分组毫无意义。解决办法是固定顺序先set_index(timestamp).groupby(src)再.resample(window)。写完顺手打印一下g.index.names应该看到[src, timestamp]两层索引。坑三缺失值填 0 之后模型学到了攻击有数据这种伪规律。现象是模型在测试集上 AUC 高得离谱一换数据就崩。原因是物理量测断采的时段恰好也常常是攻击时段填 0 让模型把数值为 0直接等同为发生攻击学到的是采集故障的规律而不是入侵行为。解决办法是按列语义分流填补同时显式保留__isna标记列让模型有机会区分真 0和缺测。另外可以做一个对照实验去掉所有__isna列再跑一遍如果指标掉得厉害说明模型确实在依赖缺失模式需要回头看采集侧的时序是否存在系统性偏差。5.2 模型与评估层面的坑坑四在划分之前做标准化或 SMOTE指标虚高五到十个点。现象是训练完 F1 到 0.95 以上换个随机种子就掉到 0.7。原因是不小心对全量数据做了StandardScaler().fit_transform(X)或者SMOTE().fit_resample(X, y)测试集的分布信息提前泄漏。解决办法是把split_and_scale放在流程最前面所有拟合类操作只接受训练集。更硬的办法是用sklearn.pipeline.Pipeline把缩放、降维、分类器串成一条fit时天然只在训练集上走一遍从结构上杜绝泄漏。SMOTE 尤其要注意它属于训练阶段的数据增强绝不能在测试集上做否则测的就不是真实分布了。坑五D-S 冲突样本被静默吞掉导致漏报集中在特定攻击类型上。现象是整体召回看着不错但把混淆矩阵按攻击子类型拆开看某一类比如低速慢扫描几乎全漏。原因是这类攻击在三路传感器上的证据高度一致地正常conflict很小合成结果被正常证据压过阈值判断直接放过。解决办法是把1 - conflict的分布画出来看同时把合成后m_attack落在 0.3 到 0.5 之间的样本单独捞出来人工过一遍。这批边界样本往往就是融合策略最该优化的地方比盲目调分类器阈值收获大得多。工程上也建议给这一档单独设一个待复核标签别硬塞进二分类里。6. 二次开发换数据、加传感器、往下游走代码跑通只是起点多数人要的是把它改成自己的东西。三条最常走的改造路线按改动成本从低到高排一下。换数据集。只要新数据的字段能映射到timestamp / src / 标签这三样改COL_MAPS加一条映射、在settings.yaml里指定标签列名就能跑。麻烦的是标签粒度公开数据集通常是连接级标签一条连接一个标签而这套代码是按 60 秒窗口聚合的一个窗口里可能同时有正常连接和攻击连接。我一般用窗口内只要有任一攻击连接就标为攻击这种偏召回的规则并在报告里写明这个假设——不同假设下的指标没法横向比较写清楚比数字好看重要。加传感器。比如想接入继电保护装置的动作记录或者安全设备侧的告警流。改动点固定在三处COL_MAPS加映射、build_window_features加一个分支的聚合口径、fusion.py里给它一个初始权重和alpha。初始值建议取 0.5 到 0.7 的保守区间等做了消融再往上调。# 新增一路传感器时的最小改动示意 COL_MAPS[relay] {action_time: timestamp, dev_id: src, action_type: act, trip_ms: trip} # preprocess 的聚合分支里追加 elif name relay: agg {trip: [mean, max], cnt: sum} # 融合侧给保守先验 FUSION_WEIGHTS {netflow: 1.0, hostlog: 0.8, pmu: 0.9, relay: 0.6}往下游走。科研代码和可用系统的距离通常在两件事上增量更新和可解释性。增量更新指的是别每次都全量重训——用joblib存下scaler、pca、clf新数据来的时候只做transform和predict模型按周重训一次就够了。可解释性指的是给每条告警附上哪几路证据贡献最大实现上可以取随机森林的feature_importances_按来源聚合或者在 D-S 合成时记录每一路的m_attack与最终结果的差值作为归因分数输出。我做这类融合项目有个固定习惯每次改完融合策略先不跑全量而是固定抽 200 条样本手工核对合成结果逐条看三路原始证据和最终判定把判断明显不合理的记下来。这一步花不了半小时但能拦住绝大多数指标涨了、逻辑错了的情况——尤其是加权权重这类参数很容易调出一组在测试集上好看、在业务上荒谬的值。从那以后我每次调融合参数都强制走一遍这个小样本核对比盯着 AUC 曲线靠谱得多。希望这套代码和上面这些经验能帮你在自己的场景里少绕几个弯。本文还有配套的精品资源点击获取