Modications and Tests for Moran process

Author: marcharper

axelrod/moran.py

@@ -40,7 +40,11 @@ def __init__(self, players, turns=100, noise=0, deterministic_cache=None, mutati
 
         If the mutation_rate is 0, the population will eventually fixate on exactly one player type. In this
         case a StopIteration exception is raised and the play stops. If mutation_rate is not zero, then
-        the process will iterate indefinitely.
+        the process will iterate indefinitely, so mp.play() will never exit, and you should use the class as an
+        iterator instead.
+
+        When a player mutates, it chooses a random player type from the initial population. This is not the only
+        method yet emulates the common method in the literature.
 
         Parameters
         ----------
@@ -65,11 +69,21 @@ def __init__(self, players, turns=100, noise=0, deterministic_cache=None, mutati
         self.mutation_rate = mutation_rate
         assert (mutation_rate >= 0) and (mutation_rate <= 1)
         assert (noise >= 0) and (noise <= 1)
-
         if deterministic_cache is not None:
             self.deterministic_cache = deterministic_cache
         else:
             self.deterministic_cache = DeterministicCache()
+        # Build the set of mutation targets
+        # Determine the number of unique types (players)
+        keys = set([str(p) for p in players])
+        # Create a dictionary mapping each type to a set of representatives of the other types
+        d = dict()
+        for p in players:
+            d[str(p)] = p
+        mt = dict()
+        for key in keys:
+            mt[key] = [v for (k, v) in d.items() if k != key]
+        self.mutation_targets = mt
 
     def set_players(self):
         """Copy the initial players into the first population."""
@@ -88,10 +102,23 @@ def _stochastic(self):
         """
         return is_stochastic(self.players, self.noise) or (self.mutation_rate > 0)
 
+    def mutate(self, index):
+        # If mutate, choose another strategy at random from the initial population
+        r = random.random()
+        if r < self.mutation_rate:
+            s = str(self.players[index])
+            p = random.choice(self.mutation_targets[s])
+            new_player = p.clone()
+        else:
+            # Just clone the player
+            new_player = self.players[index].clone()
+        return new_player
+
     def __next__(self):
         """Iterate the population:
         - play the round's matches
         - chooses a player proportionally to fitness (total score) to reproduce
+        - mutate, if appropriate
         - choose a player at random to be replaced
         - update the population
         """
@@ -107,15 +134,15 @@ def __next__(self):
         j = fitness_proportionate_selection(scores)
         # Mutate?
         if self.mutation_rate:
-            r = random.random()
-            # If mutate, choose another strategy at random
-            if r < self.mutation_rate:
-                j = randrange(0, len(self.players))
+            new_player = self.mutate(j)
+        else:
+            new_player = self.players[j].clone()
         # 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.players[i] = new_player
         self.populations.append(self.population_distribution())
+        return self
 
     def _play_next_round(self):
         """Plays the next round of the process. Every player is paired up

axelrod/tests/unit/test_moran.py

@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+from collections import Counter
+import itertools
 import random
 import unittest
 
@@ -46,6 +48,24 @@ def test_two_players(self):
         self.assertEqual(populations, mp.populations)
         self.assertEqual(mp.winning_strategy_name, str(p2))
 
+    def test_two_players_with_mutation(self):
+        p1, p2 = axelrod.Cooperator(), axelrod.Defector()
+        random.seed(5)
+        mp = MoranProcess((p1, p2), mutation_rate=0.2)
+        self.assertEqual(mp._stochastic, True)
+        self.assertEqual(mp.mutation_targets, {str(p1): [p2], str(p2): [p1]})
+        # Test that mutation causes the population to alternate between fixations
+        counters = [
+            Counter({'Cooperator': 2}),
+            Counter({'Defector': 2}),
+            Counter({'Cooperator': 2}),
+            Counter({'Defector': 2})
+        ]
+        for counter in counters:
+            for _ in itertools.takewhile(lambda x: x.population_distribution() != counter, mp):
+                pass
+            self.assertEqual(mp.population_distribution(), counter)
+
     def test_three_players(self):
         players = [axelrod.Cooperator(), axelrod.Cooperator(),
                    axelrod.Defector()]
@@ -57,6 +77,24 @@ def test_three_players(self):
         self.assertEqual(populations, mp.populations)
         self.assertEqual(mp.winning_strategy_name, str(axelrod.Defector()))
 
+    def test_three_players_with_mutation(self):
+        p1 = axelrod.Cooperator()
+        p2 = axelrod.Random()
+        p3 = axelrod.Defector()
+        players = [p1, p2, p3]
+        mp = MoranProcess(players, mutation_rate=0.2)
+        self.assertEqual(mp._stochastic, True)
+        self.assertEqual(mp.mutation_targets, {str(p1): [p2, p3], str(p2): [p1, p3], str(p3): [p1, p2]})
+        # Test that mutation causes the population to alternate between fixations
+        counters = [
+            Counter({'Cooperator': 3}),
+            Counter({'Defector': 3}),
+        ]
+        for counter in counters:
+            for _ in itertools.takewhile(lambda x: x.population_distribution() != counter, mp):
+                pass
+            self.assertEqual(mp.population_distribution(), counter)
+
     def test_four_players(self):
         players = [axelrod.Cooperator() for _ in range(3)]
         players.append(axelrod.Defector())