简介面向西安地铁2号线车辆受电弓无法升弓故障诊断需求这份资源提供基于贝叶斯网络的完整实现方案与可运行Python代码适合具备一定编程基础的地铁车辆维护工程师、故障诊断研究人员以及对贝叶斯网络应用感兴趣的开发者。资源为单个docx文档约42KB内容既有论文核心思路梳理也有利用pgmpy库完成网络结构构建、EM算法参数学习、缺失数据处理与诊断推理的逐步讲解并覆盖故障树向贝叶斯网络转换、条件概率表设定、模拟故障数据生成及后验概率查询等细节。目前已有65人浏览学习。学习者可以结合理论与实践调整网络结构和参数以适配其他受电弓或车辆子系统故障场景进而为检修团队快速定位故障原因、优化检修流程提供直接可复用的技术参考。1. 贝叶斯网络为什么适合地铁受电弓故障诊断地铁车辆每天跑完一整天受电弓碳滑板磨到限位、气囊漏气导致弓网接触压力不足、升弓机构卡滞导致拉弧——这些故障在运行数据里往往不是单一传感器超限而是多个参数同时偏离正常范围。传统的阈值告警会报告一排红灯维修人员面对几十条报警反而定位不到根因。贝叶斯网络的思路刚好反过来先把“故障原因”和“可观测信号”之间的因果依赖画成一张有向图再用条件概率描述每个节点受父节点影响的程度。一旦收到新的传感器证据网络就能算出“当前最可能的故障是什么”输出一个带概率的排序结果而不是孤立的一堆报警。这篇文章从受电弓的故障机理开始带你一步步完成建模、参数学习、推理和工程化落地全部核心代码用 Python pgmpy 实现可直接复制修改。适合做车辆运维诊断、PHM 方向的一线工程师也适合刚接触贝叶斯网络的开发者在真实场景里快速上手。2. 受电弓故障机理与贝叶斯网络建模从故障树到有向无环图2.1 先从故障树出发找到受电弓的可观测证据受电弓的常见故障模式在不同线路上略有差异但基本可以分成三类气路故障气囊漏气、减压阀失效、机械结构故障铰链卡滞、升弓弹簧断裂、电气接触故障碳滑板磨损、弓网接触压力不足导致的拉弧。做贝叶斯网络之前我习惯先列出每个故障原因对应的“可观测量”因为网络里每个叶子节点必须对应一个能拿到数据的传感器或检测项。常见故障模式与观测变量的对应关系如下表故障原因父节点可观测信号子节点传感器/检测手段气囊漏气气路压力过低、接触压力不足压力变送器、压力传感器碳滑板磨损滑板剩余厚度偏小、接触电阻偏大激光测距、厚度检测仪升弓机构卡滞升弓时间过长、接触压力波动电流变送器、升弓时间记录接触压力不足拉弧次数增多、牵引电流波动燃弧检测装置、电流互感器这一步的关键在于不要把所有测点都直接连到故障原因上中间通常还有“中间状态”。比如“接触压力不足”既是气囊漏气的结果又是拉弧的原因。把它单独拿出来作为中间节点能让网络更贴合物理过程也让后验概率的解释更自然。2.2 故障树转化为贝叶斯网络时的三类节点故障树里的“顶事件”“中间事件”“底事件”对应到贝叶斯网络里就是三类节点根节点无父节点代表独立的故障原因比如“气囊漏气”“碳滑板磨损”“机构卡滞”每个根节点有自己的先验概率。中间节点代表物理状态的中间变化比如“接触压力不足”“弓网接触电阻异常”作用是连接根因和更下游的症状。叶子节点无子节点代表可以直接测到的量比如“气路压力低”“拉弧次数高”“升弓时间长”。边有向箭头的方向必须遵循因果方向从“原因”指向“结果”。不能因为“气路压力低时通常能观察到气囊漏气”就把箭头反过来。贝叶斯网络推理虽然允许反向传播证据但结构上必须是物理因果结构。以常见做法为例受电弓诊断网络的邻接关系可以设置成四组因果链气囊漏气 → 气路压力低 接触压力不足接触压力不足 → 拉弧次数高 牵引电流波动碳滑板磨损 → 滑板厚度偏小 接触电阻偏大升弓机构卡滞 → 升弓时间长 接触压力波动2.3 用 pgmpy 搭建受电弓诊断网络的最小框架理论部分落到代码上我用 pgmpy 的BayesianNetwork来定义网络结构。以下代码可以直接复制运行前提是环境里已经装好 pgmpyfrom pgmpy.models import BayesianNetwork from pgmpy.factors.discrete import TabularCPD # 定义网络结构每个元组表示一条有向边方向为 父节点 - 子节点 model BayesianNetwork([ (airbag_leak, air_pressure), # 气囊漏气 - 气路压力低 (airbag_leak, contact_force), # 气囊漏气 - 接触压力不足 (contact_force, arcing), # 接触压力不足 - 拉弧次数高 (slider_wear, slider_thickness), # 碳滑板磨损 - 滑板厚度偏小 (slider_wear, contact_resistance), # 碳滑板磨损 - 接触电阻偏大 (mechanism_stuck, raise_time), # 机构卡滞 - 升弓时间长 (mechanism_stuck, contact_force) # 机构卡滞 - 接触压力波动 ]) # 检查结构是否合法不允许有环每个节点必须有定义好的状态 print(网络节点:, model.nodes()) print(网络边:, model.edges())逻辑说明BayesianNetwork构造函数的参数是一个包含所有有向边的列表每条边写成(父节点, 子节点)的元组。节点名建议直接用带下划线的简写避免在后面的代码里反复写中文变量名。构造完成后调用model.check_model()可以检查图结构是否是合法的有向无环图如果建边时不小心形成了环这一步会直接报错。参数说明这里引入了 7 个节点其中airbag_leak、slider_wear、mechanism_stuck是根节点contact_force是中间节点air_pressure、arcing、slider_thickness、contact_resistance、raise_time是叶节点。实际项目中节点数往往会更多比如会增加“环境湿度”“接触网硬点”等外部因素但结构搭建的方式完全一样。2.4 条件概率表CPT的两种来源专家经验与历史数据贝叶斯网络除了结构还必须给每个节点配备条件概率表。根节点需要先验概率非根节点需要给定父节点状态下各取值的概率。两种主流做法第一种是纯专家经验。比如根据某线路历史统计气囊漏气的先验概率是 0.02碳滑板磨损是 0.05。CPT 里的数值可以来自故障树分析中的概率计算也可以来自检修记录里故障频率的统计。这种方式适合“故障样本少、机理清晰”的场景新线路没有历史数据时基本只能靠它。第二种是用历史数据学习。把运行记录和最终的故障定责结果对齐形成一张“传感器证据 故障标签”的数据表然后调用 pgmpy 的fit()方法从数据里估计条件概率。这种方式适合线路已经运营了足够长时间、积累了大量故障样本的情况。通常我会先用专家经验建一个初始的贝叶斯网络上线跑跑上三个月后再用积累的数据重新训练 CPT形成“先专家后数据”的演进路径。手动设置 CPT 的代码片段如下# 根节点先验概率airbag_leak 的先验为 P(no)0.98, P(yes)0.02 cpd_leak TabularCPD(airbag_leak, 2, [[0.98, 0.02]]) # 中间节点 contact_force依赖 airbag_leak 和 mechanism_stuck 两个父节点 # 状态顺序为 [no, yes]父节点取值顺序从左到右为 # (airbag_leakno, mechanism_stuckno), (no, yes), (yes, no), (yes, yes) cpd_force TabularCPD( contact_force, 2, [[0.96, 0.55, 0.30, 0.05], [0.04, 0.45, 0.70, 0.95]], evidence[airbag_leak, mechanism_stuck], evidence_card[2, 2] )逻辑说明TabularCPD第一个参数是节点名第二个参数是该节点的状态数量第三个参数是概率矩阵。矩阵的行对应contact_force的每个取值第一行是 no第二行是 yes矩阵的列对应父节点取值的所有组合顺序由evidence_card[2, 2]决定。这里“气囊漏气且机构卡滞同时存在”时接触压力正常的概率只有 0.05对应矩阵最后一列能反映多故障并发时的叠加效果。注意每一列的概率之和必须等于 1否则check_model()校验不通过。参数里的 0.04、0.45、0.70、0.95 这些数值直接影响后验推理结果的灵敏度不建议拍脑袋乱填至少要跟故障树里的底层事件概率对上。3. 诊断系统的核心代码实现参数学习、推理与特征处理3.1 数据准备把传感器记录转成贝叶斯网络能吃的离散样本贝叶斯网络在 pgmpy 里的标准输入是一个 pandas DataFrame每一行是一条样本每一列是一个节点变量取值必须是离散的字符串或整数。但是受电弓的传感器数据大多是连续量比如气路压力是 460 kPa、升弓时间是 6.8 秒。直接丢进网络之前必须先做离散化。习惯做法是给每个连续量设定阈值或分箱区间import pandas as pd def preprocess_sensor_data(raw_df): df raw_df.copy() # 气路压力低于 450 kPa 视为 low否则 normal df[air_pressure] df[air_pressure].apply( lambda v: low if v 450 else normal ) # 滑板剩余厚度小于 22mm 时视为 thin df[slider_thickness] df[slider_thickness].apply( lambda v: thin if v 22 else normal ) # 升弓时间大于 9 秒视为 slow df[raise_time] df[raise_time].apply( lambda v: slow if v 9 else normal ) # 拉弧次数日累计超过 15 次视为 high df[arcing] df[arcing].apply( lambda v: high if v 15 else normal ) # 接触电阻和接触压力按相关系数分箱这里用中位数做二值化 df[contact_resistance] pd.qcut( df[contact_resistance], q2, labels[small, large] ) df[contact_force] pd.qcut( df[contact_force], q2, labels[low, normal] ) return df逻辑说明上限阈值比如气路压力 450 kPa、升弓时间 9 秒一般来自受电弓检修规程可以在部署时配置成 JSON 参数文件方便不同线路按自己的标准调整。pd.qcut的分箱方式是按样本分布等频切分这适合没有明确物理阈值的量有些项目也会改成固定阈值cut两者各有好处。注意离散化之后必须保证每个节点在训练集和推理集里出现的取值类别完全一致否则fit()会报 “ValueError: Data must be a DataFrame” 或状态数量不匹配的异常。3.2 参数学习有数据时用最大似然估计数据少时用贝叶斯估计当历史数据里有明确的故障标签时就可以跳过手动设置 CPT 的繁琐步骤直接学到整张概率表。下面这段代码把刚才定义好的空模型model只包含结构不包含 CPD和样本数据train_df放在一起调用fit()from pgmpy.estimators import MaximumLikelihoodEstimator, BayesianEstimator # 方法一最大似然估计样本量大时推荐 model_mle model.copy() model_mle.fit(train_df, estimatorMaximumLikelihoodEstimator) # 方法二贝叶斯估计加入 BDeu 先验避免零概率 model_bdeu model.copy() model_bdeu.fit( train_df, estimatorBayesianEstimator, prior_typeBDeu, equivalent_sample_size10 ) # 检查学习到的 CPT print(model_bdeu.get_cpds(contact_force))逻辑说明MaximumLikelihoodEstimator的做法是直接统计每个父节点组合下子节点取值的频率。这种方法在样本充足时无偏但是当某个父节点组合下没有任何样本时CPT 里会出现 0 概率后续推理一旦落到这个分支就会直接得出不可能的结论。BayesianEstimator加了一个equivalent_sample_size10的 BDeu 先验相当于给每个父节点组合都预置了 10 条虚拟样本从数学上完全消除零概率问题。在受电弓故障样本很少的场景下我一般直接用第二种方法。参数说明equivalent_sample_size表示虚拟样本量的强度设得越大先验对最终概率的影响越大设得太小平滑效果不明显。一般取 5 到 20 之间即可。如果现场故障样本特别稀疏可以调大这个值让专家先验继续主导一小段时间。3.3 故障推理变元消去算法与证据更新网络结构和 CPT 都齐了之后核心工作就是推理给定一组观测到的传感器证据计算每个故障原因的后验概率。pgmpy 的VariableElimination提供精确推理适合几十个节点以内的网络。from pgmpy.inference import VariableElimination infer VariableElimination(model_bdeu) # 假设当前列车检测到气路压力低、拉弧次数高、滑板厚度偏小 evidence { air_pressure: low, arcing: high, slider_thickness: thin } # 分别查询三个根节点的后验概率 result_leak infer.query(variables[airbag_leak], evidenceevidence) result_wear infer.query(variables[slider_wear], evidenceevidence) result_stuck infer.query(variables[mechanism_stuck], evidenceevidence) print(气囊漏气概率:, result_leak) print(碳滑板磨损概率:, result_wear) print(机构卡滞概率:, result_stuck)逻辑说明query()的第一个参数是想要查询的目标节点列表第二个参数是证据字典。推理引擎会利用贝叶斯网络的条件独立性把联合分布拆成多个因子相乘再通过消元逐步边缘化掉无关变量从而高效算出后验分布。在这个例子里“拉弧次数高”并不是直接证据对应的根因但它会通过contact_force这一中间节点间接提高“气囊漏气”和“机构卡滞”的后验概率这正是贝叶斯网络相对阈值判据的核心优势能利用间接证据互相印证。参数说明evidence字典里的 key 必须和网络里设置的节点名完全一致value 必须是该节点状态列表里的合法取值。如果跑出KeyError优先检查预处理函数里的取值是否需要做字符串归一化。如果跑出ValueError: Can not use the given evidence说明状态类别和 CPT 里的类别对不上需要查看model_bdeu.get_cpds(节点名).state_names来确认真实的取值集合。3.4 连续量节点处理的另一种思路高斯贝叶斯网络二值化离散化的优点是简单可解释缺点是对阈值敏感。如果怕阈值选不好导致推理结果不稳定可以改用LinearGaussianBayesianNetwork让连续传感器变量直接作为节点然后让离散故障节点作为父节点、连续变量作为子节点为每个父节点状态组合拟合一个高斯分布。不过这个路线在工程上比纯离散网络麻烦因为它的 CPT 变成了均值方差参数表推理用的是联合树算法结果也不如离散网络直观。我的建议是第一阶段先用离散化方案上线等确认了稳定阈值再考虑连续高斯方案。4. 实际故障诊断流程与验证从后验概率到检修决策4.1 完整的诊断判断逻辑同时输出多条可能原因单次推理只给出一个后验概率还不够现场维修需要的是一个可解释的结论列表。实际系统里我不会只依赖某一个概率阈值而是先给根节点排序再结合一个置信度阈值做筛选def diagnose(model, evidence, top_k3, threshold0.3): infer VariableElimination(model) root_nodes [airbag_leak, slider_wear, mechanism_stuck] result {} for node in root_nodes: q infer.query(variables[node], evidenceevidence) # 取出 yes 状态对应的后验概率 state_index q.state_names[node].index(yes) result[node] q.values[state_index] # 按后验概率降序排列 sorted_result sorted(result.items(), keylambda kv: kv[1], reverseTrue) # 过滤低于阈值的项 final [(name, prob) for name, prob in sorted_result if prob threshold] return final[:top_k] # 示例某次运行数据 diagnosis diagnose( model_bdeu, {air_pressure: low, arcing: high, raise_time: slow} ) for node, prob in diagnosis: print(f故障原因: {node}, 后验概率: {prob:.2%})逻辑说明这段代码把三个根节点分别查一遍后验概率然后排序和过滤形成真正的故障诊断输出。state_index q.state_names[node].index(yes)这一步很关键因为 pgmpy 返回的values数组顺序可能和state_names列表一致但显式地用索引取值更稳妥避免代码里硬编码第几个下标。处理后输出结果就可以直接推送到检修终端或生成工单。参数说明threshold0.3是诊断灵敏度的调节旋钮。阈值调低会报告更多潜在故障召回率高但容易误报阈值调高只报概率最高的一两个故障精确率高但可能漏掉低概率并发故障。地铁运维场景里我会区分“预警”和“告警”两档0.3 以上只推送至日检列表0.6 以上直接生成故障工单。4.2 用一个完整案例把全流程串起来拿一列地铁某日早高峰的传感器数据举例。原始记录值如下气路压力值428 kPa → 离散化为low日累计拉弧次数27 次 → 离散化为high滑板剩余厚度24.5 mm → 离散化为normal升弓时间10.2 秒 → 离散化为slow用上面编号的网络推理得到的后验概率排序可能为升弓机构卡滞约 0.63气囊漏气约 0.28碳滑板磨损约 0.05。这个结果说明“升弓时间长接触压力波动”这两个证据强指向机构卡滞而滑板厚度正常又大幅降低了磨损的后验概率。把这份输出交给检修人员相当于直接把边界清晰的原因定位从几十条报警里提出来。4.3 用历史故障样本验证诊断准确率系统上线前必须回答一个问题诊断准确率到底多高验证方法和普通分类模型类似但要注意贝叶斯网络的“标签”是根节点的真实故障状态。用留一法交叉验证的思路写一个评估循环from sklearn.metrics import accuracy_score, precision_recall_fscore_support def evaluate(model, data, labels, root_node): infer VariableElimination(model) preds [] for _, row in data.iterrows(): # 把这一行的所有观测变量转换为证据 evidence { col: row[col] for col in data.columns } q infer.query(variables[root_node], evidenceevidence) state_index q.state_names[root_node].index(yes) preds.append(q.values[state_index] 0.5) acc accuracy_score(labels, preds) precision, recall, f1, _ precision_recall_fscore_support( labels, preds, averagebinary ) return acc, precision, recall, f1 acc, prec, rec, f1 evaluate( model_bdeu, test_df[observable_columns], test_df[slider_wear_true] ) print(fAccuracy: {acc:.3f}, Precision: {prec:.3f}, Recall: {rec:.3f}, F1: {f1:.3f})逻辑说明对每条测试样本把除根节点外的所有可观测列都作为证据传入推理引擎然后看yes状态的后验概率是否超过 0.5。注意这里的data.columns必须只包含模型里的观测变量不能把slider_wear_true这类标记列也放进去否则等于把答案提前告诉了推理引擎。参数说明0.5 只是默认阈值验证时可以循环换成 0.3、0.4、0.6 等数值画出不同阈值下的精确率和召回率曲线。这一步能帮助业务方决定线上系统把预警线放在哪里。如果某条线路的故障率本来就低看整体 Accuracy 意义不大更应该关注 Minority 类别的 Recall。5. 部署经验与调试技巧样本不平衡、零概率与代码运行环境5.1 样本不平衡真实故障太少模型学不到“异常模式”地铁受电弓的真实故障率非常低气囊漏气可能一年也没几次直接拿全量样本做参数学习学到的先验概率会极度偏向正常。这会导致一个典型问题即使气路压力已经低到报警线由于先验概率正常类太高后验概率依然不高诊断被“正常”主导。解决办法有三个层次。第一采样时做类平衡训练贝叶斯网络时把故障样本复制若干倍或者用 SMOTE 这类算法合成故障样本但合成样本要谨慎避免制造出不存在的父节点组合。第二调大 BDeu 的equivalent_sample_size会影响先验强度但对样本量极端的类更有效的是直接手动设置根节点先验把现场统计到的故障频率写进 CPT。第三不要在整车上训练一个“通用受电弓模型”而是按线路分组训练因为不同线路的接触网状态差异很大合并训练会把线路特征稀释掉。5.2 零概率灾难某个证据组合在训练集里从没出现过当证据组合复杂到一定程度训练数据覆盖不到某些父节点取值组合是常见现象。比如airbag_leakyes且mechanism_stuckyes的并发故障在历史样本里一次都没出现最大似然估计就会把contact_forcenormal的概率学到 0。line 推理时一旦证据命中这个分支概率结果直接失真。解决办法是前面提到的贝叶斯估计加prior_typeBDeu。此外还有一个更细的操作对所有 CPT 里为 0 的单元格做拉普拉斯平滑在拟合完成后再次检查def smooth_zero_cpds(model, alpha0.01): for cpd in model.get_cpds(): values cpd.values.copy() values[values 0] alpha # 重新归一化确保每列概率和为 1 normalized values / values.sum(axis0) cpd.values normalized return model逻辑说明这段代码在模型推理之前把所有零概率替换成一个极小值然后按列重新归一化。alpha0.01适合样本量较大的情况如果样本量小可以取 0.05 到 0.1。这个技巧在“专家搭建的 CPT 少量数据修正”混合模式下尤其好用因为专家也可能漏填某个组合。5.3 用 BIF 文件持久化模型方便跨环境复用模型训练好之后不能每次都从原始数据重新训练。pgmpy 支持把整个贝叶斯网络包括结构、CPT、节点状态写成标准 BIF 格式这样在列车诊断主机、地面服务器、云平台之间迁移时就只要拷贝文件from pgmpy.readwrite import BIFReader, BIFWriter # 保存 BIFWriter(model_bdeu).write_bif(pantograph_diagnosis.bif) # 加载 model BIFReader(pantograph_diagnosis.bif).get_model()逻辑说明BIF 文件是纯文本的 XML 格式可以通过版本管理工具统一追踪每条线路的网络版本。现场升级模型时只需要把新的 BIF 文件下发到列车的边缘计算单元重启推理服务即可生效。注意保存前确认所有 CPT 都已完成平滑否则加载后再去改会比较绕。5.4 关于代码运行环境的一个实操提示这套代码不挑运行环境最常见的执行方式有三种在 WSL Ubuntu 的终端里直接跑 Python 脚本在 VS Code 里新建 Jupyter Notebook 一步步调试或者把诊断代码封装成 FastAPI 服务部署在服务器上。我建议在本地调试阶段直接用python diagnosis.py运行把上面提到的函数按顺序组织在同一个文件里入口处加上if __name__ __main__:调用一次完整推理方便验证数据流是否畅通。把print(模型加载完成等待传感器数据...)放在入口位置测试时能清晰看到脚本执行到哪一步停了。最后想留一个实用的验证技巧当你在现场拿到一批新的运行数据不要急着把整个模型重新训练一遍而是先用旧模型对单条异常样本做推理把根节点的后验概率分布画成柱状图对比新增样本前后排序是否发生变化。如果某条证据加入后概率排序基本不变说明这条证据对区分故障原因几乎没有贡献后续可以把它从观测变量里逐步剔除简化网络的维护成本。本文还有配套的精品资源点击获取
简介面向西安地铁2号线车辆受电弓无法升弓故障诊断需求这份资源提供基于贝叶斯网络的完整实现方案与可运行Python代码适合具备一定编程基础的地铁车辆维护工程师、故障诊断研究人员以及对贝叶斯网络应用感兴趣的开发者。资源为单个docx文档约42KB内容既有论文核心思路梳理也有利用pgmpy库完成网络结构构建、EM算法参数学习、缺失数据处理与诊断推理的逐步讲解并覆盖故障树向贝叶斯网络转换、条件概率表设定、模拟故障数据生成及后验概率查询等细节。目前已有65人浏览学习。学习者可以结合理论与实践调整网络结构和参数以适配其他受电弓或车辆子系统故障场景进而为检修团队快速定位故障原因、优化检修流程提供直接可复用的技术参考。1. 贝叶斯网络为什么适合地铁受电弓故障诊断地铁车辆每天跑完一整天受电弓碳滑板磨到限位、气囊漏气导致弓网接触压力不足、升弓机构卡滞导致拉弧——这些故障在运行数据里往往不是单一传感器超限而是多个参数同时偏离正常范围。传统的阈值告警会报告一排红灯维修人员面对几十条报警反而定位不到根因。贝叶斯网络的思路刚好反过来先把“故障原因”和“可观测信号”之间的因果依赖画成一张有向图再用条件概率描述每个节点受父节点影响的程度。一旦收到新的传感器证据网络就能算出“当前最可能的故障是什么”输出一个带概率的排序结果而不是孤立的一堆报警。这篇文章从受电弓的故障机理开始带你一步步完成建模、参数学习、推理和工程化落地全部核心代码用 Python pgmpy 实现可直接复制修改。适合做车辆运维诊断、PHM 方向的一线工程师也适合刚接触贝叶斯网络的开发者在真实场景里快速上手。2. 受电弓故障机理与贝叶斯网络建模从故障树到有向无环图2.1 先从故障树出发找到受电弓的可观测证据受电弓的常见故障模式在不同线路上略有差异但基本可以分成三类气路故障气囊漏气、减压阀失效、机械结构故障铰链卡滞、升弓弹簧断裂、电气接触故障碳滑板磨损、弓网接触压力不足导致的拉弧。做贝叶斯网络之前我习惯先列出每个故障原因对应的“可观测量”因为网络里每个叶子节点必须对应一个能拿到数据的传感器或检测项。常见故障模式与观测变量的对应关系如下表故障原因父节点可观测信号子节点传感器/检测手段气囊漏气气路压力过低、接触压力不足压力变送器、压力传感器碳滑板磨损滑板剩余厚度偏小、接触电阻偏大激光测距、厚度检测仪升弓机构卡滞升弓时间过长、接触压力波动电流变送器、升弓时间记录接触压力不足拉弧次数增多、牵引电流波动燃弧检测装置、电流互感器这一步的关键在于不要把所有测点都直接连到故障原因上中间通常还有“中间状态”。比如“接触压力不足”既是气囊漏气的结果又是拉弧的原因。把它单独拿出来作为中间节点能让网络更贴合物理过程也让后验概率的解释更自然。2.2 故障树转化为贝叶斯网络时的三类节点故障树里的“顶事件”“中间事件”“底事件”对应到贝叶斯网络里就是三类节点根节点无父节点代表独立的故障原因比如“气囊漏气”“碳滑板磨损”“机构卡滞”每个根节点有自己的先验概率。中间节点代表物理状态的中间变化比如“接触压力不足”“弓网接触电阻异常”作用是连接根因和更下游的症状。叶子节点无子节点代表可以直接测到的量比如“气路压力低”“拉弧次数高”“升弓时间长”。边有向箭头的方向必须遵循因果方向从“原因”指向“结果”。不能因为“气路压力低时通常能观察到气囊漏气”就把箭头反过来。贝叶斯网络推理虽然允许反向传播证据但结构上必须是物理因果结构。以常见做法为例受电弓诊断网络的邻接关系可以设置成四组因果链气囊漏气 → 气路压力低 接触压力不足接触压力不足 → 拉弧次数高 牵引电流波动碳滑板磨损 → 滑板厚度偏小 接触电阻偏大升弓机构卡滞 → 升弓时间长 接触压力波动2.3 用 pgmpy 搭建受电弓诊断网络的最小框架理论部分落到代码上我用 pgmpy 的BayesianNetwork来定义网络结构。以下代码可以直接复制运行前提是环境里已经装好 pgmpyfrom pgmpy.models import BayesianNetwork from pgmpy.factors.discrete import TabularCPD # 定义网络结构每个元组表示一条有向边方向为 父节点 - 子节点 model BayesianNetwork([ (airbag_leak, air_pressure), # 气囊漏气 - 气路压力低 (airbag_leak, contact_force), # 气囊漏气 - 接触压力不足 (contact_force, arcing), # 接触压力不足 - 拉弧次数高 (slider_wear, slider_thickness), # 碳滑板磨损 - 滑板厚度偏小 (slider_wear, contact_resistance), # 碳滑板磨损 - 接触电阻偏大 (mechanism_stuck, raise_time), # 机构卡滞 - 升弓时间长 (mechanism_stuck, contact_force) # 机构卡滞 - 接触压力波动 ]) # 检查结构是否合法不允许有环每个节点必须有定义好的状态 print(网络节点:, model.nodes()) print(网络边:, model.edges())逻辑说明BayesianNetwork构造函数的参数是一个包含所有有向边的列表每条边写成(父节点, 子节点)的元组。节点名建议直接用带下划线的简写避免在后面的代码里反复写中文变量名。构造完成后调用model.check_model()可以检查图结构是否是合法的有向无环图如果建边时不小心形成了环这一步会直接报错。参数说明这里引入了 7 个节点其中airbag_leak、slider_wear、mechanism_stuck是根节点contact_force是中间节点air_pressure、arcing、slider_thickness、contact_resistance、raise_time是叶节点。实际项目中节点数往往会更多比如会增加“环境湿度”“接触网硬点”等外部因素但结构搭建的方式完全一样。2.4 条件概率表CPT的两种来源专家经验与历史数据贝叶斯网络除了结构还必须给每个节点配备条件概率表。根节点需要先验概率非根节点需要给定父节点状态下各取值的概率。两种主流做法第一种是纯专家经验。比如根据某线路历史统计气囊漏气的先验概率是 0.02碳滑板磨损是 0.05。CPT 里的数值可以来自故障树分析中的概率计算也可以来自检修记录里故障频率的统计。这种方式适合“故障样本少、机理清晰”的场景新线路没有历史数据时基本只能靠它。第二种是用历史数据学习。把运行记录和最终的故障定责结果对齐形成一张“传感器证据 故障标签”的数据表然后调用 pgmpy 的fit()方法从数据里估计条件概率。这种方式适合线路已经运营了足够长时间、积累了大量故障样本的情况。通常我会先用专家经验建一个初始的贝叶斯网络上线跑跑上三个月后再用积累的数据重新训练 CPT形成“先专家后数据”的演进路径。手动设置 CPT 的代码片段如下# 根节点先验概率airbag_leak 的先验为 P(no)0.98, P(yes)0.02 cpd_leak TabularCPD(airbag_leak, 2, [[0.98, 0.02]]) # 中间节点 contact_force依赖 airbag_leak 和 mechanism_stuck 两个父节点 # 状态顺序为 [no, yes]父节点取值顺序从左到右为 # (airbag_leakno, mechanism_stuckno), (no, yes), (yes, no), (yes, yes) cpd_force TabularCPD( contact_force, 2, [[0.96, 0.55, 0.30, 0.05], [0.04, 0.45, 0.70, 0.95]], evidence[airbag_leak, mechanism_stuck], evidence_card[2, 2] )逻辑说明TabularCPD第一个参数是节点名第二个参数是该节点的状态数量第三个参数是概率矩阵。矩阵的行对应contact_force的每个取值第一行是 no第二行是 yes矩阵的列对应父节点取值的所有组合顺序由evidence_card[2, 2]决定。这里“气囊漏气且机构卡滞同时存在”时接触压力正常的概率只有 0.05对应矩阵最后一列能反映多故障并发时的叠加效果。注意每一列的概率之和必须等于 1否则check_model()校验不通过。参数里的 0.04、0.45、0.70、0.95 这些数值直接影响后验推理结果的灵敏度不建议拍脑袋乱填至少要跟故障树里的底层事件概率对上。3. 诊断系统的核心代码实现参数学习、推理与特征处理3.1 数据准备把传感器记录转成贝叶斯网络能吃的离散样本贝叶斯网络在 pgmpy 里的标准输入是一个 pandas DataFrame每一行是一条样本每一列是一个节点变量取值必须是离散的字符串或整数。但是受电弓的传感器数据大多是连续量比如气路压力是 460 kPa、升弓时间是 6.8 秒。直接丢进网络之前必须先做离散化。习惯做法是给每个连续量设定阈值或分箱区间import pandas as pd def preprocess_sensor_data(raw_df): df raw_df.copy() # 气路压力低于 450 kPa 视为 low否则 normal df[air_pressure] df[air_pressure].apply( lambda v: low if v 450 else normal ) # 滑板剩余厚度小于 22mm 时视为 thin df[slider_thickness] df[slider_thickness].apply( lambda v: thin if v 22 else normal ) # 升弓时间大于 9 秒视为 slow df[raise_time] df[raise_time].apply( lambda v: slow if v 9 else normal ) # 拉弧次数日累计超过 15 次视为 high df[arcing] df[arcing].apply( lambda v: high if v 15 else normal ) # 接触电阻和接触压力按相关系数分箱这里用中位数做二值化 df[contact_resistance] pd.qcut( df[contact_resistance], q2, labels[small, large] ) df[contact_force] pd.qcut( df[contact_force], q2, labels[low, normal] ) return df逻辑说明上限阈值比如气路压力 450 kPa、升弓时间 9 秒一般来自受电弓检修规程可以在部署时配置成 JSON 参数文件方便不同线路按自己的标准调整。pd.qcut的分箱方式是按样本分布等频切分这适合没有明确物理阈值的量有些项目也会改成固定阈值cut两者各有好处。注意离散化之后必须保证每个节点在训练集和推理集里出现的取值类别完全一致否则fit()会报 “ValueError: Data must be a DataFrame” 或状态数量不匹配的异常。3.2 参数学习有数据时用最大似然估计数据少时用贝叶斯估计当历史数据里有明确的故障标签时就可以跳过手动设置 CPT 的繁琐步骤直接学到整张概率表。下面这段代码把刚才定义好的空模型model只包含结构不包含 CPD和样本数据train_df放在一起调用fit()from pgmpy.estimators import MaximumLikelihoodEstimator, BayesianEstimator # 方法一最大似然估计样本量大时推荐 model_mle model.copy() model_mle.fit(train_df, estimatorMaximumLikelihoodEstimator) # 方法二贝叶斯估计加入 BDeu 先验避免零概率 model_bdeu model.copy() model_bdeu.fit( train_df, estimatorBayesianEstimator, prior_typeBDeu, equivalent_sample_size10 ) # 检查学习到的 CPT print(model_bdeu.get_cpds(contact_force))逻辑说明MaximumLikelihoodEstimator的做法是直接统计每个父节点组合下子节点取值的频率。这种方法在样本充足时无偏但是当某个父节点组合下没有任何样本时CPT 里会出现 0 概率后续推理一旦落到这个分支就会直接得出不可能的结论。BayesianEstimator加了一个equivalent_sample_size10的 BDeu 先验相当于给每个父节点组合都预置了 10 条虚拟样本从数学上完全消除零概率问题。在受电弓故障样本很少的场景下我一般直接用第二种方法。参数说明equivalent_sample_size表示虚拟样本量的强度设得越大先验对最终概率的影响越大设得太小平滑效果不明显。一般取 5 到 20 之间即可。如果现场故障样本特别稀疏可以调大这个值让专家先验继续主导一小段时间。3.3 故障推理变元消去算法与证据更新网络结构和 CPT 都齐了之后核心工作就是推理给定一组观测到的传感器证据计算每个故障原因的后验概率。pgmpy 的VariableElimination提供精确推理适合几十个节点以内的网络。from pgmpy.inference import VariableElimination infer VariableElimination(model_bdeu) # 假设当前列车检测到气路压力低、拉弧次数高、滑板厚度偏小 evidence { air_pressure: low, arcing: high, slider_thickness: thin } # 分别查询三个根节点的后验概率 result_leak infer.query(variables[airbag_leak], evidenceevidence) result_wear infer.query(variables[slider_wear], evidenceevidence) result_stuck infer.query(variables[mechanism_stuck], evidenceevidence) print(气囊漏气概率:, result_leak) print(碳滑板磨损概率:, result_wear) print(机构卡滞概率:, result_stuck)逻辑说明query()的第一个参数是想要查询的目标节点列表第二个参数是证据字典。推理引擎会利用贝叶斯网络的条件独立性把联合分布拆成多个因子相乘再通过消元逐步边缘化掉无关变量从而高效算出后验分布。在这个例子里“拉弧次数高”并不是直接证据对应的根因但它会通过contact_force这一中间节点间接提高“气囊漏气”和“机构卡滞”的后验概率这正是贝叶斯网络相对阈值判据的核心优势能利用间接证据互相印证。参数说明evidence字典里的 key 必须和网络里设置的节点名完全一致value 必须是该节点状态列表里的合法取值。如果跑出KeyError优先检查预处理函数里的取值是否需要做字符串归一化。如果跑出ValueError: Can not use the given evidence说明状态类别和 CPT 里的类别对不上需要查看model_bdeu.get_cpds(节点名).state_names来确认真实的取值集合。3.4 连续量节点处理的另一种思路高斯贝叶斯网络二值化离散化的优点是简单可解释缺点是对阈值敏感。如果怕阈值选不好导致推理结果不稳定可以改用LinearGaussianBayesianNetwork让连续传感器变量直接作为节点然后让离散故障节点作为父节点、连续变量作为子节点为每个父节点状态组合拟合一个高斯分布。不过这个路线在工程上比纯离散网络麻烦因为它的 CPT 变成了均值方差参数表推理用的是联合树算法结果也不如离散网络直观。我的建议是第一阶段先用离散化方案上线等确认了稳定阈值再考虑连续高斯方案。4. 实际故障诊断流程与验证从后验概率到检修决策4.1 完整的诊断判断逻辑同时输出多条可能原因单次推理只给出一个后验概率还不够现场维修需要的是一个可解释的结论列表。实际系统里我不会只依赖某一个概率阈值而是先给根节点排序再结合一个置信度阈值做筛选def diagnose(model, evidence, top_k3, threshold0.3): infer VariableElimination(model) root_nodes [airbag_leak, slider_wear, mechanism_stuck] result {} for node in root_nodes: q infer.query(variables[node], evidenceevidence) # 取出 yes 状态对应的后验概率 state_index q.state_names[node].index(yes) result[node] q.values[state_index] # 按后验概率降序排列 sorted_result sorted(result.items(), keylambda kv: kv[1], reverseTrue) # 过滤低于阈值的项 final [(name, prob) for name, prob in sorted_result if prob threshold] return final[:top_k] # 示例某次运行数据 diagnosis diagnose( model_bdeu, {air_pressure: low, arcing: high, raise_time: slow} ) for node, prob in diagnosis: print(f故障原因: {node}, 后验概率: {prob:.2%})逻辑说明这段代码把三个根节点分别查一遍后验概率然后排序和过滤形成真正的故障诊断输出。state_index q.state_names[node].index(yes)这一步很关键因为 pgmpy 返回的values数组顺序可能和state_names列表一致但显式地用索引取值更稳妥避免代码里硬编码第几个下标。处理后输出结果就可以直接推送到检修终端或生成工单。参数说明threshold0.3是诊断灵敏度的调节旋钮。阈值调低会报告更多潜在故障召回率高但容易误报阈值调高只报概率最高的一两个故障精确率高但可能漏掉低概率并发故障。地铁运维场景里我会区分“预警”和“告警”两档0.3 以上只推送至日检列表0.6 以上直接生成故障工单。4.2 用一个完整案例把全流程串起来拿一列地铁某日早高峰的传感器数据举例。原始记录值如下气路压力值428 kPa → 离散化为low日累计拉弧次数27 次 → 离散化为high滑板剩余厚度24.5 mm → 离散化为normal升弓时间10.2 秒 → 离散化为slow用上面编号的网络推理得到的后验概率排序可能为升弓机构卡滞约 0.63气囊漏气约 0.28碳滑板磨损约 0.05。这个结果说明“升弓时间长接触压力波动”这两个证据强指向机构卡滞而滑板厚度正常又大幅降低了磨损的后验概率。把这份输出交给检修人员相当于直接把边界清晰的原因定位从几十条报警里提出来。4.3 用历史故障样本验证诊断准确率系统上线前必须回答一个问题诊断准确率到底多高验证方法和普通分类模型类似但要注意贝叶斯网络的“标签”是根节点的真实故障状态。用留一法交叉验证的思路写一个评估循环from sklearn.metrics import accuracy_score, precision_recall_fscore_support def evaluate(model, data, labels, root_node): infer VariableElimination(model) preds [] for _, row in data.iterrows(): # 把这一行的所有观测变量转换为证据 evidence { col: row[col] for col in data.columns } q infer.query(variables[root_node], evidenceevidence) state_index q.state_names[root_node].index(yes) preds.append(q.values[state_index] 0.5) acc accuracy_score(labels, preds) precision, recall, f1, _ precision_recall_fscore_support( labels, preds, averagebinary ) return acc, precision, recall, f1 acc, prec, rec, f1 evaluate( model_bdeu, test_df[observable_columns], test_df[slider_wear_true] ) print(fAccuracy: {acc:.3f}, Precision: {prec:.3f}, Recall: {rec:.3f}, F1: {f1:.3f})逻辑说明对每条测试样本把除根节点外的所有可观测列都作为证据传入推理引擎然后看yes状态的后验概率是否超过 0.5。注意这里的data.columns必须只包含模型里的观测变量不能把slider_wear_true这类标记列也放进去否则等于把答案提前告诉了推理引擎。参数说明0.5 只是默认阈值验证时可以循环换成 0.3、0.4、0.6 等数值画出不同阈值下的精确率和召回率曲线。这一步能帮助业务方决定线上系统把预警线放在哪里。如果某条线路的故障率本来就低看整体 Accuracy 意义不大更应该关注 Minority 类别的 Recall。5. 部署经验与调试技巧样本不平衡、零概率与代码运行环境5.1 样本不平衡真实故障太少模型学不到“异常模式”地铁受电弓的真实故障率非常低气囊漏气可能一年也没几次直接拿全量样本做参数学习学到的先验概率会极度偏向正常。这会导致一个典型问题即使气路压力已经低到报警线由于先验概率正常类太高后验概率依然不高诊断被“正常”主导。解决办法有三个层次。第一采样时做类平衡训练贝叶斯网络时把故障样本复制若干倍或者用 SMOTE 这类算法合成故障样本但合成样本要谨慎避免制造出不存在的父节点组合。第二调大 BDeu 的equivalent_sample_size会影响先验强度但对样本量极端的类更有效的是直接手动设置根节点先验把现场统计到的故障频率写进 CPT。第三不要在整车上训练一个“通用受电弓模型”而是按线路分组训练因为不同线路的接触网状态差异很大合并训练会把线路特征稀释掉。5.2 零概率灾难某个证据组合在训练集里从没出现过当证据组合复杂到一定程度训练数据覆盖不到某些父节点取值组合是常见现象。比如airbag_leakyes且mechanism_stuckyes的并发故障在历史样本里一次都没出现最大似然估计就会把contact_forcenormal的概率学到 0。line 推理时一旦证据命中这个分支概率结果直接失真。解决办法是前面提到的贝叶斯估计加prior_typeBDeu。此外还有一个更细的操作对所有 CPT 里为 0 的单元格做拉普拉斯平滑在拟合完成后再次检查def smooth_zero_cpds(model, alpha0.01): for cpd in model.get_cpds(): values cpd.values.copy() values[values 0] alpha # 重新归一化确保每列概率和为 1 normalized values / values.sum(axis0) cpd.values normalized return model逻辑说明这段代码在模型推理之前把所有零概率替换成一个极小值然后按列重新归一化。alpha0.01适合样本量较大的情况如果样本量小可以取 0.05 到 0.1。这个技巧在“专家搭建的 CPT 少量数据修正”混合模式下尤其好用因为专家也可能漏填某个组合。5.3 用 BIF 文件持久化模型方便跨环境复用模型训练好之后不能每次都从原始数据重新训练。pgmpy 支持把整个贝叶斯网络包括结构、CPT、节点状态写成标准 BIF 格式这样在列车诊断主机、地面服务器、云平台之间迁移时就只要拷贝文件from pgmpy.readwrite import BIFReader, BIFWriter # 保存 BIFWriter(model_bdeu).write_bif(pantograph_diagnosis.bif) # 加载 model BIFReader(pantograph_diagnosis.bif).get_model()逻辑说明BIF 文件是纯文本的 XML 格式可以通过版本管理工具统一追踪每条线路的网络版本。现场升级模型时只需要把新的 BIF 文件下发到列车的边缘计算单元重启推理服务即可生效。注意保存前确认所有 CPT 都已完成平滑否则加载后再去改会比较绕。5.4 关于代码运行环境的一个实操提示这套代码不挑运行环境最常见的执行方式有三种在 WSL Ubuntu 的终端里直接跑 Python 脚本在 VS Code 里新建 Jupyter Notebook 一步步调试或者把诊断代码封装成 FastAPI 服务部署在服务器上。我建议在本地调试阶段直接用python diagnosis.py运行把上面提到的函数按顺序组织在同一个文件里入口处加上if __name__ __main__:调用一次完整推理方便验证数据流是否畅通。把print(模型加载完成等待传感器数据...)放在入口位置测试时能清晰看到脚本执行到哪一步停了。最后想留一个实用的验证技巧当你在现场拿到一批新的运行数据不要急着把整个模型重新训练一遍而是先用旧模型对单条异常样本做推理把根节点的后验概率分布画成柱状图对比新增样本前后排序是否发生变化。如果某条证据加入后概率排序基本不变说明这条证据对区分故障原因几乎没有贡献后续可以把它从观测变量里逐步剔除简化网络的维护成本。本文还有配套的精品资源点击获取
原理与C语言实现详解)
