本文目录导读：
我来为你展示如何使用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()
主要功能特点
- 基础级联模型：简单的阈值级联传播
- 参数敏感性分析：研究不同阈值的影响
- 网络结构比较：比较不同网络中的传播效果
- 动画演示：可视化级联传播过程
- 高级变体：支持独立级联和线性阈值模型
应用场景
- 信息传播：模拟社交媒体中的信息扩散
- 流行病建模：疾病传播模拟
- 技术采纳：新产品或技术的市场扩散
- 意见形成：社会舆论形成过程
这些代码提供了完整的级联模型实现,适合教学、研究和实际应用，你可以根据需要调整参数和网络结构。
