Python多传感器融合电力系统网络入侵检测实战-平芜编程栈

简介：面向电力系统网络安全方向的多源数据融合实践资料，适合具备Python基础、熟悉数据处理与机器学习的研究生及工业控制、物联网安全技术人员，用于提升网络入侵检测中的异常识别与响应能力。内容围绕数据预处理、特征提取、融合算法设计与入侵检测模型构建展开，涵盖从异构传感器数据接入、清洗到融合特征构造、分类模型训练与异常判别的完整链路，重点说明如何融合多源数据以增强检测精度和系统鲁棒性。压缩包共1个docx文档，约77KB，以文档形式集中呈现方法说明与Python代码实现，便于按模块对照阅读。已有32人学习，可作为关键基础设施安全监控中多源协同检测的复现参考；读者可结合实际数据集或仿真环境测试、调参，并进一步优化融合策略。

1. 凌晨的误报告警：多传感器融合怎么落到电力系统网络入侵检测上

某配电站凌晨两点连续弹出十几条流量告警，值班人员排查到天亮，最后发现大半是主站批量对时带来的正常抖动。这类场景在电力系统网络入侵检测里太常见了：单看网络流量的异常模型误报率高到没人愿意盯。多传感器融合要解决的就是这件事，把网络流量、主机日志、量测数据和保护事件放到同一条时间轴上互相印证，一个异常只有在多个源上同时露头才升级成告警。这份 Python 代码实现给了一条能跑通的多源数据融合流水线，从预处理对齐、特征提取、特征级与决策级融合一直到模型训练和评估，适合有 Python 基础、想把融合逻辑真正写进入侵检测流程的开发者。融合不是把几份 CSV 横向拼一下就完事，对齐口径、标签时序、评估方式，每一步都有坑。

2. 项目架构与技术选型：一条从多源采报到入侵告警的流水线

这个项目的骨架是一套离线批处理流水线，输入是四类异构的电力系统侧数据（网络流量聚合、主机侧日志、量测数据、保护与事件记录），输出是每条时间窗样本的入侵概率和告警标签。它不追求把模型做得多深，重点在于把「多源怎么合、在哪个层级合、合完怎么评」讲清楚，所以代码分层比较克制，每一层都能单独替换。

理解这套分层比记住函数名更重要。很多人拿到这类代码直接改模型，结果发现指标上不去，问题往往不在模型，而在上游的时间对齐和标签挂接。后面几章的代码都是围绕这个判断展开的。

2.1 目录结构与模块职责

# 项目目录（按模块职责切分，不按文件类型堆） power_ids_fusion/ ├── configs/ │ └── default.yaml # 数据路径、重采样粒度、窗口、模型超参 ├── data/ │ ├── raw/ # 各传感器源原始 CSV，一源一文件 │ ├── labels/ # 攻击事件起止时间表 │ └── processed/ # 对齐后的宽表，落盘便于复用 ├── src/ │ ├── preprocess.py # 时间对齐、缺失填补、标签挂接 │ ├── features.py # 分源特征提取、滑动统计、熵特征 │ ├── fusion.py # 特征级融合与决策级融合 │ ├── models.py # 单源基线模型工厂 │ ├── evaluate.py # 指标、阈值搜索、混淆矩阵 │ └── run_pipeline.py # 命令行入口，串起全流程 ├── notebooks/ │ └── explore_fusion.ipynb # 看数据分布、看融合前后对比 └── requirements.txt

preprocess.py只管一件事：把不同采样率、不同时间精度的源变成同一张以秒为索引的宽表，顺手把标签挂上去。features.py按源分别造特征，不跨源，保证后面对比单源和融合时口径一致。fusion.py是核心，特征级融合和决策级融合各写一套，方便你二选一时直接切。models.py只负责按配置名返回模型实例，不涉及训练流程。

