简介：围绕电力系统网络入侵检测中的多传感器融合方法，这份文档面向具备Python编程基础、从事电力系统安全或工业控制系统安全研究的科研人员与研究生，帮助其理解如何整合来自不同传感器的异构数据，提升异常行为识别精度与响应效率。内容覆盖数据预处理、特征提取、融合算法设计与入侵检测模型构建等关键环节，可用于潜在入侵的早期预警、算法复现以及融合策略优化等场景。压缩包共1个文件，为docx文档，大小约77KB，结构紧凑，便于查阅与整理。已有35人学习，适合作为多源数据协同检测的入门参考与课题实践素材，读者可据此梳理检测流程、把握分类模型设计要点，并在仿真环境或实际数据中进一步测试与调优。

1. 一路告警压不住的时候：多传感器融合在电力系统网络入侵检测里到底做什么

电力监控场景里有个老毛病：只看流量探针，一次规约异常可能只是主站正常的遥控下发；只看主机日志，一次集中登录可能只是运维在做例行巡检。告警刷屏、值班员麻木、真事件埋在噪声里，这是不少现场的血泪经验。多传感器融合落到电力系统的网络入侵检测上，说白了就是把流量统计、主机日志、设备事件、安全设备告警这几路异构数据按同一根时间轴对齐，再在特征层或决策层合成一个判断。这份 Python 代码实现给的是能跑的骨架：数据接入、时基对齐、特征拼接、决策级加权、评估与阈值调节，一条链走通。适合有 Python 基础、又要在工控安全或电力监控方向做落地的研究生和工程师，也适合拿它当自己项目的融合基线。

2. 把四路数据拧到一根时间轴上：重采样、缺失标记与标签对齐

融合模型效果差，十次里有六次不是算法问题，是数据没对齐。流量是按秒来的，主机日志是事件触发的，保护装置的动作记录可能几分钟才有一条，安全设备告警还带着自己的时区配置。直接按行 concat，等于把不同时刻发生的事强行说成同时发生，模型学到的全是噪声。这一章把「对齐」这件事拆成三步：先认清楚各路数据的时间口径，再定重采样网格，最后处理缺失和标签窗口。

2.1 四类数据源与它们各自的时间口径

在电力监控网络里，能拿到的融合输入通常是这四类，字段名各家不同，但语义差不多：

真正麻烦的是第四类。很多安全设备的告警时间戳记的是规则匹配完成的时刻，一条会话持续几十秒，告警可能落在会话结束之后。拿这个时间去对齐流量窗口，特征和标签就错位了。常见做法是优先用会话开始时间，拿不到就退回采集时间，并在特征表里单独留一列标记这个来源用的是哪种时间口径。

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"], utc=True) 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"], utc=True) # 用正则粗筛失败/拒绝类关键字，比上 NLP 模型省事，也够用 df["is_fail"] = df["msg"].str.contains( r"failed|denied|invalid", case=False, regex=True ).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，混用会让特征的物理含义变味。utc=True这个参数不要省，本地时区加上夏令时会让某些窗口凭空多出或少掉一小时的数据。如果你的数据源时标是带偏移的字符串，先统一转 UTC 再做任何聚合，顺序不能反。

2.3 缺失不是零：缺失标记与「距上次事件时间」

稀疏源重采样之后会出现大量 NaN。直接fillna(0)是最常见的做法，也是最容易埋雷的做法。深夜两点没有登录失败日志，和「这个窗口日志采集器掉线了」，在模型眼里都会变成log_fail_cnt=0，含义完全不同。我的处理方式是两个都保留：数值填 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(start=start, end=end, freq=WIN, tz="UTC") X = pd.DataFrame(index=idx) for name, df in parts.items(): sub = df.reindex(idx) # 观测标记：本窗口该来源是否有真实数据 X[f"{name}_seen"] = sub.notna().any(axis=1).astype(int) # 距上次事件的秒数，第一次出现前用一个大常数占位 has_evt = sub.notna().any(axis=1) last_seen = pd.Series( np.where(has_evt, idx.view("int64") // 10**9, np.nan), index=idx ).ffill() X[f"{name}_since_last_s"] = ( idx.view("int64") // 10**9 - last_seen ).fillna(86400).clip(upper=86400) # 数值列填 0，但 seen 列已经保留了「这是填充值」的信息 X = X.join(sub.fillna(0), how="left") # 标签按窗口右边界对齐，详见 2.4 X = X.join(label, how="left") return X

since_last_s用clip(upper=86400)是为了防止极端值把标准化拉偏。seen列在特征重要性里往往排得很靠前，不是因为它多有业务含义，而是因为它替模型区分了「真零」和「假零」。

