Coevolving voter
The coevolving voter system, which fragments into consensual components when rewiring is fast.
- Name:
coevolving_voter - Field:
social/opinion_dynamics - Formalism: discrete map
- References: Holme and Newman 2006; Vazquez, Eguíluz, and San Miguel 2008; Thurner et al. 2018
Select this system
from skcomplex.systems import create_simulation, create_system, describe
system = create_system('coevolving_voter')
entry = describe('coevolving_voter')
trajectory = create_simulation('coevolving_voter').prepare().run().trajectory_
Factory
@register_system(
field='social/opinion_dynamics',
tags=('opinion', 'voter', 'adaptive network', 'rewiring', 'fragmentation'),
references=('Holme and Newman 2006', 'Vazquez, Eguíluz, and San Miguel 2008', 'Thurner et al. 2018'),
defaults={'until': 200},
)
def coevolving_voter(
n_voters: int = 200,
*,
mean_degree: float = 4.0,
rewiring: float = 0.3,
activity: float = 0.5,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return the coevolving voter system, which fragments into consensual components when rewiring is fast.
Voters on an Erdős-Rényi graph copy discordant neighbours or rewire the discordant edges to like-minded voters.
Below a critical rewiring probability, near 0.38 for mean degree 4, the network stays connected and discordant
edges survive for long times, and above it the network splits into components of one opinion.
Args:
n_voters:
The number of voters.
mean_degree:
The mean degree of the initial random graph.
rewiring:
The probability that an active voter with a discordant neighbour rewires instead of copying.
activity:
The probability that a voter acts in a step.
random_state:
The seed or generator of the graph and of the uniform random opinions.
Returns:
system:
The system, with the groups `voters` and `ties` and the interaction `separation`.
Examples:
>>> from skcomplex.core.engine.runtime import Simulation
>>> from skcomplex.systems.social.opinion_dynamics import coevolving_voter
>>> trajectory = Simulation(coevolving_voter(30), random_state=0).prepare().run(until=5).trajectory_
>>> int(trajectory.network(-1).sources.size) == int(trajectory.network(0).sources.size)
True
"""
rng = np.random.default_rng(random_state)
space = NetworkSpace(nx.gnp_random_graph(n_voters, mean_degree / (n_voters - 1), seed=rng))
opinions = (rng.random(n_voters) < FIRST_OPINION_SHARE).astype(float)
copying = MapInteraction('copying', target='voters', state='opinion', function=_copy_discordant)
return _coevolving(CoevolvingVoterElement, space, opinions, copying, rewiring=rewiring, activity=activity)
Components
Elements:
CoevolvingVoterElement, fromElementElement
Interactions:
MapInteractionSeparationInteraction, fromTopologyInteraction
class CoevolvingVoterElement(Element):
"""Voter of a coevolving network, with a binary opinion, its degree, and its number of discordant neighbours.
Every step a voter is active with probability `activity` and picks a random neighbour. When the neighbour
disagrees, the voter copies its opinion with probability `1 - rewiring`, and otherwise the interaction
`SeparationInteraction` moves the edge to a random voter who agrees with it.
Args:
label:
The label of the voter or the group.
visual:
How the agent is drawn, or `None` for `default_visual_`.
opinion:
The opinion, 0 or 1.
degree:
The number of neighbours, read at the start of every step.
activity:
The probability that a voter acts in a step.
rewiring:
The probability that an active voter with a discordant neighbour rewires instead of copying.
Examples:
>>> from skcomplex.systems.social.opinion_dynamics import CoevolvingVoterElement
>>> CoevolvingVoterElement('voters', rewiring=0.3).rewiring
0.3
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = OPINION_VISUAL,
*,
opinion: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
degree: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
activity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
rewiring: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
) -> None:
super().__init__(label, visual=visual)
self.opinion = opinion
self.degree = degree
self.activity = activity
self.rewiring = rewiring
class Element(BaseElement):
"""Base class of entities that carry state, such as particles, populations, agents, or firms.
A subclass takes its states, its parameters, and `visual` as arguments of its own `__init__`, annotating each state
with a `StateSpec` and each parameter with a `ParameterSpec` inside `Annotated`, and returns its intrinsic dynamics
from `_get_laws`, which returns no laws by default. A subclass drawn in its own way gives `visual` a default in its
signature.
An element whose laws have the agent trait, such as rules, alone or in a group, is an agent: under discrete time it
runs on the agent stepper, whose rules receive every member as an agent with its states as attributes. Its states
then hold any value, not only numbers. A member is removed only when its states tagged with a quantity are settled
to zero, and its states and parameters cannot take the names of the agent view, such as `location` or `alive`, or
of the attributes Mesa reads, such as `position` or `system`.
A subclass that connects to other elements declares its ports in the class variable `ports`, a mapping from the name
of a port to its `ConnectorSpec`, and takes one algebraic state per variable of a port, named `<port>_<variable>`.
A flow is positive into the element.
Args:
label:
The label of the element, unique among the parts of its system.
visual:
How the element is drawn, or `None` for `default_visual_`.
Attributes:
default_visual_:
The visual of the element when it sets none.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.components.elements import Element
>>> from skcomplex.core.specs import StateSpec
>>> from skcomplex.core.specs import VisualSpec
>>> from sage.all__sagemath_symbolics import oo
>>> from sage.sets.real_set import RealSet
>>> class PopulationElement(Element):
... def __init__(
... self,
... label,
... n: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
... visual=VisualSpec('line'),
... ):
... super().__init__(label, visual=visual)
... self.n = n
>>> PopulationElement('rabbits', n=40)
PopulationElement(label='rabbits', n=40)
"""
ports: ClassVar[Mapping[str, ConnectorSpec]] = {}
@classmethod
def _get_view(cls) -> type | None:
"""Return the class of the views of the element's agents, or `None` for the engine's plain view.
An agent element names a subclass of the engine's `AgentView` that annotates its states and parameters, such as
`class WordView(AgentView): count: float`, so its rules read them with checked types.
"""
return None
def _get_laws(self, namespace: SimpleNamespace) -> Sequence[BasePart]:
"""Return the laws of the element for a namespace of symbols, none by default.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the element.
"""
return []
class MapInteraction(Interaction):
"""Map of a state of discrete time, whose next value is a function, or the state plus it.
In the mode `'replace'` the next value of the state is `f`, and in the mode `'increment'` it is the state plus `f`.
Under the `'synchronous'` schedule every member of a bound group updates at once, from the states at the start of
the step. Under the `'sequential'` schedule the members update one after the other, in their order, and under the
`'random-sequential'` schedule in a new random order every step, each member reading the values the members before
it set. Under the `'member'` schedule of the agent form, every member updates in its own activation, together with
the rules of its element and the other member interactions, in the order the system lists them.
In the `'symbolic'` form, `function(namespace)` returns a symbolic expression, or a vector of them for a vector
state, where `namespace.target` holds the states and parameters of the target, `namespace.sources` the namespaces
of the sources, and `namespace.t` the time, and may hold draws such as `normal('noise')`. In the `'agent'` form,
the target is an agent group and `function(agent, system)` returns the value as a number, or as an array of the
shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the system view.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state the map sets.
state:
The name of the state of the target.
function:
The function `f(namespace)` in the symbolic form, or `f(agent, system)` in the agent form.
sources:
The paths of the elements, or members of groups such as `cells[0]`, that the function reads.
mode:
`'replace'` or `'increment'`.
schedule:
`'synchronous'`, `'sequential'`, `'random-sequential'`, or `'member'` for the agent form.
form:
`'symbolic'` for update laws of the states of the target, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, the
function is not callable, or the agent form returns a value that does not match the state, raised by
`run` of the simulation.
Examples:
>>> from skcomplex.core.components.elements import Element
>>> from skcomplex.core.components.interactions import MapInteraction
>>> from skcomplex.core.components.spaces import DiscreteTime
>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.runtime import Simulation
>>> from skcomplex.core.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
... super().__init__(label)
... self.x = x
>>> logistic = MapInteraction(
... 'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.components.elements import GroupElement
>>> class WeightsElement(Element):
... def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
... super().__init__(label, visual=visual)
... self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
... 'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (RuleLaw, UpdateLaw)
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec(elements=(ElementKind.ANY,), states=('state',))],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True, elements=(ElementKind.ANY,))] = (),
mode: str = 'replace',
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.function = function
self.sources = sources
self.mode = mode
self.schedule = schedule
self.form = form
self.phase = phase
def _check(self) -> None:
"""Raise an error for an unknown mode, schedule, or form, or a function that cannot be called.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, or the
function is not callable.
"""
super()._check()
self._check_variant('mode', self.mode, MODES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
):
if value not in variants:
error_msg = f'The {name} of map {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
if self.schedule == 'member' and self.form != 'agent':
error_msg = (
f'The map {self.label!r} updates every member in its activation, which only the agent form does.'
)
raise ValidationError(error_msg)
if not callable(self.function):
error_msg = f'The function of map {self.label!r} must be callable, but it is {self.function!r}.'
raise ValidationError(error_msg)
def _apply_mode(self, current: Value, value: Value) -> Value:
"""Return the next value of a state from its current value and the value of the function, by the mode."""
if self.mode == 'increment':
return current + value
if self.mode == 'replace':
return value
raise self._get_variant_error(self.mode, 'mode', '_apply_mode')
def _get_vector(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> Value:
"""Return the next value of a vector state of an agent."""
current = np.asarray(getattr(agent, self.state), dtype=np.float64)
value = np.asarray(self.function(agent, system), dtype=np.float64)
if value.shape != current.shape:
error_msg = (
f'The function of map {self.label!r} returns the shape {value.shape} for the state {self.state!r} of '
f'the shape {current.shape}. Return an array of the shape of the state.'
)
raise ValidationError(error_msg)
return self._apply_mode(current, value)
def _get_value(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> Value:
"""Return the next value of the state of an agent, a number or the array of a vector state."""
if np.ndim(getattr(agent, self.state)) > 0:
return self._get_vector(agent, system)
found = self.function(agent, system)
if not isinstance(found, SupportsFloat):
error_msg = f'The function of map {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
value = float(found)
return self._apply_mode(float(getattr(agent, self.state)), value)
def _set_value(self, agent: AgentViewProtocol, value: Value) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _update_member(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, system))
def update(self, system: SystemViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
system:
The system view of the interaction.
"""
self._update_agents(system, self.schedule, self._get_value, self._set_value)
def _get_order(self) -> object:
"""Return the order of the members of the symbolic form: none, their own order, or a uniform draw."""
if self.schedule == 'synchronous':
return None
if self.schedule == 'sequential':
return SR(0)
return uniform(re.sub(r'\W', '_', f'order_{self.label}'))
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its next value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.sources`.
Returns:
laws:
The update law of the state, or the rule that runs the map of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._update_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.update, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
mapped = cast(Value, self.function(namespace))
next_value = self._apply_mode(value, mapped)
return [UpdateLaw(value, next_value, increment=self.mode == 'increment', order=self._get_order())]
class SeparationInteraction(TopologyInteraction):
"""Rewiring of discordant edges by one end, which keeps the edge and moves its other end to a like-minded node.
It is a `TopologyInteraction` that rewires the edges of the voters. The end `keep` acts on an edge with
probability `activity / degree` per step, and when it holds the opinion `opinion` and the other end does not, it
rewires the edge with probability `rewiring`, drawing a new end among the voters of the opinion `opinion`.
Args:
label:
The label of the interaction.
voters:
The path of the voters on the nodes.
keep:
The end of an edge that acts and keeps the edge, `'source'` or `'target'`.
opinion:
The opinion of the end that acts, 0 or 1.
Examples:
>>> from skcomplex.systems.social.opinion_dynamics import SeparationInteraction
>>> SeparationInteraction('separation', voters='voters', keep='target', opinion=0).action
'rewire'
"""
def __init__(
self,
label: str,
*,
voters: Annotated[str, RoleSpec()],
keep: str = 'source',
opinion: float = 1.0,
) -> None:
super().__init__(
label,
target=voters,
action='rewire',
probability=self._get_chance,
toward=self._get_weight,
keep=keep,
)
self.voters = voters
self.keep = keep
self.opinion = opinion
def _get_weight(self, node: SimpleNamespace) -> object:
"""Return the weight of a candidate node as a new end, 1 when it holds the opinion of the acting end."""
return node.opinion if self.opinion == 1 else 1 - node.opinion
def _get_chance(self, namespace: SimpleNamespace) -> object:
"""Return the probability that the acting end rewires the edge, when the ends hold different opinions."""
actor, other = (namespace.source, namespace.target)[:: 1 if self.keep == 'source' else -1]
chance = actor.activity * actor.rewiring / (actor.degree + SMALLEST_DEGREE)
if self.opinion == 1:
return chance * actor.opinion * (1 - other.opinion)
return chance * (1 - actor.opinion) * other.opinion
Equations
voters.opinion[i]: opinion_voters(t + 1) == -(active_voters_topology - 1)*opinion_voters(t) + active_voters_topology*where(uniform_u_copying < -_neighbours_1*activity_voters*(rewiring_voters - 1)/(_neighbours_0 + 1e-09), -opinion_voters(t) + 1, opinion_voters(t))
voters.degree[i]: degree_voters(t + 1) == _neighbours_0*active_voters_topology - (active_voters_topology - 1)*degree_voters(t)