Moran Process with Mutation

axelrod/moran.py

@@ -31,7 +31,33 @@ def fitness_proportionate_selection(scores):
 
 
 class MoranProcess(object):
-    def __init__(self, players, turns=100, noise=0, deterministic_cache=None):
+    def __init__(self, players, turns=100, noise=0, deterministic_cache=None, mutation_rate=0.):
+        """
+        An agent based Moran process class. In each round, each player plays a Match with each other
+        player. Players are assigned a fitness score by their total score from all matches in the round.
+        A player is chosen to reproduce proportionally to fitness, possibly mutated, and is cloned. The
+        clone replaces a randomly chosen player.
+
+        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, 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
+        ----------
+        players, iterable of axelrod.Player subclasses
+        turns: int, 100
+            The number of turns in each pairwise interaction
+        noise: float, 0
+            The background noise, if any. Randomly flips plays with probability `noise`.
+        deterministic_cache: axelrod.DeterministicCache, None
+            A optional prebuilt deterministic cache
+        mutation_rate: float, 0
+            The rate of mutation. Replicating players are mutated with probability `mutation_rate`
+        """
         self.turns = turns
         self.noise = noise
         self.initial_players = players  # save initial population
@@ -40,10 +66,24 @@ def __init__(self, players, turns=100, noise=0, deterministic_cache=None):
         self.set_players()
         self.score_history = []
         self.winning_strategy_name = None
+        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 sorted(keys):
+            mt[key] = [v for (k, v) in sorted(d.items()) if k != key]
+        self.mutation_targets = mt
 
     def set_players(self):
         """Copy the initial players into the first population."""
@@ -60,29 +100,48 @@ def _stochastic(self):
         A boolean to show whether a match between two players would be
         stochastic
         """
-        return is_stochastic(self.players, self.noise)
+        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
         """
         # Check the exit condition, that all players are of the same type.
-        classes = set(p.__class__ for p in self.players)
-        if len(classes) == 1:
+        classes = set(str(p) for p in self.players)
+        if (self.mutation_rate == 0) and (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)
+        # Mutate?
+        if self.mutation_rate:
+            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,34 @@ def test_two_players(self):
         self.assertEqual(populations, mp.populations)
         self.assertEqual(mp.winning_strategy_name, str(p2))
 
+    def test_two_random_players(self):
+        p1, p2 = axelrod.Random(0.5), axelrod.Random(0.25)
+        random.seed(5)
+        mp = MoranProcess((p1, p2))
+        populations = mp.play()
+        self.assertEqual(len(mp), 2)
+        self.assertEqual(len(populations), 2)
+        self.assertEqual(populations, mp.populations)
+        self.assertEqual(mp.winning_strategy_name, str(p1))
+
+    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.assertDictEqual(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 +87,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.assertDictEqual(mp.mutation_targets, {str(p1): [p3, p2], 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())