Python案例如何用Networkx做网络弹性

本文目录导读：

1. 基础设置和网络创建
2. 随机节点攻击
3. 目标攻击（基于节点重要性）
4. 边攻击分析
5. 恢复策略模拟
6. 完整分析示例
7. 高级分析功能

我来为您详细介绍如何使用NetworkX进行网络弹性分析。

基础设置和网络创建

import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
import random
# 创建不同类型的网络
def create_networks():
    networks = {}
    # 随机网络 (Erdos-Renyi)
    networks['ER'] = nx.erdos_renyi_graph(100, 0.05)
    # 小世界网络 (Watts-Strogatz)
    networks['WS'] = nx.watts_strogatz_graph(100, 4, 0.3)
    # 无标度网络 (Barabasi-Albert)
    networks['BA'] = nx.barabasi_albert_graph(100, 2)
    return networks

随机节点攻击

def random_node_attack(G, num_removals):
    """随机删除节点并观察网络特性变化"""
    G = G.copy()
    n_original = G.number_of_nodes()
    # 记录每个阶段的网络特性
    results = {
        'removed_nodes': [],
        'largest_cc_size': [],
        'num_components': [],
        'avg_shortest_path': []
    }
    for i in range(num_removals):
        if G.number_of_nodes() == 0:
            break
        # 随机选择一个节点删除
        node = random.choice(list(G.nodes()))
        G.remove_node(node)
        # 计算网络特性
        largest_cc = max(nx.connected_components(G), key=len) if G.number_of_nodes() > 0 else set()
        results['removed_nodes'].append(i + 1)
        results['largest_cc_size'].append(len(largest_cc) / n_original)
        # 计算连通组件数
        results['num_components'].append(nx.number_connected_components(G) if G.number_of_nodes() > 0 else 0)
        # 计算平均最短路径（只在最大连通子图中计算）
        if len(largest_cc) > 0:
            subgraph = G.subgraph(largest_cc)
            if len(subgraph) > 1:
                results['avg_shortest_path'].append(nx.average_shortest_path_length(subgraph))
            else:
                results['avg_shortest_path'].append(0)
        else:
            results['avg_shortest_path'].append(0)
    return results

目标攻击（基于节点重要性）

def targeted_node_attack(G, num_removals, strategy='degree'):
    """基于节点重要性的目标攻击"""
    G = G.copy()
    n_original = G.number_of_nodes()
    results = {
        'removed_nodes': [],
        'largest_cc_size': [],
        'num_components': [],
        'avg_shortest_path': []
    }
    for i in range(num_removals):
        if G.number_of_nodes() == 0:
            break
        # 根据策略选择最重要的节点
        if strategy == 'degree':
            # 基于度中心性
            centrality = nx.degree_centrality(G)
        elif strategy == 'betweenness':
            # 基于介数中心性
            centrality = nx.betweenness_centrality(G)
        elif strategy == 'closeness':
            # 基于接近中心性
            centrality = nx.closeness_centrality(G)
        elif strategy == 'eigenvector':
            # 基于特征向量中心性
            centrality = nx.eigenvector_centrality(G, max_iter=1000)
        # 选择中心性最高的节点
        target_node = max(centrality, key=centrality.get)
        G.remove_node(target_node)
        # 记录指标
        largest_cc = max(nx.connected_components(G), key=len) if G.number_of_nodes() > 0 else set()
        results['removed_nodes'].append(i + 1)
        results['largest_cc_size'].append(len(largest_cc) / n_original)
        results['num_components'].append(nx.number_connected_components(G) if G.number_of_nodes() > 0 else 0)
        if len(largest_cc) > 0:
            subgraph = G.subgraph(largest_cc)
            if len(subgraph) > 1:
                results['avg_shortest_path'].append(nx.average_shortest_path_length(subgraph))
            else:
                results['avg_shortest_path'].append(0)
        else:
            results['avg_shortest_path'].append(0)
    return results

边攻击分析

