Skip to content

Public goods game

The spatial public goods game in groups of five neighbours, with Fermi imitation.

  • Name: public_goods_game
  • Field: social/cooperation
  • Formalism: discrete map
  • References: Szabó and Hauert 2002; Szabó and Fáth 2007

Factory

@register_model(
    field='social/cooperation',
    tags=('lattice', 'games', 'cooperation', 'public goods'),
    references=('Szabó and Hauert 2002', 'Szabó and Fáth 2007'),
    defaults={'until': 50, 'dt': 1},
)
def public_goods_game(
    shape: Sequence[int] = (50, 50),
    *,
    synergy: float = 4.5,
    noise: float = 0.5,
    share: float = 0.5,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the spatial public goods game in groups of five neighbours, with Fermi imitation.

    Every step has two phases. In the phase `play` every player takes the payoff of its five groups, and in the phase
    `imitate` every player copies a random neighbour with the Fermi probability `1 / (1 + exp((P_own - P_other) / K))`
    of the payoffs of the same step, at the noise `K`.

    Args:
        shape:
            The number of rows and columns of the periodic lattice.

        synergy:
            The factor that multiplies the contributions of a group.

        noise:
            The noise of the Fermi imitation.

        share:
            The share of cooperators at the start.

        random_state:
            The seed or generator of the initial strategies.

    Returns:
        system:
            The model, with the strategies `players.strategy` and the payoffs `players.payoff`.

    Examples:
        >>> from skcomplex.models.social.cooperation import public_goods_game
        >>> public_goods_game((10, 10)).elements[0].check().prepare().size_
        100
    """
    strategies = _random_strategies(shape, share, random_state)
    players = LatticeElement(
        'players',
        element=ContributorElement,
        space=LatticeSpace(tuple(shape), neighborhood='von_neumann'),
        values={'strategy': strategies, 'synergy': synergy, 'noise': noise},
    )
    imitation = ImitationInteraction(
        'imitation',
        target='players',
        state='strategy',
        rule='fermi',
        form='probability',
        intensity=1 / noise,
        phase='imitate',
    )
    time = DiscreteTime(schedule=PhasesScheduleSpec((PhaseSpec('play'), PhaseSpec('imitate'))))
    return System(
        elements=[players],
        interactions=[PublicGoodsInteraction('play', target='players', phase='play'), imitation],
        time=time,
    )

Components

class ContributorElement(Element):
    """Player of the spatial public goods game, a cooperator at 1 and a defector at 0, with its payoff.

    The play is a `PublicGoodsInteraction` of the lattice or a `HyperedgeGoodsInteraction` of a hypergraph, and the
    imitation an `ImitationInteraction` that reads the payoffs of the same step, which `public_goods_game` and
    `hypergraph_public_goods` add.

    Args:
        label:
            The label of the player or the lattice group.

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

        strategy:
            The initial strategy, 1 or 0.

        payoff:
            The payoff of the last play, the sum of the shares of its groups minus its contributions.

        synergy:
            The factor that multiplies the contributions of a group.

        noise:
            The noise of the Fermi imitation.

    Examples:
        >>> from skcomplex.models.social.cooperation import ContributorElement
        >>> ContributorElement('player').synergy
        4.5
    """

    def __init__(
        self,
        label: str,
        visual: VisualSpec | None = None,
        *,
        strategy: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 1.0,
        payoff: Annotated[float, StateSpec()] = 0.0,
        synergy: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 4.5,
        noise: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
    ) -> None:
        super().__init__(label, visual=visual)
        self.strategy = strategy
        self.payoff = payoff
        self.synergy = synergy
        self.noise = noise

class PublicGoodsInteraction(GroupingInteraction):
    """Play of the public goods game in the groups of a player and its four neighbours, which sets the payoffs.

    It is a symbolic `GroupingInteraction` of the von Neumann neighbourhoods of a lattice, a cooperator at 1 and a
    defector at 0. Every player joins five groups, the one centred on itself and those centred on its four neighbours.
    In each group a cooperator pays 1, and the pot multiplied by `synergy` is shared by the members, so the payoff of a
    player is the sum of its shares minus its contributions. The synergy is a parameter of the players.

    Args:
        label:
            The label of the interaction.

        target:
            The path of the players on the lattice.

        phase:
            The phase of the step the play runs in, or `None` for the default phase.

    Examples:
        >>> from skcomplex.models.social.cooperation import PublicGoodsInteraction
        >>> PublicGoodsInteraction('play', target='players').check().write
        'payoff'
    """

    def __init__(self, label: str, *, target: Annotated[str, RoleSpec()], phase: str | None = None) -> None:
        super().__init__(
            label,
            target=target,
            groups='neighborhood',
            form='symbolic',
            write='payoff',
            function=self._get_share,
            phase=phase,
        )

    def _get_share(self, group: SimpleNamespace) -> Expression:
        """Return the share of the pot of a group for its member, minus the contribution of the member."""
        pot = group.member.synergy * total(group.members.strategy, over=group.members)
        return pot / count(over=group.members) - group.member.strategy

class ImitationInteraction(NeighborInteraction):
    """Imitation of a discrete state of the neighbours of every member, at a rate or with a probability per step.

    By the rule `'random'` a member copies a neighbour chosen at random, by `'majority'` it takes the option that more
    of its neighbours hold than any other, and by `'best'` it copies the neighbour whose payoff is largest, when that
    payoff is larger than its own. By `'better'` it picks a neighbour at random and copies it when the neighbour's
    payoff is larger than its own. By `'fermi'` it picks a neighbour at random and copies it with probability
    `1 / (1 + exp(-beta (P_j - P_i)))`, where `P_j` and `P_i` are the payoffs of the neighbour and of the member and
    `beta` is the intensity. A member without such a neighbour keeps its state. The rules read the neighbourhood of
    the space, the graph of a network or the neighbourhood of a lattice.

    In the `'rate'` form every member imitates at the rate `rate`, by jumps between every pair of options in continuous
    time. In the `'probability'` form every member imitates with probability `probability` in every step of a
    discrete-time system, all members at once in its `phase`, reading the states the phases before it set. In the
    `'agent'` form the target is an agent group, and every agent imitates its neighbours in the space with probability
    `probability` in its `phase`. The agents imitate at once, reading the states of the step before, under the schedule
    `'synchronous'`, one after the other in their order or in a new random order every step under `'sequential'` or
    `'random-sequential'`, each reading the states the agents before it set, or each in its own activation under
    `'member'`. The rule `'better'` runs in the agent form only. Among neighbours of equal largest payoff, the rule
    `'best'` of the agent form copies the first under the ties `'first'` and one drawn uniformly under `'random'`.

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

        target:
            The path of the group on the nodes or sites.

        state:
            The name of the discrete state of the members.

        payoff:
            The name of the state that holds the payoff of the members, which the rules `'best'` and `'fermi'` read.

        options:
            The values the state takes.

        rate:
            The rate at which every member imitates, in the rate form.

        rule:
            `'random'`, `'majority'`, `'best'`, or `'fermi'`.

        intensity:
            The intensity `beta` of the Fermi rule.

        form:
            `'rate'` for imitation in continuous time, `'probability'` for imitation in discrete time, or `'agent'`
            for imitation of the neighbours of the agents of an agent group.

        probability:
            The probability that a member imitates in a step, in the probability form.

        layer:
            The layer of the network whose edges the interaction runs on, or `None` for the graph of the network.

        phase:
            The phase of the step the agent and the probability forms run in, or `None` for the default phase.

        schedule:
            The order of the agents in the agent form, `'synchronous'`, `'sequential'`, `'random-sequential'`, or
            `'member'`.

        ties:
            The neighbour the rule `'best'` of the agent form copies among ties, `'first'` or `'random'`.

    Raises:
        ValidationError:
            The rule, the form, the schedule, or the ties are unknown, a schedule or ties other than the defaults are
            given outside the agent form, or the options are not distinct.

    Examples:
        >>> from skcomplex.core.interactions import ImitationInteraction
        >>> ImitationInteraction('imitation', target='voters', rule='majority').check().options
        (0.0, 1.0)
        >>> ImitationInteraction('imitation', target='players', rule='fermi', form='probability').check().form
        'probability'
        >>> ImitationInteraction('imitation', target='traders', rule='better', form='agent', phase='learn').check().rule
        'better'
    """

    def __init__(
        self,
        label: str,
        *,
        target: Annotated[str, RoleSpec()],
        state: str = 'state',
        payoff: str = 'payoff',
        options: Sequence[float] = (0.0, 1.0),
        rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
        rule: str = 'random',
        intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
        form: str = 'rate',
        probability: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
        layer: str | None = None,
        phase: str | None = None,
        schedule: str = 'synchronous',
        ties: str = 'first',
    ) -> None:
        super().__init__(label, layer=layer)
        self.target = target
        self.state = state
        self.payoff = payoff
        self.options = options
        self.rate = rate
        self.rule = rule
        self.intensity = intensity
        self.form = form
        self.probability = probability
        self.phase = phase
        self.schedule = schedule
        self.ties = ties

    def _check_agent_options(self) -> None:
        """Raise an error for an unknown schedule or ties, or either of them not default outside the agent form."""
        self._check_variant('schedule', self.schedule, SCHEDULES)
        self._check_variant('ties', self.ties, TIES)
        if self.form != 'agent' and (self.schedule != 'synchronous' or self.ties != 'first'):
            error_msg = (
                f'The imitation {self.label!r} has the schedule {self.schedule!r} and the ties {self.ties!r}, which '
                'only the agent form runs. Use form="agent", or schedule="synchronous" and ties="first".'
            )
            raise ValidationError(error_msg)

    def _check(self) -> None:
        """Raise an error for an unknown rule, form, schedule, or ties, or options that are not distinct.

        Raises:
            ValidationError:
                The rule, the form, the schedule, or the ties are unknown, a schedule or ties other than the defaults
                are given outside the agent form, or the options are not distinct.
        """
        super()._check()
        self._check_variant('rule', self.rule, RULES, (BaseSpec,))
        if self.form not in FORMS:
            error_msg = f'The form of imitation is {self.form!r}. Use one of {", ".join(FORMS)}.'
            raise ValidationError(error_msg)
        if self.rule == 'better' and self.form != 'agent':
            error_msg = f'The rule of imitation {self.rule!r} runs in the agent form only. Use form `agent`.'
            raise ValidationError(error_msg)
        self._check_agent_options()
        if len(set(self.options)) < max(len(self.options), 2):
            error_msg = f'The options of imitation are {self.options!r}. Give at least two distinct values.'
            raise ValidationError(error_msg)

    def _get_neighborhood(self) -> AdjacencyNeighborhood | None:
        """Return the neighbourhood of the layer, or `None` for the neighbourhood of the space."""
        return None if self.layer is None else AdjacencyNeighborhood(layer=self.layer)

    def _get_fermi(self, member: SimpleNamespace, other: SimpleNamespace, intensity: Expression) -> Expression:
        """Return the Fermi probability that a member copies another of a different payoff."""
        return 1 / (1 + exp(-intensity * (getattr(other, self.payoff) - getattr(member, self.payoff))))

    def _get_counts(self, namespace: SimpleNamespace) -> dict[float, Expression]:
        """Return the number of the neighbours of the target that hold every option."""
        source, over = self._get_member(namespace, 'source'), self._get_neighborhood()
        return {
            option: neighbors(where(getattr(source, self.state) == option, 1, 0), over=over) for option in self.options
        }

    def _get_majority(self, namespace: SimpleNamespace) -> dict[float, Expression]:
        """Return the indicator of every option that more of the neighbours of the target hold than any other."""
        counts = self._get_counts(namespace)
        leads = {}
        for option, count in counts.items():
            lead = SR(1)
            for other, other_count in counts.items():
                if other != option:
                    lead = lead * where(count > other_count, 1, 0)
            leads[option] = lead
        return leads

    def _get_best(self, namespace: SimpleNamespace) -> tuple[Expression, Expression]:
        """Return the option of the neighbour of largest payoff and whether that payoff is larger than the target's."""
        source, over = self._get_member(namespace, 'source'), self._get_neighborhood()
        payoff = getattr(source, self.payoff)
        best = neighbors(getattr(source, self.state), over=over, reduce='argmax', weight=payoff)
        largest = neighbors(payoff, over=over, reduce='max')
        return best, where(largest > getattr(self._get_member(namespace, 'target'), self.payoff), 1, 0)

    def _switches(self, value: Expression, rate: Expression, chosen: Expression) -> list[BaseLaw]:
        """Return the jumps of a state from every option to the chosen option, at a rate when it is chosen."""
        return [
            JumpLaw(rate * where(value == first, 1, 0) * where(chosen == second, 1, 0), {value: second - first})
            for first, second in itertools.permutations(self.options, 2)
        ]

    def _copies(self, member: SimpleNamespace, other: SimpleNamespace, namespace: SimpleNamespace) -> list[BaseLaw]:
        """Return the copies of the state of the other end of an edge by a member, chosen among its neighbours."""
        value, rate = getattr(member, self.state), self._get_parameter_symbol('rate', namespace) / member.degree
        if self.rule == 'fermi':
            rate = rate * self._get_fermi(member, other, self._get_parameter_symbol('intensity', namespace))
        return self._switches(value, rate, getattr(other, self.state))

    def _get_jumps(self, namespace: SimpleNamespace) -> list[BaseLaw]:
        """Return the jumps of the state at the rate of imitation, in continuous time."""
        value = getattr(self._get_member(namespace, 'target'), self.state)
        if self.rule == 'majority':
            leads = self._get_majority(namespace)
            return [
                JumpLaw(
                    self._get_parameter_symbol('rate', namespace) * where(value == first, 1, 0) * leads[second],
                    {value: second - first},
                )
                for first, second in itertools.permutations(self.options, 2)
            ]
        if self.rule == 'best':
            best, better = self._get_best(namespace)
            return self._switches(value, self._get_parameter_symbol('rate', namespace) * better, best)
        if self.rule not in ('random', 'fermi'):
            raise self._get_variant_error(self.rule, 'rule', '_get_jumps')
        source, target = self._get_member(namespace, 'source'), self._get_member(namespace, 'target')
        return [*self._copies(target, source, namespace), *self._copies(source, target, namespace)]

    def _draw_copy(self, namespace: SimpleNamespace) -> Expression:
        """Return the next state of the target by the random or the Fermi rule, from the option it copies or its own."""
        source, target = self._get_member(namespace, 'source'), self._get_member(namespace, 'target')
        value = getattr(target, self.state)
        probability, over = self._get_parameter_symbol('probability', namespace), self._get_neighborhood()
        counts = self._get_counts(namespace)
        degree = sum(counts.values(), SR(0))
        if self.rule == 'fermi':
            fermi = self._get_fermi(target, source, self._get_parameter_symbol('intensity', namespace))
            weights = [
                probability * neighbors(where(getattr(source, self.state) == option, 1, 0) * fermi, over=over)
                for option in self.options
            ]
        else:
            weights = [probability * counts[option] for option in self.options]
        stay = degree - sum(weights, SR(0)) + where(degree == 0, 1, 0)
        index = categorical(f'{value}_imitation', [stay, *weights])
        return value + sum(
            (where(index == position + 1, option - value, 0) for position, option in enumerate(self.options)),
            SR(0),
        )

    def _get_update(self, namespace: SimpleNamespace) -> list[BaseLaw]:
        """Return the update of the state of the target with the probability of imitation, in discrete time."""
        value = getattr(self._get_member(namespace, 'target'), self.state)
        if self.rule in ('random', 'fermi'):
            return [UpdateLaw(value, self._draw_copy(namespace), phase=self.phase)]
        if self.rule == 'majority':
            leads = self._get_majority(namespace)
            imitated = value + sum((lead * (option - value) for option, lead in leads.items()), SR(0))
        elif self.rule != 'best':
            raise self._get_variant_error(self.rule, 'rule', '_get_update')
        else:
            best, better = self._get_best(namespace)
            imitated = where(better == 1, best, value)
        return [
            UpdateLaw(
                value,
                where(
                    uniform(f'{value}_imitation') < self._get_parameter_symbol('probability', namespace),
                    imitated,
                    value,
                ),
                phase=self.phase,
            ),
        ]

    def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
        """Return the jumps or the update of the state by the rule of the imitation.

        Args:
            namespace:
                The namespace of an edge.

        Returns:
            laws:
                The jumps of the state between every pair of options, or the update of the state of every member.
        """
        if self.form == 'agent' and self.schedule == 'member':
            return [RuleLaw(self._imitate_member, phase=self.phase, members=self._get_role('target'))]
        if self.form == 'agent':
            return [RuleLaw(self._imitate, phase=self.phase)]
        if self.form == 'probability':
            return self._get_update(namespace)
        return self._get_jumps(namespace)

    def _choose(
        self,
        agent: AgentViewProtocol,
        neighbours: Sequence[AgentViewProtocol],
        model: ModelViewProtocol,
    ) -> float | None:
        """Return the option an agent copies from its neighbours by the rule, or `None` to keep its own."""
        rng = model.rng
        if self.rule == 'majority':
            counts = [sum(getattr(other, self.state) == option for other in neighbours) for option in self.options]
            leaders = np.flatnonzero(np.asarray(counts) == max(counts))
            return float(self.options[int(leaders[0])]) if len(leaders) == 1 else None
        if self.rule == 'random':
            return float(getattr(neighbours[int(rng.integers(len(neighbours)))], self.state))
        own = float(getattr(agent, self.payoff))
        if self.rule == 'best':
            payoffs = np.array([getattr(other, self.payoff) for other in neighbours], dtype=float)
            best = self._pick_best(payoffs, self.ties, model)
            return float(getattr(neighbours[best], self.state)) if payoffs[best] > own else None
        other = neighbours[int(rng.integers(len(neighbours)))]
        difference = float(getattr(other, self.payoff)) - own
        if self.rule == 'better':
            return float(getattr(other, self.state)) if difference > 0 else None
        if self.rule != 'fermi':
            raise self._get_variant_error(self.rule, 'rule', '_choose')
        fermi = 1 / (1 + np.exp(-self._get_parameter_value('intensity', model) * difference))
        return float(getattr(other, self.state)) if rng.random() < fermi else None

    def _get_choice(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
        """Return the option an agent copies with the probability of imitation, or `None` to keep its own."""
        groups = set(model.groups(self._get_role('target')))
        neighbours = [other for other in agent.neighbors(layer=self.layer) if other.group in groups]
        if not neighbours or model.rng.random() >= self._get_parameter_value('probability', model):
            return None
        return self._choose(agent, neighbours, model)

    def _set_choice(self, agent: AgentViewProtocol, choice: float | None) -> None:
        """Set the state of an agent to the option it copies, if any."""
        if choice is not None:
            setattr(agent, self.state, choice)

    def _imitate_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
        """Let one agent imitate its neighbours with the probability of imitation, in its own activation."""
        if agent.alive:
            self._set_choice(agent, self._get_choice(agent, model))

    def _imitate(self, model: ModelViewProtocol) -> None:
        """Let every agent imitate its neighbours with the probability of imitation, by the schedule."""
        self._update_agents(model, self.schedule, self._get_choice, self._set_choice)

Equations

players.strategy[i]: strategy_players(t + 1) == strategy_players(t) + where(where((_neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0))*categorical_strategy_players__target_imitation_imitation >= -_neighbours_2*probability_imitation + _neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0), 1, 0) + where((_neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0))*categorical_strategy_players__target_imitation_imitation >= -_neighbours_2*probability_imitation - _neighbours_3*probability_imitation + _neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0), 1, 0) == 1, -strategy_players(t), 0) + where(where((_neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0))*categorical_strategy_players__target_imitation_imitation >= -_neighbours_2*probability_imitation + _neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0), 1, 0) + where((_neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0))*categorical_strategy_players__target_imitation_imitation >= -_neighbours_2*probability_imitation - _neighbours_3*probability_imitation + _neighbours_0 + _neighbours_1 + where(_neighbours_0 + _neighbours_1 == 0, 1, 0), 1, 0) == 2, -strategy_players(t) + 1.0, 0)
players.payoff[i]: payoff_players(t + 1) == 1/5*(strategy_players__member + strategy_players__other0 + strategy_players__other1 + strategy_players__other2 + strategy_players__other3)*synergy_players__member - strategy_players__member