simulations.py

from collections import Counter
from math import inf

from numpy.random import choice
from numpy.random import random

EPS = .00005  # Gurobi gives very bad precision in some cases


class Graph:
    """Underlying graph of potential delegations generated by preferential attachment.

    Attributes:
        gamma (float): preferential attachment chooses a node with probability proportional to (node weight)^gamma
        outdegree (int/None): positive number of potential delegations for every non-voting nodes or None if unspecified
        d (float between 0 and 1): the probability of an incoming voter delegating
        potential_delegations (list of ((list of int) / None)): adjacency list representation of the graph
        observers (list of Mechanism): list of Observer objects to be notified of changes to the graph
        degrees (list of int): list of degrees of the nodes
        probability_weights (list of float): list of probability weights for choosing nodes:
                                             self.probability_weights[n] = self.degrees[n] ** self.gamma
    """

    def __init__(self, gamma, d, outdegree=None):
        """Initialize graph with single voter."""
        self.gamma = gamma
        assert outdegree is None or 1 <= outdegree
        self.outdegree = outdegree
        self.d = d
        self.potential_delegations = [None]
        self.observers = []
        self.degrees = [1]
        self.probability_weights = [1 ** self.gamma]

    def number_of_nodes(self):
        return len(self.potential_delegations)

    def is_voter(self, node):
        assert 0 <= node < self.number_of_nodes()
        return self.potential_delegations[node] is None

    def add_preferential_node(self, outdegree=None):
        """Add a new node to the graph, delegating to ``self.outdegree`` many nodes chosen via preferential attachment.

        Probability of attachment to a node ``n`` is proportional to ``self.probability_weights[n] = self.degrees[n] ** self.gamma``.

        Args:
            outdegree (int/None): Number of outgoing edges, overriding self.outdegree if not None. If self.outdegree is
                                  None, this parameter must be present and an integer.
        """

        if outdegree is None:
            if self.outdegree is None:
                raise ValueError("Outdegree must be specified and an integer if self.outdegree is None.")
            outdegree = self.outdegree
        else:
            assert outdegree >= 1

        pots = []

        for _ in range(outdegree):
            # with updated weights
            total_weight = sum(self.probability_weights)
            normalized_weights = [weight / total_weight for weight in self.probability_weights]

            endpoint = choice(self.number_of_nodes(), p=normalized_weights)
            assert 0 <= endpoint < self.number_of_nodes()

            pots.append(endpoint)
            self.degrees[endpoint] += 1
            self.probability_weights[endpoint] = self.degrees[endpoint] ** self.gamma

        self.potential_delegations.append(pots)
        self.degrees.append(1)
        self.probability_weights.append(1 ** self.gamma)

        for observer in self.observers:
            observer.notify_of_added_delegating_node(pots)

    def add_voter(self):
        """Add a voting vertex."""
        self.potential_delegations.append(None)
        self.degrees.append(1)
        self.probability_weights.append(1 ** self.gamma)
        for observer in self.observers:
            observer.notify_of_added_voting_node()

    def add_node(self, outdegree=None):
        if random() < self.d:
            self.add_preferential_node(outdegree)
        else:
            self.add_voter()

    def to_dot(self):
        s = "digraph {\n"
        s += "node [shape=circle color=\"#3F51B5\" style=filled label=\"\"];"  # delegators
        for i in range(self.number_of_nodes()):
            if not self.is_voter(i):
                s += f" {i};"
        s += "\nnode [shape=circle color=\"#F44336\" style=filled label=\"\"];"  # voters
        for i in range(self.number_of_nodes()):
            if self.is_voter(i):
                s += f" {i};"
        s += "\nedge [color=\"#000000\" arrowhead=open penwidth=1];\n"
        for i in range(self.number_of_nodes()):
            if not self.is_voter(i):
                for j in set(self.potential_delegations[i]):
                    s += f"{i} -> {j};\n"
        s += "}\n"
        return s


class Observer:
    """Class being notified of nodes being added to a graph."""
    def __init__(self, graph):
        """Register instance as an observer of ``graph``.

        Args:
            graph (Graph): The observed graph
        """
        self.graph = graph
        graph.observers.append(self)

    def notify_of_added_voting_node(self):
        """React to a voter being added to the graph."""
        pass

    def notify_of_added_delegating_node(self, potential_delegations):
        """React to a delegating node being added to the graph.

        Node that, since observers are not yet registered at initialization of ``self.graph``, they will not be notified
        of the initial voting node.
        """
        assert len(potential_delegations) > 0
        assert all(0 <= pot < self.graph.number_of_nodes() - 1 for pot in potential_delegations)


