Python案例如何用Networkx做阈值模型

wen python案例 2026-07-15 284

本文目录导读：

基本阈值模型实现
高级阈值模型版本
主要功能说明

我来为你详细讲解如何使用Networkx实现阈值模型,阈值模型是一种经典的传播模型，用于模拟信息、行为或疾病的传播。

基本阈值模型实现

import networkx as nx
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import random
class ThresholdModel:
    def __init__(self, graph, initial_infected, threshold=0.5):
        """
        初始化阈值模型
        参数:
        graph: NetworkX图对象
        initial_infected: 初始感染节点列表
        threshold: 感染阈值 (默认0.5)
        """
        self.graph = graph.copy()
        self.threshold = threshold
        self.nodes = list(graph.nodes())
        self.n = len(self.nodes)
        # 记录节点状态: 0-易感, 1-已感染
        self.state = {node: 0 for node in self.nodes}
        for node in initial_infected:
            self.state[node] = 1
        # 记录传播历史
        self.history = [self.state.copy()]
    def step(self):
        """执行一次传播步骤"""
        new_infected = []
        for node in self.nodes:
            if self.state[node] == 0:  # 易感节点
                neighbors = list(self.graph.neighbors(node))
                if len(neighbors) > 0:
                    # 计算已感染的邻居比例
                    infected_neighbors = sum(self.state[neighbor] for neighbor in neighbors)
                    infected_ratio = infected_neighbors / len(neighbors)
                    # 如果超过阈值，节点被感染
                    if infected_ratio >= self.threshold:
                        new_infected.append(node)
        # 更新状态
        for node in new_infected:
            self.state[node] = 1
        self.history.append(self.state.copy())
        return len(new_infected) > 0  # 返回是否还有新感染
    def propagate(self, max_steps=50):
        """执行完整的传播过程"""
        steps = 0
        while steps < max_steps:
            has_new = self.step()
            if not has_new:
                break
            steps += 1
        return steps
    def get_infection_status(self):
        """获取感染状态"""
        infected = sum(1 for state in self.state.values() if state == 1)
        return infected / self.n  # 返回感染比例
# 示例1：在随机图上运行阈值模型
def example_random_graph():
    print("=== 随机图阈值模型示例 ===")
    # 创建随机图
    G = nx.erdos_renyi_graph(n=100, p=0.05)
    # 初始化模型
    initial_infected = random.sample(list(G.nodes()), 5)
    model = ThresholdModel(G, initial_infected, threshold=0.3)
    # 传播
    steps = model.propagate()
    infection_rate = model.get_infection_status()
    print(f"传播了 {steps} 步")
    print(f"最终感染率: {infection_rate:.2%}")
    # 可视化
    visualize_model(G, model.history[-1])
    return model
# 示例2：在小世界网络上运行
def example_small_world():
    print("\n=== 小世界网络阈值模型示例 ===")
    # 创建小世界网络
    G = nx.watts_strogatz_graph(n=100, k=4, p=0.1)
    # 不同阈值的实验
    thresholds = [0.2, 0.5, 0.8]
    results = []
    for threshold in thresholds:
        initial_infected = random.sample(list(G.nodes()), 5)
        model = ThresholdModel(G, initial_infected, threshold)
        steps = model.propagate()
        infection_rate = model.get_infection_status()
        results.append((threshold, steps, infection_rate))
        print(f"阈值 {threshold}: 传播 {steps} 步, 感染率 {infection_rate:.2%}")
    return results
# 示例3：多初始感染源实验
def example_multiple_sources():
    print("\n=== 多初始源感染实验 ===")
    G = nx.scale_free_graph(n=100, alpha=1.5).to_undirected()
    G = nx.Graph(G)  # 去除多重边
    threshold = 0.4
    for num_sources in [1, 3, 5, 10]:
        initial_infected = random.sample(list(G.nodes()), min(num_sources, len(G.nodes())))
        model = ThresholdModel(G, initial_infected, threshold)
        steps = model.propagate()
        infection_rate = model.get_infection_status()
        print(f"初始源 {num_sources}个: 传播 {steps} 步, 感染率 {infection_rate:.2%}")
# 可视化函数
def visualize_model(G, state, title="Threshold Model"):
    """可视化网络状态"""
    plt.figure(figsize=(10, 8))
    # 设置节点颜色
    colors = ['red' if state[node] == 1 else 'lightblue' for node in G.nodes()]
    # 布局
    pos = nx.spring_layout(G, k=1, iterations=50)
    # 绘制网络
    nx.draw(G, pos, node_color=colors, node_size=100, 
            with_labels=False, alpha=0.7, edge_color='gray')
    plt.title(title)
    plt.tight_layout()
    plt.show()
