Python案例如何用Networkx做网络鲁棒性

wen python案例 2026-07-15 177

本文目录导读：

基础环境准备
创建网络模型
随机故障模拟
蓄意攻击模拟
边攻击模拟
可视化对比分析
综合鲁棒性分析
鲁棒性指标计算
完整分析流程
高级功能：级联失效模拟

我来详细介绍如何使用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网络鲁棒性分析案例包含了：