Spreading

Spreading: rumours, cascades, contagions on hypergraphs, and the growth of texts.

ExposureInteraction(label, *, target, rate=0.06, exponent=2.0)

Bases: CouplingInteraction

Infection of susceptible participants by the doses of their environments, and recovery, in discrete time.

It is a CouplingInteraction over the hyperedges of a network in the map form. An environment, a hyperedge, gives each susceptible member the dose rate * rho ** exponent, where rho is the share of the other members that are infected, and the doses of the environments of a participant add up to D. In every step a susceptible participant is infected with probability 1 - exp(-D), and an infected one recovers with probability recovery, a parameter of the participants.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the participants on the network.	required
rate	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The dose of an environment whose other members are all infected.	0.06
exponent	Annotated[float, ParameterSpec(domain=open(0, oo))]	The exponent of the infection kernel, sublinear below 1 and supralinear above.	2.0

Examples:

>>> from skcomplex.models.social.spreading import ExposureInteraction
>>> ExposureInteraction('exposure', target='participants', exponent=1.5).check().exponent
1.5

Source code in src/skcomplex/models/social/spreading/_exposure.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.06,
    exponent: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
) -> None:
    super().__init__(
        label,
        target=target,
        mode='hyperedges',
        form='map',
        write='infected',
        function=self._get_dose,
        combine=self._get_next,
    )
    self.rate = rate
    self.exponent = exponent

GrowingWordElement(label, visual=None, *, count=1.0, innovation=0.1)

Bases: Element

Distinct word of the Simon model, with the number of times it occurs in the text.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the agent is drawn, or None for default_visual_.	None
count	Annotated[float, StateSpec(domain=NN)]	The number of occurrences.	1.0
innovation	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that the next word of the text is new.	0.1

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.social.spreading import GrowingWordElement
>>> group = GroupElement('words', element=GrowingWordElement, size=1, values={'count': 1.0})
>>> group.label, group.element.__name__, group.size
('words', 'GrowingWordElement', 1)

Source code in src/skcomplex/models/social/spreading/_growing_word.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    count: Annotated[float, StateSpec(domain=NN)] = 1.0,
    innovation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, visual=visual)
    self.count = count
    self.innovation = innovation

GrowingWordView(agent, stepper)

Bases: AgentView

View of a word of the Simon model, with its states and parameters as attributes.

Source code in src/skcomplex/engine/solvers/_mesa.py

def __init__(self, agent: mesa.Agent, stepper: 'MesaSolver') -> None:
    self._agent = agent
    self._stepper = stepper

InfectiveElement(label, visual=None, *, infected=0.0, recovery=1.0)

Bases: Element

Node of a contagion, susceptible at 0 and infected at 1, that recovers at a rate.

Parameters:

Name	Type	Description	Default
label	str	The label of the node or the group.	required
visual	VisualSpec | None	How the node is drawn, or None for default_visual_.	None
infected	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	The initial state, 0 or 1.	0.0
recovery	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which an infected node recovers.	1.0

Examples:

>>> from skcomplex.models.social.spreading import InfectiveElement
>>> InfectiveElement('node').recovery
1.0

Source code in src/skcomplex/models/social/spreading/_infective.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    infected: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    recovery: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.infected = infected
    self.recovery = recovery

ListenersElement(label, visual=None, *, n=0)

Bases: Element

Number of people in one state of a rumour: ignorant, spreading, or stifling.

Parameters:

Name	Type	Description	Default
label	str	The label of the element.	required
visual	VisualSpec | None	How the agent is drawn, or None for default_visual_.	None
n	Annotated[float, StateSpec(domain=NN)]	The initial number of people.	0

Examples:

>>> from skcomplex.models.social.spreading import ListenersElement
>>> ListenersElement('spreaders', n=1).n
1

Source code in src/skcomplex/models/social/spreading/_listeners.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    n: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.n = n

ParticipantElement(label, visual=None, *, infected=0.0, recovery=0.05)

Bases: Element

Participant in the environments of a contagion, susceptible at 0 and infected at 1.