def edge_attack_analysis(G, num_removals, strategy='random'):
    """分析边删除对网络的影响"""
    G = G.copy()
    results = {
        'removed_edges': [],
        'largest_cc_size': [],
        'efficiency': []
    }
    for i in range(num_removals):
        if G.number_of_edges() == 0:
            break
        if strategy == 'random':
            # 随机删除边
            edge = random.choice(list(G.edges()))
        elif strategy == 'bridge':
            # 优先删除桥接边（关键连接）
            bridges = list(nx.bridges(G))
            if bridges:
                edge = random.choice(bridges)
            else:
                edge = random.choice(list(G.edges()))
        G.remove_edge(*edge)
        # 计算效率
        largest_cc = max(nx.connected_components(G), key=len) if G.number_of_nodes() > 0 else set()
        results['removed_edges'].append(i + 1)
        results['largest_cc_size'].append(len(largest_cc) / (G.number_of_nodes() + 1))
        # 全局效率
        if len(largest_cc) > 1:
            subgraph = G.subgraph(largest_cc)
            results['efficiency'].append(nx.global_efficiency(subgraph))
        else:
            results['efficiency'].append(0)
    return results

恢复策略模拟

def network_recovery_simulation(G, attack_ratio=0.2, recovery_strategy='random'):
    """模拟网络攻击后的恢复"""
    G = G.copy()
    n_nodes = G.number_of_nodes()
    n_remove = int(n_nodes * attack_ratio)
    # 攻击阶段
    if attack_ratio > 0:
        G_attacked = targeted_node_attack(G, n_remove, 'betweenness')
    else:
        G_attacked = G.copy() if False else G
    # 恢复阶段
    G_recovered = G.copy()
    removed_nodes = random.sample(list(G.nodes()), n_remove)
    for node in removed_nodes:
        G_recovered.remove_node(node)
    # 尝试恢复（添加回节点）
    recovery_results = []
    for i, node in enumerate(removed_nodes):
        if recovery_strategy == 'random':
            # 随机恢复
            G_recovered.add_node(node)
            # 恢复部分边
            original_neighbors = list(G.neighbors(node))
            for neighbor in original_neighbors[:max(1, len(original_neighbors)//2)]:
                if neighbor in G_recovered:
                    G_recovered.add_edge(node, neighbor)
        elif recovery_strategy == 'preferential':
            # 优先连接到高度数节点
            G_recovered.add_node(node)
            degrees = dict(G_recovered.degree())
            if degrees:
                # 选择度最高的几个邻居
                potential_neighbors = sorted(degrees.items(), key=lambda x: x[1], reverse=True)[:3]
                for neighbor, _ in potential_neighbors:
                    if neighbor != node and G.has_edge(node, neighbor):
                        G_recovered.add_edge(node, neighbor)
        # 记录恢复效果
        largest_cc = max(nx.connected_components(G_recovered), key=len)
        recovery_results.append({
            'step': i + 1,
            'n_nodes': G_recovered.number_of_nodes(),
            'largest_cc_ratio': len(largest_cc) / n_nodes
        })
    return recovery_results

完整分析示例

def complete_resilience_analysis():
    """完整的网络弹性分析"""
    # 创建网络
    networks = create_networks()
    # 攻击参数
    num_attacks = 30
    # 存储结果
    results = {}
    for name, G in networks.items():
        print(f"\n分析 {name} 网络:")
        print(f"  节点数: {G.number_of_nodes()}")
        print(f"  边数: {G.number_of_edges()}")
        # 随机攻击
        random_results = random_node_attack(G, num_attacks)
        # 目标攻击（基于度）
        targeted_results = targeted_node_attack(G, num_attacks, 'degree')
        results[name] = {
            'random_attack': random_results,
            'targeted_attack': targeted_results
        }
        # 计算临界点（最大的连通组件下降到50%以下）
        for attack_type, attack_results in [('随机', random_results), ('目标', targeted_results)]:
            critical_point = None
            for i, size in enumerate(attack_results['largest_cc_size']):
                if size < 0.5:
                    critical_point = i + 1
                    break
            if critical_point:
                print(f"  {attack_type}攻击 - 临界点: 删除 {critical_point} 个节点")
            else:
                print(f"  {attack_type}攻击 - 未达到临界点")
    return results
# 可视化结果
def plot_resilience_analysis(results):
    """绘制网络弹性分析结果"""
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    for idx, (name, data) in enumerate(results.items()):
        # 随机攻击结果
        ax1 = axes[0, idx]
        ax1.plot(data['random_attack']['removed_nodes'], 
                data['random_attack']['largest_cc_size'], 
                'b-', label='随机攻击', linewidth=2)
        # 目标攻击结果
        ax1.plot(data['targeted_attack']['removed_nodes'], 
                data['targeted_attack']['largest_cc_size'], 
                'r-', label='目标攻击', linewidth=2)
        ax1.set_xlabel('删除节点数')
        ax1.set_ylabel('最大连通组件比例')
        ax1.set_title(f'{name} 网络弹性分析')
        ax1.legend()
        ax1.grid(True, alpha=0.3)
        ax1.axhline(y=0.5, color='gray', linestyle='--', alpha=0.5)
    plt.tight_layout()
    plt.show()
# 运行分析
if __name__ == "__main__":
    # 运行完整分析
    analysis_results = complete_resilience_analysis()
    # 可视化结果
    plot_resilience_analysis(analysis_results)
    # 额外分析：比较不同攻击策略
    G = nx.barabasi_albert_graph(100, 2)
    strategies = ['degree', 'betweenness', 'closeness', 'eigenvector']
    plt.figure(figsize=(12, 8))
    for strategy in strategies:
        results = targeted_node_attack(G, 30, strategy)
        plt.plot(results['removed_nodes'], 
                results['largest_cc_size'], 
                label=f'{strategy}攻击', 
                linewidth=2, 
                marker='o', 
                markersize=4)
    plt.xlabel('删除节点数')
    plt.ylabel('最大连通组件比例')
    plt.title('不同攻击策略对无标度网络的影响')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()

高级分析功能

def advanced_resilience_metrics(G):
    """计算高级弹性指标"""
    metrics = {}
    # 1. 网络效率
    metrics['global_efficiency'] = nx.global_efficiency(G)
    # 2. 自然连通度
    if G.number_of_nodes() > 0:
        eigenvalues = nx.adjacency_spectrum(G)
        metrics['natural_connectivity'] = np.log(np.abs(eigenvalues)).sum().real / G.number_of_nodes()
    # 3. 鲁棒性系数
    try:
        metrics['robustness_coefficient'] = nx.robustness_coefficient(G)
    except:
        metrics['robustness_coefficient'] = None
    # 4. 度分布熵
    degrees = [d for n, d in G.degree()]
    degree_counts = np.bincount(degrees)
    degree_probs = degree_counts / len(degrees)
    degree_probs = degree_probs[degree_probs > 0]  # 移除零概率
    metrics['degree_entropy'] = -np.sum(degree_probs * np.log2(degree_probs))
    return metrics
# 比较不同网络的弹性指标
def compare_network_elasticity():
    """比较不同网络拓扑的弹性"""
    networks = create_networks()
    print("\n网络弹性指标比较:")
    print("-" * 50)
    for name, G in networks.items():
        metrics = advanced_resilience_metrics(G)
        print(f"\n{name} 网络:")
        for key, value in metrics.items():
            if value is not None:
                print(f"  {key}: {value:.4f}")
            else:
                print(f"  {key}: N/A")

这个完整的NetworkX网络弹性分析案例提供了：

1. 基础框架：创建不同类型的网络
2. 攻击模拟：随机攻击和目标攻击
3. 恢复策略：节点恢复模拟
4. 可视化：结果图表展示
5. 高级指标：综合弹性评估

通过这些工具,您可以系统地评估网络在面对攻击时的弹性表现。