简介：面向电力系统网络安全研究与工业控制系统安全方向的技术资料，聚焦网络入侵场景下的多源数据融合检测方法，适合具备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.md

io_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 pyarrow

pyarrow是我额外建议装的，中间产物用 parquet 存比 CSV 快一个数量级，体积也小得多。整条链路在 8G 内存的机器上跑几十万条记录没有压力，真要上到千万级，再考虑把对齐那层换成按天分片处理，而不是一上来就上集群。

2.3 六层链路：每一层的输入、输出和失败点

整条链路可以拆成下面六层，每层都有明确的输入输出，出问题时能快速定位是哪一层断的：

采集接入层的核心是"不改原始文件"，所有转换都在内存里做，data/raw永远保持只读。这听起来是老生常谈，但真出过事故：有人在原始目录上做了一次就地清洗，后来想回滚发现原始数据已经没了，后悔药没地方买。

时间对齐层是整个项目的技术难点，也是最有价值的部分，第 3 章会展开。这里先记一句：对齐的基准一定要选采样最密的那一路，用稀疏的那一路当基准，等于把高频信息全部扔掉。

清洗层要注意的是"不要静默丢数据"。缺失值超过一定比例的记录应该被标记出来单独看，而不是dropna()一删了事——被删掉的那批记录里，很可能恰好就是攻击导致的采集中断。

特征提取层要防的是标签泄漏。滑动窗口如果跨过了攻击结束的时间点，窗口里的统计量就会混入"攻击后"的信息，模型在训练集上表现极好，上线后一塌糊涂。常见做法是在标签变化点前后各留一段隔离带，隔离带内的窗口直接丢弃。

融合层和判别层是策略问题不是代码问题，第 4 章细讲。这里只强调一点：融合权重和判定阈值必须是配置项，不能硬编码在函数里，否则每换一个场景就要改代码重新打包。

3. 预处理与特征工程：把三种采样率的数据对齐到同一张宽表

多源融合项目里，八成的功夫花在这一章。模型再花哨，喂进去的是错位的数据，出来的就是玄学结果。

3.1 三类数据源的字段画像与时间对齐

电力监控场景下能拿到的三类数据，采样节奏差别很大：

三类数据的共同点是都有时间戳，难点是精度和时区不一致。对齐的常见做法是以采样最密的那一路为基准，用 pandas 的merge_asof做就近匹配，并且用一个显式的容忍度参数卡死偏差范围。容忍度一般取最慢那一路采样周期的二分之一，超过就留空，而不是硬凑一个值过去。

import pandas as pd def align_sources(frames, ts_col="ts", tolerance="2s"): """ frames: dict，键为数据源名，值为带时间戳列的 DataFrame ts_col: 统一后的时间戳列名（已转成 datetime64） tolerance: 允许的最大时间偏差，超过则匹配为空 返回: 以采样最密的一路为基准的宽表 """ # 基准选行数最多的那一路，避免用稀疏源上采样造出假数据 base_name = max(frames, key=lambda k: len(frames[k])) base = frames[base_name].sort_values(ts_col).reset_index(drop=True) 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, on=ts_col, direction="nearest", tolerance=pd.Timedelta(tolerance), suffixes=("", f"_{name}"), ) return base

这段逻辑有三个参数值得说清。tolerance决定了多宽松地认为"这两条记录是同一时刻发生的"，设太大，会把几分钟前的日志贴到当前流量上，造成因果倒置；设太小，宽表里会大面积 NaN，模型学不到跨源关系。direction="nearest"相比默认的backward更适合事件型日志，因为日志写入往往比实际行为晚几十毫秒，只往前找会系统性地错配。suffixes必须显式给，否则两路数据里同名字段（比如都有status）会在无提示的情况下被覆盖，这类问题排查起来极其难受。

对齐完还要检查一件事：每路数据的匹配率。匹配率低于 60% 的数据源，要么是时间戳单位错了（秒当成毫秒），要么是时区没统一。这两个原因占了不对齐问题的绝大多数。

3.2 特征提取：时域统计、频域能量、行为窗口

对齐后的宽表是逐条的，但入侵检测真正关心的是"一段时间里发生了什么"，所以要切滑动窗口。窗口长度按业务定：网络侧的扫描行为通常持续十几秒，物理侧的异常量测可能只持续几百毫秒，一个窗口装不下两种节奏，通常要同时保留长短两套窗口，再在对齐层合并。

下面的函数对每个数值列产出时域统计和频域能量占比：