The infection and the recovery are an ExposureInteraction of the hypergraph, which hypergraph_exposure_sis adds.

Parameters:

Name	Type	Description	Default
label	str	The label of the participant or the group.	required
visual	VisualSpec | None	How the participant is drawn, or None for default_visual_.	None
infected	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	The initial state, 0 or 1.	0.0
recovery	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that an infected participant recovers in a step.	0.05

Examples:

>>> from skcomplex.models.social.spreading import ParticipantElement
>>> ParticipantElement('participant').recovery
0.05

Source code in src/skcomplex/models/social/spreading/_participant.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    infected: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    recovery: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.05,
) -> None:
    super().__init__(label, visual=visual)
    self.infected = infected
    self.recovery = recovery

RumourContactsInteraction(label, *, ignorants, spreaders, stiflers, spread=1.0, stifle=1.0)

Bases: Interaction

Contacts of the Maki-Thompson rumour model in a well-mixed population of size N.

A spreader who meets an ignorant tells the rumour at rate spread * X * Y / N, and a spreader who meets another spreader or a stifler stops spreading at rate stifle * Y * (Y - 1 + Z) / N.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
ignorants	Annotated[str, RoleSpec()]	The path of the ignorants X.	required
spreaders	Annotated[str, RoleSpec()]	The path of the spreaders Y.	required
stiflers	Annotated[str, RoleSpec()]	The path of the stiflers Z.	required
spread	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The contact rate that spreads the rumour.	1.0
stifle	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The contact rate that stifles a spreader.	1.0

Examples:

>>> from skcomplex.models.social.spreading import RumourContactsInteraction
>>> RumourContactsInteraction('contacts', ignorants='X', spreaders='Y', stiflers='Z').spread
1.0

Source code in src/skcomplex/models/social/spreading/_rumour_contacts.py

def __init__(
    self,
    label: str,
    *,
    ignorants: Annotated[str, RoleSpec()],
    spreaders: Annotated[str, RoleSpec()],
    stiflers: Annotated[str, RoleSpec()],
    spread: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    stifle: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label)
    self.ignorants = ignorants
    self.spreaders = spreaders
    self.stiflers = stiflers
    self.spread = spread
    self.stifle = stifle

SimplicialInfectionInteraction(label, *, target, beta=0.1, beta_triangle=0.1)

Bases: HyperedgeInteraction

Infection of a susceptible node by the other members of a link or a filled triangle, when all are infected.

A susceptible node of a link is infected at rate beta when its neighbour is infected, and a susceptible node of a triangle at rate beta_triangle when both other members are infected.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group of nodes.	required
beta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate of infection through a link.	0.1
beta_triangle	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate of infection through a triangle.	0.1

Examples:

>>> from skcomplex.models.social.spreading import SimplicialInfectionInteraction
>>> SimplicialInfectionInteraction('infection', target='nodes', beta_triangle=0.5).beta_triangle
0.5

Source code in src/skcomplex/models/social/spreading/_simplicial_infection.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    beta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
    beta_triangle: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label)
    self.target = target
    self.beta = beta
    self.beta_triangle = beta_triangle

ThresholdActivationInteraction(label, *, target)

Bases: MapInteraction

Activation of the Watts cascade model, where a node becomes active once enough of its neighbours are active.

A node becomes active for good once the share of active nodes among its neighbouring nodes reaches its threshold. The nodes update one after the other in a new random order every step, each reading the activations the nodes before it set, so a cascade can cross several nodes in one step.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group of nodes.	required

Examples:

>>> from skcomplex.models.social.spreading import ThresholdActivationInteraction
>>> ThresholdActivationInteraction('activation', target='nodes').check().state
'active'

Source code in src/skcomplex/models/social/spreading/_threshold_activation.py

def __init__(self, label: str, *, target: Annotated[str, RoleSpec()]) -> None:
    super().__init__(
        label,
        target=target,
        state='active',
        function=self._get_active,
        schedule='random-sequential',
        form='agent',
    )

