Algorithm for Memory Depth in FSM

axelrod/strategies/finite_state_machines.py

@@ -1,7 +1,220 @@
 from axelrod.action import Action
 from axelrod.player import Player
+from typing import DefaultDict, Iterator
+from collections import defaultdict, namedtuple
 
 C, D = Action.C, Action.D
+ALL_ACTIONS = [C, D]
+
+
+"""
+Memit = unit of memory.
+
+This represents the amount of memory that we gain with each new piece of
+history.  It includes a state, our_response that we make on our way into that
+state (in_act), and the opponent's action that makes us move out of that state
+(out_act).
+
+For example for this FSM:
+(0, C, 0, C),
+(0, D, 1, C),
+(1, C, 0, D),
+(1, D, 0, D)
+
+Has the memits:
+(C, 0, C),
+(C, 0, D),
+(D, 0, C),
+(D, 0, D),
+(C, 1, C),
+(C, 1, D)
+"""
+Memit = namedtuple("Memit", ["in_act", "state", "out_act"])
+
+def memits_match(x, y):
+    """In action and out actions are the same."""
+    return x.in_act == y.in_act and x.out_act == y.out_act
+def memit_sort(x, y):
+    """Returns a tuple of x in y, sorted so that (x, y) are viewed as the
+    same as (y, x).
+    """
+    if repr(x) <= repr(y):
+        return (x, y)
+    else:
+        return (y, x)
+
+
+def get_accessible_transitions(transitions: dict, initial_state: int) -> dict:
+  """Gets all transitions from the list that can be reached from the
+  initial_state.
+  """
+  edge_dict: DefaultDict[int, list] = defaultdict(list)
+  visited = dict()
+  for k, v in transitions.items():
+      visited[k[0]] = False
+      edge_dict[k[0]].append(v[0])
+  accessible_edges = [initial_state]
+
+  edge_queue = [initial_state]
+  visited[initial_state] = True
+  while len(edge_queue) > 0:
+      edge = edge_queue.pop()
+      for successor in edge_dict[edge]:
+          if not visited[successor]:
+              visited[successor] = True
+              edge_queue.append(successor)
+              accessible_edges.append(successor)
+
+  accessible_transitions = dict()
+  for k, v in transitions.items():
+      if k[0] in accessible_edges:
+          accessible_transitions[k] = v
+
+  return accessible_transitions
+
+
+def longest_path(edges: dict, starting_at: Memit) -> int:
+    """Returns the number of nodes in the longest path that starts at the given
+    node.  Returns infinity if a loop is encountered.
+    """
+    visited = dict()
+    for k, v in edges.items():
+        visited[k] = False
+        for vi in v:
+            visited[vi] = False
+
+    # This is what we'll recurse on.  visited dict is shared between calls.
+    def recurse(at_node):
+        visited[at_node] = True
+        record = 1  # Count the nodes, not the edges.
+        for successor in edges[at_node]:
+            if visited[successor]:
+                return float("inf")
+            successor_length = recurse(successor)
+            if successor_length == float("inf"):
+                return float("inf")
+            if record < successor_length + 1:
+                record = successor_length + 1
+        return record
+
+    return recurse(starting_at)
+
+
+Transition = namedtuple("Transition", ["state", "last_opponent_action",
+                                       "next_state", "next_action"])
+def transition_iterator(transitions: dict) -> Iterator[Transition]:
+    """Changes the transition dictionary into a iterator on namedtuples, because
+    we use repeatedly.
+    """
+    for k, v in transitions.items():
+        yield Transition(k[0], k[1], v[0], v[1])
+
+
+def get_memory_from_transitions(transitions: dict,
+                                initial_state: int = None) -> int:
+    """This function calculates the memory of an FSM from the transitions.
+
+    Assume that transitions are a dict with entries like
+    (state, last_opponent_action): (next_state, next_action)
+
+    We first break down the transitions into memits (see above).  We also create
+    a graph of memits, where the successor to a given memit are all possible
+    memits that could occur in the memory immediately before the given memit.
+
+    Then we pair up memits with different states, but same in and out actions.
+    These represent points in time that we can't determine which state we're in.
+    We also create a graph of memit-pairs, where memit-pair, Y, succedes a
+    memit-pair, X, if the two memits in X are succeded by the two memits in Y.
+    These edges reperesent consecutive points in time that we can't determine
+    which state we're in.
+
+    Then for all memit-pairs that disagree, in the sense that they imply
+    different next_action, we find the longest chain starting at that
+    memit-pair.  [If a loop is encountered then this will be infinite.]  We take
+    the maximum over all sugh memit-pairs.  This represents the longest possible
+    chain of memory for which we wouldn't know what to do next.  We return this.
+    """
+    # If initial_state is set, use this to determine which transitions are
+    # reachable from the initial_state and restrict to those.
+    if initial_state is not None:
+        transitions = get_accessible_transitions(transitions, initial_state)
+
+    # Get the incoming actions for each state.
+    incoming_action_by_state: DefaultDict[int, set] = defaultdict(set)
+    for trans in transition_iterator(transitions):
+        incoming_action_by_state[trans.next_state].add(trans.next_action)
+
+    # Keys are starting memit, and values are all possible terminal memit.
+    # Will walk backwards through the graph.
+    memit_edges: DefaultDict[Memit, set] = defaultdict(set)
+    for trans in transition_iterator(transitions):
+        # Since all actions are out-paths for each state, add all of these.
+        # That is to say that your opponent could do anything
+        for out_action in ALL_ACTIONS:
+            # More recent in action history
+            starting_node = Memit(trans.next_action, trans.next_state,
+                                  out_action)
+            # All incoming paths to current state
+            for in_action in incoming_action_by_state[trans.state]:
+                # Less recent in action history
+                ending_node = Memit(in_action, trans.state,
+                                    trans.last_opponent_action)
+                memit_edges[starting_node].add(ending_node)
+
+    all_memits = memit_edges.keys()
+
+    pair_nodes = set()
+    pair_edges: DefaultDict[tuple, set] = defaultdict(set)
+    # Loop through all pairs of memits.
+    for x, y in [(x, y) for x in all_memits for y in all_memits]:
+        if x == y:
+            continue
+        if not memits_match(x, y):
+            continue
+
+        # If the memits match, then the strategy can't tell the difference
+        # between the states.  We call this a pair of matched memits (or just a
+        # pair).
+        pair_nodes.add(memit_sort(x, y))
+        # When two memits in matched pair have successors that are also matched,
+        # then we draw an edge.  This represents consecutive historical times
+        # that we can't tell which state we're in.
+        for x_successor in memit_edges[x]:
+            for y_successor in memit_edges[y]:
+                if memits_match(x_successor, y_successor):
+                    pair_edges[memit_sort(x, y)].add(memit_sort(x_successor,
+                                                                y_successor))
+
+    if len(pair_nodes) == 0:
+        # If there are no pair of tied memits, then either no memits are needed
+        # to break a tie (i.e. all next_actions are the same) or the first memit
+        # breaks a tie (i.e. memory 1)
+        next_action_set = set()
+        for trans in transition_iterator(transitions):
+            next_action_set.add(trans.next_action)
+        if len(next_action_set) == 1:
+            return 0
+        return 1
+
+    # Get next_action for each memit.  Used to decide if they are in conflict,
+    # because we only have undecidability if next_action doesn't match.
+    next_action_by_memit = dict()
+    for trans in transition_iterator(transitions):
+        for in_action in incoming_action_by_state[trans.state]:
+            memit_key = Memit(in_action, trans.state,
+                              trans.last_opponent_action)
+            next_action_by_memit[memit_key] = trans.next_action
+
+    # Calculate the longest path.
+    record = 0
+    for pair in pair_nodes:
+        if next_action_by_memit[pair[0]] != next_action_by_memit[pair[1]]:
+            # longest_path is the longest chain of tied states.  We add one to
+            # get the memory length needed to break all ties.
+            path_length = longest_path(pair_edges, pair) + 1
+            if record < path_length:
+                record = path_length
+    return record
 
 
 class SimpleFSM(object):