这种切法的好处是替换成本低。比如你想把随机森林换成梯度提升树，改models.py一个映射就行；想换数据源，只要在preprocess.py的源配置里加一项、在features.py加一个提取函数，其余模块不用动。坏处是文件数偏多，小项目里会显得啰嗦，但对二次开发来说这点啰嗦是值得的——我踩过把所有逻辑塞一个脚本的坑，改到第三版就看不懂自己写了什么。

2.2 技术栈与依赖取舍

组件	在流水线里的位置	选它的理由
pandas / numpy	数据对齐、滑动窗口、宽表合并	时间序列重采样和`merge_asof`是现成的，自己写容易出错
scikit-learn	标准化、特征选择、分类器、交叉验证	Pipeline 能把预处理和模型绑在一起，防泄漏
scipy	频域特征、统计检验	量测数据做 FFT 提主频能量很轻量
pyyaml	配置管理	超参不进代码，做对比实验时改 yaml 就行
joblib	模型和 scaler 落盘	小模型序列化够用，不引入额外依赖
matplotlib	混淆矩阵、PR 曲线	汇报和自查都要图

项目没有把深度学习框架放进主依赖，这不是做不了，而是刻意的取舍。电力系统侧的攻击样本往往只有几百到几千条，上来就堆 CNN/LSTM 很容易过拟合，而且训练不动、调参成本高。常见做法是先用树模型和逻辑回归把基线跑出来，确认融合真的带来增益，再考虑上时序模型。代码里给模型留了统一的fit/predict_proba接口，后面想接 PyTorch 的序列模型，包一层适配器就能塞进决策级融合。

依赖这块有个血泪经验：pandas 的大版本变动在时间索引和merge_asof的容差处理上有过行为差异，requirements.txt里最好锁主版本，别让「昨天还能跑，今天就报错」这种事发生在复现实验的时候。

2.3 数据流与运行入口

整条链路是单向的：原始 CSV → 统一时间轴 → 缺失填补 → 挂标签 → 分源特征 → 融合 → 训练评估 → 产物落盘。入口脚本把每一步的参数都从配置里读，避免命令行参数泛滥。

# src/run_pipeline.py import argparse from pathlib import Path import yaml from preprocess import load_sources, unify_timebase, fill_missing, attach_labels, load_labels from features import build_features from fusion import FeatureFusion, DecisionFusion from models import make_model from evaluate import evaluate_and_report def main(cfg_path: str): cfg = yaml.safe_load(Path(cfg_path).read_text(encoding="utf-8")) # 1) 读入四类源，key 为源名，value 为 DataFrame frames = load_sources(cfg["data"]["raw_dir"]) # 2) 统一到秒级时间轴，base_freq 决定告警的决策粒度 aligned = unify_timebase(frames, base_freq=cfg["preprocess"]["base_freq"]) # 3) 缺失填补，同时生成缺失指示列（缺失本身就是一种信号） wide = fill_missing(aligned, max_gap=cfg["preprocess"]["max_gap"]) # 4) 把攻击事件的起止时间挂到宽表上，未命中区间一律标 normal labeled = attach_labels(wide, load_labels(cfg["data"]["label_file"]), tol=cfg["preprocess"]["label_tol"]) # 5) 分源特征 + 融合，feature_level 与 decision_level 二选一 X_src, y = build_features(labeled, cfg["features"]) model = make_model(cfg["model"]["name"], **cfg["model"]["params"]) if cfg["fusion"]["level"] == "feature": pipe = FeatureFusion(model, cfg["features"]["select_topk"]) result = evaluate_and_report(pipe, X_src, y, cfg) else: base = {name: make_model(cfg["model"]["name"], **cfg["model"]["params"]) for name in X_src} pipe = DecisionFusion(base, weights=cfg["fusion"]["weights"]) result = evaluate_and_report(pipe, X_src, y, cfg) # 6) 产物落盘：宽表、模型、指标 Path(cfg["output"]["dir"]).mkdir(parents=True, exist_ok=True) wide.to_parquet(Path(cfg["output"]["dir"]) / "aligned.parquet") result.save(cfg["output"]["dir"]) if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--config", default="configs/default.yaml") main(ap.parse_args())