ThresholdNodeElement(label, visual=VisualSpec('agents', {'color_by': 'active', 'color_range': (0.0, 1.0)}), *, active=0.0, threshold=0.18)

Bases: Element

Node of the Watts cascade model, which the activation turns active once enough of its neighbours are active.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the agent is drawn, or None for default_visual_.	VisualSpec('agents', {'color_by': 'active', 'color_range': (0.0, 1.0)})
active	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	Whether the node starts active.	0.0
threshold	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The share of active neighbours at which the node becomes active.	0.18

Examples:

>>> from skcomplex.models.social.spreading import ThresholdNodeElement
>>> group = GroupElement('nodes', element=ThresholdNodeElement, size=10, values={'threshold': 0.18})
>>> group.label, group.element.__name__, group.size
('nodes', 'ThresholdNodeElement', 10)

Source code in src/skcomplex/models/social/spreading/_cascades.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'active', 'color_range': (0.0, 1.0)}),
    *,
    active: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    threshold: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.18,
) -> None:
    super().__init__(label, visual=visual)
    self.active = active
    self.threshold = threshold

ThresholdNodeView(agent, stepper)

Bases: AgentView

View of a node of the Watts cascade model, with its states and parameters as attributes.

Source code in src/skcomplex/engine/solvers/_mesa.py

def __init__(self, agent: mesa.Agent, stepper: 'MesaSolver') -> None:
    self._agent = agent
    self._stepper = stepper

WritingInteraction(label, *, target, words_per_step=100)

Bases: Interaction

Writing of the Simon model, a number of words per step.

Every next word is new with probability innovation, and otherwise repeats a random earlier occurrence, so a word is repeated with probability proportional to its count.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group of words.	required
words_per_step	int	The number of words written per step.	100

Examples:

>>> from skcomplex.models.social.spreading import WritingInteraction
>>> WritingInteraction('writing', target='words').words_per_step
100

Source code in src/skcomplex/models/social/spreading/_writing.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    words_per_step: int = 100,
) -> None:
    super().__init__(label)
    self.target = target
    self.words_per_step = words_per_step

write(model)

Write the words of one step.

Parameters:

Name	Type	Description	Default
model	ModelView	The model.	required

Source code in src/skcomplex/models/social/spreading/_writing.py

def write(self, model: ModelView) -> None:
    """Write the words of one step.

    Args:
        model:
            The model.
    """
    words = model.agents(self.target, GrowingWordView)
    if not words:
        return
    innovation = words[0].innovation
    counts = [int(word.count) for word in words]
    tokens = np.repeat(np.arange(len(counts)), counts).tolist()
    for _ in range(self.words_per_step):
        index = len(counts) if model.rng.random() < innovation else tokens[int(model.rng.integers(len(tokens)))]
        if index == len(counts):
            counts.append(0)
        counts[index] += 1
        tokens.append(index)
    for word, count in zip(words, counts, strict=False):
        word.count = float(count)
    for count in counts[len(words) :]:
        model.create(self.target, count=float(count))

hypergraph_exposure_sis(n_nodes=1000, *, sizes=(10, 20, 30), hyperdegree=5.0, rate=0.06, exponent=2.0, recovery=0.05, share=0.3, seed=0)

Return the SIS contagion of St-Onge et al. in the environments of a hypergraph, with a nonlinear kernel.

Participants meet in environments, the hyperedges of a random hypergraph. In every step an environment gives a susceptible member the dose rate * rho ** exponent from the share rho of its other members that are infected, the member is infected with probability 1 - exp(-D) of the sum D of its doses, and an infected participant recovers with probability recovery. Heterogeneous exposure times give the power law rho ** exponent to the infection kernel. With a sublinear or linear kernel the epidemic sets in continuously, and with a supralinear kernel, exponent > 1, the healthy state stays stable: an endemic state coexists with it above a critical rate, so a small seed dies out and a large one persists. In mean field the prevalence I is endemic where k rate I ** (exponent - 1) (1 - I) = recovery for k environments per participant.

Parameters:

Name	Type	Description	Default
n_nodes	int	The number of participants.	1000
sizes	Sequence[int]	The sizes of the environments.	(10, 20, 30)
hyperdegree	float	The mean number of environments of a participant, split evenly over the sizes, whose members are drawn at random.	5.0
rate	float	The dose of an environment whose other members are all infected.	0.06
exponent	float	The exponent of the infection kernel.	2.0
recovery	float	The probability of recovery in a step.	0.05
share	float	The share of infected participants at the start.	0.3
seed	int	The seed of the hypergraph and the infected participants.	0

Returns:

Name	Type	Description
system	System	The model, with the states participants.infected and the prevalence prevalence.

Examples:

>>> from skcomplex.models.social.spreading import hypergraph_exposure_sis
>>> hypergraph_exposure_sis(40).elements[0].check().prepare().size_
40

Source code in src/skcomplex/models/social/spreading/_exposure.py

@register_model(
    field='social/spreading',
    tags=('contagion', 'hypergraph', 'exposure', 'bistability'),
    references=('St-Onge et al. 2021', 'Majhi et al. 2022'),
    defaults={'until': 200, 'dt': 1},
    observables=lambda system: [ObservableSpec('prevalence', group='participants', variable='infected')],
)
def hypergraph_exposure_sis(
    n_nodes: int = 1000,
    *,
    sizes: Sequence[int] = (10, 20, 30),
    hyperdegree: float = 5.0,
    rate: float = 0.06,
    exponent: float = 2.0,
    recovery: float = 0.05,
    share: float = 0.3,
    seed: int = 0,
) -> System:
    """Return the SIS contagion of St-Onge et al. in the environments of a hypergraph, with a nonlinear kernel.

    Participants meet in environments, the hyperedges of a random hypergraph. In every step an environment gives a
    susceptible member the dose `rate * rho ** exponent` from the share `rho` of its other members that are infected,
    the member is infected with probability `1 - exp(-D)` of the sum `D` of its doses, and an infected participant
    recovers with probability `recovery`. Heterogeneous exposure times give the power law `rho ** exponent` to the
    infection kernel. With a sublinear or linear kernel the epidemic sets in continuously, and with a supralinear
    kernel, `exponent > 1`, the healthy state stays stable: an endemic state coexists with it above a critical rate,
    so a small seed dies out and a large one persists. In mean field the prevalence `I` is endemic where `k rate
    I ** (exponent - 1) (1 - I) = recovery` for `k` environments per participant.

    Args:
        n_nodes:
            The number of participants.

        sizes:
            The sizes of the environments.

        hyperdegree:
            The mean number of environments of a participant, split evenly over the sizes, whose members are drawn at
            random.

        rate:
            The dose of an environment whose other members are all infected.

        exponent:
            The exponent of the infection kernel.

        recovery:
            The probability of recovery in a step.

        share:
            The share of infected participants at the start.

        seed:
            The seed of the hypergraph and the infected participants.

    Returns:
        system:
            The model, with the states `participants.infected` and the prevalence `prevalence`.

    Examples:
        >>> from skcomplex.models.social.spreading import hypergraph_exposure_sis
        >>> hypergraph_exposure_sis(40).elements[0].check().prepare().size_
        40
    """
    rng = np.random.default_rng(seed)
    hypergraph = xgi.Hypergraph()
    hypergraph.add_nodes_from(range(n_nodes))
    for size in sizes:
        number = round(n_nodes * hyperdegree / (len(sizes) * size))
        hypergraph.add_edges_from(rng.choice(n_nodes, size, replace=False).tolist() for _ in range(number))
    infected = (rng.permutation(n_nodes) < round(share * n_nodes)).astype(float)
    participants = LatticeElement(
        'participants',
        element=ParticipantElement,
        space=NetworkSpace(xgi.to_graph(hypergraph), hyperedges=hypergraph),
        values={'infected': infected, 'recovery': recovery},
        visual=VisualSpec('sites', {'states': ['infected'], 'low': 0.0, 'high': 1.0}),
    )
    exposure = ExposureInteraction('exposure', target='participants', rate=rate, exponent=exponent)
    return System(elements=[participants], interactions=[exposure])

