本文目录导读：
我来详细介绍如何使用NetworkX进行网络鲁棒性分析,包含多个实用案例。
基础环境准备
import networkx as nx import numpy as np import matplotlib.pyplot as plt import random from collections import Counter # 设置中文显示 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False
创建网络模型
def create_networks():
"""创建不同类型的网络"""
networks = {}
# 随机网络 (ER模型)
networks['ER'] = nx.erdos_renyi_graph(100, 0.05, seed=42)
# 小世界网络 (WS模型)
networks['WS'] = nx.watts_strogatz_graph(100, 4, 0.1, seed=42)
# 无标度网络 (BA模型)
networks['BA'] = nx.barabasi_albert_graph(100, 3, seed=42)
return networks
随机故障模拟
def random_failure_simulation(G, steps=100):
"""随机故障模拟"""
G_copy = G.copy()
n_initial = G_copy.number_of_nodes()
results = {
'nodes_removed': [],
'largest_component': [],
'efficiency': []
}
nodes = list(G_copy.nodes())
for i in range(min(steps, len(nodes))):
# 记录当前状态
results['nodes_removed'].append(i)
# 计算最大连通分量
if nx.is_connected(G_copy):
largest_cc = G_copy.number_of_nodes()
else:
largest_cc = len(max(nx.connected_components(G_copy), key=len))
results['largest_component'].append(largest_cc / n_initial)
# 计算网络效率
try:
efficiency = nx.global_efficiency(G_copy)
except:
efficiency = 0
results['efficiency'].append(efficiency)
# 随机移除一个节点
if nodes:
node = random.choice(nodes)
G_copy.remove_node(node)
nodes.remove(node)
return results
蓄意攻击模拟
def targeted_attack_simulation(G, steps=100):
"""蓄意攻击模拟（按度中心性）"""
G_copy = G.copy()
n_initial = G_copy.number_of_nodes()
results = {
'nodes_removed': [],
'largest_component': [],
'efficiency': []
}
for i in range(min(steps, n_initial)):
results['nodes_removed'].append(i)
# 计算最大连通分量
if nx.is_connected(G_copy):
largest_cc = G_copy.number_of_nodes()
else:
largest_cc = len(max(nx.connected_components(G_copy), key=len))
results['largest_component'].append(largest_cc / n_initial)
# 计算网络效率
try:
efficiency = nx.global_efficiency(G_copy)
except:
efficiency = 0
results['efficiency'].append(efficiency)
# 按度中心性移除节点
if G_copy.number_of_nodes() > 0:
degrees = dict(G_copy.degree())
target = max(degrees, key=degrees.get)
G_copy.remove_node(target)
return results
边攻击模拟
def edge_attack_simulation(G, steps=200):
"""边攻击模拟"""
G_copy = G.copy()
n_initial = G_copy.number_of_nodes()
results = {
'edges_removed': [],
'largest_component': [],
'efficiency': []
}
edges = list(G_copy.edges())
for i in range(min(steps, len(edges))):
results['edges_removed'].append(i)
# 计算最大连通分量
if nx.is_connected(G_copy):
largest_cc = G_copy.number_of_nodes()
else:
largest_cc = len(max(nx.connected_components(G_copy), key=len))
results['largest_component'].append(largest_cc / n_initial)
# 计算网络效率
try:
efficiency = nx.global_efficiency(G_copy)
except:
efficiency = 0
results['efficiency'].append(efficiency)
# 移除边介数中心性最高的边
if G_copy.number_of_edges() > 0:
edge_betweenness = nx.edge_betweenness_centrality(G_copy)
target = max(edge_betweenness, key=edge_betweenness.get)
G_copy.remove_edge(*target)
return results
可视化对比分析
def plot_robustness_comparison(results_dict, metric='largest_component'):
"""绘制鲁棒性对比图"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# 随机故障对比
ax1 = axes[0]
for network_name, results in results_dict.items():
data = results['random'][metric]
ax1.plot(data['nodes_removed'], data,
label=f"{network_name}", linewidth=2)
ax1.set_xlabel('移除节点数', fontsize=12)
ax1.set_ylabel('最大连通分量比例', fontsize=12)
ax1.set_title('随机故障下的网络鲁棒性', fontsize=14)
ax1.legend()
ax1.grid(True, alpha=0.3)
# 蓄意攻击对比
ax2 = axes[1]
for network_name, results in results_dict.items():
data = results['targeted'][metric]
ax2.plot(data['nodes_removed'], data,
label=f"{network_name}", linewidth=2)
ax2.set_xlabel('移除节点数', fontsize=12)
ax2.set_ylabel('最大连通分量比例', fontsize=12)
ax2.set_title('蓄意攻击下的网络鲁棒性', fontsize=14)
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
综合鲁棒性分析
def comprehensive_robustness_analysis(networks):
"""综合分析不同网络的鲁棒性"""
results = {}
for name, G in networks.items():
results[name] = {
'random': random_failure_simulation(G),
'targeted': targeted_attack_simulation(G),
'edge': edge_attack_simulation(G)
}
# 绘制对比图
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
for idx, (network_name, network_results) in enumerate(results.items()):
ax = axes[idx // 2, idx % 2]
# 绘制随机故障和蓄意攻击
random_data = network_results['random']['largest_component']
targeted_data = network_results['targeted']['largest_component']
ax.plot(random_data['nodes_removed'], random_data,
label='随机故障', linewidth=2, color='blue')
ax.plot(targeted_data['nodes_removed'], targeted_data,
label='蓄意攻击', linewidth=2, color='red')
ax.set_xlabel('移除节点数', fontsize=10)
ax.set_ylabel('最大连通分量比例', fontsize=10)
ax.set_title(f'{network_name}网络鲁棒性', fontsize=12)
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
return results
鲁棒性指标计算
def calculate_robustness_metrics(G, attack_strategy='random'):
"""计算网络鲁棒性指标"""
metrics = {}
# 基础指标
metrics['nodes'] = G.number_of_nodes()
metrics['edges'] = G.number_of_edges()
metrics['density'] = nx.density(G)
metrics['avg_degree'] = np.mean([d for n, d in G.degree()])
# 连通性指标
if nx.is_connected(G):
metrics['diameter'] = nx.diameter(G)
metrics['avg_path_length'] = nx.average_shortest_path_length(G)
metrics['global_efficiency'] = nx.global_efficiency(G)
else:
# 计算最大连通分量
largest_cc = max(nx.connected_components(G), key=len)
G_largest = G.subgraph(largest_cc).copy()
metrics['diameter'] = nx.diameter(G_largest)
metrics['avg_path_length'] = nx.average_shortest_path_length(G_largest)
metrics['global_efficiency'] = nx.global_efficiency(G_largest)
metrics['largest_cc_ratio'] = len(largest_cc) / G.number_of_nodes()
# 度分布指标
degrees = [d for n, d in G.degree()]
metrics['degree_variance'] = np.var(degrees)
# 聚类系数
metrics['avg_clustering'] = nx.average_clustering(G)
# 计算网络鲁棒性R值
if attack_strategy == 'random':
attack_results = random_failure_simulation(G)
else:
attack_results = targeted_attack_simulation(G)
# R值 = 归一化的连通分量面积
component_sizes = attack_results['largest_component']
R = np.trapz(component_sizes, dx=1) / len(component_sizes)
metrics['R_value'] = R
return metrics
完整分析流程
def complete_robustness_analysis():
"""完整的网络鲁棒性分析流程"""
# 1. 创建网络
print("创建网络模型...")
networks = create_networks()
# 2. 基本统计
print("\n网络基本统计:")
for name, G in networks.items():
print(f"\n{name}网络:")
print(f" 节点数: {G.number_of_nodes()}")
print(f" 边数: {G.number_of_edges()}")
print(f" 平均度: {np.mean([d for n, d in G.degree()]):.2f}")
# 3. 鲁棒性分析
print("\n进行鲁棒性分析...")
results = comprehensive_robustness_analysis(networks)
# 4. 计算详细指标
print("\n详细鲁棒性指标:")
for name in networks:
G = networks[name]
print(f"\n{name}网络:")
metrics_random = calculate_robustness_metrics(G, 'random')
metrics_targeted = calculate_robustness_metrics(G, 'targeted')
print(f" 随机攻击R值: {metrics_random['R_value']:.3f}")
print(f" 蓄意攻击R值: {metrics_targeted['R_value']:.3f}")
print(f" 全局效率: {metrics_random['global_efficiency']:.3f}")
print(f" 平均聚类系数: {metrics_random['avg_clustering']:.3f}")
# 5. 分析边攻击的影响
print("\n边攻击分析:")
for name in networks:
G = networks[name]
edge_results = edge_attack_simulation(G)
print(f"{name}网络: 移除{len(edge_results['edges_removed'])}条边后"
f"连通分量比例: {edge_results['largest_component'][-1]:.3f}")
return results
# 运行完整分析
if __name__ == "__main__":
results = complete_robustness_analysis()
高级功能：级联失效模拟
def cascading_failure_simulation(G, initial_failure_nodes=1, load_factor=1.5):
"""模拟级联失效过程"""
G_copy = G.copy()
n_initial = G_copy.number_of_nodes()
# 计算初始节点负载（基于度中心性）
degrees = dict(G_copy.degree())
max_degree = max(degrees.values())
# 节点容量 = 负载 * 负载因子
capacities = {node: degree * load_factor
for node, degree in degrees.items()}
# 记录级联过程
cascade_history = [n_initial]
failed_nodes = set()
# 初始故障
initial_failures = random.sample(list(G_copy.nodes()),
min(initial_failure_nodes, len(G_copy)))
# 级联传播
new_failures = set(initial_failures)
while new_failures and G_copy.number_of_nodes() > 0:
failed_nodes.update(new_failures)
# 移除故障节点
for node in new_failures:
if node in G_copy:
G_copy.remove_node(node)
# 计算新的负载分布
if G_copy.number_of_nodes() > 0:
new_degrees = dict(G_copy.degree())
# 检查哪些节点过载
new_failures = set()
for node, degree in new_degrees.items():
if degree > capacities.get(node, float('inf')):
new_failures.add(node)
else:
new_failures = set()
cascade_history.append(G_copy.number_of_nodes())
return {
'failed_nodes_percentage': (n_initial - G_copy.number_of_nodes()) / n_initial,
'cascade_size': len(failed_nodes),
'cascade_history': cascade_history,
'surviving_nodes': G_copy.number_of_nodes()
}
# 级联失效示例
def demonstrate_cascading_failure():
networks = create_networks()
print("级联失效模拟结果:")
for name, G in networks.items():
result = cascading_failure_simulation(G, initial_failure_nodes=5)
print(f"\n{name}网络:")
print(f" 初始故障节点: 5")
print(f" 最终失效节点: {result['cascade_size']}")
print(f" 存活节点比例: {1 - result['failed_nodes_percentage']:.2f}")
这个完整的NetworkX网络鲁棒性分析案例包含了：
- 多种网络模型（ER、WS、BA）
- 随机故障和蓄意攻击模拟
- 边攻击分析
- 级联失效模拟
- 可视化对比
- 鲁棒性指标计算
通过这些代码,您可以全面评估不同类型网络在不同攻击策略下的鲁棒性表现。
