Python案例如何用Networkx做传播模型

本文目录导读：

1. 安装和导入必要的库
2. 创建网络图
3. SIR传播模型实现
4. SI传播模型实现
5. 带免疫的SIRS模型
6. 网络结构对比实验
7. 实时传播可视化
8. 参数敏感性分析

我来详细介绍如何使用NetworkX实现传播模型,包括几种常见的传播模型案例。

安装和导入必要的库

import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
import random
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

创建网络图

def create_network(n_nodes, model='small_world'):
    """创建不同结构的网络"""
    if model == 'small_world':
        # 小世界网络
        G = nx.watts_strogatz_graph(n=n_nodes, k=4, p=0.2, seed=42)
    elif model == 'scale_free':
        # 无标度网络
        G = nx.barabasi_albert_graph(n=n_nodes, m=2, seed=42)
    elif model == 'random':
        # 随机网络
        G = nx.erdos_renyi_graph(n=n_nodes, p=0.1, seed=42)
    else:
        # 完全图
        G = nx.complete_graph(n_nodes)
    return G
# 创建示例网络
G = create_network(100, 'small_world')
print(f"网络节点数: {G.number_of_nodes()}")
print(f"网络边数: {G.number_of_edges()}")

SIR传播模型实现

class SIRModel:
    """SIR传播模型"""
    def __init__(self, G, beta, gamma):
        """
        G: 网络图
        beta: 感染概率
        gamma: 恢复概率
        """
        self.G = G
        self.beta = beta
        self.gamma = gamma
        self.S = set(G.nodes())
        self.I = set()
        self.R = set()
    def initialize(self, initial_infected=None):
        """初始化感染节点"""
        self.S = set(self.G.nodes())
        self.I = set()
        self.R = set()
        if initial_infected:
            # 指定初始感染节点
            for node in initial_infected:
                self.S.remove(node)
                self.I.add(node)
        else:
            # 随机选择一个初始感染节点
            initial = random.choice(list(self.G.nodes()))
            self.S.remove(initial)
            self.I.add(initial)
    def step(self):
        """执行一个时间步的传播"""
        new_infected = set()
        new_recovered = set()
        # 感染过程
        for infected in self.I:
            neighbors = set(self.G.neighbors(infected))
            susceptible_neighbors = neighbors & self.S
            for neighbor in susceptible_neighbors:
                if random.random() < self.beta:
                    new_infected.add(neighbor)
        # 恢复过程
        for infected in self.I:
            if random.random() < self.gamma:
                new_recovered.add(infected)
        # 更新状态
        self.I -= new_recovered
        self.I |= new_infected
        self.S -= new_infected
        self.R |= new_recovered
        return len(self.S), len(self.I), len(self.R)
    def simulate(self, n_steps):
        """模拟传播过程"""
        history = []
        for step in range(n_steps):
            s, i, r = self.step()
            history.append((s, i, r))
            # 如果没有感染节点，停止模拟
            if i == 0:
                break
        return history
    def get_state(self):
        """获取当前状态"""
        return {
            'S': len(self.S),
            'I': len(self.I),
            'R': len(self.R),
            'S_nodes': self.S,
            'I_nodes': self.I,
            'R_nodes': self.R
        }
# 运行SIR模型
G = create_network(100, 'small_world')
sir = SIRModel(G, beta=0.2, gamma=0.1)
# 初始化感染
initial_infected = [0]  # 从节点0开始感染
sir.initialize(initial_infected)
# 模拟100步
history = sir.simulate(100)
# 可视化结果
def plot_sir_history(history):
    """绘制SIR传播历史"""
    steps = range(len(history))
    S_vals = [h[0] for h in history]
    I_vals = [h[1] for h in history]
    R_vals = [h[2] for h in history]
    plt.figure(figsize=(10, 6))
    plt.plot(steps, S_vals, 'b-', label='易感者 (S)', linewidth=2)
    plt.plot(steps, I_vals, 'r-', label='感染者 (I)', linewidth=2)
    plt.plot(steps, R_vals, 'g-', label='康复者 (R)', linewidth=2)
    plt.xlabel('时间步')
    plt.ylabel('节点数量')
    plt.title('SIR传播模型模拟结果')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()
plot_sir_history(history)

SI传播模型实现