2.4 标签窗口与特征窗口必须错开

这是最隐蔽的一个坑。假设标签定义是「窗口 [t, t+10s) 内是否发生入侵」，而特征用的是同一个窗口 [t, t+10s) 的统计量。攻击发生时流量必然突增，这个突增同时进了特征和标签，模型离线能刷到 0.99，上线直接崩。正确做法是把特征窗口整体前移：特征取 [t-W, t)，标签取 (t, t+H]，中间留一段缓冲，两边不重叠。

def make_label(event_ts: pd.Series, idx: pd.DatetimeIndex, horizon: str = "10s") -> pd.DataFrame: """事件时间戳 -> 窗口标签；只在 (t, t+horizon] 内标记为 1""" y = pd.Series(0, index=idx, dtype=int) for t in event_ts: # 找事件之后的第一个网格点，避免把事件当刻算进来 pos = idx.searchsorted(t, side="right") 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用side="right"保证事件不会被算到自己所在的窗口里。

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_estimators=n_estimators, max_depth=max_depth, class_weight="balanced", # 入侵样本少，先压住多数类 n_jobs=-1, random_state=42, )), ]) model = build_feature_level_model() model.fit(X_train, y_train) score = model.predict_proba(X_test)[:, 1] # 取「攻击」类概率

class_weight="balanced"是按类别频率反比给权重，参数不用手调，但对极端不平衡（正样本低于 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], dtype=float) w = w / w.sum() # 归一化，防止权重和不为 1 导致分数越界 stack = np.vstack([proba[k] for k in keys]) return (stack * w[:, None]).sum(axis=0) def soft_vote(proba: dict, thr: float = 0.5) -> np.ndarray: """等权软投票，作为加权融合的对照基线""" stack = np.vstack(list(proba.values())) return (stack.mean(axis=0) >= 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_iter=1000) lr.fit(Xtr, y_train) return lr.predict_proba(Xte)[:, 1], dict(zip(keys, lr.coef_[0]))

堆叠的系数能直接读出来当权重参考，但要注意它是在验证集上训的，最终评估还得留一段完全没碰过的时间段。系数出现负数也别慌，说明那一路在当前数据上起到了反向指示作用，通常是该路数据质量有问题，值得回头查采集链路。

4. 融合模型的训练与评估：类别不平衡下的可信指标

入侵检测的数据集天然不平衡，一个月的流量里真正的攻击窗口可能只有几十个。这个比例下，准确率是废指标，99% 的准确率可能意味着模型把所有样本都判成正常。评估体系搭不对，融合做得再花哨也证明不了价值。

4.1 采样与代价敏感怎么选

三种手段：过采样、欠采样、代价敏感。欠采样会丢掉大量正常样本的时间结构，对时序场景伤害大，一般不单独用。过采样里 SMOTE 这类插值方法在时序特征上要谨慎，它会在两个不相邻的时间点之间造样本，物理上不成立。我的默认组合是class_weight="balanced"加阈值调节，只有在正样本极少（比如少于 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是占位写法，实际算假阳性率要用混淆矩阵，别照抄。更稳的写法是遍历阈值直接数 FP：

def 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 best

max_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(..., utc=True)，并在特征表里留一列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[f"seen_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(ascending=False) report["top_corr"] = corr.head(3).round(3).to_dict() report["tz"] = str(X.index.tzinfo) return report

top_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, index=feature_df.index[::step])

6.2 阈值加迟滞，压住分数抖动

单阈值判定在边界附近必然抖。工程上的做法是双阈值迟滞：分数超过高位才置为告警，要降到低位以下才解除，中间区间保持上一状态。

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.state

min_hold这个参数别设太小，设成 1 等于没做迟滞。high和low的间距决定灵敏度，间距太窄照样抖，太宽会漏掉短时攻击，我一般在回放数据上把间距从 0.1 扫到 0.3，看「同一事件重复告警次数」这个指标落到多少。

还有一个习惯值得养成：把融合分数当排序用，而不是只当判定用。值班员先看分数最高的前 20 条，比看一个二值告警列表信息量大得多，尤其在阈值还没调准的初期。资源包里的 Python 脚本按数据接入、特征构造、融合、评估拆成了独立模块，样例数据的字段和文中列名对得上，照着第 2 章的列名改就能接自己的数据。从那以后我每接一个融合场景，都强制先把时区对齐、标签窗口重叠、按时间切分的基线分数这三件事跑一遍，再谈模型选型——顺序反了，后面全是白工。希望帮到你。

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