扩散模型

class pynetim.diffusion_model.BaseCallbackDiffusionModel[source]

Bases: PyDiffusionModelBase

C++回调版自定义传播模型基类。

继承 C++ 的 PyDiffusionModelBase，通过回调机制实现自定义传播逻辑。 单线程模式下可用，比纯 Python 实现快。 多线程模式受 Python GIL 限制，无法真正并行。

graph

图对象。

seeds

种子节点集合。

num_nodes

节点数量。

Example

>>> from pynetim.diffusion_model import BaseCallbackDiffusionModel
>>>
>>> class MyICModel(BaseCallbackDiffusionModel):
...     def run_single_trial(self, seeds, random_seed):
...         import random
...         random.seed(random_seed)
...         activated = set(seeds)
...         current = list(seeds)
...         count = len(seeds)
...         frequency = [0] * self.graph.num_nodes
...
...         while current:
...             new_active = []
...             for node in current:
...                 for neighbor, weight in self.graph.out_neighbors(node):
...                     if neighbor not in activated and random.random() < weight:
...                         activated.add(neighbor)
...                         new_active.append(neighbor)
...                         count += 1
...                         frequency[neighbor] += 1
...             current = new_active
...
...         return count, activated, frequency
...
>>> model = MyICModel(graph, {0, 1})
>>> avg = model.run_monte_carlo_diffusion(mc_rounds=1000, random_seed=42)
run_single_trial(seeds, random_seed)[source]

执行单次传播试验。

子类必须重写此方法实现自定义传播逻辑。

Parameters:

  • seeds (List[int]) – 初始种子节点列表。

  • random_seed (int) – 随机数种子，用于确保结果可重现。

Returns:

包含三个元素的元组：

  • 激活节点总数

  • 激活的节点集合

  • 每个节点的激活频数列表

Return type:

Tuple[int, Set[int], List[int]]

Raises:

NotImplementedError – 子类未实现此方法时抛出。

Note

此方法会被 C++ 层回调，因此需要保持签名一致。

class pynetim.diffusion_model.BaseMultiprocessDiffusionModel(graph, seeds)[source]

Bases: ABC

多进程版自定义传播模型基类。

纯 Python 实现，使用 multiprocessing 实现真正的并行计算。 不受 Python GIL 限制，多进程可真正并行加速。 适合大规模模拟和需要并行加速的自定义模型。

Parameters:
graph

图对象。

seeds

种子节点集合。

num_nodes

节点数量。

Example

>>> from pynetim.diffusion_model import BaseMultiprocessDiffusionModel
>>>
>>> class MyICModel(BaseMultiprocessDiffusionModel):
...     def run_single_trial(self, seeds, random_seed):
...         import random
...         random.seed(random_seed)
...         activated = set(seeds)
...         current = list(seeds)
...         count = len(seeds)
...         frequency = [0] * self.graph.num_nodes
...
...         while current:
...             new_active = []
...             for node in current:
...                 for neighbor, weight in self.graph.out_neighbors_with_weights(node):
...                     if neighbor not in activated and random.random() < weight:
...                         activated.add(neighbor)
...                         new_active.append(neighbor)
...                         count += 1
...                         frequency[neighbor] += 1
...             current = new_active
...
...         return count, activated, frequency
...
>>> model = MyICModel(graph, {0, 1})
>>> avg = model.run_monte_carlo_diffusion(mc_rounds=1000, num_processes=4, random_seed=42)
__init__(graph, seeds)[source]

初始化传播模型。

Parameters:

  • graph (IMGraph) – 图对象。

  • seeds (Set[int]) – 种子节点集合。

run_monte_carlo_diffusion(mc_rounds, num_processes=None, show_progress=False, random_seed=None, normalize=False)[source]

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds (int) – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • num_processes (int | None) – 进程数，默认使用 CPU 核心数。

  • show_progress (bool) – 是否显示进度条（暂未实现）。

  • random_seed (int | None) – 随机数种子，默认为 None（每次结果不同）。

  • normalize (bool) – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

Return type:

float

Note

当模拟次数较多时（>= 进程数 * 10），自动启用多进程并行。

run_monte_carlo_with_frequency(mc_rounds, num_processes=None, random_seed=None, normalize=False)[source]

运行蒙特卡洛模拟，返回平均影响力和激活频数。

