Python案例如何用Networkx做网络压缩

wen python案例 2026-07-15 152

本文目录导读：

基于度的节点筛选
社区检测压缩
骨架提取（生成树）
基于介数中心性的压缩
综合压缩示例
应用场景示例

我来介绍使用NetworkX进行网络压缩的几种常用方法：

基于度的节点筛选

import networkx as nx
import matplotlib.pyplot as plt
import random
# 创建示例图
G = nx.barabasi_albert_graph(100, 2)  # 无标度网络
# 方法1：按节点度筛选
def compress_by_degree(G, threshold=5):
    """保留度大于阈值的节点"""
    nodes_to_keep = [n for n, d in G.degree() if d > threshold]
    G_compressed = G.subgraph(nodes_to_keep).copy()
    return G_compressed
G_degree = compress_by_degree(G, threshold=10)
print(f"原始节点数: {G.number_of_nodes()}, 压缩后: {G_degree.number_of_nodes()}")

社区检测压缩

# 方法2：基于社区检测进行压缩
def compress_by_community(G, method='louvain'):
    """使用社区检测合并节点"""
    try:
        import community as community_louvain
        # 使用Louvain算法
        partition = community_louvain.best_partition(G)
    except ImportError:
        # 使用Girvan-Newman
        from networkx.algorithms.community import girvan_newman
        communities = girvan_newman(G)
        first_level_communities = next(communities)
        partition = {}
        for i, comm in enumerate(first_level_communities):
            for node in comm:
                partition[node] = i
    # 创建压缩图
    G_compressed = nx.Graph()
    # 为每个社区创建超级节点
    community_nodes = {}
    for node, community_id in partition.items():
        if community_id not in community_nodes:
            community_nodes[community_id] = []
        community_nodes[community_id].append(node)
    # 添加超级节点
    for comm_id in community_nodes:
        G_compressed.add_node(f"Community_{comm_id}", 
                            size=len(community_nodes[comm_id]))
    # 添加边（社区间的连接）
    for (u, v) in G.edges():
        comm_u = partition[u]
        comm_v = partition[v]
        if comm_u != comm_v:
            G_compressed.add_edge(f"Community_{comm_u}", 
                                f"Community_{comm_v}")
    return G_compressed, partition
G_community, partition = compress_by_community(G)
print(f"社区数量: {len(G_community.nodes())}")

骨架提取（生成树）

# 方法3：使用最小生成树或最大生成树
def compress_by_spanning_tree(G, weight='weight', mode='maximum'):
    """使用生成树压缩网络"""
    # 如果没有权重，添加随机权重
    if not weight in G.edges(data=True)[0]:
        for (u, v) in G.edges():
            G[u][v][weight] = random.random()
    if mode == 'maximum':
        # 最大生成树（保留重要连接）
        G_compressed = nx.maximum_spanning_tree(G, weight=weight)
    else:
        # 最小生成树（保留紧密连接）
        G_compressed = nx.minimum_spanning_tree(G, weight=weight)
    return G_compressed
G_tree = compress_by_spanning_tree(G, mode='maximum')
print(f"压缩后边数: {G_tree.number_of_edges()}")

基于介数中心性的压缩

# 方法4：保留重要节点
def compress_by_centrality(G, percentile=20):
    """保留中心性最高的节点"""
    # 计算介数中心性
    betweenness = nx.betweenness_centrality(G)
    # 确定阈值
    threshold = sorted(betweenness.values(), reverse=True)[
        min(len(betweenness)-1, int(len(betweenness) * percentile / 100))
    ]
    # 保留重要节点
    nodes_to_keep = [n for n, c in betweenness.items() if c >= threshold]
    G_compressed = G.subgraph(nodes_to_keep).copy()
    return G_compressed
G_centrality = compress_by_centrality(G, percentile=30)
print(f"原始节点数: {G.number_of_nodes()}, 压缩后: {G_centrality.number_of_nodes()}")

综合压缩示例