# 动画可视化
def create_animation(G, history):
    """创建传播过程的动画"""
    fig, ax = plt.subplots(figsize=(10, 8))
    pos = nx.spring_layout(G, k=1, iterations=50)
    def update(frame):
        ax.clear()
        state = history[frame]
        colors = ['red' if state[node] == 1 else 'lightblue' for node in G.nodes()]
        nx.draw(G, pos, node_color=colors, node_size=100, 
                with_labels=False, alpha=0.7, edge_color='gray', ax=ax)
        ax.set_title(f"Step {frame}: 感染率 {sum(state.values())/len(G.nodes):.2%}")
    anim = FuncAnimation(fig, update, frames=len(history), interval=500, repeat=False)
    return anim
# 高级分析：阈值的影响
def threshold_sensitivity_analysis():
    """分析阈值对传播的影响"""
    print("\n=== 阈值敏感性分析 ===")
    G = nx.erdos_renyi_graph(n=100, p=0.06)
    thresholds = np.linspace(0.1, 1.0, 20)
    infection_rates = []
    for threshold in thresholds:
        # 多次实验取平均
        rates = []
        for _ in range(10):
            initial_infected = random.sample(list(G.nodes()), 5)
            model = ThresholdModel(G, initial_infected, threshold)
            model.propagate()
            rates.append(model.get_infection_status())
        avg_rate = np.mean(rates)
        infection_rates.append(avg_rate)
        if len(thresholds) <= 10:
            print(f"阈值 {threshold:.1f}: 平均感染率 {avg_rate:.2%}")
    # 绘制阈值-感染率曲线
    plt.figure(figsize=(10, 6))
    plt.plot(thresholds, infection_rates, 'b-', linewidth=2)
    plt.xlabel('阈值')
    plt.ylabel('感染率')
    plt.title('阈值对传播的影响')
    plt.grid(True, alpha=0.3)
    plt.show()
    return thresholds, infection_rates
# 运行示例
if __name__ == "__main__":
    print("NetworkX 阈值模型演示")
    print("="*50)
    # 运行基本示例
    model1 = example_random_graph()
    # 小世界网络实验
    results = example_small_world()
    # 多源感染实验
    example_multiple_sources()
    # 阈值敏感性分析
    thresholds, rates = threshold_sensitivity_analysis()

高级阈值模型版本

class AdvancedThresholdModel(ThresholdModel):
    def __init__(self, graph, initial_infected, thresholds=None):
        """
        高级阈值模型，支持节点级阈值
        参数:
        graph: NetworkX图对象
        initial_infected: 初始感染节点
        thresholds: 节点阈值字典，如果为None则使用随机阈值
        """
        super().__init__(graph, initial_infected, threshold=0.5)
        if thresholds is None:
            # 为每个节点生成随机阈值
            self.node_thresholds = {
                node: random.uniform(0.2, 0.8) 
                for node in self.nodes
            }
        else:
            self.node_thresholds = thresholds
    def step(self):
        """重写step方法，使用节点级阈值"""
        new_infected = []
        for node in self.nodes:
            if self.state[node] == 0:
                neighbors = list(self.graph.neighbors(node))
                if len(neighbors) > 0:
                    infected_neighbors = sum(self.state[neighbor] for neighbor in neighbors)
                    infected_ratio = infected_neighbors / len(neighbors)
                    # 使用节点特定的阈值
                    node_threshold = self.node_thresholds.get(node, self.threshold)
                    if infected_ratio >= node_threshold:
                        new_infected.append(node)
        for node in new_infected:
            self.state[node] = 1
        self.history.append(self.state.copy())
        return len(new_infected) > 0
# 使用示例
def advanced_example():
    print("\n=== 高级阈值模型示例 ===")
    G = nx.karate_club_graph()  # 使用经典的空手道俱乐部网络
    # 设置节点阈值（基于节点度）
    thresholds = {}
    for node in G.nodes():
        degree = G.degree(node)
        # 度数越高的节点，阈值越高（更难被感染）
        thresholds[node] = min(0.8, degree / max(G.degree()))
    initial_infected = [0]  # 从节点0开始
    model = AdvancedThresholdModel(G, initial_infected, thresholds)
    steps = model.propagate()
    print(f"传播步数: {steps}")
    print(f"最终感染率: {model.get_infection_status():.2%}")
    # 可视化
    visualize_model(G, model.history[-1], 
                   title=f"Advanced Threshold Model (Final Step)")
    # 显示传播过程动画
    anim = create_animation(G, model.history)
    plt.show()
    return model
# 运行高级示例
model_advanced = advanced_example()