这段入口做三件事：按配置串流程、按fusion.level走不同融合分支、把中间产物和指标落盘。几个参数值得单独说。base_freq是整条链路的地基，设成1s表示模型每秒出一个判断，设成5s会降低样本量和噪声，但也可能把短时攻击窗口糊掉，数据量小时我一般先用 1s 看效果再调。max_gap控制前向填充的最大跨度，超过这个跨度的空洞不填，交给后续的中位数兜底。label_tol是标签挂接的时间容差，攻击事件的起止时间戳精度通常和传感器数据不一致，容差太小会导致大量攻击窗口被标成正常。

X_src是个字典而不是矩阵，这一点是刻意的：决策级融合需要各源独立的特征矩阵，特征级融合在内部再拼。如果一上来就拼成一个大矩阵，后面想拆开做单源对比就得重跑一遍特征工程。

3. 预处理与时间对齐：把四类异构传感器数据拼到一条时间轴上

预处理章节的代码量不大，但决定了后面所有指标可不可信。四类源的采样率差得很远：网络流量聚合可能秒级，主机日志是事件驱动、稀疏且时间戳精度只到秒，量测数据往往几十毫秒一条，保护事件更是偶发。直接concat的结果是一张稀疏得离谱的表，或者时间戳对不齐导致行数骤减。

判断预处理做没做对，有一个很朴素的检查方法：打印每个源的行数、时间跨度，以及对齐后宽表的行数。如果宽表行数远小于任意一个源的行数，说明某一步的索引合并出了问题，别急着往下跑模型。

3.1 统一时间基准与重采样

# src/preprocess.py（片段一：时间轴统一） import pandas as pd import numpy as np # 每个源的聚合语义不同，不能用一套规则硬套 SENSOR_RULES = { "netflow": {"how": "sum"}, # 流量计数类，累加才有意义 "host_log": {"how": "sum"}, # 登录失败次数等计数 "pmu": {"how": "mean"}, # 量测值取窗口均值 "soe": {"how": "max"}, # 保护事件用 max 保留"是否发生" } def unify_timebase(frames: dict, base_freq: str = "1s") -> dict: """把各源重采样到统一基准频率，返回同频的 dict。""" out = {} for name, df in frames.items(): df = df.copy() # 关键：统一成 UTC 且精度到秒，避免时区和纳秒尾数导致对齐失败 df["ts"] = pd.to_datetime(df["ts"], utc=True).dt.floor("s") df = df.drop_duplicates(subset="ts", keep="last").set_index("ts") num_cols = df.select_dtypes(include=[np.number]).columns.tolist() other_cols = [c for c in df.columns if c not in num_cols] agg = SENSOR_RULES[name]["how"] res = df[num_cols].resample(base_freq).agg(agg) # 非数值列（设备号、事件类型）取窗口内最后一个 if other_cols: res = res.join(df[other_cols].resample(base_freq).last()) out[name] = res return out

这段代码解决的是「不同频率怎么落到同一格」的问题。pd.to_datetime(..., utc=True)和.dt.floor("s")是两步保险：前者消除时区差异，后者把微秒、纳秒的尾数砍掉，否则两个源的时间戳看起来一样、实际相差几十微秒，merge_asof的容差判断会失效。SENSOR_RULES里给每个源单独配聚合方式，是因为计数类字段用均值会把「一次爆发式扫描」稀释成「持续小流量」，而量测类字段用 sum 又会把物理量叠成没有意义的数，这个映射关系弄反了，模型学到的就是假的规律。

resample之后必然会引入空行——原始数据在某些秒上本来就没有记录。这是正常现象，不是 bug，下一节的缺失填补就是给这些空行收尾的。drop_duplicates保留最后一条，是防止采集端重复上报同一条记录，不去重的话计数类特征会被系统性放大。

3.2 缺失值、异常值与标签挂接