# 完整的压缩流程示例
def complete_compression_pipeline(G, methods=['degree', 'community', 'tree'], 
                                  degree_threshold=10, percentile=30):
    """综合使用多种压缩方法"""
    results = {}
    for method in methods:
        if method == 'degree':
            # 基于度的压缩
            compressed = compress_by_degree(G.copy(), degree_threshold)
            results['degree'] = {
                'graph': compressed,
                'nodes': compressed.number_of_nodes(),
                'edges': compressed.number_of_edges()
            }
        elif method == 'community':
            # 基于社区的压缩
            compressed, _ = compress_by_community(G.copy())
            results['community'] = {
                'graph': compressed,
                'nodes': compressed.number_of_nodes(),
                'edges': compressed.number_of_edges()
            }
        elif method == 'tree':
            # 生成树压缩
            compressed = compress_by_spanning_tree(G.copy())
            results['tree'] = {
                'graph': compressed,
                'nodes': compressed.number_of_nodes(),
                'edges': compressed.number_of_edges(),
                'type': 'tree'
            }
        elif method == 'centrality':
            # 中心性压缩
            compressed = compress_by_centrality(G.copy(), percentile)
            results['centrality'] = {
                'graph': compressed,
                'nodes': compressed.number_of_nodes(),
                'edges': compressed.number_of_edges()
            }
    return results
# 执行压缩
compression_results = complete_compression_pipeline(G)
# 可视化比较
def visualize_compression_results(original_G, results):
    """可视化压缩结果"""
    fig, axes = plt.subplots(2, 3, figsize=(15, 10))
    # 原始图
    pos = nx.spring_layout(original_G)
    nx.draw(original_G, pos, ax=axes[0, 0], node_size=20, alpha=0.6)
    axes[0, 0].set_title(f"原始网络\n{original_G.number_of_nodes()} nodes\n"
                        f"{original_G.number_of_edges()} edges")
    # 各种压缩结果
    for i, (method, data) in enumerate(results.items()):
        row = (i + 1) // 3
        col = (i + 1) % 3
        compressed_G = data['graph']
        nx.draw(compressed_G, nx.spring_layout(compressed_G), 
                ax=axes[row, col], node_size=20, alpha=0.6)
        axes[row, col].set_title(f"{method}压缩\n{data['nodes']} nodes\n"
                               f"{data['edges']} edges")
    plt.tight_layout()
    plt.show()
# 可视化结果
visualize_compression_results(G, compression_results)
# 打印统计信息
print("\n=== 网络压缩统计 ===")
print(f"{'方法':<15} {'节点数':<10} {'边数':<10} {'压缩率(%)':<12}")
print("-" * 50)
print(f"{'原始网络':<15} {G.number_of_nodes():<10} {G.number_of_edges():<10} {'100':<12}")
for method, data in compression_results.items():
    compression_rate = (data['nodes'] / G.number_of_nodes()) * 100
    print(f"{method:<15} {data['nodes']:<10} {data['edges']:<10} "
          f"{compression_rate:<12.2f}")

应用场景示例

# 实际应用：社交网络压缩
def social_network_compression_demo():
    """社交网络压缩示例"""
    # 创建模拟社交网络
    G = nx.watts_strogatz_graph(200, 6, 0.3)
    print("=== 社交网络压缩 ===")
    # 1. 基于活动度压缩（保留活跃用户）
    activity = {n: random.randint(1, 100) for n in G.nodes()}
    nx.set_node_attributes(G, activity, 'activity')
    active_users = [n for n in G.nodes() if G.nodes[n]['activity'] > 50]
    G_active = G.subgraph(active_users).copy()
    print(f"活跃用户网络: {G_active.number_of_nodes()} 用户, "
          f"{G_active.number_of_edges()} 关系")
    # 2. 社区结构压缩
    G_community, partition = compress_by_community(G)
    print(f"社区压缩网络: {G_community.number_of_nodes()} 社区, "
          f"{G_community.number_of_edges()} 社区间连接")
    return G_active, G_community
# 运行示例
G_active, G_community = social_network_compression_demo()

这些方法可以根据具体需求选择使用,常见的应用场景包括：