Merge pull request #534 from marcharper/moran-process

Moran process

Author: meatballs

axelrod/__init__.py

@@ -8,6 +8,7 @@
 from .player import init_args, is_basic, obey_axelrod, update_history, Player
 from .mock_player import MockPlayer, simulate_play
 from .match import Match
+from .moran import MoranProcess
 from .strategies import *
 from .deterministic_cache import DeterministicCache
 from .match_generator import *

axelrod/match.py

@@ -7,6 +7,11 @@
 C, D = Actions.C, Actions.D
 
 
+def is_stochastic(players, noise):
+    """Determines if a match is stochastic -- true if there is noise or if any
+    of the players involved is stochastic."""
+    return (noise or any(p.classifier['stochastic'] for p in players))
+
 class Match(object):
 
     def __init__(self, players, turns, deterministic_cache=None, noise=0):
@@ -38,10 +43,7 @@ def _stochastic(self):
         A boolean to show whether a match between two players would be
         stochastic
         """
-        return (
-            self._noise or
-            any(p.classifier['stochastic'] for p in self.players)
-            )
+        return is_stochastic(self.players, self._noise)
 
     @property
     def _cache_update_required(self):

axelrod/moran.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+from collections import Counter
+import random
+
+import numpy as np
+
+from .deterministic_cache import DeterministicCache
+from .match import Match, is_stochastic
+from .player import Player
+from .random_ import randrange
+
+
+def fitness_proportionate_selection(scores):
+    """Randomly selects an individual proportionally to score.
+
+    Parameters
+    ----------
+    scores: Any sequence of real numbers
+
+    Returns
+    -------
+    An index of the above list selected at random proportionally to the list
+    element divided by the total.
+    """
+    csums = np.cumsum(scores)
+    total = csums[-1]
+    r = random.random() * total
+
+    for i, x in enumerate(csums):
+        if x >= r:
+            return i
+
+class MoranProcess(object):
+    def __init__(self, players, turns=100, noise=0):
+        self.turns = turns
+        self.noise = noise
+        self.players = list(players) # initial population
+        self.winning_strategy_name = None
+        self.populations = []
+        self.populations.append(self.population_distribution())
+        self.score_history = []
+        self.num_players = len(self.players)
+
+    @property
+    def _stochastic(self):
+        """
+        A boolean to show whether a match between two players would be
+        stochastic
+        """
+        return is_stochastic(self.players, self.noise)
+
+    def __next__(self):
+        """Iterate the population:
+        - play the round's matches
+        - chooses a player proportionally to fitness (total score) to reproduce
+        - choose a player at random to be replaced
+        - update the population
+        """
+        # Check the exit condition, that all players are of the same type.
+        population = self.populations[-1]
+        classes = set(p.__class__ for p in self.players)
+        if len(classes) == 1:
+            self.winning_strategy_name = str(self.players[0])
+            raise StopIteration
+        scores = self._play_next_round()
+        # Update the population
+        # Fitness proportionate selection
+        j = fitness_proportionate_selection(scores)
+        # Randomly remove a strategy
+        i = randrange(0, len(self.players))
+        # Replace player i with clone of player j
+        self.players[i] = self.players[j].clone()
+        self.populations.append(self.population_distribution())
+
+    def _play_next_round(self):
+        """Plays the next round of the process. Every player is paired up
+        against every other player and the total scores are recorded."""
+        N = self.num_players
+        scores = [0] * N
+        for i in range(N):
+            for j in range(i + 1, N):
+                player1 = self.players[i]
+                player2 = self.players[j]
+                player1.reset()
+                player2.reset()
+                match = Match((player1, player2), self.turns, noise=self.noise)
+                match.play()
+                match_scores = np.sum(match.scores(), axis=0) / float(self.turns)
+                scores[i] += match_scores[0]
+                scores[j] += match_scores[1]
+        self.score_history.append(scores)
+        return scores
+
+    def population_distribution(self):
+        """Returns the population distribution of the last iteration."""
+        player_names = [str(player) for player in self.players]
+        counter = Counter(player_names)
+        return counter
+
+    next = __next__  # Python 2
+
+    def __iter__(self):
+        return self
+
+    def reset(self):
+        """Reset the process to replay."""
+        self.winning_strategy_name = None
+        self.populations = [self.populations[0]]
+        self.score_history = []
+
+    def play(self):
+        """Play the process out to completion."""
+        while True:
+            try:
+                self.__next__()
+            except StopIteration:
+                break
+
+    def __len__(self):
+        return len(self.populations)

axelrod/random_.py

@@ -12,3 +12,10 @@ def random_choice(p=0.5):
     if r < p:
         return Actions.C
     return Actions.D
+
+def randrange(a, b):
+    """Python 2 / 3 compatible randrange. Returns a random integer uniformly
+    between a and b (inclusive)"""
+    c = b - a
+    r = c * random.random()
+    return a + int(r)

axelrod/tests/unit/test_moran.py

@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+import random
+import unittest
+
+import axelrod
+from axelrod import MoranProcess
+from axelrod.moran import fitness_proportionate_selection
+
+from hypothesis import given, example, settings
+from hypothesis.strategies import integers, lists, sampled_from, random_module, floats
+
+
+class TestMoranProcess(unittest.TestCase):
+
+    def test_fps(self):
+        self.assertEqual(fitness_proportionate_selection([0, 0, 1]), 2)
+        random.seed(1)
+        self.assertEqual(fitness_proportionate_selection([1, 1, 1]), 0)
+        self.assertEqual(fitness_proportionate_selection([1, 1, 1]), 2)
+
+    def test_stochastic(self):
+        p1, p2 = axelrod.Cooperator(), axelrod.Cooperator()
+        mp = MoranProcess((p1, p2))
+        self.assertFalse(mp._stochastic)
+        p1, p2 = axelrod.Cooperator(), axelrod.Cooperator()
+        mp = MoranProcess((p1, p2), noise=0.05)
+        self.assertTrue(mp._stochastic)
+        p1, p2 = axelrod.Cooperator(), axelrod.Random()
+        mp = MoranProcess((p1, p2))
+        self.assertTrue(mp._stochastic)
+
+    def test_exit_condition(self):
+        p1, p2 = axelrod.Cooperator(), axelrod.Cooperator()
+        mp = MoranProcess((p1, p2))
+        mp.play()
+        self.assertEqual(len(mp), 1)
+
+    def test_two_players(self):
+        p1, p2 = axelrod.Cooperator(), axelrod.Defector()
+        random.seed(5)
+        mp = MoranProcess((p1, p2))
+        mp.play()
+        self.assertEqual(len(mp), 5)
+        self.assertEqual(mp.winning_strategy_name, str(p2))
+
+    def test_three_players(self):
+        players = [axelrod.Cooperator(), axelrod.Cooperator(),
+                   axelrod.Defector()]
+        random.seed(5)
+        mp = MoranProcess(players)
+        mp.play()
+        self.assertEqual(len(mp), 7)
+        self.assertEqual(mp.winning_strategy_name, str(axelrod.Defector()))
+
+    def test_four_players(self):
+        players = [axelrod.Cooperator() for _ in range(3)]
+        players.append(axelrod.Defector())
+        random.seed(10)
+        mp = MoranProcess(players)
+        mp.play()
+        self.assertEqual(len(mp), 9)
+        self.assertEqual(mp.winning_strategy_name, str(axelrod.Defector()))
+
+    @given(strategies=lists(sampled_from(axelrod.strategies),
+                   min_size=2,  # Errors are returned if less than 2 strategies
+                   max_size=5, unique=True),
+           rm=random_module())
+    @settings(max_examples=5, timeout=0)  # Very low number of examples
+
+    # Two specific examples relating to cloning of strategies
+    @example(strategies=[axelrod.BackStabber, axelrod.MindReader],
+             rm=random.seed(0))
+    @example(strategies=[axelrod.ThueMorse, axelrod.MindReader],
+             rm=random.seed(0))
+    def test_property_players(self, strategies, rm):
+        """Hypothesis test that randomly checks players"""
+        players = [s() for s in strategies]
+        mp = MoranProcess(players)
+        mp.play()
+        self.assertIn(mp.winning_strategy_name, [str(p) for p in players])
+
+    def test_reset(self):
+        p1, p2 = axelrod.Cooperator(), axelrod.Defector()
+        random.seed(8)
+        mp = MoranProcess((p1, p2))
+        mp.play()
+        self.assertEqual(len(mp), 4)
+        self.assertEqual(len(mp.score_history), 3)
+        mp.reset()
+        self.assertEqual(len(mp), 1)
+        self.assertEqual(mp.winning_strategy_name, None)
+        self.assertEqual(mp.score_history, [])

docs/tutorials/further_topics/index.rst

@@ -11,6 +11,7 @@ Contents:
 
    classification_of_strategies.rst
    creating_matches.rst
+   moran.rst
    morality_metrics.rst
    probabilistict_end_tournaments.rst
    reading_and_writing_interactions.rst

docs/tutorials/further_topics/moran.rst

@@ -0,0 +1,56 @@
+Moran Process
+=============
+
+The strategies in the library can be pitted against one another in the
+[Moran process](https://en.wikipedia.org/wiki/Moran_process), a population
+process simulating natural selection. Given the evolutionary basis of the Moran
+process it can be compared to the :ref:`ecological-variant`.
+While that variant was used by Axelrod in his original works, the Moran process
+is now much more widely studied in the literature.
+
+The process works as follows. Given an
+initial population of players, the population is iterated in rounds consisting
+of:
+- matches played between each pair of players, with the cumulative total
+scores recored
+- a player is chosen to reproduce proportional to the player's score in the
+round
+- a player is chosen at random to be replaced
+
+The process proceeds in rounds until the population consists of a single player
+type. That type is declared the winner. To run an instance of the process with
+the library, proceed as follows::
+
+    >>> import axelrod as axl
+    >>> players = [axl.Cooperator(), axl.Defector(),
+    ...               axl.TitForTat(), axl.Grudger()]
+    >>> mp = axl.MoranProcess(players)
+    >>> mp.play()
+    >>> mp.winning_strategy_name   # doctest: +SKIP
+    Defector
+
+You can access some attributes of the process, such as the number of rounds::
+
+    >>> len(mp)  # doctest: +SKIP
+    6
+
+The sequence of populations::
+
+    >>> import pprint
+    >>> pprint.pprint(mp.populations)  # doctest: +SKIP
+    [Counter({'Defector': 1, 'Cooperator': 1, 'Grudger': 1, 'Tit For Tat': 1}),
+    Counter({'Defector': 1, 'Cooperator': 1, 'Grudger': 1, 'Tit For Tat': 1}),
+    Counter({'Defector': 2, 'Cooperator': 1, 'Grudger': 1}),
+    Counter({'Defector': 3, 'Grudger': 1}),
+    Counter({'Defector': 3, 'Grudger': 1}),
+    Counter({'Defector': 4})]
+
+The scores in each round::
+
+    >>> for row in mp.score_history: # doctest: +SKIP
+    ...     print([round(element, 1) for element in row])
+    [[6.0, 7.0800000000000001, 6.9900000000000002, 6.9900000000000002],
+    [6.0, 7.0800000000000001, 6.9900000000000002, 6.9900000000000002],
+    [3.0, 7.04, 7.04, 4.9800000000000004],
+    [3.04, 3.04, 3.04, 2.9699999999999998],
+    [3.04, 3.04, 3.04, 2.9699999999999998]]