# src/preprocess.py（片段二：填补与标签） def fill_missing(frames: dict, max_gap: str = "10s") -> pd.DataFrame: """宽表合并 + 缺失填补 + 缺失指示列。""" wide = pd.concat( [df.add_prefix(f"{name}__") for name, df in frames.items()], axis=1, join="outer" ).sort_index() num_cols = wide.select_dtypes(include=[np.number]).columns # 缺失本身携带信息（比如日志断流可能意味着主机被拿下），先记下来 miss_flags = wide[num_cols].isna().astype(np.int8).add_suffix("__isna") # 前向填充有上限，跨度太大的空洞交给中位数 limit = max(1, int(pd.Timedelta(max_gap).total_seconds())) filled = wide[num_cols].ffill(limit=limit) filled = filled.fillna(filled.median(numeric_only=True)) out = wide.copy() out[num_cols] = filled return pd.concat([out, miss_flags], axis=1) def attach_labels(wide: pd.DataFrame, labels: pd.DataFrame, tol: str = "5s") -> pd.DataFrame: """把攻击事件起止时间映射到样本上，direction=backward 表示取最近一条。""" wide = wide.sort_index() labels = labels.copy() labels["ts"] = pd.to_datetime(labels["ts"], utc=True).dt.floor("s") labels = labels.sort_values("ts") merged = pd.merge_asof( wide.reset_index().rename(columns={"index": "ts"}), labels[["ts", "label"]], on="ts", direction="backward", tolerance=pd.Timedelta(tol), ) # 攻击窗口外的样本标 normal，这是二分类的默认口径 merged["label"] = merged["label"].fillna("normal") return merged.set_index("ts").sort_index()

fill_missing里最容易被忽略的是miss_flags。很多人只做填充，把「这个源在那一秒断供」这一事实抹掉了，可断供在安全场景里往往是强信号——主机日志突然没了，可能是采集进程被杀，也可能是攻击者清日志。把缺失转成 0/1 指示列喂给模型，等于把这类信号显式交出去，代价只是特征维度翻倍，值。

ffill(limit=...)的limit单位是行数不是时间，重采样到秒级之后行数约等于秒数，所以这里直接用total_seconds()换算，换成 5s 粒度的配置时记得同步改这段，否则填充跨度会被放大五倍。

attach_labels用merge_asof而不是区间判断，是因为标签表通常只记录攻击开始时间，用「向后找最近一条」的规则能把后续窗口都归到这次攻击上。tolerance设小了，标签表里的时间戳和传感器时间戳差一秒就匹配不上；设大了，攻击结束后一大段正常流量会被误标成攻击。我一般从数据里抽几个攻击事件，手工比对一下前后窗口，再定这个值。

3.3 数据级融合与标准化边界

# src/fusion.py（片段一：特征级融合管线） from sklearn.pipeline import Pipeline from sklearn.preprocessing import RobustScaler from sklearn.feature_selection import VarianceThreshold, SelectKBest, f_classif from sklearn.impute import SimpleImputer class FeatureFusion: """特征级融合：各源特征先拼成一个大矩阵，再统一进模型。""" def __init__(self, model, topk: int = 60): self.pipe = Pipeline([ ("imputer", SimpleImputer(strategy="median")), # 流量和量测特征都是长尾分布，用分位数缩放比 z-score 稳 ("scaler", RobustScaler()), ("var", VarianceThreshold(1e-6)), ("select", SelectKBest(f_classif, k=topk)), ("clf", model), ]) def fit(self, X_src: dict, y): # 只在这里做横向拼接，保证切分逻辑在外面控制 X = pd.concat(X_src, axis=1) self.pipe.fit(X, y) return self def predict_proba(self, X_src: dict): X = pd.concat(X_src, axis=1) return self.pipe.predict_proba(X)[:, 1]