maki_thompson(n_people=1000, *, spread=1.0, stifle=1.0)

Return the Maki-Thompson rumour model started by one spreader, which leaves about 20% of people ignorant.

Parameters:

Name	Type	Description	Default
n_people	int	The size of the population.	1000
spread	float	The contact rate that spreads the rumour.	1.0
stifle	float	The contact rate that stifles a spreader.	1.0

Returns:

Name	Type	Description
system	System	The model, with the counts ignorants.n, spreaders.n, and stiflers.n.

Examples:

>>> from skcomplex.models.social.spreading import maki_thompson
>>> [element.n for element in maki_thompson(100).elements]
[99, 1, 0]

Source code in src/skcomplex/models/social/spreading/_rumours.py

@register_model(
    field='social/spreading',
    tags=('rumours', 'jump process', 'stochastic'),
    references=('Maki and Thompson 1973', 'Pastor-Satorras et al. 2015'),
    defaults={'until': 20.0, 'dt': 0.5},
)
def maki_thompson(n_people: int = 1000, *, spread: float = 1.0, stifle: float = 1.0) -> System:
    """Return the Maki-Thompson rumour model started by one spreader, which leaves about 20% of people ignorant.

    Args:
        n_people:
            The size of the population.

        spread:
            The contact rate that spreads the rumour.

        stifle:
            The contact rate that stifles a spreader.

    Returns:
        system:
            The model, with the counts `ignorants.n`, `spreaders.n`, and `stiflers.n`.

    Examples:
        >>> from skcomplex.models.social.spreading import maki_thompson
        >>> [element.n for element in maki_thompson(100).elements]
        [99, 1, 0]
    """
    return System(
        elements=[
            ListenersElement('ignorants', n=n_people - 1),
            ListenersElement('spreaders', n=1),
            ListenersElement('stiflers', n=0),
        ],
        interactions=[
            RumourContactsInteraction(
                'contacts',
                ignorants='ignorants',
                spreaders='spreaders',
                stiflers='stiflers',
                spread=spread,
                stifle=stifle,
            ),
        ],
    )

simon_yule(*, innovation=0.1, words_per_step=100)

Return the Simon model of a text that grows from one word, whose word counts follow a power law.

The rank-frequency exponent is 1 / (1 - innovation), and the number of distinct words grows as innovation times the length of the text.

Parameters:

Name	Type	Description	Default
innovation	float	The probability that the next word is new.	0.1
words_per_step	int	The number of words written per step.	100

Returns:

Name	Type	Description
system	System	The model, with the group words.

Examples:

>>> from skcomplex.models.social.spreading import simon_yule
>>> simon_yule().elements[0].check().prepare().size_
1

Source code in src/skcomplex/models/social/spreading/_growth.py

@register_model(
    field='social/spreading',
    tags=('agents', 'growth', 'power law', 'language'),
    references=('Simon 1955', 'Yule 1925', 'Thurner, Hanel, and Klimek 2018'),
    defaults={'until': 20},
)
def simon_yule(*, innovation: float = 0.1, words_per_step: int = 100) -> System:
    """Return the Simon model of a text that grows from one word, whose word counts follow a power law.

    The rank-frequency exponent is `1 / (1 - innovation)`, and the number of distinct words grows as `innovation`
    times the length of the text.

    Args:
        innovation:
            The probability that the next word is new.

        words_per_step:
            The number of words written per step.

    Returns:
        system:
            The model, with the group `words`.

    Examples:
        >>> from skcomplex.models.social.spreading import simon_yule
        >>> simon_yule().elements[0].check().prepare().size_
        1
    """
    words = GroupElement(
        'words',
        element=GrowingWordElement,
        size=1,
        values={'count': 1.0, 'innovation': innovation},
        variable=True,
    )
    return System(
        elements=[words],
        interactions=[WritingInteraction('writing', target='words', words_per_step=words_per_step)],
    )

