简介:面向经济学研究者、供应链分析人员与数据科学实践者,这份资源包完整复现丁浩员等发表于《经济研究》2024年第8期的跨国供应链断裂与重构测度方法。围绕FactSet Revere全球供应链数据库,给出从数据读取、关系筛选到断裂(Break)、恢复(Recover)、转移(Transfer)三大指标计算的Python实现路线,并提供数据预处理、时间顺序调整和异常值处理等关键技术细节。资源共25个文件,压缩包仅1.27MB,以8个Python脚本、8个Excel数据与结果文件、3个TXT说明文档、2个HTML可视化看板及分析图为主体,覆盖数据处理、指标测算、结果演示与图文说明。目前已有875人学习下载,适合需要复现该指标或将其迁移到自身研究场景的读者。除完整可运行代码外,还包含扩展数据脚本、供应链关系筛选工具、分析图表和Dashboard,可直接对照操作,为供应链稳定性、抗风险能力和动态变化研究提供一套可落地的技术参考。
1. 供应链断裂与重构测度:把“断供风险”从Excel透视表里捞出来
供应链断裂与重构测度,并不是一套只能写论文的数学模型。它解决的是更具体的业务问题:核心供应商停摆时,产线要多久能恢复、有多少替代路径可用、要不要提前备一条备胎。很多团队只看采购金额占比做风险排序,结果断供之后才发现风险藏在网络结构里而不是金额里。这篇文章用Python加NetworkX,把多层供应链网络拆成可复现的断裂指标和重构指标,从数据准备、参数调整到失效模拟完整走一遍,适合需要给供应链韧性做量化评估的分析师、数据工程师和供应链规划人员。
2. 断裂测度与重构测度的基础:为什么金额占比不是风险的真相
2.1 金额占比高不等于风险高,网络结构才是断裂风险的真相
先看一个实际评估过的场景。某组件厂给下游总装厂供货,金额占该总装厂采购额的80%,看起来是“命脉供应商”。但深入看网络,这家组件厂有四个经过认证的替代供应商,其中两个已经在小批量供货,这时候它的真实断裂风险远没有金额占比看起来那么高。反过来,另一个供应商金额占比只有15%,但它是某种特殊材料的唯一认证来源,替代认证周期要18个月,一旦断掉整条产线直接停摆。
这就是网络视角和金额视角的区别。供应链在本质上是一张有向带权图:节点是工厂或贸易商,边是供货关系,权重可以是金额、数量或采购份额。断裂测度要回答的是“某些节点失效后,网络里还有多少条路能走通”;重构测度要回答的是“走不通之后,需要付出多大代价才能重新走通”。这两个问题,Excel透视表和常规BI看板都答不了,因为它们没有图结构的概念。我一般会把供应链风险拆成三个层次:节点层看某家供应商本身的依赖度、集中度和枢纽程度;路径层看原材料到总装厂之间有多少条带权重的可达路径;网络层看整张图在移除若干节点后,最大连通分量和交易量保留率的变化。
这三个层次正好对应三组不同测度指标。节点层适合做供应商评级,路径层适合做关键物料分析,网络层适合做整体韧性评估。多数团队用网络层指标做汇报,用节点层指标做日常监控,路径层指标在年度复盘时集中用。
另一个容易被忽视的问题是边的方向。供应链关系天然是有向的——原材料流向组件厂,组件流向总装厂,反向没有意义。但很多人在建图时习惯用Graph而不是DiGraph,这是个隐患。用无向图算介数中心性,会把本身并不关键的中间节点捧到高位,因为它们“看起来”连接了左右两边的所有节点。我遇到过有团队用无向图跑完关键节点识别,结果排第一的是一个只有两条边的中转仓库,原因是它的位置恰好在两个密集社区之间。这就是方向性没处理好带来的误导。
在所有节点级指标里,我使用频率最高的是HHI(赫芬达尔-赫希曼指数)和单一依赖度。HHI把供应商份额的平方求和,份额越集中的网络数值越接近1,完全分散的网络趋近于0。选它而不是简单的CR3(前三大供应商份额之和),是因为HHI对份额分布的均匀程度更敏感:三个各占33%的供应商,和“一个占90%、两个各占5%”的情况下,CR3都是99%,但HHI能明显区分这两种结构——后者一旦那个90%份额的供应商出问题,短期内完全没有补位能力。
2.2 重构测度不是“能不能恢复”,而是“花多大代价恢复”
说完断裂,再来说重构。重构测度在学术上有很多复杂定义,工程落地时我只需要回答三个问题。第一个是结构冗余:失效发生后,网络中还有多少条替代路径能把物料送到目标工厂,衡量方式是最短替代路径长度、替代路径数量,以及最大连通分量大小。第二个是替换成本:要恢复原有可达性,最少需要新增或修复几条边,这个指标可以直接换算成资金和时间预估。第三个是恢复优先级:先修哪条边效果最好,做法是对每条候选边做恢复模拟,看哪条边的恢复对网络连通性改善最大。
选这三个维度做重构测度,不是因为指标多显得专业,而是因为它们正好对应业务上的三句话:“有没有备胎”、“备胎要多久上”、“先修哪条最划算”。把这三句话翻译成代码,就是第4章要写的失效模拟和边恢复排序。在设计测度代码时,我强烈建议把断裂测度和重构测度分开实现。断裂测度是静态的,一次数据快照就能算完;重构测度是动态的,需要对失效场景做枚举模拟。两者混在一个脚本里,后续改参数时会非常痛苦,这是很多公开示例代码的通病。
数据粒度是另一个决定测度结果质量的因素。如果边表里的金额是月度采购金额,而节点属性表里的产能是年度产能,两者混在一起算出来的权重比例会失真。我一般要求数据至少达到“供应商-客户-物料”的三级粒度,即一条边记录是哪家供应商给哪个客户供哪种物料,金额单独列。有了物料维度之后,同一个供应商和客户之间即使有多条边,也能通过物料品类区分风险场景,而不是用一个汇总金额把不同类型风险揉在一起。
最后补一个指标说明。介数中心性我在断裂测度里定位为“结构级风险”而不是“业务级风险”。它衡量的是一个节点出现在多少条最短路径上,介数高的节点一旦失效,大量路径会被切断。这个指标在原材料层和组件层最有用,因为总装厂通常是网络的汇聚端,介数天然偏高,排出来意义不大,反而会干扰判断。
3. 用NetworkX在本地计算断裂风险:从边表到HHI与关键节点
3.1 供应链网络数据的最小字段与组织方式
开始写代码之前,先把数据格式定下来。我见过很多用Excel管理供应链数据的团队,字段千奇百怪,但真正喂给测度模型的最少只需要一张边表、三个字段:supplier是供应商节点ID,customer是客户节点ID,amount_ratio是该客户向该供应商采购金额占其总采购金额的比例。这里有个细节:amount_ratio用比例而不是绝对金额。不同规模的客户之间绝对金额不可比,比例天然做了归一化,HHI和依赖度算出来才有跨节点比较的意义。如果源数据只有绝对金额,读入后按customer分组求和,再除以组内总和即可。
除了边表,建议再准备一张节点属性表,至少包含node_id、node_type(原材料/组件/总装)、region(区域)、critical_flag(是否关键物料)。下面的演示代码直接生成一张12节点、14条边的三层供应链网络,不需要额外创建文件,拿到就能跑。我故意在结构里埋了两个风险点:总装厂plant_4只依赖组件厂comp_4,而comp_4的上游又只依赖raw_3和raw_4两家,形成一条相对狭窄的链路。
import networkx as nx import pandas as pd import numpy as np # 12个节点、14条有向边,weight表示采购金额比例(小数形式) # 每一行:supplier -> customer, amount_ratio edges = [ ('raw_1', 'comp_1', 0.80), ('raw_2', 'comp_1', 0.20), ('raw_1', 'comp_2', 0.50), ('raw_3', 'comp_2', 0.50), ('raw_2', 'comp_3', 0.65), ('raw_4', 'comp_3', 0.35), ('raw_3', 'comp_4', 0.40), ('raw_4', 'comp_4', 0.60), ('comp_1', 'plant_1', 0.70), ('comp_2', 'plant_1', 0.30), ('comp_2', 'plant_2', 0.45), ('comp_3', 'plant_2', 0.55), ('comp_3', 'plant_3', 0.50), ('comp_4', 'plant_3', 0.50), ('comp_4', 'plant_4', 1.00), ] G = nx.DiGraph() for supplier, customer, ratio in edges: G.add_edge(supplier, customer, weight=ratio) # 校验:每个客户节点的入边权重之和应等于1 # 不为1说明源数据有漏记,后续HHI会失真 for node in G.nodes: in_weight = sum(d['weight'] for _, _, d in G.in_edges(node, data=True)) if in_weight > 0 and abs(in_weight - 1.0) > 1e-6: print(f'警告: {node} 的入边比例之和为 {in_weight:.4f}') print('网络节点数:', G.number_of_nodes()) print('网络边数:', G.number_of_edges())这段代码做了三件事:定义边表,用networkx的DiGraph建一张有向带权图,最后检查每个下游节点是否满足“入边权重合计为1”的约束。最后一步的校验脚本很重要,上游漏记一条边就会导致比例失衡,直接让后续所有指标失真。我在实际项目中会把这个校验做成数据流水线的一部分,而不是只在初始建模时跑一次。
weight参数是比例值的小数形式,范围0到1。如果你的源数据里是百分比数字(比如80而不是0.8),在add_edge之前统一除以100,否则后面算HHI时平方和会异常偏大。这个错误在真实数据里出现频率比想象中高,我见过有团队用百分比跑完发现HHI全在0.9以上,还以为是风险极高,其实是量纲错了。
3.2 计算节点级断裂风险:依赖度与HHI
节点级断裂风险固定用两个指标:单一供应商依赖度和供应商集中度HHI。前一个回答“某个客户有多依赖它的最大供应商”,后一个回答“整个供应商结构是分散还是集中”。两个指标的实现都很短,但格式上有讲究。
def supplier_dependency(G, customer): """返回某客户所有供应商的份额,按降序排列。 份额最大的供应商就是单一依赖的源头。 """ in_edges = [(s, d['weight']) for s, _, d in G.in_edges(customer, data=True)] if not in_edges: return [] total = sum(w for _, w in in_edges) sorted_edges = sorted(in_edges, key=lambda x: x[1], reverse=True) return [(s, w / total) for s, w in sorted_edges] def hhi_index(G, customer): """计算某客户的供应商集中度HHI,取值介于0和1之间。 HHI越接近1,供应商结构越集中,断裂风险越高。 """ in_edges = [(s, d['weight']) for s, _, d in G.in_edges(customer, data=True)] if not in_edges: return 0.0 total = sum(w for _, w in in_edges) return sum((w / total) ** 2 for _, w in in_edges) for customer in ['plant_1', 'plant_2', 'plant_3', 'plant_4']: deps = supplier_dependency(G, customer) hhi = hhi_index(G, customer) print(f'{customer}: HHI={hhi:.3f}, 最大供应商份额={deps[0][1]:.2%}') print(f' 供应商明细: {deps}')这段代码的关键逻辑是:先通过G.in_edges取出某客户的全部上游边,再按权重归一化得到份额。supplier_dependency返回排序后的供应商列表,hhi_index直接对份额平方求和。两者都写了防御式逻辑——没有入边时返回空列表或0,避免后续流程崩溃。
customer参数直接传节点ID字符串。G必须是networkx的有向图对象,如果传无向图,in_edges和out_edges的语义就乱了,算出来的依赖度和HHI都会出错。注意:演示网络中plant_4的入边只有comp_4一条,所以它的HHI必然是1.0。这不是bug,正是我们要暴露的极端集中风险——单一供应商在金额口径下就是100%依赖。
3.3 定位结构性断裂源:介数中心性
做完节点级风险,第二步是找出“断了之后影响面最大”的节点。介数中心性衡量一个节点出现在多少条最短路径上,排在越前面,说明该节点失效会切断大量上下游物料流动路径。
在带权有向图上计算介数中心性时,NetworkX会把weight参数当成“距离”来处理,所以金额比例要取倒数作为距离——份额越大说明关系越紧密,对应的路径代价越小。
# 权重是金额比例,数值越大关系越紧密 # 取倒数转换为“距离”,份额越大距离越短 distance_G = G.copy() for u, v, d in distance_G.edges(data=True): distance_G[u][v]['distance'] = 1.0 / d['weight'] bc = nx.betweenness_centrality(distance_G, weight='distance', normalized=True) for node, value in sorted(bc.items(), key=lambda x: x[1], reverse=True): print(f'{node}: 介数中心性={value:.4f}')这里有一个容易踩坑的细节:如果直接给betweenness_centrality传weight='weight',NetworkX会把占比当距离,那么一条0.8份额的边会被当成短距离,一条0.2份额的边被当成“更长”的距离,计算出来的最短路径会偏向高份额边。取倒数后才符合“高份额=低成本=更常用路径”的业务含义。
normalized=True表示把结果归一化到0到1之间,方便跨网络比较。介数排名出来后,我一般结合3.2的HHI结果一起看。比如在这个演示网络中,comp_4的介数排名未必最高,但plant_4对它的依赖是100%,这时候业务上要优先盯comp_4的连续性和替代方案,而不是只看介数排行榜。
4. 重构测度:用失效模拟量化“断链后要花多大代价恢复”
4.1 失效模拟:把“断链”从故事变成数字
重构测度的起点是失效模拟。常见做法是枚举失效场景:单节点失效、双节点同时失效、按概率随机选N个节点失效。对每个场景,删除失效节点及其关联的所有边,再计算网络关键指标的变化。单节点失效的枚举规模是O(N),双节点是O(N²),当N超过200时双节点全枚举要跑四万次。工程上我一般先用单节点枚举做初筛,把介数排名前20%的节点拎出来做双节点组合,而不是全量枚举。
def simulate_failure(G, failed_nodes): """返回失效后的交易量保留率、最大弱连通分量节点数。 failed_nodes: 需要模拟失效的节点ID列表。 """ H = G.copy() H.remove_nodes_from(failed_nodes) # 交易量保留率 = 剩余边的总权重 / 原始边的总权重 original_weight = sum(d['weight'] for _, _, d in G.edges(data=True)) if original_weight == 0: return 0.0, 0 remaining_weight = sum(d['weight'] for _, _, d in H.edges(data=True)) weight_retention = remaining_weight / original_weight # 最大弱连通分量节点数:衡量整体连通骨架的残存规模 if H.number_of_nodes() == 0: largest_cc_size = 0 else: largest_cc_size = max(len(cc) for cc in nx.weakly_connected_components(H)) return weight_retention, largest_cc_size scenarios = ['comp_4', 'raw_1', 'raw_4', 'comp_2'] for node in scenarios: retention, cc_size = simulate_failure(G, [node]) print(f'失效节点 {node}: 交易量保留率={retention:.2%}, ' f'最大连通分量节点数={cc_size}')这个函数先复制原图再remove_nodes_from,确保每次模拟不污染原始图。remove_nodes_from会连带删除该节点所有入边和出边——一家供应商倒闭,它的采购和销售关系同时断掉。weight_retention直接反映经济损失面,比如模拟结果中raw_1失效时保留率约84%,comp_4失效时保留率会低不少,因为comp_4带着两条入边和两条出边一起消失,影响面是双层的。
最大弱连通分量节点数用的是weakly_connected_components,这个参数在有向图里很关键。有向图存在强连通和弱连通两种语义:强连通要求路径方向一致才能互通,弱连通只要求忽略方向后仍然连接。供应链评估关注的是“物料能不能绕过断点到达”,方向是核心约束,但弱连通分量能反映骨架存留,所以我习惯两个都看,函数里先返回弱连通规模作为整体连通性的粗略指标。
4.2 重构成本:最少需要恢复几条边
失效模拟只回答了“断得有多重”,重构测度要回答“要花多大代价走通”。我常用的量化方式是计算最小恢复边数:从失效后的网络出发,找出最少需要新增或修复几条边,才能让每一个总装厂都能到达至少一个原材料节点。
要注意的是,全局最优恢复方案是NP-hard问题,工程上退一步用贪婪算法逼近。每个总装厂独立检查可达性,如果不可达就补一条从最近上游到它的直连边。
def min_recovery_edges(G, failed_nodes, downstream_nodes, upstream_nodes): """贪婪估算恢复全部下游到上游可达性所需的最小边数。 返回: 新增边列表, 所需边数。 """ H = G.copy() H.remove_nodes_from(failed_nodes) # 找出当前不可达的下游节点 unreachable = [] for d_node in downstream_nodes: if d_node not in H: unreachable.append(d_node) continue # 用has_path检查从任意上游节点到该下游是否可达 reachable = any(nx.has_path(H, u, d_node) for u in upstream_nodes if u in H) if not reachable: unreachable.append(d_node) recovery_edges = [] for d_node in unreachable: # 找一个距离最短的上游节点,直接补一条边 best_u = None best_dist = float('inf') for u in upstream_nodes: if u not in H: continue # 在原图G里计算距离,用于排序候选 if G.has_edge(u, d_node): dist = 1.0 / G[u][d_node]['weight'] if dist < best_dist: best_dist = dist best_u = u if best_u is not None: recovery_edges.append((best_u, d_node)) # 0.5作为新合约份额占位,不影响“需要补几条边”的结论 H.add_edge(best_u, d_node, weight=0.5) return recovery_edges, len(recovery_edges) downstream = ['plant_1', 'plant_2', 'plant_3', 'plant_4'] upstream = ['raw_1', 'raw_2', 'raw_3', 'raw_4'] for node in ['comp_4', 'raw_1', 'comp_3']: rec_edges, num = min_recovery_edges(G, [node], downstream, upstream) print(f'失效 {node} 后,最少需恢复 {num} 条边: {rec_edges}')这里逐节点判断可达性,has_path用BFS遍历,单次复杂度O(V+E)。如果网络规模到几千个节点,建议先用all_pairs_shortest_path预计算所有节点对的最短路径,避免在循环里反复调用has_path。
参数说明:failed_nodes是失效节点列表;downstream_nodes是下游需求点集合,通常是总装厂或核心客户;upstream_nodes是上游起点集合,通常是原材料或一级供应商。0.5的权重占位只是一个符号,它会影响后续介数计算,但对“需要补几条边”的估算没有影响。真实业务中,新增一个供应商往往要经过认证、试产、小批量验证多个环节,最小恢复边数乘以平均认证周期就能粗略换算成恢复时间。
4.3 输出一张可汇报的测度结果表
把断裂和重构指标整合成一张宽表,是给业务方汇报时最实用的格式。每行一个失效场景,列包含场景名称、交易量保留率、最大连通分量节点数、最小恢复边数。这样看板直接展示“如果X断了,会损失多少、需要多久恢复”这两个决策者真正关心的数字。
def build_risk_report(G, scenarios, downstream, upstream): rows = [] for scn in scenarios: # 兼容单节点和双节点失效场景 failed = scn.split('+') retention, cc_size = simulate_failure(G, failed) rec_edges, num = min_recovery_edges(G, failed, downstream, upstream) rows.append({ '失效场景': scn, '交易量保留率': round(retention, 4), '最大连通分量节点数': cc_size, '最小恢复边数': num, }) return pd.DataFrame(rows) scenarios = ['raw_1', 'raw_4', 'comp_1', 'comp_4', 'raw_1+comp_4'] report_df = build_risk_report(G, scenarios, downstream, upstream) print(report_df.to_string(index=False))这份报告可以直接放进周会材料。我一般会再按“最小恢复边数”和“交易量保留率”两个维度做四象限:恢复边数少且保留率高的场景是低风险,恢复边数多且保留率低的场景必须提前做备份预案。用matplotlib画出来就是一张可以给管理层看的图,这部分不用额外开发,从上述DataFrame取数即可。
5. 供应链测度避坑清单:五个翻车现场的现象、原因与解决
5.1 幽灵节点:同一家供应商被拆成多个节点
现象:跑出来的依赖度普遍偏低,HHI整体失真,某个明明很关键的供应商在结果里排不进前五。
原因:数据清洗环节没有做名称标准化。供应链数据经常从多个系统导出,同一家公司在采购系统里叫“某科技”,在物流系统里叫“某科技(一厂)”,在Excel台账里又叫“某科技股份”,被当成三个节点建图后,它真实的份额被拆散了。
解决:在建图之前先做节点归一化。常见做法是把名称里的括号、地域后缀、公司类型后缀去掉再做精确匹配;更稳妥的是维护一张供应商别名映射表,把多个名称映射到同一个标准ID。我在项目里会在读入边表后立刻跑一遍重复检测,输出疑似重复的节点对人工审核,而不是直接进图。
5.2 权重口径混用:金额、数量、份额一锅炖
现象:HHI算出来接近1甚至等于1,单一依赖度全部超过90%,结果一看就不可信。
原因:边表的weight字段有的是金额,有的是物料数量,有的是手工填的份额。金额大但用量少的物料在金额口径下被放大,数量大但单价低的物料又被低估。不同客户之间的weight完全不在一个量纲上。
解决:统一口径再做归一化。如果数据源给的是金额,就按金额比例归一;给的是数量,就按数量比例归一,绝对不能混用。加一个自动化校验:每个客户的入边权重之和应该等于1或接近1,误差超过2%就在日志里标记。第3章代码里的校验雏形可以直接扩展成数据流水线规则。
5.3 失效模拟修改了原图,导致后续场景全被污染
现象:模拟多个失效场景时,第二个场景的计算结果明显偏大,像是被上一次模拟“传染”了。
原因:simulate_failure函数内部如果直接在原始图G上执行remove_nodes_from,第一次模拟就会把节点永久删掉。后续所有场景都在残缺图上计算,结果当然失真。更隐蔽的一个坑是:G.copy()是浅拷贝,如果边属性里嵌套了dict或list,浅拷贝只复制引用,修改内部key仍然会污染原图。
解决:所有失效模拟都必须在图副本上进行。函数开头写H = G.copy(),所有修改只作用于H。如果权重字段是简单浮点数,copy()够用;一旦权重或属性包含嵌套对象,直接用copy.deepcopy。这也是我在4.1代码里为什么特意先用G.copy()再remove_nodes_from的原因。
5.4 在无向图上算介数中心性,把中转仓库排到第一
现象:关键节点排序结果里,一个只有两条边的小仓库排在最前面,业务同事看了完全不信。
原因:用了Graph建图,所有边变成双向,方向语义丢失。供应链是原材料流向总装厂的单向网络,无向化之后每条最短路径的中间节点激增,处于两个密集社区之间的小仓库在无向图上“看起来”连接了全局。
解决:一律使用DiGraph建图。计算介数中心性时传给betweenness_centrality的必须是同一张有向图,weight参数传的是距离而不是份额(取倒数的原因第3章已经详述)。NetworkX在DiGraph上默认计算有向最短路径,不需要额外参数,只要图类型正确即可。
5.5 时间窗口错配:月度快照和年度汇总混算
现象:每个月风险报告波动剧烈,某个月HHI突然升高,下个月又恢复正常,业务团队觉得指标不稳定不可信。
原因:数据更新频率不一致。有些边表来自月度采购数据,有些来自年度框架协议汇总。年度汇总的边权重在每个月里恒定,月度数据波动大,混合后HHI自然跟着月度波动上下跳动,掩盖了真实的结构性变化。
解决:统一以月度为采样频率做快照,每个月的边表都从当月采购数据生成,确保所有边的统计周期一致。如果某些品类只有年度数据,就在月度快照中沿用最近一期年度值,但单独加一列data_source标记,方便后续排查该值是本月实测还是年度沿用。
6. 从静态快照到滚动监测:把测度指标变成一个每周跑一次的信号
单一快照的测度结果只能说明“当前网络在什么样的风险状态下”。供应链是动态的:供应商可能被替换、新增、淘汰,采购份额也在逐月变化。我最后会把这个测度脚本封装成每周自动运行的函数,输出一张随时间变化的指标趋势表,用两个固定阈值做触发告警。
做法是维护一个历史目录,每周导出一次边表快照,文件名带周序号,然后循环读入所有历史快照,逐周计算HHI和关键节点,汇总成趋势DataFrame。代码结构大致如下:
def weekly_snapshot_metrics(snapshot_dir): """读取一周一张的边表快照,返回每周风险指标趋势。""" import glob rows = [] for path in sorted(glob.glob(f'{snapshot_dir}/edges_*.csv')): week = path.split('_')[-1].replace('.csv', '') df = pd.read_csv(path) G_week = nx.DiGraph() for _, r in df.iterrows(): # 同时存weight和distance,避免后面重复取倒数 G_week.add_edge( r['supplier'], r['customer'], weight=r['amount_ratio'], distance=1.0 / r['amount_ratio'] ) # 全网络平均HHI hhi_vals = [ hhi_index(G_week, n) for n in G_week.nodes if G_week.in_degree(n) > 0 ] avg_hhi = np.mean(hhi_vals) if hhi_vals else 0.0 # 关键节点:先算一次全图介数,再取最大值 bc_full = nx.betweenness_centrality(G_week, weight='distance') top_node = max(bc_full, key=bc_full.get) rows.append({ 'week': week, 'avg_hhi': round(avg_hhi, 4), 'top_node': top_node, }) return pd.DataFrame(rows)这段代码有两点可以按场景调整。第一,avg_hhi的聚合方式,如果管理层更关心重点物料供应商的集中度,可以把全网络平均改成加权平均,权重取节点入边的总金额。第二,top_node这里每行都重新算一次全图介数,网络上千节点时会比较慢。建议每周快照时只算一次并落库,而不是每次展示时现算。这里我直接把distance属性在add_edge时写入,就是为了避免每周快照时再遍历一遍边做倒数转换。
我自己的习惯是设置两个阈值:avg_hhi超过0.4认为整体供应链集中度过高,触发复查;top_node连续两周不变且该节点HHI超过0.6,触发专项备份预案。阈值本身需要结合业务规模标定,0.4和0.6只是经验起点。跑几周历史数据后回看,找出“指标异常但实际没出事”和“指标正常但实际出事了”两类周次,再反向调整阈值。
这个滚动监测的收益在于把断裂与重构测度从一个一次性分析项目变成持续运营的能力。供应商风险是缓慢累积的——份额的微小移动、新供应商的悄然进入,单看任何一周都不起眼,连续几个月看趋势,结构变化就很明显。我最早做这套东西时也是频繁翻车,后来靠每周固定跑、固定留档、固定对比趋势,才真正把指标用起来。希望这套思路和代码能帮你把手里的供应链数据盘活,不只在断供发生后做复盘,而是提前看见风险。希望帮到你。
本文还有配套的精品资源,点击获取