数据级融合（early fusion）说白了就是把所有特征拼成一个大矩阵，交给一个模型学。这套代码把它包进Pipeline，目的不是好看，是防泄漏。标准化、方差过滤、SelectKBest全都必须只在训练折上fit，如果在切分之前对整个数据集做一遍，测试集的信息就漏进训练过程，交叉验证分数会漂亮得离谱，上线直接打回原形——这个坑我在不止一个项目里见过。

RobustScaler换掉StandardScaler的理由很实际：网络流量的字节数、连接时长都是重尾分布，几万个正常样本里夹一个超大值，均值和标准差会被拉偏，z-score 之后正常样本全挤在零附近，模型分不开。用中位数和四分位距缩放，异常值的影响被压住。VarianceThreshold放在缩放之后，是为了先干掉那些常数列——某些源在整段时间里没变化，这类特征对分类没有任何贡献，留着只会拖慢训练。

topk参数控制保留多少特征。源越多，拼出来的维度越大，四类源全展开轻松破千维，样本量却只有几千条，这时候不做筛选，树模型会记住噪声。我一般先把topk设大一点看特征重要性排序，再往下砍，而不是一上来就设个很小的值。

4. 融合策略与检测模型：特征级与决策级两条路的具体写法

融合层级的选择不是学术问题，是工程问题。特征级融合实现简单、能捕捉跨源的组合模式，但对缺失和量纲敏感，一个源质量差会拖累整体；决策级融合各源独立建模、互不干扰，某个源掉线只影响它自己的那票，代价是丢了跨源交互信息。

比较务实的做法是两条都跑一遍，看哪个在你的数据上更稳。代码里两条路共用同一套特征和同一个评估函数，切换只改配置。

4.1 分源特征提取

# src/features.py import numpy as np import pandas as pd def rolling_stats(df: pd.DataFrame, cols, windows=(10, 60), stats=("mean", "std", "max")) -> pd.DataFrame: """多窗口滑动统计，捕捉短时突变和长时基线。""" parts = [] for w in windows: r = df[cols].rolling(w, min_periods=max(2, w // 5)) for s in stats: part = getattr(r, s)() part.columns = [f"{c}__w{w}_{s}" for c in cols] parts.append(part) return pd.concat(parts, axis=1) def shannon_entropy(series: pd.Series, bins: int = 16) -> float: """分箱熵，用来刻画流量分布在时间窗内是集中还是发散。""" vals = series.dropna().to_numpy() if vals.size == 0: return 0.0 hist, _ = np.histogram(vals, bins=bins) p = hist / max(hist.sum(), 1) p = p[p > 0] return float(-(p * np.log2(p)).sum()) def build_features(labeled: pd.DataFrame, cfg: dict): """按源分别造特征，返回 {源名: 特征矩阵} 和标签。""" y = (labeled["label"] != "normal").astype(int).to_numpy() X_src = {} for src, cols in cfg["source_columns"].items(): cols = [c for c in cols if c in labeled.columns] feats = rolling_stats(labeled, cols, windows=tuple(cfg["windows"]), stats=tuple(cfg["stats"])) # 每个源追加一列分布熵，作为该源在窗口内的"发散度" feats[f"{src}__entropy"] = ( labeled[cols[0]].rolling(cfg["entropy_window"]) .apply(lambda s: shannon_entropy(s), raw=False) ) X_src[src] = feats.reset_index(drop=True) return X_src, y

rolling_stats用两个窗口（比如 10 秒和 60 秒）分别刻画短时突变和长时基线。这个设计有明确目的：端口扫描这类攻击在 10 秒窗口里表现为连接数暴涨，在 60 秒窗口里可能一点都不显眼；而慢速横向移动正好反过来。单窗口只能抓住一类，多窗口让模型自己挑。

min_periods设成w // 5是为了序列开头不产生大片 NaN。如果设成w，前 60 行全是空值，样本量本来就小，这一下砍掉一截很肉疼；设成 1 又会让开头的统计量极不稳定，噪声很大。

