简介本资源面向电力系统安全、工业控制系统防护及物联网安全方向的科研人员与研究生聚焦多传感器融合技术在电力网络入侵检测中的落地应用解决异构数据协同感知难、攻击识别精度低、响应滞后等关键问题。压缩包为1个77KB的docx文档完整涵盖数据预处理、多源特征提取、融合算法设计如加权平均、D-S证据理论或卡尔曼滤波变体、入侵检测模型构建与Python代码实现细节含核心函数说明、参数配置逻辑及典型攻击场景测试结果分析。目前已有31人学习下载文档结构清晰按“问题建模—方法设计—代码实现—实验验证”递进展开附有可直接复现的模块化Python脚本注释与调用示例支持在仿真环境或实际SCADA数据上快速验证融合策略有效性并为后续优化提供可扩展接口与改进路径提示。1. 电力系统里“看不见的刀”当SCADA被悄悄篡改多传感器融合是唯一能听见心跳的听诊器你有没有试过——在某次微电网攻防演练中IDS入侵检测系统日志风平浪静防火墙策略纹丝不动但后台数据库里某台断路器的状态字段却在凌晨3:17被连续写入了57次“CLOSE→OPEN→CLOSE”而物理侧继电保护装置毫无响应这不是误报也不是丢包是攻击者用合法协议绕过边界防护在OT层“静默翻转”设备状态。这类攻击在IEC 61850/GOOSE通信场景下极难捕捉单靠流量分析或日志审计几乎失效。而这个Python项目干了一件很实在的事它不指望一个模型包打天下而是把SCADA遥信、PMU相量、网络探针NetFlow、工控协议解析器Modbus/TCP解包模块、甚至边缘网关的CPU温度与内存波动——五类异构传感器数据全拉进同一个时间对齐框架用动态加权证据融合D-S理论自适应置信度衰减生成“设备行为可信度分”再喂给轻量级XGBoost分类器做最终判决。实测在IEEE 14节点仿真环境中对重放攻击、命令注入、时序扰动三类典型OT层入侵F1-score从单源检测的0.62提升到0.89漏报率压到3.7%。它不是炫技的AI玩具是给调度员看的“第二双眼睛”——适合正在啃《IEC 62443》标准、手头有RTDS或OPAL-RT仿真平台、想把论文落到真实IED设备日志上的电力安全方向研究生也适合在能源集团安全部门做红蓝对抗的工程师拿来直接跑通再插进自己的SIEM管道。2. 项目结构与技术栈为什么选Python而不是MATLAB因为实时性要卡在毫秒级而MATLAB Runtime部署太重这个项目不是MATLAB全家桶式的一键仿真它从设计第一天就锚定“可嵌入边缘网关”的目标。所以整个架构是典型的三层松耦合数据采集层Sensor Ingestion、融合推理层Fusion Engine、决策输出层Alert Dispatcher。所有模块都用纯Python实现依赖项控制在12个以内核心是numpy、pandas、scikit-learn、pydantic做配置校验、concurrent.futures多线程流水线完全避开tensorflow/pytorch这种大块头——毕竟在ARM64边缘盒子上跑GPU模型是玄学而用XGBoost做10ms级推理才是工程现实。2.1 目录结构拒绝“src/main/python”式迷宫每个文件夹名就是它的战场multi_sensor_fusion_power/ ├── config/ # 全局配置传感器采样周期、时间对齐窗口、D-S证据权重初始值 │ ├── sensor_config.yaml # 每类传感器的IP、端口、协议类型、字段映射如pmu_phase_a: V_angle │ └── fusion_config.yaml # 融合策略参数置信度衰减系数α0.92、冲突阈值γ0.35、最小证据数k3 ├── data/ # 数据沙盒含3组实测样本含攻击注入标记 │ ├── normal_20231015/ # 正常运行72小时原始数据CSV格式带时间戳毫秒级精度 │ ├── attack_replay_20231016/ # 重放攻击注入数据含GOOSE帧篡改标记列 │ └── attack_inject_20231017/ # 命令注入数据含Modbus功能码异常标记 ├── ingestion/ # 采集层5个独立模块各守一摊 │ ├── scada_reader.py # 读取IEC 104主站日志提取遥信变位SOE事件 │ ├── pmu_parser.py # 解析IEEE C37.118格式PMU数据提取相角差、频率偏移 │ ├── netflow_collector.py # 抓取交换机镜像端口NetFlow v9统计协议分布与连接数突增 │ ├── modbus_analyzer.py # 深度解析Modbus/TCP PDU识别非常规功能码如0x11诊断、非法地址访问 │ └── edge_health.py # 读取网关/RTU的/sys/class/thermal/thermal_zone*/temp等Linux系统指标 ├── fusion/ # 核心融合层不是简单拼接是带“信任投票”的协同判断 │ ├── time_aligner.py # 关键用PTPv2时间戳对齐5路数据容忍±15ms偏差电力系统允许范围 │ ├── evidence_builder.py # 将每路数据转化为D-S基本概率分配BPA如SCADA变位→attack0.7, normal0.3 │ ├── ds_fuser.py # 实现改进Dempster-Shafer合成规则加入冲突抑制项避免一票否决 │ └── confidence_calculator.py # 动态计算每路传感器当前置信度基于历史准确率当前负载波动率 ├── model/ # 检测模型轻量但够用 │ ├── feature_engineer.py # 构造23维特征如SCADA变位频次/分钟、PMU相角标准差、NetFlow TCP重传率 │ └── xgb_detector.py # XGBoost二分类器max_depth4, n_estimators80导出为.onnx供C调用 ├── output/ # 输出层不止告警更要可追溯 │ ├── alert_dispatcher.py # 生成STIX 2.1格式告警含溯源字段攻击路径、受影响IED、置信度链 │ └── dashboard_api.py # FastAPI接口返回JSON格式融合结果供前端可视化 ├── tests/ # 真实测试用例非mock │ ├── test_time_alignment.py # 验证15ms内5路数据对齐精度用真实RTDS时间戳比对 │ └── test_ds_fusion.py # 注入已知冲突证据验证融合后置信度是否合理衰减 └── run_pipeline.py # 主入口按config启动全链路支持--moderealtime接Kafka或--modetest跑data/下样本提示别急着跑run_pipeline.py。先看config/sensor_config.yaml——里面pmu_parser的ip和port默认是127.0.0.1:4444这是模拟PMU发送端如果你没起模拟器会卡在连接超时。真实部署时这里要填你现场PMU设备的真实IP。2.2 技术选型深挖为什么D-S理论比简单加权平均更抗干扰很多新手看到“多源融合”第一反应是np.average([scada_score, pmu_score, netflow_score], weights[0.4,0.3,0.3])。这在实验室OK但在电厂现场会翻车。举个真实案例某次APT攻击中攻击者先用漏洞让SCADA主站失联导致SCADA数据流中断再伪造PMU数据制造“系统稳定”假象同时用NetFlow探针抓到异常DNS隧道流量。如果用静态加权平均SCADA权重0.4直接归零PMU权重0.3又给出高分NetFlow那0.3权重根本拉不回整体分数——结果是漏报。而D-S理论的核心是处理不确定性。它不直接输出“攻击概率”而是输出三元组m(attack), m(normal), m(unknown)。当SCADA中断时它的BPA变成m(unknown)1.0表示“我不知道”PMU伪造数据时它的BPA可能变成m(normal)0.8, m(attack)0.2表示“大概率正常但有点可疑”NetFlow则给出m(attack)0.9, m(normal)0.1。Dempster合成规则会自动压制m(normal)放大m(attack)并把m(unknown)作为缓冲——最终融合结果可能是m(attack)0.75, m(normal)0.15, m(unknown)0.10。你看它没说“100%是攻击”但把unknown留出来就是给运维人员留的“人工复核”入口。这才是工业场景需要的稳健性。2.3 代码即文档time_aligner.py里的PTP对齐逻辑毫秒级精度怎么保电力系统对时间同步要求苛刻SCADA事件记录SOE精度需≤1msPMU相量时间戳需≤1μs。本项目取折中所有传感器数据统一到PTPv2主时钟Grandmaster Clock对齐窗口设为15ms满足IEC 61850-9-3 Class T1要求。关键代码如下# fusion/time_aligner.py import pandas as pd from datetime import datetime, timedelta def align_timestamps(df_list: list[pd.DataFrame], master_clock_col: str ptp_timestamp, tolerance_ms: int 15) - pd.DataFrame: 将多源DataFrame按PTP主时钟对齐采用最近邻插值策略 :param df_list: 每个DataFrame必须含master_clock_col列datetime64[ns]格式 :param tolerance_ms: 最大允许时间偏差毫秒超出则丢弃该行 :return: 对齐后的宽表DataFrame索引为master_clock_col列名带源前缀 # 步骤1统一转换为纳秒级整数时间戳避免浮点误差 for i, df in enumerate(df_list): if not pd.api.types.is_datetime64_any_dtype(df[master_clock_col]): raise ValueError(fColumn {master_clock_col} must be datetime64) # 转为ns整数便于后续计算 df_list[i][master_clock_col _ns] df[master_clock_col].astype(int64) # 步骤2取所有数据中最早和最晚时间生成对齐基准时间轴 all_ns np.concatenate([df[master_clock_col _ns].values for df in df_list]) start_ns np.min(all_ns) end_ns np.max(all_ns) # 以1ms为步长生成基准时间轴足够覆盖15ms窗口 base_timeline_ns np.arange(start_ns, end_ns 1_000_000, 1_000_000) # 1ms 1e6 ns # 步骤3对每个数据源找到基准时间点最近的原始数据点±tolerance_ms内 aligned_dfs [] for i, df in enumerate(df_list): src_name fsrc_{i} # 计算每个基准时间点到该源数据的最近距离 dist_matrix np.abs(base_timeline_ns[:, None] - df[master_clock_col _ns].values[None, :]) # 找到距离最小的索引且距离≤tolerance_ms*1e6 min_dist_idx np.argmin(dist_matrix, axis1) min_dists np.min(dist_matrix, axis1) valid_mask min_dists (tolerance_ms * 1_000_000) # 构建对齐后子DataFrame aligned_df pd.DataFrame({ base_timestamp: pd.to_datetime(base_timeline_ns, unitns), f{src_name}_value: [df.iloc[j][value] if valid_mask[k] else np.nan for k, j in enumerate(min_dist_idx)] }) aligned_dfs.append(aligned_df) # 步骤4合并所有源按base_timestamp索引 result_df aligned_dfs[0] for i in range(1, len(aligned_dfs)): result_df pd.merge(result_df, aligned_dfs[i], onbase_timestamp, howouter, suffixes(, f_src{i})) return result_df.set_index(base_timestamp) # 使用示例在run_pipeline.py中调用 if __name__ __main__: # 假设已从5个采集模块拿到DataFrame列表 scada_df pd.read_csv(data/normal_20231015/scada.csv, parse_dates[ptp_timestamp], date_parserlambda x: pd.to_datetime(x)) pmu_df pd.read_csv(data/normal_20231015/pmu.csv, parse_dates[ptp_timestamp]) # ... 其他df同理 aligned_data align_timestamps([scada_df, pmu_df, netflow_df, modbus_df, edge_df]) print(f对齐后数据形状: {aligned_data.shape}, 时间范围: {aligned_data.index.min()} ~ {aligned_data.index.max()})这段代码的关键在于不用pd.merge_asof它只支持两表对齐且对“最近邻”定义僵硬我们用np.abs矩阵运算显式控制容差逻辑透明时间戳转int64纳秒避免datetime64[ns]在merge时因时区或精度损失引入毫秒级漂移tolerance_ms15是硬约束超出即np.nan绝不强行插值——电力系统宁可丢数据也不要错误时间关联。实测在i7-11800H32GB内存机器上对齐10万行5源数据耗时800ms满足边缘网关实时性要求。3. 核心融合算法拆解D-S证据理论不是数学游戏是给每路传感器发“信用分”的机制很多资料把Dempster-Shafer理论讲成一堆集合论符号但工程落地时它本质就是一个动态信用评估系统。本项目没用教科书式m(A)而是把每路传感器看作一个“裁判”它每次判罚输出一个BPA都要交“保证金”——这个保证金就是它的实时置信度。下面逐段拆解fusion/ds_fuser.py的核心逻辑。3.1evidence_builder.py如何把原始数据翻译成“裁判的判决书”每类传感器的数据特征天差地别不能用同一套公式。本项目为5类传感器定制了BPA生成规则# fusion/evidence_builder.py import numpy as np from typing import Dict, List, Tuple def build_bpa_from_scada(df: pd.DataFrame, confidence: float) - Dict[str, float]: SCADA遥信变位BPA重点看变位频次与SOE事件一致性 # 计算1分钟内变位次数正常应5次 change_count len(df[df[change_type] state_change]) # SOE事件中同一IED在1秒内出现3次变位视为异常 soe_anomaly (df.groupby(ied_id)[soe_timestamp] .apply(lambda x: (x.diff().dt.total_seconds() 1).sum() 3) .any()) if change_count 10 or soe_anomaly: # 异常高攻击置信度 return {attack: 0.8 * confidence, normal: 0.15 * confidence, unknown: 0.05 * confidence} else: # 正常高正常置信度 return {attack: 0.05 * confidence, normal: 0.9 * confidence, unknown: 0.05 * confidence} def build_bpa_from_pmu(df: pd.DataFrame, confidence: float) - Dict[str, float]: PMU相量BPA看相角差标准差与频率偏移 # 计算相角差A-B, B-C, C-A的标准差单位度 angle_std df[[angle_a, angle_b, angle_c]].std(axis1).mean() freq_dev abs(df[frequency].mean() - 50.0) # 偏离50Hz程度 if angle_std 2.5 or freq_dev 0.15: return {attack: 0.75 * confidence, normal: 0.2 * confidence, unknown: 0.05 * confidence} else: return {attack: 0.03 * confidence, normal: 0.94 * confidence, unknown: 0.03 * confidence} def build_bpa_from_netflow(df: pd.DataFrame, confidence: float) - Dict[str, float]: NetFlow BPA看协议异常分布与连接数突增 # TCP重传率 5% 或 DNS请求数/分钟 2000 视为异常 retrans_rate df[tcp_retrans].sum() / max(df[tcp_sent].sum(), 1) dns_qps df[df[protocol] DNS][packet_count].sum() / 60.0 if retrans_rate 0.05 or dns_qps 2000: return {attack: 0.85 * confidence, normal: 0.1 * confidence, unknown: 0.05 * confidence} else: return {attack: 0.02 * confidence, normal: 0.95 * confidence, unknown: 0.03 * confidence} # 其他modbus_analyzer.py和edge_health.py的BPA构建逻辑类似略参数说明confidence参数来自confidence_calculator.py它不是固定值。例如当edge_health.py读到网关CPU温度85℃时它会主动降低自身置信度因为高温可能导致传感器读数漂移此时confidence可能从0.95降到0.6。这就是“动态信用”的体现——传感器自己承认“我现在状态不好”。3.2ds_fuser.py改进Dempster合成规则解决经典D-S的“一票否决”陷阱经典Dempster规则在证据高度冲突时归一化因子K趋近于1导致合成结果崩溃。本项目加入两项改进冲突抑制项当K 0.7即冲突严重不直接归一化而是将冲突质量K按比例分摊到attack和normal上未知集保留强制保留至少5%的unknown质量防止过度自信。代码实现如下# fusion/ds_fuser.py import numpy as np from typing import List, Dict, Any def dempster_combination(evidences: List[Dict[str, float]], conflict_threshold: float 0.7) - Dict[str, float]: 改进Dempster-Shafer合成规则 :param evidences: 列表每个元素是{attack:0.x, normal:0.y, unknown:0.z}字典 :param conflict_threshold: 冲突阈值超过则启用抑制机制 :return: 合成后的BPA字典 # 初始化假设只有attack/normal/unknown三个命题 frame [attack, normal, unknown] # 初始化mass函数矩阵 M[i][j] 表示第i个证据对第j个命题的分配 M np.zeros((len(evidences), len(frame))) for i, ev in enumerate(evidences): for j, prop in enumerate(frame): M[i, j] ev.get(prop, 0.0) # 经典Dempster合成M1 ⊕ M2 ⊕ ... ⊕ Mn # 迭代合成每次合成两个证据 current_mass M[0].copy() for i in range(1, len(evidences)): next_mass M[i].copy() # 计算冲突K sum(current_mass[a] * next_mass[b] for a!b) K 0.0 for a in range(len(frame)): for b in range(len(frame)): if a ! b: # 只计算不同命题间的冲突 K current_mass[a] * next_mass[b] # 如果冲突过高启用抑制机制 if K conflict_threshold: # 将冲突质量K按当前mass比例分摊到attack和normal上 # unknown部分保持不变因其本就是不确定性 alloc_to_attack K * (current_mass[0] next_mass[0]) / (current_mass[0] current_mass[1] next_mass[0] next_mass[1] 1e-8) alloc_to_normal K * (current_mass[1] next_mass[1]) / (current_mass[0] current_mass[1] next_mass[0] next_mass[1] 1e-8) # 更新massattack alloc_to_attack, normal alloc_to_normal, unknown不变 current_mass[0] alloc_to_attack current_mass[1] alloc_to_normal # 重置next_mass跳过经典合成 continue # 经典合成m(C) sum(m1(A)*m2(B)) for A∩BC, then normalize new_mass np.zeros(len(frame)) for a in range(len(frame)): for b in range(len(frame)): # A∩B的结果attack∩attackattack, normal∩normalnormal, unknown∩anythingunknown if a 0 and b 0: # attack ∩ attack c 0 elif a 1 and b 1: # normal ∩ normal c 1 else: # 其他交集都映射到unknown c 2 new_mass[c] current_mass[a] * next_mass[b] # 归一化除以1-K if 1 - K 1e-8: current_mass new_mass / (1 - K) else: current_mass new_mass # 退化情况 # 强制保留至少5% unknown工程鲁棒性 unknown_min 0.05 if current_mass[2] unknown_min: # 从attack和normal中按比例扣减补足unknown deficit unknown_min - current_mass[2] total_normal_attack current_mass[0] current_mass[1] if total_normal_attack 1e-8: current_mass[0] - deficit * (current_mass[0] / total_normal_attack) current_mass[1] - deficit * (current_mass[1] / total_normal_attack) current_mass[2] unknown_min return {frame[i]: current_mass[i] for i in range(len(frame))} # 使用示例 if __name__ __main__: # 模拟5路证据已带置信度衰减 evidences [ build_bpa_from_scada(scada_df, conf_scada), build_bpa_from_pmu(pmu_df, conf_pmu), build_bpa_from_netflow(netflow_df, conf_netflow), build_bpa_from_modbus(modbus_df, conf_modbus), build_bpa_from_edge(edge_df, conf_edge) ] final_bpa dempster_combination(evidences) print(f融合后BPA: {final_bpa}) # 输出如{attack: 0.72, normal: 0.18, unknown: 0.10}这段代码的血泪经验是永远不要相信K接近0的合成结果。我们在某次测试中发现当所有传感器都一致认为“normal”时K极小但此时若有一路传感器因硬件故障输出了错误BPA比如把attack写成0.9经典规则会因K≈0而完全忽略冲突导致合成结果仍偏向normal——这就是“群体盲思”。改进版通过conflict_threshold主动拦截并把冲突质量反向强化到attack上反而提升了鲁棒性。3.3confidence_calculator.py传感器的“健康报告”怎么写用滑动窗口指数衰减传感器置信度不是拍脑袋定的它基于历史表现动态调整。本项目用双滑动窗口机制准确率窗口过去100次判决中该传感器BPA主导命题max(m(attack), m(normal))与最终真实标签一致的比例稳定性窗口过去50个时间片内该传感器BPA的entropy信息熵标准差熵越稳定置信度越高然后用指数衰减融合两者# fusion/confidence_calculator.py import numpy as np from collections import deque class ConfidenceCalculator: def __init__(self, window_size_acc: int 100, window_size_stab: int 50): self.acc_history deque(maxlenwindow_size_acc) # 存储(bool)准确率 self.entropy_history deque(maxlenwindow_size_stab) # 存储(float)熵值 def update_accuracy(self, is_correct: bool): 更新准确率历史 self.acc_history.append(is_correct) def update_entropy(self, bpa: Dict[str, float]): 计算当前BPA熵值并更新历史 probs np.array(list(bpa.values())) # 避免log(0)加极小值 probs np.clip(probs, 1e-8, 1.0) entropy -np.sum(probs * np.log(probs)) self.entropy_history.append(entropy) def get_confidence(self) - float: 计算综合置信度准确率 * (1 - 熵稳定性系数) if len(self.acc_history) 0: return 0.5 # 默认中性 acc_rate np.mean(self.acc_history) if len(self.entropy_history) 10: stab_factor 0.5 # 数据不足稳定性权重减半 else: # 熵的标准差越小说明BPA越稳定总是集中在一个命题上 stab_std np.std(self.entropy_history) # 稳定性系数 1 - 归一化标准差std越小系数越高 stab_factor 1.0 - min(stab_std / 0.5, 1.0) # 0.5是经验值上限 # 指数衰减融合准确率权重0.7稳定性权重0.3 confidence 0.7 * acc_rate 0.3 * stab_factor return max(0.3, min(0.95, confidence)) # 截断到[0.3, 0.95] # 在pipeline中每次获得真实标签如人工标注或仿真注入标记后调用 # calc_scada.update_accuracy(predicted_label true_label) # calc_scada.update_entropy(scada_bpa) # conf_scada calc_scada.get_confidence()这个设计让传感器有了“自我认知”一个长期准确但近期熵值飙升比如PMU受电磁干扰的传感器置信度会自然下降反之一个一直平庸但突然连续10次精准捕获攻击的传感器置信度会快速爬升。这才是活的融合系统。4. 运行与调试从零跑通的完整链路以及那些让你怀疑人生的5个坑别被“Python实现”四个字骗了——这项目对环境敏感度堪比精密仪器。我第一次在Ubuntu 22.04上跑通花了整整两天踩的坑全记在这儿了。下面是从下载资源到看到第一条告警的完整路径附赠避坑指南。4.1 环境准备为什么必须用Python 3.9因为NumPy 1.23的int64精度项目requirements.txt明确要求numpy1.23.5 pandas1.5.3 scikit-learn1.2.2 pydantic1.10.12 fastapi0.104.1 uvicorn0.23.2注意不要用pip install -r requirements.txt一键安装scikit-learn 1.2.2依赖numpy1.21.0,1.24.0而最新numpy已到1.26装了会报ImportError: cannot import name ABCIndexClass。必须严格锁定版本。正确步骤# 创建干净虚拟环境 python3.9 -m venv venv_fusion source venv_fusion/bin/activate # 逐个安装顺序很重要 pip install --upgrade pip pip install numpy1.23.5 pip install pandas1.5.3 pip install scikit-learn1.2.2 pip install pydantic1.10.12 pip install fastapi uvicorn # 这俩用最新版即可4.2 数据准备别直接跑data/下的CSV先用generate_sample.py理解数据结构项目data/目录下是实测数据但字段名和格式可能与你的现场不一致。建议先跑自带的样本生成器它会创建符合IEC 61850语义的模拟数据# 进入项目根目录 cd multi_sensor_fusion_power # 生成1小时模拟数据含正常攻击片段 python utils/generate_sample.py --duration_hours 1 --output_dir data/simulated_20231018 # 查看生成的数据结构 head -n 5 data/simulated_20231018/scada.csv # 输出timestamp,ied_id,point_id,value,change_type,soe_timestamp,ptp_timestamp # 2023-10-18 00:00:00.000,IED101,ST101,0,state_change,2023-10-18 00:00:00.001,2023-10-18 00:00:00.000123456关键点ptp_timestamp列必须是datetime64[ns]格式且精度到纳秒注意末尾6位数字。如果用Excel打开CSV再保存会丢失纳秒精度导致time_aligner.py报ValueError: cannot convert to datetime64[ns]。4.3 启动Pipeline两种模式别混用项目支持两种运行模式通过--mode参数切换--modetest离线模式读取data/下指定目录的CSV文件适合调试算法逻辑--moderealtime实时模式从Kafka Topic消费数据需提前部署Kafka适合生产部署推荐新手从test模式开始# 运行测试模式指定数据目录 python run_pipeline.py --modetest --data_dir data/simulated_20231018 # 你会看到滚动日志 # [INFO] Loading SCADA data from data/simulated_20231018/scada.csv... # [INFO] Aligning 5 sources with tolerance 15ms... # [INFO] Fusion result: {attack: 0.042, normal: 0.915, unknown: 0.043} # [INFO] No alert triggered (attack prob 0.5)提示默认告警阈值是attack_prob 0.5。如果你想看攻击告警编辑config/fusion_config.yaml把alert_threshold: 0.5改成0.05再跑一次。4.4 避坑 / 常见问题 / 排查那些让我重启三次IDE的瞬间现象1time_aligner.py报错ValueError: cannot convert to datetime64[ns]原因CSV中的ptp_timestamp列包含空值、非标准时间格式如2023/10/18 00:00:00而非2023-10-18 00:00:00.000或Excel保存时转成了字符串。解决用pandas.read_csv时强制指定parse_dates和date_parser# 在ingestion/scada_reader.py中 df pd.read_csv(filepath, parse_dates[ptp_timestamp], date_parserlambda x: pd.to_datetime(x, format%Y-%m-%d %H:%M:%S.%f, errorscoerce))现象2ds_fuser.py合成后attack概率恒为0.0原因某路传感器BPA生成函数返回了空字典如build_bpa_from_scada中df为空导致m(attack)0参与合成。解决在evidence_builder.py每个函数末尾加兜底# 所有BPA构建函数末尾统一加 if not bpa_dict: # 空字典 return {attack: 0.01, normal: 0.98, unknown: 0.01} # 默认保守估计现象3XGBoost训练时报ValueError: Input contains NaN原因feature_engineer.py构造特征本文还有配套的精品资源点击获取