class SIModel:
    """SI传播模型（不考虑恢复）"""
    def __init__(self, G, beta):
        self.G = G
        self.beta = beta
        self.S = set(G.nodes())
        self.I = set()
    def initialize(self, initial_infected=None):
        """初始化感染节点"""
        self.S = set(self.G.nodes())
        self.I = set()
        if initial_infected:
            for node in initial_infected:
                self.S.remove(node)
                self.I.add(node)
        else:
            initial = random.choice(list(self.G.nodes()))
            self.S.remove(initial)
            self.I.add(initial)
    def step(self):
        """执行一个时间步的传播"""
        new_infected = set()
        for infected in self.I:
            neighbors = set(self.G.neighbors(infected))
            susceptible_neighbors = neighbors & self.S
            for neighbor in susceptible_neighbors:
                if random.random() < self.beta:
                    new_infected.add(neighbor)
        self.I |= new_infected
        self.S -= new_infected
        return len(self.S), len(self.I)
    def simulate(self, n_steps):
        """模拟传播过程"""
        history = []
        for step in range(n_steps):
            s, i = self.step()
            history.append((s, i))
            if len(self.I) == len(self.G):
                break
        return history
# 运行SI模型
G = create_network(100, 'scale_free')
si = SIModel(G, beta=0.15)
si.initialize([0])
history_si = si.simulate(100)
# 可视化SI模型结果
plt.figure(figsize=(10, 6))
steps = range(len(history_si))
S_vals = [h[0] for h in history_si]
I_vals = [h[1] for h in history_si]
plt.plot(steps, S_vals, 'b-', label='易感者 (S)', linewidth=2)
plt.plot(steps, I_vals, 'r-', label='感染者 (I)', linewidth=2)
plt.xlabel('时间步')
plt.ylabel('节点数量')'SI传播模型模拟结果')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

带免疫的SIRS模型

class SIRSModel:
    """SIRS传播模型（带有免疫丧失）"""
    def __init__(self, G, beta, gamma, delta):
        """
        G: 网络图
        beta: 感染概率
        gamma: 恢复概率
        delta: 免疫丧失概率
        """
        self.G = G
        self.beta = beta
        self.gamma = gamma
        self.delta = delta
        self.S = set(G.nodes())
        self.I = set()
        self.R = set()
    def initialize(self, initial_infected=None):
        """初始化感染节点"""
        self.S = set(self.G.nodes())
        self.I = set()
        self.R = set()
        if initial_infected:
            for node in initial_infected:
                self.S.remove(node)
                self.I.add(node)
        else:
            initial = random.choice(list(self.G.nodes()))
            self.S.remove(initial)
            self.I.add(initial)
    def step(self):
        """执行一个时间步的传播"""
        new_infected = set()
        new_recovered = set()
        new_susceptible = set()
        # 感染过程
        for infected in self.I:
            neighbors = set(self.G.neighbors(infected))
            susceptible_neighbors = neighbors & self.S
            for neighbor in susceptible_neighbors:
                if random.random() < self.beta:
                    new_infected.add(neighbor)
        # 恢复过程
        for infected in self.I:
            if random.random() < self.gamma:
                new_recovered.add(infected)
        # 免疫丧失过程
        for recovered in self.R:
            if random.random() < self.delta:
                new_susceptible.add(recovered)
        # 更新状态
        self.I -= new_recovered
        self.I |= new_infected
        self.S -= new_infected
        self.S |= new_susceptible
        self.R -= new_susceptible
        self.R |= new_recovered
        return len(self.S), len(self.I), len(self.R)
# 运行SIRS模型
G = create_network(100, 'small_world')
sirs = SIRSModel(G, beta=0.3, gamma=0.1, delta=0.05)
sirs.initialize([0])
# 模拟200步
history_sirs = []
for _ in range(200):
    s, i, r = sirs.step()
    history_sirs.append((s, i, r))
# 可视化SIRS模型结果
plt.figure(figsize=(10, 6))
steps = range(len(history_sirs))
S_vals = [h[0] for h in history_sirs]
I_vals = [h[1] for h in history_sirs]
R_vals = [h[2] for h in history_sirs]
plt.plot(steps, S_vals, 'b-', label='易感者 (S)', linewidth=2)
plt.plot(steps, I_vals, 'r-', label='感染者 (I)', linewidth=2)
plt.plot(steps, R_vals, 'g-', label='康复者 (R)', linewidth=2)
plt.xlabel('时间步')
plt.ylabel('节点数量')'SIRS传播模型模拟结果（带免疫丧失）')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