Parameters:

  • mc_rounds (int) – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • num_processes (int | None) – 进程数，默认使用 CPU 核心数。

  • random_seed (int | None) – 随机数种子，默认为 None（每次结果不同）。

  • normalize (bool) – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

包含两个元素：

  • 平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

  • 每个节点的激活频数列表

Return type:

Tuple[float, List[int]]

abstract run_single_trial(seeds, random_seed)[source]

执行单次传播试验。

子类必须重写此方法实现自定义传播逻辑。

Parameters:

  • seeds (List[int]) – 初始种子节点列表。

  • random_seed (int) – 随机数种子，用于确保结果可重现。

Returns:

包含三个元素的元组：

  • 激活节点总数

  • 激活的节点集合

  • 每个节点的激活频数列表

Return type:

Tuple[int, Set[int], List[int]]

set_seeds(seeds)[source]

设置种子节点集合。

Parameters:

seeds (Set[int]) – 新的种子节点集合。

class pynetim.diffusion_model.IndependentCascadeModel

Bases: pybind11_object

__init__(graph: IMGraph, seeds: set[int], record_activated: bool = False, record_activation_frequency: bool = False) → None

构造独立级联模型。

Parameters:

  • graph – 图对象。

  • seeds – 初始种子节点集合。

  • record_activated – 是否记录激活节点，默认为 False。

  • record_activation_frequency – 是否记录激活频数，默认为 False。

get_activated_nodes() → set[int]

获取上次模拟的激活节点集合。

Returns:

激活节点集合。仅在 record_activated 为 True 时有效。

Return type:

Set[int]

get_activation_frequency() → list[int]

获取各节点的激活频数。

Returns:

激活频数列表，索引 i 表示节点 i 被激活的次数。

Return type:

List[int]

run_monte_carlo_diffusion(mc_rounds: int, random_seed: int | None = None, use_multithread: bool = False, num_threads: int = 0, normalize: bool = False) → float

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

  • use_multithread – 是否启用多线程，默认为 False。

  • num_threads – 线程数，当 use_multithread=True 时必须大于 0。

  • normalize – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

Return type:

float

Raises:

ValueError – 当 use_multithread=True 但 num_threads <= 0 时抛出。

run_single_simulation(random_seed: int | None = None) → int

执行单次传播模拟。

Parameters:

random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

Returns:

本次模拟激活的节点数。

Return type:

int

set_record_activated(record: bool) → None

启用或禁用激活节点记录。

Parameters:

record – 是否记录激活节点。

set_record_activation_frequency(record: bool) → None

启用或禁用激活频数记录。

Parameters:

record – 是否记录激活频数。

set_seeds(new_seeds: set[int]) → None

更新种子节点集合。

Parameters:

new_seeds – 新的种子节点集合。

class pynetim.diffusion_model.LinearThresholdModel

Bases: pybind11_object

__init__(graph: IMGraph, seeds: set[int], theta_l: float = 0.0, theta_h: float = 1.0, record_activated: bool = False, record_activation_frequency: bool = False) → None

构造线性阈值模型。

Parameters:

  • graph – 图对象。

  • seeds – 初始种子节点集合。

  • theta_l – 阈值下界，默认为 0.0。

  • theta_h – 阈值上界，默认为 1.0。

  • record_activated – 是否记录激活节点，默认为 False。

  • record_activation_frequency – 是否记录激活频数，默认为 False。

get_activated_nodes() → set[int]

获取上次模拟的激活节点集合。

Returns:

激活节点集合。仅在 record_activated 为 True 时有效。

Return type:

Set[int]

get_activation_frequency() → list[int]

获取各节点的激活频数。

Returns:

激活频数列表，索引 i 表示节点 i 被激活的次数。

Return type:

List[int]

run_monte_carlo_diffusion(mc_rounds: int, random_seed: int | None = None, use_multithread: bool = False, num_threads: int = 0, normalize: bool = False) → float

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

  • use_multithread – 是否启用多线程，默认为 False。

  • num_threads – 线程数，当 use_multithread=True 时必须大于 0。

  • normalize – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

Return type:

float

Raises:

ValueError – 当 use_multithread=True 但 num_threads <= 0 时抛出。

run_single_simulation(random_seed: int | None = None) → int

执行单次传播模拟。

Parameters:

random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

Returns:

本次模拟激活的节点数。

Return type:

int

set_record_activated(record: bool) → None

启用或禁用激活节点记录。

Parameters:

record – 是否记录激活节点。

set_record_activation_frequency(record: bool) → None

启用或禁用激活频数记录。

Parameters:

record – 是否记录激活频数。

set_seeds(new_seeds: set[int]) → None

更新种子节点集合。

Parameters:

new_seeds – 新的种子节点集合。

class pynetim.diffusion_model.PyDiffusionModelBase

Bases: pybind11_object

__init__(graph: IMGraph, seeds: set[int], py_override: object = None, record_activated: bool = False, record_activation_frequency: bool = False) → None

构造 Python 兼容的传播模型基类。

Parameters:

  • graph – 图对象。

  • seeds – 初始种子节点集合。

  • py_override – Python 重写对象，默认为 None。

  • record_activated – 是否记录激活节点，默认为 False。

  • record_activation_frequency – 是否记录激活频数，默认为 False。

get_activated_nodes() → set[int]

获取上次模拟的激活节点集合。

Returns:

激活节点集合。

Return type:

Set[int]

get_activation_frequency() → list[int]

获取各节点的激活频数。

Returns:

激活频数列表。

Return type:

List[int]

get_py_override() → object

获取 Python 重写对象。

Returns:

Python 重写对象。

Return type:

object

property graph

获取图对象。

run_monte_carlo_diffusion(mc_rounds: int, random_seed: int | None = None, use_multithread: bool = False, num_threads: int = 0, normalize: bool = False) → float

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

  • use_multithread – 是否启用多线程，默认为 False。

  • num_threads – 线程数，当 use_multithread=True 时必须大于 0。

  • normalize – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

Return type:

float

Raises:

ValueError – 当 use_multithread=True 但 num_threads <= 0 时抛出。

Note

多线程模式下受 Python GIL 限制，无法真正并行。 建议使用 BaseMultiprocessDiffusionModel 实现真正的并行计算。

run_single_simulation(random_seed: int | None = None) → int

执行单次传播模拟。

Parameters:

random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

Returns:

本次模拟激活的节点数。

Return type:

int

set_py_override(obj: object) → None

设置 Python 重写对象。

Parameters:

obj – Python 重写对象，用于自定义传播逻辑。

set_record_activated(record: bool) → None

启用或禁用激活节点记录。

Parameters:

record – 是否记录激活节点。

set_record_activation_frequency(record: bool) → None

启用或禁用激活频数记录。

Parameters:

record – 是否记录激活频数。

set_seeds(new_seeds: set[int]) → None

更新种子节点集合。

Parameters:

new_seeds – 新的种子节点集合。

class pynetim.diffusion_model.SusceptibleInfectedModel

Bases: pybind11_object

__init__(graph: IMGraph, seeds: set[int], beta: float, max_steps: int, record_activated: bool = False, record_activation_frequency: bool = False) → None

构造 SI 模型。

Parameters:

  • graph – 图对象。

  • seeds – 初始感染节点集合。

  • beta – 感染概率，必须在 (0, 1] 范围内。

  • max_steps – 最大传播步数。

  • record_activated – 是否记录感染节点，默认为 False。

  • record_activation_frequency – 是否记录感染频数，默认为 False。

get_activated_nodes() → set[int]

获取上次模拟的感染节点集合。

Returns:

感染节点集合。仅在 record_activated 为 True 时有效。

Return type:

Set[int]

get_activation_frequency() → list[int]

获取各节点的感染频数。

Returns:

感染频数列表，索引 i 表示节点 i 被感染的次数。

Return type:

List[int]

run_monte_carlo_diffusion(mc_rounds: int, random_seed: int | None = None, use_multithread: bool = False, num_threads: int = 0, normalize: bool = False) → float

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

  • use_multithread – 是否启用多线程，默认为 False。

  • num_threads – 线程数，当 use_multithread=True 时必须大于 0。

  • normalize – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均感染节点数。若 normalize=True，返回归一化后的感染比例。

Return type:

float

Raises:

ValueError – 当 use_multithread=True 但 num_threads <= 0 时抛出。

run_single_simulation(random_seed: int | None = None) → int

执行单次传播模拟。

Parameters:

random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

Returns:

本次模拟感染的节点数。

Return type:

int

set_beta(beta: float) → None

设置感染概率。

Parameters:

beta – 感染概率，必须在 (0, 1] 范围内。

set_max_steps(max_steps: int) → None

设置最大传播步数。

Parameters:

max_steps – 最大传播步数。

set_record_activated(record: bool) → None

启用或禁用感染节点记录。

Parameters:

record – 是否记录感染节点。

set_record_activation_frequency(record: bool) → None

启用或禁用感染频数记录。

Parameters:

record – 是否记录感染频数。

set_seeds(new_seeds: set[int]) → None

更新种子节点集合。

Parameters:

new_seeds – 新的种子节点集合。

class pynetim.diffusion_model.SusceptibleInfectedRecoveredModel

Bases: pybind11_object

__init__(graph: IMGraph, seeds: set[int], beta: float, gamma: float, record_activated: bool = False, record_activation_frequency: bool = False) → None

构造 SIR 模型。

Parameters:

  • graph – 图对象。

  • seeds – 初始感染节点集合。

  • beta – 感染概率，必须在 (0, 1] 范围内。

  • gamma – 恢复概率，必须在 (0, 1] 范围内。

  • record_activated – 是否记录感染/恢复节点，默认为 False。

  • record_activation_frequency – 是否记录感染/恢复频数，默认为 False。

get_activated_nodes() → set[int]

获取上次模拟的感染节点集合。

Returns:

感染节点集合。仅在 record_activated 为 True 时有效。

Return type:

Set[int]

get_activation_frequency() → list[int]

获取各节点的感染频数。

Returns:

感染频数列表，索引 i 表示节点 i 被感染的次数。

Return type:

List[int]

run_monte_carlo_diffusion(mc_rounds: int, random_seed: int | None = None, use_multithread: bool = False, num_threads: int = 0, normalize: bool = False) → float

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

  • use_multithread – 是否启用多线程，默认为 False。

  • num_threads – 线程数，当 use_multithread=True 时必须大于 0。

  • normalize – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均感染节点数。若 normalize=True，返回归一化后的感染比例。

Return type:

float

Raises:

ValueError – 当 use_multithread=True 但 num_threads <= 0 时抛出。

run_single_simulation(random_seed: int | None = None) → int

执行单次传播模拟。

Parameters:

random_seed – 随机种子，用于结果可重现。若为 None 则使用真随机种子。

Returns:

本次模拟感染的节点数。

Return type:

int

set_beta(beta: float) → None

设置感染概率。

Parameters:

beta – 感染概率，必须在 (0, 1] 范围内。

set_gamma(gamma: float) → None

设置恢复概率。

Parameters:

gamma – 恢复概率，必须在 (0, 1] 范围内。

set_record_activated(record: bool) → None

启用或禁用感染/恢复节点记录。

Parameters:

record – 是否记录感染/恢复节点。

set_record_activation_frequency(record: bool) → None

启用或禁用感染/恢复频数记录。

Parameters:

record – 是否记录感染/恢复频数。

set_seeds(new_seeds: set[int]) → None

更新种子节点集合。

Parameters:

new_seeds – 新的种子节点集合。

自定义扩散模型

class pynetim.diffusion_model.base_callback_diffusion_model.BaseCallbackDiffusionModel[source]

Bases: PyDiffusionModelBase

C++回调版自定义传播模型基类。

继承 C++ 的 PyDiffusionModelBase，通过回调机制实现自定义传播逻辑。 单线程模式下可用，比纯 Python 实现快。 多线程模式受 Python GIL 限制，无法真正并行。

graph

图对象。

seeds

种子节点集合。

num_nodes

节点数量。

Example

>>> from pynetim.diffusion_model import BaseCallbackDiffusionModel
>>>
>>> class MyICModel(BaseCallbackDiffusionModel):
...     def run_single_trial(self, seeds, random_seed):
...         import random
...         random.seed(random_seed)
...         activated = set(seeds)
...         current = list(seeds)
...         count = len(seeds)
...         frequency = [0] * self.graph.num_nodes
...
...         while current:
...             new_active = []
...             for node in current:
...                 for neighbor, weight in self.graph.out_neighbors(node):
...                     if neighbor not in activated and random.random() < weight:
...                         activated.add(neighbor)
...                         new_active.append(neighbor)
...                         count += 1
...                         frequency[neighbor] += 1
...             current = new_active
...
...         return count, activated, frequency
...
>>> model = MyICModel(graph, {0, 1})
>>> avg = model.run_monte_carlo_diffusion(mc_rounds=1000, random_seed=42)
run_single_trial(seeds, random_seed)[source]