class ProtocollingObserver(Observer):
    def __init__(self, graph):
        super().__init__(graph)
        self.protocol = ["Initial voter"]

    def notify_of_added_voting_node(self):
        self.protocol.append("New voter")

    def notify_of_added_delegating_node(self, potential_delegations):
        self.protocol.append(potential_delegations)


class Mechanism(Observer):
    """Algorithm instance observing generation of a ``Graph`` instance and making delegation decisions.

    Class attributes:
        PLOT_COLOR (string): color for plotting instances of mechanism
        PLOT_ABBREVIATION (string): one-letter abbreviation, used for file names
        PLOT_LABEL (string): label to appear in legend
        PLOT_PATTERN (string): dash pattern for plotting

    Attributes:
        graph (Graph): The observed graph
    """

    PLOT_COLOR = None
    PLOT_ABBREVIATION = None
    PLOT_LABEL = None
    PLOT_PATTERN = None

    @staticmethod
    def is_splittable():
        """Say whether the class returns splittable or confluent delegations.

        This function must be overwritten by every inheriting class.

        Returns:
            bool:
            True for splittable delegations, i.e. get_delegations() returns list of (dict of int → float).
            False for confluent delegations, i.e. get_delegations() returns list of (None / int).
        """
        assert False

    @staticmethod
    def max_weight_from_delegations(delegations):
        assert False

    def get_delegations(self, time_out=None):
        """Compute the resolved delegations for the current state of the graph.

        Args:
            time_out (float): Time out in seconds. Currently only respected by ConfluentFlow.

        Returns:
            if is_splittable():
                list of ((dict of int → float) / None):
                For every node n, a dictionary mapping delegates to the weight of delegation, or None for voters.
            else:
                list of (int / None):
                For every node n, the index of the node it delegates to or None if it is a voter.
        """
        pass


class ConfluentMechanism(Mechanism):
    @staticmethod
    def _max_weight_from_confluent_delegations(delegations):
        """

        >>> ConfluentMechanism._max_weight_from_confluent_delegations([None, 0, None, 2, 3])
        3
        >>> ConfluentMechanism._max_weight_from_confluent_delegations([None, None, None])
        1

        Args:
             delegations (list of (None / int))
        """
        num_agents = len(delegations)

        transitive_delegations = []
        for node, delegate in enumerate(delegations):
            if delegate is None:
                transitive_delegations.append(node)
            else:
                transitive_delegations.append(delegate)

        def get_transitive_delegate(node):
            nonlocal transitive_delegations
            delegate = transitive_delegations[node]
            if transitive_delegations[delegate] == delegate:  # Delegate is voter
                return delegate
            transitive_delegations[node] = get_transitive_delegate(delegate)
            return transitive_delegations[node]

        weight_counter = Counter(get_transitive_delegate(node) for node in range(num_agents))

        # weight_counter.most_common(1) might e.g. be [(max_weight_voter, max_weight)]
        return weight_counter.most_common(1)[0][1]

    @staticmethod
    def is_splittable():
        False

    @staticmethod
    def max_weight_from_delegations(delegations):
        return ConfluentMechanism._max_weight_from_confluent_delegations(delegations)


class SplittableMechanism(Mechanism):
    @staticmethod
    def _max_weight_from_splittable_delegations(delegations):
        """

        >>> SplittableMechanism._max_weight_from_splittable_delegations([{}, {}, {0: 0.25, 1: 0.75}])
        1.75
        >>> SplittableMechanism._max_weight_from_splittable_delegations([{}, {}, {0: 0.75, 1: 0.5}, {1: 0.75, 2: 0.25}])
        2.25
        >>> SplittableMechanism._max_weight_from_splittable_delegations([{}, {}, {0: 1.5, 1: 1.5}, {2: 2.0}, {3: 1.0}])
        2.5

        Args:
             delegations (list of ((dict of int → float) / None))
        """
        num_agents = len(delegations)

        congestions = [1. for _ in range(num_agents)]
        for succs in delegations:
            if succs is not None:
                for neighbor in succs:
                    congestions[neighbor] += succs[neighbor]

        max_congestion = -inf
        for i, succs in enumerate(delegations):
            if succs is None:
                max_congestion = max(max_congestion, congestions[i])
        assert max_congestion > -inf

        return max_congestion

    @staticmethod
    def is_splittable():
        True

    @staticmethod
    def max_weight_from_delegations(delegations):
        return SplittableMechanism._max_weight_from_splittable_delegations(delegations)


if __name__ == "__main__":
    from doctest import testmod
    testmod()