简介：面向电力系统网络安全方向的科研人员与研究生，这份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

这套划分的关键在fusion.py被单独拎出来。很多同类代码把融合逻辑硬塞在训练函数里，改一种融合策略就得动模型代码，改一次跑一次，最后谁也不敢动。把融合拆成独立模块之后，融合策略和分类器变成两个正交维度，可以自由组合成「特征级 + 随机森林」「决策级 + D-S + XGBoost」这样的矩阵做消融。我在做对比实验时最看重的就是这种可正交拆解的结构，它决定了你后面能不能在论文或报告里讲清楚"提升到底来自融合还是来自换模型"。

2.2 依赖清单与运行环境

依赖不多，都是数据处理和机器学习的老面孔，版本上不追求最新，追求能复现：

# requirements.txt 的核心部分 python==3.9 # 3.8~3.11 均可，3.12 上部分科学计算包编译会折腾 numpy==1.24.3 pandas==2.0.3 scikit-learn==1.3.0 imbalanced-learn==0.11.0 # SMOTE 等过采样，处理攻击样本稀疏 pyyaml==6.0 # 读 settings.yaml matplotlib==3.7.2 # 画混淆矩阵与融合前后对比 joblib==1.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，如果本机已经装了旧版 sklearn，pip 会静默降级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(columns=COL_MAPS[source_name]) df["source"] = source_name # 时间解析：数值型按单位换算，字符串型直接 to_datetime if ts_unit is None: df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce") else: df["timestamp"] = pd.to_datetime(df["timestamp"], unit=ts_unit, errors="coerce") # 解析失败的行直接丢，留着会在后续 resample 里变成幽灵分组 return df.dropna(subset=["timestamp"]).sort_values("timestamp").reset_index(drop=True)

ts_unit参数看着不起眼，实际上是最容易翻车的地方。pd.to_datetime带unit="s"时，如果传进来的是毫秒值，解析结果会跑到 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(cnt=1.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(), axis=1) return wide.sort_index()

这段的核心是groupby("src").resample(window)的组合。注意顺序：先按实体分组，再在组内做时间重采样，这样每台设备、每个源地址都拥有自己独立且连续的时间轴，不会出现 A 设备的数据被聚合到 B 设备窗口里的情况。用反了（先 resample 再 groupby）就会得到一张实体被打散的表，而且不报错，是最阴的那种 bug。

对齐之后，三路数据在时间上依然是错开的——网络侧每分钟都有数据，主机日志可能连续几个小时空窗。pd.concat(axis=1)按索引对齐，空窗处自动补 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_direction="both") 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_size=test_size, stratify=y, random_state=seed) scaler = StandardScaler().fit(X_tr) # 只在训练集上 fit return (scaler.transform(X_tr), scaler.transform(X_te), y_tr, y_te, scaler)

stratify=y保证了训练集和测试集的正负样本比例一致，样本量小的时候不加这个参数，很可能出现测试集里攻击样本只有个位数的情况，评估出来的 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_periods=1).mean() out[f"{c}__diff"] = out[c].diff().fillna(0.0) return out

min_periods=1是必须的，否则序列开头 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_components=var_ratio, svd_solver="full", random_state=42) return pca.fit_transform(X_tr), pca.transform(X_te), pca

svd_solver="full"在特征数几百这个量级完全够用，别急着上随机化版本，随机化求解会引入额外的随机性，做对比实验时会干扰你判断提升到底来自哪里。另外提醒一句：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 result

alpha这个参数是整套融合逻辑里最值得调的一个。它表达的是"我有多信这个传感器"。物理侧数据质量高可以给 0.95，日志侧噪声大给 0.7，剩下 30% 质量留给不确定集——这样即使某一路给出了极端错误的判断，在合成时也会因为不确定质量的存在而被稀释，不至于一票否决。

conflict的处理也很关键。当两路传感器给出完全相反的证据（一个说攻击概率接近 1，一个说接近 0），1 - 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_estimators=300, # 树数量，300 在该数据规模下收益已趋平 max_depth=None, class_weight="balanced", # 双重保险，与 SMOTE 二选一即可 n_jobs=-1, random_state=seed) 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输出的对比表大致长这样：

这张表能说明的东西比自己脑补多得多。单源量测侧 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 曲线靠谱得多。希望这套代码和上面这些经验，能帮你在自己的场景里少绕几个弯。

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