shannon_entropy按源各算一列，原因是电流、流量这类连续量的「分布形状」比「绝对值」更能反映异常。正常运行时量测值集中在窄区间，熵低；一旦出现异常注入或者设备异常，分布被拉散，熵会跳。这列用rolling().apply()计算会慢，数据量大时建议改成 numpy 滑窗或者按批算，代码里保留这种写法是为了可读性。

4.2 决策级融合：加权投票与元学习器

# src/fusion.py（片段二：决策级融合） import numpy as np class DecisionFusion: """决策级融合：每个源独立训练一个子模型，输出概率后再融合。""" def __init__(self, base_models: dict, weights: dict = None, meta=None): self.base_models = base_models # {源名: 未训练模型} self.weights = weights # {源名: 权重}，None 表示等权 self.meta = meta # 可选：二层学习器 def _stack(self, X_src: dict) -> np.ndarray: # 列顺序必须与 base_models 的插入顺序一致，否则权重会错位 proba = [self.base_models[name].predict_proba(X_src[name])[:, 1] for name in self.base_models] return np.column_stack(proba) def fit(self, X_src: dict, y): for name, model in self.base_models.items(): model.fit(X_src[name], y) if self.meta is not None: # 元学习器吃的是各源的概率输出，不是原始特征 self.meta.fit(self._stack(X_src), y) return self def predict_proba(self, X_src: dict) -> np.ndarray: Z = self._stack(X_src) if self.meta is not None: return self.meta.predict_proba(Z)[:, 1] if self.weights: w = np.array([self.weights[n] for n in self.base_models], dtype=float) else: w = np.ones(Z.shape[1]) / Z.shape[1] w = w / w.sum() # 归一化，避免权重和不为 1 导致概率越界 return Z @ w

决策级融合的核心是「每个源说一句话，再按可信度加权」。_stack用字典的插入顺序决定列顺序，这是代码里唯一一处隐式依赖，写的时候务必保证weights的 key 和base_models的 key 完全一致，我见过因为顺序错位导致「权重最高的源实际投给了另一个源」的事故，指标看着还行，逻辑全错。

权重怎么定是个关键选择。等权是最稳的起点，至少不会因为调参引入偏差；按单源验证集上的 F1 或 AUC 定权，比等权通常有小幅提升；如果非要走元学习器（stacking），一定要用嵌套交叉验证，把元学习器的训练放在外层训练折内部，否则它会把测试集的信息吃进去，分数虚高。

meta这一路在样本量少的时候不建议开。二层模型的输入只有四列概率，参数量却不少，几百条攻击样本根本喂不饱，最后大概率不如等权。

4.3 分类模型训练与评估口径

# src/evaluate.py import numpy as np from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import (precision_recall_curve, average_precision_score, f1_score, confusion_matrix) def pick_threshold(y_true, proba, max_fpr: float = 0.01): """在验证集上按 FPR 上限挑阈值，返回 (阈值, 对应 F1)。""" prec, rec, thr = precision_recall_curve(y_true, proba) best = (0.5, 0.0) for p, r, t in zip(prec[:-1], rec[:-1], thr): pred = (proba >= t).astype(int) tn, fp, fn, tp = confusion_matrix(y_true, pred, labels=[0, 1]).ravel() fpr = fp / max(fp + tn, 1) if fpr <= max_fpr: f1 = 2 * p * r / max(p + r, 1e-9) if f1 > best[1]: best = (float(t), float(f1)) return best def evaluate_and_report(pipe, X_src, y, cfg): """时序切分 + PR-AUC + 固定 FPR 下的阈值选择。""" tscv = TimeSeriesSplit(n_splits=cfg["eval"]["n_splits"]) scores, pr_aucs = [], [] for tr, te in tscv.split(y): X_tr = {k: v.iloc[tr].reset_index(drop=True) for k, v in X_src.items()} X_te = {k: v.iloc[te].reset_index(drop=True) for k, v in X_src.items()} pipe.fit(X_tr, y[tr]) proba = pipe.predict_proba(X_te) pr_aucs.append(average_precision_score(y[te], proba)) thr, f1 = pick_threshold(y[te], proba, max_fpr=cfg["eval"]["max_fpr"]) scores.append((thr, f1)) return {"pr_auc": float(np.mean(pr_aucs)), "f1_at_fpr": float(np.mean([s[1] for s in scores])), "threshold": float(np.mean([s[0] for s in scores]))}

