Spreading models

Contents

Spreading models#

See dedicated propagation model guide for these functions.

Base structures for concrete models#

class BaseModel#

Bases: ABC

Base abstract propagation model.

_abc_impl = <_abc._abc_data object>#
abstract property _compartmental_graph: CompartmentalGraph#

Compartmental model that defines allowed transitions and states.

abstract property _seed_selector: BaseSeedSelector#

A method of selecting seed agents.

abstract agent_evaluation_step(agent: Any, layer_name: str, net: MultilayerNetwork) → str#

Try to change state of given node of the network according to model.

Parameters:

  • agent – id of the node or the actor to evaluate

  • layer_name – a layer where the node exists

  • net – a network where the node exists

Returns:

state of the model after evaluation

abstract determine_initial_states(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Determine initial states in the network according to seed selector.

Parameters:

net – network to initialise seeds for

Returns:

list of nodes with their states

abstract get_allowed_states(net: MultilayerNetwork) → dict[str, tuple[str, ...]]#

Return dict with allowed states in each layer of net if applied model.

Parameters:

net – a network to determine allowed nodes’ states for

static get_states_num(net: MultilayerNetwork) → dict[str, tuple[tuple[Any, int], ...]]#

Return states in the network with number of agents that adopted them.

It is the most basic function which assumes that field “status” in the network is self explaining and there is no need to decode it (e.g. to separate hidden state from public one).

Returns:

dictionary with items representing each of layers and with summary of nodes states in values

abstract network_evaluation_step(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Evaluate the network at one time stamp according to the model.

Parameters:

network – a network to evaluate

Returns:

list of nodes that changed state after the evaluation

static update_network(net: MultilayerNetwork, activated_nodes: list[NetworkUpdateBuffer]) → list[dict[str, str]]#

Update the network global state by list of already activated nodes.

Parameters:

  • net – network to update

  • activated_nodes – already activated nodes

class CompartmentalGraph#

Bases: object

Class which encapsulates model of processes spread in the network.

__init__() → None#

Create empty object.

_get_desctiprion_str() → str#

Print out parameters of the compartmental model.

Returns:

returns string describing object,

static _int_to_bins(bins: tuple[int | float | number, ...], base_num: int) → list[int]#
add(process_name: str, states: list[str]) → None#

Add process with allowed states to the compartmental graph.

Parameters:

  • layer – name of process, e.g. “Illness”

  • type – names of states like [‘s’, ‘i’, ‘r’]

compile(background_weight: float = 0.0, track_changes: bool = False) → None#

Create transition matrices for models of propagation in each layer.

All transition probabilities are set to 0. To be more specific, transitions matrices are stored as a networkx one-directional graph. After compilation user is able to set certain transitions in model.

Parameters:

  • background_weight – [0,1] describes default weight of transition to make propagation more realistic by default it is set to 0

  • track_changes – a flag to track progress of matrices creation

get_compartments() → dict[str, tuple[str, ...]]#

Get model parameters, i.e. names of layers and states in each layer.

Returns:

dictionary keyed by names of layer, valued by tuples of states labels

get_possible_transitions(state: tuple[str, ...], layer: str) → dict[str, float]#

Return possible transitions from given state in given layer of model.

Note that possible transition is a transition with weight > 0.

Parameters:

  • state – state of the propagation model, i.e. (‘awareness.UA’, ‘illness.I’, ‘vaccination.V’)

  • layer – name of the layer of propagation model from which possible transitions are being returned

Returns:

dict with possible transitions in shape of: {possible different state in given layer: weight}

get_seeding_budget_for_network(net: MultilayerNetwork, actorwise: bool = False) → dict[str, dict[Any, int]]#

Transform seeding budget from %s to numbers according to nodes/actors.

Parameters:

  • net – input network to convert seeding budget for

  • actorwise – compute seeding budget for actors, else for nodes

Returns:

dictionary in form as e.g.: {“ill”: {“suspected”: 45, “infected”: 4, “recovered”: 1}, “vacc”: {“unvaccinated”: 35, “vaccinated”: 15}} for seeding_budget dict: {“ill”: (90, 8, 2), “vacc”: (70, 30)} and 50 nodes in each layer and nodewise mode.

reserved_names = {'_seeding_budget', 'background_weight', 'graph'}#
property seeding_budget: dict[str, tuple[int | float | number, ...]]#

Get seeding budget as % of the nodes in form of compartments as a dict.

E.g. something like that: {“ill”: (90, 8, 2), “aware”: (60, 40), “vacc”: (70, 30)} for compartments such as: “ill”: [s, i, r], “aware”: [u, a], “vacc”: [n, v]

set_transition_canonical(layer: str, transition: EdgeView, weight: float) → None#

Set weight of certain transition in propagation model.

Parameters:

  • layer – name of the later in model

  • transition – name of transition to be activated, edge in propagation model graph

  • weight – in range (number [0, 1]) of activation

set_transition_fast(initial_layer_attribute: str, final_layer_attribute: str, constraint_attributes: tuple[str, ...], weight: float) → None#

Set weight of certain transition in propagation model.

Parameters:

  • initial_layer_attribute – value of initial attribute which is being transited

  • final_layer_attribute – value of final attribute which is being transition

  • constraint_attributes – other attributes available in the propagation model

  • weight – weight (in range [0, 1]) of activation

set_transitions_in_random_edges(weights: list[list[float]]) → None#

Set out random transitions in propagation model using given weights.

Parameters:

weights – list of weights to be set in random nodes e.g. for model of 3 layers that list [[0.1, 0.2], [0.03, 0.45], [0.55]] will change 2 weights in first layer, 2, i second and 1 in third

class NetworkUpdateBuffer(node_name: str, layer_name: str, new_state: str)#

Bases: object

Auxiliary class to keep info about nodes that needs to be updated.

__init__(node_name: str, layer_name: str, new_state: str) → None#
layer_name: str#
new_state: str#
node_name: str#
to_json() → dict[str, str]#

Return dict writable to JSON.

Concrete propagation models#

Import from network_diffusion.models.

class DSAAModel(compartmental_graph: CompartmentalGraph)#

Bases: BaseModel

This model implements algorithm presented at DSAA 2022.

__init__(compartmental_graph: CompartmentalGraph) → None#

Create the object.

agent_evaluation_step(agent: Any, layer_name: str, net: MultilayerNetwork) → str#

Try to change state of given node of the network according to model.

Parameters:

  • agent – id of the node (here agent) to evaluate

  • layer_name – a layer where the node exists

  • network – a network where the node exists

Returns:

state of the model after evaluation

property compartments: CompartmentalGraph#

Return defined compartments and allowed transitions.

determine_initial_states(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Set initial states in the network according to seed selection method.

Parameters:

net – network to initialise seeds for

Returns:

a list of state of the network after initialisation

get_allowed_states(net: MultilayerNetwork) → dict[str, tuple[str, ...]]#

Return dict with allowed states in each layer of net if applied model.

In this model each process is binded with network’s layer, hence we return just the compartments and allowed states.

Parameters:

net – a network to determine allowed nodes’ states for

network_evaluation_step(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Evaluate the network at one time stamp according to the model.

We ae updating nodes ‘on the fly’, hence the activated_nodes list is empty. This behaviour is due to intention to very reflect te algorithm presented at DSAA

Parameters:

network – a network to evaluate

Returns:

list of nodes that changed state after the evaluation

class MLTModel(seeding_budget: tuple[int | float | number, int | float | number], seed_selector: BaseSeedSelector, protocol: str, mi_value: float)#

Bases: BaseModel

This model implements Multilayer Linear Threshold Model.

The model has been presented in paper: “Influence Spread in the Heterogeneous Multiplex Linear Threshold Model” by Yaofeng Desmond Zhong, Vaibhav Srivastava, and Naomi Ehrich Leonard. This implementation extends it to multilayer cases.

ACTIVE_STATE = '1'#
INACTIVE_STATE = '0'#
PROCESS_NAME = 'MLTM'#
__init__(seeding_budget: tuple[int | float | number, int | float | number], seed_selector: BaseSeedSelector, protocol: str, mi_value: float) → None#

Create the object.

Parameters:

  • seeding_budget – a proportion of INACTIVE and ACTIVE nodes in each layer

  • seed_selector – class that selects initial seeds for simulation

  • protocol – logical operator that determines how to activate actor can be OR (then actor gets activated if it gets positive input in one layer) or AND (then actor gets activated if it gets positive input in all layers)

  • mi – activation threshold to transit from INACTIVE to ACTIVE in evaluation of the actor in particular layer

agent_evaluation_step(agent: MLNetworkActor, layer_name: str, net: MultilayerNetwork) → str#

Try to change state of given actor of the network according to model.

Parameters:

  • agent – actor to evaluate in given layer

  • layer_name – a layer where the actor exists

  • net – a network where the actor exists

Returns:

state of the actor in particular layer to be set after epoch

determine_initial_states(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Determine initial states in the net according to seed selection method.

Parameters:

net – network to initialise seeds for

Returns:

a list of nodes with their initial states

get_allowed_states(net: MultilayerNetwork) → dict[str, tuple[str, ...]]#

Return dict with allowed states in each layer of net if applied model.

Parameters:

net – a network to determine allowed nodes’ states for

network_evaluation_step(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Evaluate the network at one time stamp with MLTModel.

Parameters:

network – a network to evaluate

Returns:

list of nodes that changed state after the evaluation

class MICModel(seeding_budget: tuple[int | float | number, int | float | number, int | float | number], seed_selector: BaseSeedSelector, protocol: str, probability: float)#

Bases: BaseModel

This model implements Multilayer Independent Cascade Model.

ACTIVATED_NODE = '-1'#
ACTIVE_NODE = '1'#
INACTIVE_NODE = '0'#
PROCESS_NAME = 'MICM'#
__init__(seeding_budget: tuple[int | float | number, int | float | number, int | float | number], seed_selector: BaseSeedSelector, protocol: str, probability: float) → None#

Create the object.

Parameters:

  • seeding_budget – a proportion of INACTIVE, A`CTIVE and ACTIVATED nodes in each layer

  • seed_selector – class that selects initial seeds for simulation

  • protocol – logical operator that determines how to activate actor can be OR (then actor gets activated if it gets positive input in one layer) or AND (then actor gets activated if it gets positive input in all layers)

  • probability – threshold parameter which activate actor (a random variable must be lesser than this param to result in activation)

agent_evaluation_step(agent: MLNetworkActor, layer_name: str, net: MultilayerNetwork) → str#

Try to change state of given actor of the network according to model.

Parameters:

  • agent – actor to evaluate in given layer

  • layer_name – a layer where the actor exists

  • net – a network where the actor exists

Returns:

state of the actor in particular layer to be set after epoch

determine_initial_states(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Set initial states in the network according to seed selection method.

Parameters:

net – network to initialise seeds for

Returns:

a list of state of the network after initialisation

get_allowed_states(net: MultilayerNetwork) → dict[str, tuple[str, ...]]#

Return dict with allowed states in each layer of net if applied model.

Parameters:

net – a network to determine allowed nodes’ states for

network_evaluation_step(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Evaluate the network at one time stamp with MICModel.

Parameters:

net – a network to evaluate

Returns:

list of nodes that changed state after the evaluation

class TemporalNetworkEpistemologyModel(seeding_budget: tuple[int | float | number, int | float | number], seed_selector: BaseSeedSelector, trials_nr: int, epsilon: float)#

Bases: BaseModel

Generalized version of Temporal Network Epistemology Model.

A_STATE = 'A'#
B_STATE = 'B'#
PROCESS_NAME = 'TNEM'#
__init__(seeding_budget: tuple[int | float | number, int | float | number], seed_selector: BaseSeedSelector, trials_nr: int, epsilon: float) → None#

Create the object.

Parameters:

  • seeding_budget – a proportion of INACTIVE and ACTIVE agents

  • seed_selector – class that selects initial seeds for simulation

  • trials_nr – number of trials an agent performs when experimenting with environment by drawing a sample from the binomial distribution

  • epsilon – the difference between expected value of B action, and A action equal to 0.5

agent_evaluation_step(agent: MLNetworkActor, layer_name: str, net: MultilayerNetwork) → str#

Try to change state of given actor of the network according to model.

Parameters:

  • agent – actor to evaluate in given layer

  • net – a network where the actor exists

  • snapshot_id – currently processed snapshot

Returns:

state of the actor to be set in the next snapshot

static decode_actor_status(encoded_status: str) → tuple[str, float, int]#

Decode agent features from str form.

Parameters:

encoded_status – a string representation of agent status

Returns:

a tuple with agent state, belief level and evidence

determine_initial_states(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Set initial states in the network according to seed selection method.

Parameters:

net – network to initialise seeds for

Returns:

a list of state of the network after initialisation

static encode_actor_status(state: str, belief: float, evidence: int) → str#

Encode agent features to str form.

Parameters:

  • state – state of an actor

  • belief – level of agent’s belief

  • evidence – nr of successes drawn from binomial distribution in an experiment

Returns:

a string representation of agent status

get_allowed_states(net: MultilayerNetwork) → dict[str, tuple[str, ...]]#

Return dict with allowed states of net if applied model.

Parameters:

net – a network to determine allowed nodes’ states for

static get_states_num(net: MultilayerNetwork) → dict[str, tuple[tuple[Any, int], ...]]#

Return states in the network with number of agents that adopted them.

Vector of states for each agent is following: <state of an actor> <agent’s belief><evidence>; and we are interested only in the state attribute.

Parameters:

net – a network to get states number for

Returns:

dictionary with items representing each of layers and with summary of nodes states in values

network_evaluation_step(net: MultilayerNetwork) → list[NetworkUpdateBuffer]#

Evaluate the given snapshot of the network.

Parameters:

net – a network to evaluate

Returns:

list of nodes that changed state after the evaluation

Contents