Sample-space-reducing process

Sample-space-reducing processes, whose state visits follow Zipf's law 1 / i for lam = 1.

• Name: sample_space_reducing
• Field: mathematics/stochastic_processes
• Formalism: agent-based model
• References: Corominas-Murtra, Hanel, and Thurner 2015; Thurner et al. 2018

Factory

@register_model(
    field='mathematics/stochastic_processes',
    tags=('sample space reduction', 'agents', 'zipf', 'power law'),
    references=('Corominas-Murtra, Hanel, and Thurner 2015', 'Thurner et al. 2018'),
    defaults={'until': 100},
)
def sample_space_reducing(n_balls: int = 20, *, n_states: int = 10, lam: float = 1.0) -> System:
    """Return sample-space-reducing processes, whose state visits follow Zipf's law `1 / i` for `lam = 1`.

    Args:
        n_balls:
            The number of independent balls, all started at the top state.

        n_states:
            The number of states.

        lam:
            The probability of a reducing jump, the exponent of the visiting distribution.

    Returns:
        system:
            The model, with the group `balls` of state `level`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import sample_space_reducing
        >>> sample_space_reducing(4).elements[0].check().prepare().size_
        4
    """
    return System(
        elements=[
            GroupElement(
                'balls',
                element=ReducingBallElement,
                size=n_balls,
                values={'level': float(n_states), 'n_states': float(n_states), 'lam': lam},
                variable=True,
            ),
        ],
        interactions=[ReducingJumpInteraction('jump', target='balls')],
    )

Components

class ReducingBallElement(Element):
    """Ball of the sample-space-reducing process, whose state the reducing jump moves to a random lower state.

    With probability `lam` the jump takes the ball uniformly to a state below its own, and otherwise uniformly to any
    state, and the ball restarts from the top state after it reaches state 1.

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

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

        level:
            The initial state, from 1 to `n_states`.

        n_states:
            The number of states.

        lam:
            The probability of a reducing jump.

    Examples:
        >>> from skcomplex.core.elements import GroupElement
        >>> from skcomplex.models.mathematics.stochastic_processes import ReducingBallElement
        >>> group = GroupElement('balls', element=ReducingBallElement, size=3, values={'n_states': 10.0})
        >>> group.label, group.element.__name__, group.size
        ('balls', 'ReducingBallElement', 3)
    """

    def __init__(
        self,
        label: str,
        visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'level'}),
        *,
        level: Annotated[float, StateSpec(domain=NN)] = 10.0,
        n_states: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
        lam: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    ) -> None:
        super().__init__(label, visual=visual)
        self.level = level
        self.n_states = n_states
        self.lam = lam

    @classmethod
    def _get_view(cls) -> type[ReducingBallView]:
        """Return the view of the agents, which types their states and parameters."""
        return ReducingBallView

class ReducingJumpInteraction(MapInteraction):
    """Jump of the sample-space-reducing process, to a random lower state, to any state, or back to the top state.

    A ball at state 1 restarts from the top state `n_states`. Otherwise it jumps with probability `lam` uniformly to a
    state below its own, and else uniformly to any state, so the states are visited with probabilities proportional to
    `i^(-lam)`.

    Args:
        label:
            The label of the interaction.

        target:
            The path of the group of balls.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import ReducingJumpInteraction
        >>> ReducingJumpInteraction('jump', target='balls').check().state
        'level'
    """

    def __init__(self, label: str, *, target: Annotated[str, RoleSpec()]) -> None:
        super().__init__(label, target=target, state='level', function=self._get_level, form='agent')

    def _get_level(self, agent: ReducingBallView, model: ModelViewProtocol) -> float:
        """Return the next state of a ball."""
        top = round(agent.n_states)
        if agent.level <= 1:
            return float(top)
        if model.rng.random() < agent.lam:
            return float(model.rng.integers(1, round(agent.level)))
        return float(model.rng.integers(1, top + 1))