simplicial_contagion(n_nodes=200, *, infectivity=0.75, triangle_infectivity=3.0, mean_degree=20.0, mean_triangles=6.0, recovery=1.0, share=0.4, seed=0)

Return the simplicial contagion of Iacopini et al., an SIS process on the links and triangles of a complex.

A random simplicial complex joins every pair of nodes with probability p1 and fills every triple with probability p2, chosen so that a node has mean_degree links and mean_triangles triangles on average. A susceptible node is infected at rate beta = infectivity * recovery / <k> by every infected neighbour, and at rate beta_triangle = triangle_infectivity * recovery / <k_triangle> by every triangle whose two other members are infected, and an infected node recovers at rate recovery. The mean-field prevalence follows rho' = -rho + infectivity rho (1 - rho) + triangle_infectivity rho^2 (1 - rho) in units of the recovery time. For triangle_infectivity > 1 and infectivities below the epidemic threshold 1, an endemic state coexists with the healthy one, so the epidemic dies out from a small seed and persists from a large one.

Parameters:

Name	Type	Description	Default
n_nodes	int	The number of nodes.	200
infectivity	float	The rescaled rate of infection through links.	0.75
triangle_infectivity	float	The rescaled rate of infection through triangles.	3.0
mean_degree	float	The mean number of links of a node, which sets p1 together with the sides of the triangles.	20.0
mean_triangles	float	The mean number of triangles of a node, which sets p2.	6.0
recovery	float	The rate of recovery.	1.0
share	float	The share of infected nodes at the start.	0.4
seed	int	The seed of the complex and the infected nodes.	0

Returns:

Name	Type	Description
system	System	The model, with the states nodes.infected and the prevalence prevalence.

Examples:

>>> from skcomplex.models.social.spreading import simplicial_contagion
>>> system = simplicial_contagion(30, mean_degree=6.0, mean_triangles=2.0)
>>> system.elements[0].check().prepare().size_
30

Source code in src/skcomplex/models/social/spreading/_simplicial.py

@register_model(
    field='social/spreading',
    tags=('contagion', 'network', 'hypergraph', 'simplicial complex', 'bistability'),
    references=('Iacopini et al. 2019', 'Majhi et al. 2022'),
    defaults={'until': 20.0, 'dt': 0.5},
    observables=lambda system: [ObservableSpec('prevalence', group='nodes', variable='infected')],
)
def simplicial_contagion(
    n_nodes: int = 200,
    *,
    infectivity: float = 0.75,
    triangle_infectivity: float = 3.0,
    mean_degree: float = 20.0,
    mean_triangles: float = 6.0,
    recovery: float = 1.0,
    share: float = 0.4,
    seed: int = 0,
) -> System:
    """Return the simplicial contagion of Iacopini et al., an SIS process on the links and triangles of a complex.

    A random simplicial complex joins every pair of nodes with probability `p1` and fills every triple with probability
    `p2`, chosen so that a node has `mean_degree` links and `mean_triangles` triangles on average. A susceptible node is
    infected at rate `beta = infectivity * recovery / <k>` by every infected neighbour, and at rate `beta_triangle =
    triangle_infectivity * recovery / <k_triangle>` by every triangle whose two other members are infected, and an
    infected node recovers at rate `recovery`. The mean-field prevalence follows `rho' = -rho + infectivity rho (1 -
    rho) + triangle_infectivity rho^2 (1 - rho)` in units of the recovery time. For `triangle_infectivity > 1` and
    infectivities below the epidemic threshold 1, an endemic state coexists with the healthy one, so the epidemic
    dies out from a small seed and persists from a large one.

    Args:
        n_nodes:
            The number of nodes.

        infectivity:
            The rescaled rate of infection through links.

        triangle_infectivity:
            The rescaled rate of infection through triangles.

        mean_degree:
            The mean number of links of a node, which sets `p1` together with the sides of the triangles.

        mean_triangles:
            The mean number of triangles of a node, which sets `p2`.

        recovery:
            The rate of recovery.

        share:
            The share of infected nodes at the start.

        seed:
            The seed of the complex and the infected nodes.

    Returns:
        system:
            The model, with the states `nodes.infected` and the prevalence `prevalence`.

    Examples:
        >>> from skcomplex.models.social.spreading import simplicial_contagion
        >>> system = simplicial_contagion(30, mean_degree=6.0, mean_triangles=2.0)
        >>> system.elements[0].check().prepare().size_
        30
    """
    pairs = math.comb(n_nodes - 1, PAIR - 1)
    triples = math.comb(n_nodes - 1, TRIANGLE - 1)
    triangles = min(mean_triangles / triples, 1.0)
    links = min(max(mean_degree - 2 * mean_triangles, 0.0) / pairs, 1.0)
    graph, hyperedges = _simplicial_complex(n_nodes, links, triangles, seed)
    sizes = [len(members) for members in hyperedges]
    degree = max(PAIR * sizes.count(PAIR) / n_nodes, 1.0)
    shared = max(TRIANGLE * sizes.count(TRIANGLE) / n_nodes, 1.0)
    rng = np.random.default_rng(seed + 1)
    infected = (rng.permutation(n_nodes) < round(share * n_nodes)).astype(float)
    nodes = LatticeElement(
        'nodes',
        element=InfectiveElement,
        space=NetworkSpace(graph, hyperedges=hyperedges),
        values={'infected': infected, 'recovery': recovery},
        visual=VisualSpec('sites', {'states': ['infected'], 'low': 0.0, 'high': 1.0}),
    )
    infection = SimplicialInfectionInteraction(
        'infection',
        target='nodes',
        beta=infectivity * recovery / degree,
        beta_triangle=triangle_infectivity * recovery / shared,
    )
    return System(elements=[nodes], interactions=[infection])