评估这块有两处必须说清楚。第一是切分方式，TimeSeriesSplit保证训练集永远在测试集之前，模拟的是「用历史数据预测未来攻击」的真实场景。用普通KFold随机打乱，同一个攻击会话的相邻窗口会被分到训练和测试两边，模型等于抄答案，指标能高出一大截，上线就现原形。如果数据里攻击是按会话聚集的，更严的做法是按会话 ID 做GroupKFold，一次攻击的所有窗口要么全在训练集，要么全在测试集。

第二是指标选择。入侵检测是典型的类别不平衡问题，正常样本占九成以上，accuracy没有任何参考价值。代码里用 PR-AUC（average_precision_score）看整体排序能力，再用pick_threshold在 FPR 上限内挑一个召回最高的阈值——业务上通常能接受千分之一的误报率，但不能接受漏掉一次真实攻击，这个约束比「F1 最大」更贴近实际。max_fpr这个参数没有标准答案，跟你的值班人力直接相关，误报多了没人看，设得太严又会漏。

5. 常见问题排查：融合流水线里最容易翻车的五处

这套代码我在不同数据上跑过几轮，出问题的位置高度集中。下面五条都是实际撞过的，按「现象 → 原因 → 解决」写，遇到对得上的直接照着改。

5.1 融合后样本数骤减，攻击标签几乎全丢

现象：对齐前每个源都有几万行，fill_missing出来的宽表只剩几千行，attach_labels之后攻击样本只剩个位数。

原因：九成是时间戳口径不一致。有的源是本地时间字符串、有的是带时区的时间戳、有的精度到毫秒，pd.to_datetime不加utc=True会按本地时区解释，两边差八小时，merge_asof的容差再大也匹配不上。另一种情况是direction设成forward，标签只挂在攻击结束之后，前面的窗口全成正常样本。

解决：在unify_timebase里统一utc=True加.dt.floor("s")，跑之前先打印每个源的最小/最大时间戳和重叠区间；标签方向固定用backward，容差从 5s 起调，抽三五个攻击事件手工核对。核对这一步别省，它的成本远低于后面训练出一堆废模型的成本。

5.2 准确率 0.98，实际一条攻击都没抓到

现象：模型报告accuracy高得离谱，混淆矩阵一看，攻击类全部预测成正常。

原因：正常/攻击比例可能是 97:3，模型学会「全猜正常」就能拿到 97% 的准确率。再叠加一个默认 0.5 的阈值，攻击概率普遍在 0.3 左右，全被压下去。

解决：换成 PR-AUC 和固定 FPR 下的召回率作为主指标，阈值用验证集搜出来而不是拍 0.5。训练时给分类器加class_weight="balanced"，或者在决策级融合里对少数类概率做校准。指标口径不换，后面所有优化都是在优化一个错误的目标。

5.3 交叉验证分数漂亮，换个时间段就崩

现象：五折交叉验证 F1 稳定在 0.9 以上，换一个月的数据掉到 0.5。

原因：预处理在切分之前做了。RobustScaler的fit、SelectKBest的fit只要碰到全量数据，测试集的分布信息就漏进训练。另一个隐蔽来源是特征里的滑动窗口——窗口跨越了训练集和测试集的边界，等于提前看到了未来。

解决：所有需要fit的步骤塞进Pipeline，保证只在训练折上拟合；切分按时间走，TimeSeriesSplit或者手动切一个时间点，测试集完全不参与任何拟合；滑动窗口的特征在切分点前后留一段 buffer，把跨越边界的窗口丢弃。缓冲区长度取最大窗口长度就行，代价是损失几十行样本。

5.4 决策级融合的权重调得越好，上线越差

