-
Notifications
You must be signed in to change notification settings - Fork 264
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Pass fitness function moran process #1198
Changes from 7 commits
7d8083b
421101b
3a9b2ab
5eed9b4
7a1a17a
d54c0c3
8824c9b
c3724f2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,33 +1,39 @@ | ||
"""Implementation of the Moran process on Graphs.""" | ||
|
||
from collections import Counter | ||
import random | ||
from collections import Counter | ||
from typing import Callable, List, Optional, Set, Tuple | ||
|
||
import matplotlib.pyplot as plt | ||
import numpy as np | ||
|
||
from axelrod import DEFAULT_TURNS, Player, Game | ||
from axelrod import DEFAULT_TURNS, Game, Player | ||
|
||
from .deterministic_cache import DeterministicCache | ||
from .graph import complete_graph, Graph | ||
from .graph import Graph, complete_graph | ||
from .match import Match | ||
from .random_ import randrange | ||
|
||
from typing import List, Tuple, Set, Optional | ||
|
||
|
||
def fitness_proportionate_selection(scores: List) -> int: | ||
def fitness_proportionate_selection( | ||
scores: List, fitness_function: Callable = None | ||
) -> int: | ||
"""Randomly selects an individual proportionally to score. | ||
|
||
Parameters | ||
---------- | ||
scores: Any sequence of real numbers | ||
fitness_function: A function mapping a score to a (non-negative) float | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Following @marcharper's suggestion can we change this to be |
||
|
||
Returns | ||
------- | ||
An index of the above list selected at random proportionally to the list | ||
element divided by the total. | ||
""" | ||
csums = np.cumsum(scores) | ||
if fitness_function is None: | ||
csums = np.cumsum(scores) | ||
else: | ||
csums = np.cumsum([fitness_function(s) for s in scores]) | ||
total = csums[-1] | ||
r = random.random() * total | ||
|
||
|
@@ -38,13 +44,20 @@ def fitness_proportionate_selection(scores: List) -> int: | |
|
||
|
||
class MoranProcess(object): | ||
def __init__(self, players: List[Player], turns: int = DEFAULT_TURNS, | ||
prob_end: float = None, noise: float = 0, | ||
game: Game = None, | ||
deterministic_cache: DeterministicCache = None, | ||
mutation_rate: float = 0., mode: str = 'bd', | ||
interaction_graph: Graph = None, | ||
reproduction_graph: Graph = None) -> None: | ||
def __init__( | ||
self, | ||
players: List[Player], | ||
turns: int = DEFAULT_TURNS, | ||
prob_end: float = None, | ||
noise: float = 0, | ||
game: Game = None, | ||
deterministic_cache: DeterministicCache = None, | ||
mutation_rate: float = 0., | ||
mode: str = "bd", | ||
interaction_graph: Graph = None, | ||
reproduction_graph: Graph = None, | ||
fitness_function: Callable = None, | ||
) -> None: | ||
""" | ||
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 | ||
|
@@ -92,6 +105,8 @@ def __init__(self, players: List[Player], turns: int = DEFAULT_TURNS, | |
reproduction_graph: Axelrod.graph.Graph | ||
The reproduction graph, set equal to the interaction graph if not | ||
given | ||
fitness_function: | ||
A function mapping a score to a (non-negative) float | ||
""" | ||
self.turns = turns | ||
self.prob_end = prob_end | ||
|
@@ -107,7 +122,7 @@ def __init__(self, players: List[Player], turns: int = DEFAULT_TURNS, | |
assert (mutation_rate >= 0) and (mutation_rate <= 1) | ||
assert (noise >= 0) and (noise <= 1) | ||
mode = mode.lower() | ||
assert mode in ['bd', 'db'] | ||
assert mode in ["bd", "db"] | ||
self.mode = mode | ||
if deterministic_cache is not None: | ||
self.deterministic_cache = deterministic_cache | ||
|
@@ -123,25 +138,30 @@ def __init__(self, players: List[Player], turns: int = DEFAULT_TURNS, | |
d[str(p)] = p | ||
mutation_targets = dict() | ||
for key in sorted(keys): | ||
mutation_targets[key] = [v for (k, v) in sorted(d.items()) if k != key] | ||
mutation_targets[key] = [ | ||
v for (k, v) in sorted(d.items()) if k != key | ||
] | ||
self.mutation_targets = mutation_targets | ||
|
||
if interaction_graph is None: | ||
interaction_graph = complete_graph(len(players), loops=False) | ||
if reproduction_graph is None: | ||
reproduction_graph = Graph(interaction_graph.edges(), | ||
directed=interaction_graph.directed) | ||
reproduction_graph = Graph( | ||
interaction_graph.edges(), directed=interaction_graph.directed | ||
) | ||
reproduction_graph.add_loops() | ||
# Check equal vertices | ||
v1 = interaction_graph.vertices() | ||
v2 = reproduction_graph.vertices() | ||
assert list(v1) == list(v2) | ||
self.interaction_graph = interaction_graph | ||
self.reproduction_graph = reproduction_graph | ||
self.fitness_function = fitness_function | ||
# Map players to graph vertices | ||
self.locations = sorted(interaction_graph.vertices()) | ||
self.index = dict(zip(sorted(interaction_graph.vertices()), | ||
range(len(players)))) | ||
self.index = dict( | ||
zip(sorted(interaction_graph.vertices()), range(len(players))) | ||
) | ||
|
||
def set_players(self) -> None: | ||
"""Copy the initial players into the first population.""" | ||
|
@@ -192,8 +212,11 @@ def death(self, index: int = None) -> int: | |
else: | ||
# Select locally | ||
# index is not None in this case | ||
vertex = random.choice(sorted( | ||
self.reproduction_graph.out_vertices(self.locations[index]))) | ||
vertex = random.choice( | ||
sorted( | ||
self.reproduction_graph.out_vertices(self.locations[index]) | ||
) | ||
) | ||
i = self.index[vertex] | ||
return i | ||
|
||
|
@@ -212,11 +235,15 @@ def birth(self, index: int = None) -> int: | |
# possible choices | ||
scores.pop(index) | ||
# Make sure to get the correct index post-pop | ||
j = fitness_proportionate_selection(scores) | ||
j = fitness_proportionate_selection( | ||
scores, fitness_function=self.fitness_function | ||
) | ||
if j >= index: | ||
j += 1 | ||
else: | ||
j = fitness_proportionate_selection(scores) | ||
j = fitness_proportionate_selection( | ||
scores, fitness_function=self.fitness_function | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. So this would be |
||
) | ||
return j | ||
|
||
def fixation_check(self) -> bool: | ||
|
@@ -321,11 +348,14 @@ def score_all(self) -> List: | |
for i, j in self._matchup_indices(): | ||
player1 = self.players[i] | ||
player2 = self.players[j] | ||
match = Match((player1, player2), | ||
turns=self.turns, prob_end=self.prob_end, | ||
noise=self.noise, | ||
game=self.game, | ||
deterministic_cache=self.deterministic_cache) | ||
match = Match( | ||
(player1, player2), | ||
turns=self.turns, | ||
prob_end=self.prob_end, | ||
noise=self.noise, | ||
game=self.game, | ||
deterministic_cache=self.deterministic_cache, | ||
) | ||
match.play() | ||
match_scores = match.final_score_per_turn() | ||
scores[i] += match_scores[0] | ||
|
@@ -373,7 +403,8 @@ def play(self) -> List[Counter]: | |
if self.mutation_rate != 0: | ||
raise ValueError( | ||
"MoranProcess.play() will never exit if mutation_rate is" | ||
"nonzero. Use iteration instead.") | ||
"nonzero. Use iteration instead." | ||
) | ||
while True: | ||
try: | ||
self.__next__() | ||
|
@@ -435,8 +466,13 @@ class ApproximateMoranProcess(MoranProcess): | |
Instead of playing the matches, the result is sampled | ||
from a dictionary of player tuples to distribution of match outcomes | ||
""" | ||
def __init__(self, players: List[Player], cached_outcomes: dict, | ||
mutation_rate: float = 0) -> None: | ||
|
||
def __init__( | ||
self, | ||
players: List[Player], | ||
cached_outcomes: dict, | ||
mutation_rate: float = 0, | ||
) -> None: | ||
""" | ||
Parameters | ||
---------- | ||
|
@@ -448,8 +484,12 @@ def __init__(self, players: List[Player], cached_outcomes: dict, | |
probability `mutation_rate` | ||
""" | ||
super(ApproximateMoranProcess, self).__init__( | ||
players, turns=0, noise=0, deterministic_cache=None, | ||
mutation_rate=mutation_rate) | ||
players, | ||
turns=0, | ||
noise=0, | ||
deterministic_cache=None, | ||
mutation_rate=mutation_rate, | ||
) | ||
self.cached_outcomes = cached_outcomes | ||
|
||
def score_all(self) -> List: | ||
|
@@ -466,8 +506,9 @@ def score_all(self) -> List: | |
scores = [0] * N | ||
for i in range(N): | ||
for j in range(i + 1, N): | ||
player_names = tuple([str(self.players[i]), | ||
str(self.players[j])]) | ||
player_names = tuple( | ||
[str(self.players[i]), str(self.players[j])] | ||
) | ||
|
||
cached_score = self._get_scores_from_cache(player_names) | ||
scores[i] += cached_score[0] | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Need to add
fitness_function
to the docstring (similarly forMoranProcess.__init__
):I suggest:
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ah missed that. Fixing it