网络结构对比实验

def compare_network_structures():
    """比较不同网络结构下的传播效果"""
    n_nodes = 100
    beta = 0.2
    gamma = 0.1
    # 创建不同结构的网络
    networks = {
        '小世界网络': create_network(n_nodes, 'small_world'),
        '无标度网络': create_network(n_nodes, 'scale_free'),
        '随机网络': create_network(n_nodes, 'random')
    }
    plt.figure(figsize=(12, 4))
    for idx, (name, G) in enumerate(networks.items()):
        # 运行SIR模型
        sir = SIRModel(G, beta, gamma)
        sir.initialize([0])
        history = sir.simulate(100)
        # 提取感染人数
        I_vals = [h[1] for h in history]
        plt.subplot(1, 3, idx + 1)
        plt.plot(range(len(I_vals)), I_vals, 'r-', linewidth=2)
        plt.xlabel('时间步')
        plt.ylabel('感染人数')
        plt.title(f'{name}\n峰值: {max(I_vals)}')
        plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()
# 运行对比实验
compare_network_structures()

实时传播可视化

def visualize_spread(G, states, step):
    """实时可视化传播状态"""
    plt.figure(figsize=(10, 8))
    # 为不同状态的节点设置颜色
    color_map = []
    for node in G.nodes():
        if node in states['I']:
            color_map.append('red')  # 感染者
        elif node in states['R']:
            color_map.append('green')  # 康复者
        else:
            color_map.append('blue')  # 易感者
    pos = nx.spring_layout(G, seed=42, k=2, iterations=50)
    nx.draw(G, pos, node_color=color_map, 
            node_size=100, 
            alpha=0.8,
            with_labels=False,
            edge_color='gray',
            width=0.5)
    plt.title(f'传播状态 - 步骤 {step}\n'
              f'S: {states["S"]}, I: {states["I"]}, R: {states["R"]}')
    plt.axis('off')
    plt.show()
# 示例：可视化传播过程
G = create_network(50, 'scale_free')
sir = SIRModel(G, beta=0.25, gamma=0.15)
sir.initialize([0])
# 模拟并可视化关键步骤
for step in [0, 5, 10, 20]:
    states = sir.get_state()
    visualize_spread(G, states, step)
    # 执行5步传播
    for _ in range(5):
        sir.step()

参数敏感性分析

def parameter_sensitivity_analysis():
    """分析不同参数对传播的影响"""
    G = create_network(100, 'small_world')
    initial_infected = [0]
    beta_values = [0.1, 0.2, 0.3, 0.4]
    gamma_values = [0.05, 0.1, 0.15, 0.2]
    plt.figure(figsize=(15, 5))
    # 改变感染概率beta
    plt.subplot(1, 2, 1)
    for beta in beta_values:
        sir = SIRModel(G, beta, gamma=0.1)
        sir.initialize(initial_infected)
        history = sir.simulate(100)
        I_vals = [h[1] for h in history]
        plt.plot(I_vals, label=f'β={beta}', linewidth=2)
    plt.xlabel('时间步')
    plt.ylabel('感染人数')
    plt.title('不同感染概率下的传播曲线')
    plt.legend()
    plt.grid(True, alpha=0.3)
    # 改变恢复概率gamma
    plt.subplot(1, 2, 2)
    for gamma in gamma_values:
        sir = SIRModel(G, beta=0.2, gamma=gamma)
        sir.initialize(initial_infected)
        history = sir.simulate(100)
        I_vals = [h[1] for h in history]
        plt.plot(I_vals, label=f'γ={gamma}', linewidth=2)
    plt.xlabel('时间步')
    plt.ylabel('感染人数')
    plt.title('不同恢复概率下的传播曲线')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()
# 运行参数敏感性分析
parameter_sensitivity_analysis()

代码展示了使用NetworkX实现不同传播模型的方法：

1. SIR模型：基本的感染-恢复模型
2. SI模型：只感染不恢复的模型
3. SIRS模型：带有免疫丧失的模型
4. 网络结构对比：不同网络拓扑对传播的影响
5. 实时可视化：动态展示传播过程
6. 参数敏感性分析：测试不同参数的影响

这些模型可以应用于：

• 疾病传播模拟
• 信息传播分析
• 网络影响力分析
• 免疫策略评估

根据具体需求,可以修改这些模型，比如添加隔离措施、引入时间延迟、考虑异质性等。