现象：在测试集上网格搜索权重，融合后 F1 比单源高 8 个点，换数据后反而不如最好的单源模型。

原因：权重是在测试集上调的，等于用测试集做了超参搜索，那几个数就是过拟合到这批数据上的。融合模型的参数量比单源多，过拟合空间也更大。

解决：权重只在训练折内部用交叉验证确定，测试集全程不参与任何选择；样本量小的时候直接用等权，简单但稳；实在要调，用嵌套验证——外层切训练/测试，内层在训练集上调权重。判断融合是否真的有效，得看「同一套权重在外层测试集上的表现」，而不是「最优权重下的表现」。

5.5 加了新数据源，指标反而下降

现象：从三源扩到四源，特征维度涨了三成，PR-AUC 掉了一个点。

原因：新源的缺失率高，前向填充之后大片伪造出来的值被当成真实观测；或者新源的时间戳整体偏移，融合后大部分行是空值填充的，模型学到的规律来自填充而非真实数据。维度上涨也让树模型的分裂点更难选。

解决：先单独把这个新源训练一遍，看它的单源 AUC 有没有超过 0.6，低于这个数说明它本身信息量不足，融进去只会添乱；再看缺失率，某个源缺失超过 30% 时，要么缩短填充跨度，要么给它单独的缺失指示列并且降低权重。特征是加不完的，加之前先证明加的值。

6. 二次开发：换成自己的传感器数据与在线滑窗推理

换成自己的数据，改动集中在三个地方，顺序别乱。第一，在preprocess.py的SENSOR_RULES里登记新源，写清楚它的聚合语义是 sum 还是 mean，这一条决定了重采样之后数值还代不代表原来的物理意义。第二，在features.py的cfg["source_columns"]里列出该源要参与特征构造的列，顺手加一个源级别的分布熵。第三，决定这个源走特征级还是决策级——如果它缺失率高、可信度存疑，走决策级给它一个独立的小权重，别让它污染整张特征矩阵。

从离线批处理挪到在线推理，关键是维护一个定长滑窗缓冲。下面这段是最小实现，塞在任何流式框架的update回调里就能用。

# src/online.py from collections import deque import numpy as np import pandas as pd class OnlineFusionDetector: def __init__(self, model, window=60, threshold=0.5, warmup=20): self.model = model # 已训练好、含预处理的 Pipeline self.buf = deque(maxlen=window) # 定长滑窗，自动淘汰旧样本 self.threshold = threshold self.warmup = warmup # 不足 warmup 条时不出结果 def update(self, row: dict): self.buf.append(row) if len(self.buf) < self.warmup: return None df = pd.DataFrame(list(self.buf)) # 在线场景只有单源特征矩阵，走特征级融合那条路 proba = self.model.predict_proba({"stream": df})[0] return {"score": float(proba), "alarm": bool(proba >= self.threshold)}

window要和训练时的最大窗口对齐，训练用 60 秒窗口、线上只缓冲 30 条，特征分布直接错位，模型输出的概率没有任何意义，这种错位排查起来像玄学，因为程序不报错、只是结果不对。warmup是给滑动统计预留的预热期，缓冲区没填满之前统计量不稳，宁可不输出也别送假告警。

验证改造是否有效，别只看整体指标。我通常跑一组增量实验：单源基线、两源融合、全源融合、全源去掉最差的那个，每次只动一个变量，把 PR-AUC 和固定 FPR 下的召回记成一张表。融合带来的提升如果在两个不同时间段上都成立，才算真的有效；只在一个时间段上好看，大概率是那批数据本身的特性。

实验组	参与融合的源	PR-AUC	FPR=1% 时召回	结论
A	仅网络流量	基线	基线	单源上限参照
B	流量 + 主机日志	高于 A	高于 A	跨层关联有增益
C	流量 + 日志 + 量测	与 B 接近	略高	量测贡献有限
D	C 去掉最差单源	与 C 接近	持平	验证冗余源