Python案例如何用Networkx做级联模型

wen python案例 2026-07-15 209

本文目录导读：

基础级联模型实现
高级级联模型变体
主要功能特点
应用场景

我来为你展示如何使用Networkx实现级联模型（Cascade Model），这是一个用于模拟信息传播的经典模型。

基础级联模型实现

import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
import random
class CascadeModel:
    def __init__(self, graph, threshold=0.5):
        """
        初始化级联模型
        参数:
        graph: Networkx图对象
        threshold: 激活阈值（0-1之间）
        """
        self.graph = graph.copy()
        self.threshold = threshold
        # 节点状态: 0=未激活, 1=激活
        self.node_states = {node: 0 for node in graph.nodes()}
        self.active_nodes = set()
    def initialize_seeds(self, seeds):
        """初始化种子节点"""
        for node in seeds:
            self.node_states[node] = 1
            self.active_nodes.add(node)
    def step(self):
        """执行一步级联传播"""
        new_active = set()
        # 检查每个未激活节点的邻居激活比例
        for node in self.graph.nodes():
            if self.node_states[node] == 0:  # 未激活节点
                neighbors = list(self.graph.neighbors(node))
                if len(neighbors) > 0:
                    active_neighbors = sum(1 for n in neighbors if self.node_states[n] == 1)
                    active_ratio = active_neighbors / len(neighbors)
                    # 如果激活比例超过阈值，节点被激活
                    if active_ratio >= self.threshold:
                        new_active.add(node)
        # 更新状态
        for node in new_active:
            self.node_states[node] = 1
            self.active_nodes.add(node)
        return len(new_active) > 0  # 返回是否还有新激活节点
    def run_cascade(self, max_steps=100):
        """运行级联传播直到收敛"""
        step = 0
        cascade_history = [self.active_nodes.copy()]
        while step < max_steps:
            has_new = self.step()
            cascade_history.append(self.active_nodes.copy())
            if not has_new:
                break
            step += 1
        return cascade_history
    def get_infected_ratio(self):
        """获取感染比例"""
        return len(self.active_nodes) / self.graph.number_of_nodes()
# 示例1：在小世界网络上运行级联模型
def example_small_world():
    """在小世界网络上演示级联模型"""
    # 创建小世界网络
    G = nx.watts_strogatz_graph(100, 4, 0.1, seed=42)
    # 初始化级联模型
    cascade = CascadeModel(G, threshold=0.3)
    # 设置初始种子节点
    seeds = [0, 25, 50, 75]
    cascade.initialize_seeds(seeds)
    # 运行级联
    history = cascade.run_cascade()
    # 可视化结果
    fig, axes = plt.subplots(1, 2, figsize=(12, 5))
    # 绘制网络状态
    pos = nx.spring_layout(G, seed=42)
    node_colors = ['red' if node in cascade.active_nodes else 'lightblue' 
                   for node in G.nodes()]
    nx.draw(G, pos, node_color=node_colors, 
            node_size=100, with_labels=False, ax=axes[0])
    axes[0].set_title(f"最终状态 (感染比例: {cascade.get_infected_ratio():.2f})")
    # 绘制感染过程
    sizes = [len(h) for h in history]
    axes[1].plot(sizes, 'b-o', markersize=6)
    axes[1].set_xlabel("传播步骤")
    axes[1].set_ylabel("激活节点数")
    axes[1].set_title("级联传播过程")
    axes[1].grid(True)
    plt.tight_layout()
    plt.show()
    return cascade, history
# 示例2：参数敏感性分析
def parameter_sensitivity_analysis():
    """分析不同阈值对传播的影响"""
    G = nx.barabasi_albert_graph(200, 3, seed=42)
    thresholds = np.linspace(0.1, 0.9, 9)
    results = []
    for threshold in thresholds:
        # 多次实验取平均
        infected_ratios = []
        for _ in range(10):
            cascade = CascadeModel(G, threshold=threshold)
            seeds = random.sample(list(G.nodes()), 5)
            cascade.initialize_seeds(seeds)
            cascade.run_cascade()
            infected_ratios.append(cascade.get_infected_ratio())
        results.append(np.mean(infected_ratios))
    # 可视化结果
    plt.figure(figsize=(10, 6))
    plt.plot(thresholds, results, 'bo-', linewidth=2, markersize=8)
    plt.xlabel("激活阈值")
    plt.ylabel("最终感染比例")
    plt.title("阈值对级联传播的影响")
    plt.grid(True, alpha=0.3)
    plt.show()
    return thresholds, results
# 示例3：不同网络结构的比较
def compare_network_structures():
    """比较不同网络结构中的级联传播"""
    network_types = {
        'ER随机网络': nx.erdos_renyi_graph(100, 0.05, seed=42),
        '小世界网络': nx.watts_strogatz_graph(100, 4, 0.1, seed=42),
        '无标度网络': nx.barabasi_albert_graph(100, 3, seed=42)
    }
    num_simulations = 5
    seeds_per_simulation = 3
    threshold = 0.3
    results = {}
    for name, G in network_types.items():
        infected_ratios = []
        for _ in range(num_simulations):
            cascade = CascadeModel(G, threshold=threshold)
            seeds = random.sample(list(G.nodes()), seeds_per_simulation)
            cascade.initialize_seeds(seeds)
            cascade.run_cascade()
            infected_ratios.append(cascade.get_infected_ratio())
        results[name] = {
            'mean': np.mean(infected_ratios),
            'std': np.std(infected_ratios)
        }
    # 可视化比较
    fig, ax = plt.subplots(figsize=(10, 6))
    names = list(results.keys())
    means = [results[n]['mean'] for n in names]
    stds = [results[n]['std'] for n in names]
    bars = ax.bar(names, means, yerr=stds, capsize=5, 
                  color=['#FF6B6B', '#4ECDC4', '#45B7D1'])
    ax.set_ylabel("平均感染比例")
    ax.set_title("不同网络结构中的级联传播比较")
    ax.set_ylim(0, 1)
    # 添加数值标签
    for bar, mean, std in zip(bars, means, stds):
        height = bar.get_height()
        ax.text(bar.get_x() + bar.get_width()/2., height + 0.02,
                f'{mean:.3f}±{std:.3f}', ha='center', va='bottom')
    plt.tight_layout()
    plt.show()
    return results
