Skip to content

Nonlinear three-body consensus on temporal hypergraphs

The nonlinear three-body consensus model on a random hypergraph of triangles that switch on and off.

  • Name: hypergraph_consensus
  • Field: social/opinion_dynamics
  • Formalism: ordinary differential equations
  • References: Neuhäuser, Mellor, and Lambiotte 2020; Majhi et al. 2022

Factory

@register_model(
    field='social/opinion_dynamics',
    tags=('opinion', 'consensus', 'hypergraph', 'temporal network'),
    references=('Neuhäuser, Mellor, and Lambiotte 2020', 'Majhi et al. 2022'),
    defaults={'until': 10.0, 'dt': 0.1},
    observables=lambda system: [
        ObservableSpec('mean_opinion', group='nodes', variable='x'),
        ObservableSpec('spread', _spread, group='nodes'),
    ],
)
def hypergraph_consensus(
    n_nodes: int = 30,
    *,
    triangles: float = 0.05,
    reinforcement: float = -1.0,
    period: float = 2.0,
    seed: int = 0,
) -> System:
    """Return the nonlinear three-body consensus model on a random hypergraph of triangles that switch on and off.

    Every triple of nodes is a triangle with probability `triangles`, and every triangle is active in alternate half
    periods from a random phase. On an active triangle `ijk`, node `i` follows `x_i' = s(|x_j - x_k|) (x_j + x_k - 2
    x_i)` with `s(d) = exp(reinforcement * d)`. Without reinforcement the pulls of a triangle on its three members sum
    to zero, so the mean opinion stays fixed whatever the schedules. With reinforcement the pull depends on the member
    it acts on, and the mean opinion drifts towards the pairs that agree, for a negative reinforcement, while the
    opinions still reach a consensus on a connected hypergraph.

    Args:
        n_nodes:
            The number of nodes.

        triangles:
            The probability that a triple of nodes is a triangle.

        reinforcement:
            The exponent of the modulating function `s`.

        period:
            The period of the schedules of the triangles.

        seed:
            The seed of the hypergraph, the phases, and the opinions.

    Returns:
        system:
            The model, with the opinions `nodes.x`, the mean opinion `mean_opinion`, and the mean squared opinion
            `spread`.

    Examples:
        >>> from skcomplex.models.social.opinion_dynamics import hypergraph_consensus
        >>> system = hypergraph_consensus(8, triangles=0.5)
        >>> system.elements[0].check().prepare().size_
        8
    """
    rng = np.random.default_rng(seed)
    triads = [triad for triad in itertools.combinations(range(n_nodes), TRIANGLE) if rng.random() < triangles]
    space = NetworkSpace(nx.complete_graph(n_nodes), hyperedges=triads)
    nodes = LatticeElement(
        'nodes',
        element=OpinionElement,
        space=space,
        values={'x': rng.uniform(-1.0, 1.0, n_nodes)},
        visual=VisualSpec('sites', {'states': ['x'], 'low': -1.0, 'high': 1.0}),
    )
    schedules = GroupElement(
        'triangles',
        element=ScheduleElement,
        space=space,
        on='hyperedges',
        values={'phase': rng.uniform(0.0, 2 * math.pi, len(triads))},
    )
    consensus = ThreeBodyConsensusInteraction(
        'consensus',
        target='nodes',
        triangles='triangles',
        reinforcement=reinforcement,
        period=period,
    )
    return System(elements=[nodes, schedules], interactions=[consensus])

Components

class OpinionElement(Element):
    """Node of a consensus process with a continuous opinion.

    Args:
        label:
            The label of the node or the group.

        visual:
            How the node is drawn, or `None` for `default_visual_`.

        x:
            The initial opinion.

    Examples:
        >>> from skcomplex.models.social.opinion_dynamics import OpinionElement
        >>> OpinionElement('node', x=0.5).x
        0.5
    """

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

class ScheduleElement(Element):
    """Triangle of a temporal hypergraph, active in every other half of a period from its own phase.

    Args:
        label:
            The label of the triangle or the group.

        visual:
            How the triangle is drawn, or `None` for `default_visual_`.

        phase:
            The phase of the schedule, in radians.

    Examples:
        >>> from skcomplex.models.social.opinion_dynamics import ScheduleElement
        >>> ScheduleElement('triangle', phase=1.0).phase
        1.0
    """

    def __init__(
        self,
        label: str,
        visual: VisualSpec | None = None,
        *,
        phase: Annotated[float, ParameterSpec()] = 0.0,
    ) -> None:
        super().__init__(label, visual=visual)
        self.phase = phase

class ThreeBodyConsensusInteraction(HyperedgeInteraction):
    """Nonlinear three-body consensus of Neuhäuser, Mellor, and Lambiotte, on the triangles active at the time.

    The focal node `i` of an active triangle `ijk` moves towards the two others at the rate `s(|x_j - x_k|) (x_j + x_k -
    2 x_i)`, with `s(d) = exp(reinforcement * d)`, so a pair that agrees pulls harder for a negative reinforcement. A
    triangle is active when `sin(2 pi t / period + phase) > 0`, with the phase of the triangle member of the group on
    hyperedges.

    Args:
        label:
            The label of the interaction, unique among the parts of its system.

        target:
            The path of the group of nodes.

        triangles:
            The path of the group of the schedules of the triangles.

        reinforcement:
            The exponent of the modulating function `s`.

        period:
            The period of the schedules.

    Examples:
        >>> from skcomplex.models.social.opinion_dynamics import ThreeBodyConsensusInteraction
        >>> ThreeBodyConsensusInteraction('consensus', target='nodes', triangles='triangles').check().size
        3
    """

    def __init__(
        self,
        label: str,
        *,
        target: Annotated[str, RoleSpec()],
        triangles: Annotated[str, RoleSpec()],
        reinforcement: Annotated[float, ParameterSpec()] = 0.0,
        period: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    ) -> None:
        super().__init__(label, size=TRIANGLE)
        self.target = target
        self.triangles = triangles
        self.reinforcement = reinforcement
        self.period = period

    def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
        """Return the pull of the two other members of an active triangle on the focal node."""
        x = s.member.x
        first, second = (other.x for other in s.others)
        active = where(sin(2 * pi * s.t / s.period + s.hyperedge.phase) > 0, 1, 0)
        pull = exp(s.reinforcement * abs(first - second)) * (first + second - 2 * x)
        return [RateLaw(x, active * pull)]

Equations

nodes.x[i]: diff(x_nodes(t), t) == -(2*x_nodes__member - x_nodes__other0 - x_nodes__other1)*e^(reinforcement_consensus*abs(x_nodes__other0 - x_nodes__other1))*where(sin(phase_triangles__hyperedge + 2*pi*t/period_consensus) > 0, 1, 0)