@@ -124,7 +337,7 @@ class Fortress3(FSMPlayer):
 
     name = "Fortress3"
     classifier = {
-        "memory_depth": 3,
+        "memory_depth": 2,
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -161,7 +374,7 @@ class Fortress4(FSMPlayer):
 
     name = "Fortress4"
     classifier = {
-        "memory_depth": 4,
+        "memory_depth": 3,
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -197,7 +410,7 @@ class Predator(FSMPlayer):
 
     name = "Predator"
     classifier = {
-        "memory_depth": 9,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -241,7 +454,7 @@ class Pun1(FSMPlayer):
 
     name = "Pun1"
     classifier = {
-        "memory_depth": 2,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -268,7 +481,7 @@ class Raider(FSMPlayer):
 
     name = "Raider"
     classifier = {
-        "memory_depth": 3,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -303,7 +516,7 @@ class Ripoff(FSMPlayer):
 
     name = "Ripoff"
     classifier = {
-        "memory_depth": 2,
+        "memory_depth": 3,
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -336,7 +549,7 @@ class UsuallyCooperates(FSMPlayer):
 
     name = "UsuallyCooperates"
     classifier = {
-        "memory_depth": 2,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -367,7 +580,7 @@ class UsuallyDefects(FSMPlayer):
 
     name = "UsuallyDefects"
     classifier = {
-        "memory_depth": 2,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -398,7 +611,7 @@ class SolutionB1(FSMPlayer):
 
     name = "SolutionB1"
     classifier = {
-        "memory_depth": 3,
+        "memory_depth": 2,
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -432,7 +645,7 @@ class SolutionB5(FSMPlayer):
 
     name = "SolutionB5"
     classifier = {
-        "memory_depth": 5,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -471,7 +684,7 @@ class Thumper(FSMPlayer):
 
     name = "Thumper"
     classifier = {
-        "memory_depth": 2,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -497,7 +710,7 @@ class EvolvedFSM4(FSMPlayer):
 
     name = "Evolved FSM 4"
     classifier = {
-        "memory_depth": 4,
+        "memory_depth": float("inf"),
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -533,7 +746,7 @@ class EvolvedFSM16(FSMPlayer):
 
     name = "Evolved FSM 16"
     classifier = {
-        "memory_depth": 16,  # At most
+        "memory_depth": float("inf"),  # At most
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,
@@ -589,7 +802,7 @@ class EvolvedFSM16Noise05(FSMPlayer):
 
     name = "Evolved FSM 16 Noise 05"
     classifier = {
-        "memory_depth": 16,  # At most
+        "memory_depth": float("inf"),  # At most
         "stochastic": False,
         "makes_use_of": set(),
         "long_run_time": False,