# 示例4：动画演示级联传播过程
def animate_cascade():
    """动画演示级联传播过程（需要安装matplotlib的动画支持）"""
    from matplotlib.animation import FuncAnimation
    from IPython.display import HTML
    # 创建网络
    G = nx.karate_club_graph()
    pos = nx.spring_layout(G, seed=42)
    # 初始化级联模型
    cascade = CascadeModel(G, threshold=0.25)
    cascade.initialize_seeds([0, 33])
    # 准备动画
    fig, ax = plt.subplots(figsize=(8, 6))
    def update(frame):
        ax.clear()
        if frame > 0:
            cascade.step()
        node_colors = ['red' if node in cascade.active_nodes else 'lightblue' 
                       for node in G.nodes()]
        nx.draw(G, pos, node_color=node_colors, 
                node_size=300, with_labels=True, ax=ax,
                font_size=8)
        ax.set_title(f"级联传播 - 步骤 {frame}\n激活节点: {len(cascade.active_nodes)}")
    anim = FuncAnimation(fig, update, frames=20, interval=500, repeat=False)
    plt.close()
    return anim
# 主程序
if __name__ == "__main__":
    # 运行示例
    print("示例1：小世界网络上的级联传播")
    cascade, history = example_small_world()
    print("\n示例2：参数敏感性分析")
    thresholds, results = parameter_sensitivity_analysis()
    print("\n示例3：不同网络结构的比较")
    results = compare_network_structures()
    print("\n示例4：动画演示（如果支持）")
    try:
        anim = animate_cascade()
        print("动画创建成功！")
    except:
        print("动画创建失败（可能需要额外依赖）")

高级级联模型变体

class AdvancedCascadeModel:
    """高级级联模型，支持多种传播机制"""
    def __init__(self, graph, model_type='threshold'):
        """
        参数:
        graph: Networkx图对象
        model_type: 'threshold'(阈值模型), 'independent'(独立级联), 'linear'(线性阈值)
        """
        self.graph = graph.copy()
        self.model_type = model_type
        self.reset()
    def reset(self):
        """重置模型状态"""
        self.node_states = {node: 0 for node in self.graph.nodes()}
        self.active_nodes = set()
        self.activation_times = {}
    def initialize_seeds(self, seeds):
        """初始化种子节点"""
        for node in seeds:
            self.node_states[node] = 1
            self.active_nodes.add(node)
            self.activation_times[node] = 0
    def independent_cascade_step(self, probability=0.5):
        """独立级联模型的一步传播"""
        new_active = set()
        for node in self.active_nodes:
            neighbors = list(self.graph.neighbors(node))
            for neighbor in neighbors:
                if self.node_states[neighbor] == 0:  # 未激活节点
                    if random.random() < probability:
                        new_active.add(neighbor)
        for node in new_active:
            self.node_states[node] = 1
            self.active_nodes.add(node)
            self.activation_times[node] = len(self.activation_times)
        return len(new_active) > 0
    def linear_threshold_step(self, thresholds=None):
        """线性阈值模型的一步传播"""
        if thresholds is None:
            thresholds = {node: random.random() for node in self.graph.nodes()}
        new_active = set()
        for node in self.graph.nodes():
            if self.node_states[node] == 0:
                neighbors = list(self.graph.neighbors(node))
                if len(neighbors) > 0:
                    # 计算邻居的累积影响
                    influence = 0
                    for neighbor in neighbors:
                        if self.node_states[neighbor] == 1:
                            influence += 1.0 / len(neighbors)  # 均匀分配影响
                    if influence >= thresholds[node]:
                        new_active.add(node)
        for node in new_active:
            self.node_states[node] = 1
            self.active_nodes.add(node)
        return len(new_active) > 0
    def run(self, max_steps=100, **kwargs):
        """运行级联传播"""
        history = [self.active_nodes.copy()]
        step = 0
        while step < max_steps:
            if self.model_type == 'independent':
                has_new = self.independent_cascade_step(**kwargs)
            elif self.model_type == 'linear':
                has_new = self.linear_threshold_step(**kwargs)
            else:
                has_new = self.threshold_step(**kwargs)
            history.append(self.active_nodes.copy())
            if not has_new:
                break
            step += 1
        return history
# 使用示例
def advanced_example():
    """高级级联模型使用示例"""
    G = nx.karate_club_graph()
    # 独立级联模型
    independent_cascade = AdvancedCascadeModel(G, 'independent')
    independent_cascade.initialize_seeds([0])
    independent_cascade.run(probability=0.3)
    # 线性阈值模型
    linear_cascade = AdvancedCascadeModel(G, 'linear')
    linear_cascade.initialize_seeds([0, 33])
    linear_cascade.run()
    print(f"独立级联模型 - 最终感染节点数: {len(independent_cascade.active_nodes)}")
    print(f"线性阈值模型 - 最终感染节点数: {len(linear_cascade.active_nodes)}")
# 运行高级示例
if __name__ == "__main__":
    advanced_example()