watts_cascades(graph=None, *, threshold=0.18, seeds=1, random_state=0)

Return the Watts threshold model of global cascades started by a few random seeds.

Parameters:

Name	Type	Description	Default
graph	object	The NetworkX graph, or None for a random graph of 200 nodes with mean degree 4.	None
threshold	float	The share of active neighbours at which a node becomes active.	0.18
seeds	int	The number of nodes active at the start.	1
random_state	int | Generator | None	The seed or generator of the default graph and of the seeds.	0

Returns:

Name	Type	Description
system	System	The model, with the group nodes.

Examples:

>>> import networkx as nx
>>> from skcomplex.models.social.spreading import watts_cascades
>>> watts_cascades(nx.cycle_graph(10), random_state=0).elements[0].check().prepare().size_
10

Source code in src/skcomplex/models/social/spreading/_cascades.py

@register_model(
    field='social/spreading',
    tags=('cascades', 'agents', 'network', 'threshold'),
    references=('Watts 2002', 'Pastor-Satorras et al. 2015'),
    defaults={'until': 30},
)
def watts_cascades(
    graph: object = None,
    *,
    threshold: float = 0.18,
    seeds: int = 1,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the Watts threshold model of global cascades started by a few random seeds.

    Args:
        graph:
            The NetworkX graph, or `None` for a random graph of 200 nodes with mean degree 4.

        threshold:
            The share of active neighbours at which a node becomes active.

        seeds:
            The number of nodes active at the start.

        random_state:
            The seed or generator of the default graph and of the seeds.

    Returns:
        system:
            The model, with the group `nodes`.

    Examples:
        >>> import networkx as nx
        >>> from skcomplex.models.social.spreading import watts_cascades
        >>> watts_cascades(nx.cycle_graph(10), random_state=0).elements[0].check().prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    graph = nx.gnp_random_graph(200, 4 / 199, seed=int(rng.integers(2**31))) if graph is None else graph
    size = nx.number_of_nodes(graph)
    active = np.zeros(size)
    active[rng.choice(size, seeds, replace=False)] = 1.0
    nodes = GroupElement(
        'nodes',
        element=ThresholdNodeElement,
        size=size,
        values={'active': active, 'threshold': threshold},
    )
    return System(
        elements=[nodes],
        interactions=[ThresholdActivationInteraction('activation', target='nodes')],
        space=NetworkSpace(graph),
    )