执行单次传播试验。

子类必须重写此方法实现自定义传播逻辑。

Parameters:

  • seeds (List[int]) – 初始种子节点列表。

  • random_seed (int) – 随机数种子，用于确保结果可重现。

Returns:

包含三个元素的元组：

  • 激活节点总数

  • 激活的节点集合

  • 每个节点的激活频数列表

Return type:

Tuple[int, Set[int], List[int]]

Raises:

NotImplementedError – 子类未实现此方法时抛出。

Note

此方法会被 C++ 层回调，因此需要保持签名一致。

class pynetim.diffusion_model.base_multiprocess_diffusion_model.BaseMultiprocessDiffusionModel(graph, seeds)[source]

Bases: ABC

多进程版自定义传播模型基类。

纯 Python 实现，使用 multiprocessing 实现真正的并行计算。 不受 Python GIL 限制，多进程可真正并行加速。 适合大规模模拟和需要并行加速的自定义模型。

Parameters:
graph

图对象。

seeds

种子节点集合。

num_nodes

节点数量。

Example

>>> from pynetim.diffusion_model import BaseMultiprocessDiffusionModel
>>>
>>> class MyICModel(BaseMultiprocessDiffusionModel):
...     def run_single_trial(self, seeds, random_seed):
...         import random
...         random.seed(random_seed)
...         activated = set(seeds)
...         current = list(seeds)
...         count = len(seeds)
...         frequency = [0] * self.graph.num_nodes
...
...         while current:
...             new_active = []
...             for node in current:
...                 for neighbor, weight in self.graph.out_neighbors_with_weights(node):
...                     if neighbor not in activated and random.random() < weight:
...                         activated.add(neighbor)
...                         new_active.append(neighbor)
...                         count += 1
...                         frequency[neighbor] += 1
...             current = new_active
...
...         return count, activated, frequency
...
>>> model = MyICModel(graph, {0, 1})
>>> avg = model.run_monte_carlo_diffusion(mc_rounds=1000, num_processes=4, random_seed=42)
__init__(graph, seeds)[source]

初始化传播模型。

Parameters:

  • graph (IMGraph) – 图对象。

  • seeds (Set[int]) – 种子节点集合。

run_monte_carlo_diffusion(mc_rounds, num_processes=None, show_progress=False, random_seed=None, normalize=False)[source]

运行蒙特卡洛模拟，计算平均影响力。

Parameters:

  • mc_rounds (int) – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • num_processes (int | None) – 进程数，默认使用 CPU 核心数。

  • show_progress (bool) – 是否显示进度条（暂未实现）。

  • random_seed (int | None) – 随机数种子，默认为 None（每次结果不同）。

  • normalize (bool) – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

Return type:

float

Note

当模拟次数较多时（>= 进程数 * 10），自动启用多进程并行。

run_monte_carlo_with_frequency(mc_rounds, num_processes=None, random_seed=None, normalize=False)[source]

运行蒙特卡洛模拟，返回平均影响力和激活频数。

Parameters:

  • mc_rounds (int) – 蒙特卡洛模拟次数，建议 1000-10000 次。

  • num_processes (int | None) – 进程数，默认使用 CPU 核心数。

  • random_seed (int | None) – 随机数种子，默认为 None（每次结果不同）。

  • normalize (bool) – 是否将结果归一化（除以图节点数），默认为 False。

Returns:

包含两个元素：

  • 平均激活节点数。若 normalize=True，返回归一化后的影响力比例。

  • 每个节点的激活频数列表

Return type:

Tuple[float, List[int]]

abstract run_single_trial(seeds, random_seed)[source]

执行单次传播试验。

子类必须重写此方法实现自定义传播逻辑。

Parameters:

  • seeds (List[int]) – 初始种子节点列表。

  • random_seed (int) – 随机数种子，用于确保结果可重现。

Returns:

包含三个元素的元组：

  • 激活节点总数

  • 激活的节点集合

  • 每个节点的激活频数列表

Return type:

Tuple[int, Set[int], List[int]]

set_seeds(seeds)[source]

设置种子节点集合。

Parameters:

seeds (Set[int]) – 新的种子节点集合。