import numpy as np import pandas as pd def window_features(df, value_cols, window="10s", step="5s"): """ df: 对齐后的宽表，含 ts 列 value_cols: 需要抽特征的数值列 window: 窗口长度，应覆盖一次完整攻击行为 step: 滑动步长，一般取 window 的一半，保证重叠但不冗余 """ out = [] for start in pd.date_range(df["ts"].min(), df["ts"].max(), freq=step): 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(ddof=0) 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(drop=True)

std(ddof=0)用总体标准差，是为了避免小窗口下样本标准差被放大，导致特征分布在不同窗口长度间漂移。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_neighbors=5): """ clf: 任意实现了 fit/predict_proba 的分类器 k_neighbors: SMOTE 的近邻数，样本极少时要调小，否则报错 注意必须用 imblearn 的 Pipeline，sklearn 的 Pipeline 不会在 fit 时重采样 """ return ImbPipeline([ ("imputer", SimpleImputer(strategy="median")), # 中位数比均值抗异常值 ("scaler", StandardScaler()), # 树模型可省，线性/NN 必留 ("smote", SMOTE(random_state=42, k_neighbors=k_neighbors)), ("clf", clf), ])

这里的关键细节是必须使用imblearn.pipeline.Pipeline。sklearn 自带的 Pipeline 只在fit时调用各步的fit_transform，predict时会跳过重采样步骤，但它的交叉验证不会自动把重采样限制在训练折内——如果你手动在交叉验证前对整个数据集做 SMOTE，验证集里就会混入由训练样本合成出来的假样本，评估结果虚高十几个点是常事，这就是典型的黑匣子式自欺欺人。k_neighbors在少数类样本少于 6 条时会直接报错，这时要么改成RandomOverSampler，要么换成基于代价的class_weight="balanced"，别硬撑。

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(strategy="median")), ("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), ], remainder="drop") return Pipeline([ ("pre", pre), ("clf", HistGradientBoostingClassifier( max_depth=6, learning_rate=0.05, max_iter=300, class_weight="balanced", )), ])

分组做ColumnTransformer而不是一股脑StandardScaler，是因为网络侧特征（包数、字节数）和物理侧特征（电压、电流）量纲差好几个数量级，混在一起缩放会把小量纲特征压成噪声。max_depth=6和learning_rate=0.05是这类中小样本表格数据的稳妥起点，深度再往上加，训练集 AUC 会继续涨，验证集开始掉，就是过拟合的信号。class_weight="balanced"和外面的 SMOTE 二者留一个就行，同时用会让少数类被重复加权两次。

4.2 决策级融合：三路模型各自打分再按代价加权

决策级融合（晚期融合）让每一路数据训练一个独立的子模型，各自输出"是攻击"的概率，再按权重合成一个最终分数。它对单路数据缺失更宽容——某一路没有数据，就把它的权重临时降到 0，其他两路照常工作。代价是要训练和维护多个模型。

import numpy as np def weighted_fusion(prob_list, weights, eps=1e-6): """ prob_list: 各子模型输出的攻击概率，长度一致 weights: 与 prob_list 同长度的权重，按各路在验证集上的可靠性定 返回: 融合后的攻击概率 """ w = np.asarray(weights, dtype=float) 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_fp=1.0, cost_fn=10.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% 的准确率，这个数字毫无意义。真正要盯的是三个数：

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(..., utc=True)，再.dt.tz_convert("Asia/Shanghai")，全链路只用一个时区，永远不在业务代码里转。我曾经在这上面耗掉大半天，从数据的字段名查到内存，最后发现是时区。

现象二：读一个几百兆的 CSV 直接 MemoryError。原因是pd.read_csv默认会把整张表读进内存，而且字符串列会被推成 object 类型，内存占用翻好几倍。解决方法是显式指定dtype和usecols，把用不到的大字段直接排除；或者用chunksize分块处理，每块抽完特征就丢弃，中间结果落 parquet。别指望加内存条，加完你还是会写同样的代码。

现象三：SMOTE 在训练时报n_neighbors相关的错误。原因是少数类样本数少于k_neighbors + 1。电力场景下某些攻击类型可能只有三五条记录，这很常见。解决方法是按类别的样本量动态调k_neighbors，或者直接用class_weight="balanced"替代重采样。硬把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 算例，做电力系统优化、负荷预测、信号处理的都有，接口风格不太一样，但数据准备和评估那套思路是相通的，可以对照着看融合逻辑该怎么拆。

我现在每次动特征工程，都会强制自己先跑一遍完整回放，再看指标——因为吃过一次亏，某个"改完之后验证集涨了三个点"的提交，上线第一周把一整条线路的告警刷爆了，最后查出来是新加的一个频域特征在夜间低流量时段数值异常。从那以后我给自己立了个规矩：任何特征改动，必须先在回放数据上过一遍白天的和夜间的两个时段。希望这篇拆解能帮到你，少走几步弯路。

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