diff --git a/axelrod/data/all_classifiers.yml b/axelrod/data/all_classifiers.yml index fb0b6a51c..400e818c9 100644 --- a/axelrod/data/all_classifiers.yml +++ b/axelrod/data/all_classifiers.yml @@ -2040,3 +2040,21 @@ ZD-SET-2: manipulates_state: false memory_depth: 1 stochastic: true +ZeroResp: + inspects_source: false + long_run_time: false + makes_use_of: !!set + length: null + manipulates_source: false + manipulates_state: false + memory_depth: .inf + stochastic: true +ZeroResp v2: + inspects_source: false + long_run_time: false + makes_use_of: !!set + length: null + manipulates_source: false + manipulates_state: false + memory_depth: .inf + stochastic: true diff --git a/axelrod/fingerprint.py b/axelrod/fingerprint.py index aa58609f8..232e7f46c 100644 --- a/axelrod/fingerprint.py +++ b/axelrod/fingerprint.py @@ -48,6 +48,7 @@ def _create_points(step: float, progress_bar: bool = True) -> List[Point]: points = [] for x in np.linspace(0, 1, num): for y in np.linspace(0, 1, num): + # Cast to Python float so probe names/repr stay stable across NumPy versions points.append(Point(float(x), float(y))) if progress_bar: diff --git a/axelrod/strategies/_strategies.py b/axelrod/strategies/_strategies.py index bc80eeccc..00927e05a 100644 --- a/axelrod/strategies/_strategies.py +++ b/axelrod/strategies/_strategies.py @@ -284,6 +284,8 @@ ZDMischief, ZDSet2, ) +from .zeroresp import ZeroResp +from .zeroresp_v2 import ZeroRespV2 # Note: Meta* strategies are handled in .__init__.py @@ -509,5 +511,7 @@ ZDMem2, ZDMischief, ZDSet2, + ZeroResp, + ZeroRespV2, e, ] diff --git a/axelrod/strategies/zeroresp.py b/axelrod/strategies/zeroresp.py new file mode 100644 index 000000000..82e48cc43 --- /dev/null +++ b/axelrod/strategies/zeroresp.py @@ -0,0 +1,257 @@ +""" +ZeroResp: adaptive state-machine strategy for the Iterated Prisoner's Dilemma. + +Designed to resist both heuristic exploiters and simple RL / tabular Q-learners +via delayed stochastic retaliation, epoch-based debt accounting, and a permanent +red-line ban after systemic abuse. +""" + +from __future__ import annotations + +import math +from enum import Enum, auto +from typing import List, Optional + +from axelrod.action import Action +from axelrod.player import Player + +C, D = Action.C, Action.D + + +class _State(Enum): + """Internal finite-state labels.""" + + COOPERATIVE = auto() + EQUALIZING = auto() + RED_LINE = auto() + + +class ZeroResp(Player): + """ + An adaptive state machine that balances cooperation with delayed, + randomised retaliation and a permanent ban against systemic defectors. + + Architecture + ------------ + 1. **Dynamic epochs** — interaction is partitioned into epochs of length + ``base_epoch`` (default 25). While a retaliation debt or queued strike + is outstanding the epoch is extended; once cleared the systemic-abuse + counter resets and the bot returns to cooperative mode. + + 2. **Stochastic retaliation buffer** — a defection does not trigger an + immediate mirror response. Instead a retaliatory ``D`` is scheduled + ``5 + U{1..10}`` turns later. The random delay breaks short-horizon + Markov estimates used by tabular Q-learners and reduces cascade wars + against tit-for-tat family strategies. + + 3. **Red line (ban list)** — systemic defections (defects that arrive while + debt/queue is still open, or while already equalising) raise a counter. + After a dynamic threshold (2 or 3 depending on observed hostility) the + strategy enters permanent red line (``is_red_line = True``) and defects + unconditionally for the rest of the match. + + 4. **Anti-raider** — two or more late-game defections (past ~75% of the + known match length) are treated as end-game harvest and trigger red + line immediately. + + 5. **End-game harvest** — against highly forgiving / near-pure cooperators + (and never against grim-trigger types that never defected), ZeroResp may + defect near the known end of a finite match. This is disabled when + match length is unknown. + + Names: + + - ZeroResp: Original name by EpochRedLine / SovereignStabilizer authors + - EpochRedLine: Earlier development name + - SmartTitForTat: Legacy sandbox name + """ + + name = "ZeroResp" + classifier = { + "memory_depth": float("inf"), + "stochastic": True, + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": False, + } + + # Fallback when match length is unknown / infinite. + _DEFAULT_MATCH_LENGTH = 200 + _LATE_FRACTION = 0.75 + _LIVE_INTEL_MIN_SAMPLES = 10 + _HOSTILE_COOP_THRESHOLD = 0.4 + _SOFT_HOSTILE_COOP = 0.7 + + def __init__(self) -> None: + """Initialise epoch accounting and red-line state.""" + super().__init__() + self.base_epoch = 25 + + self._state = _State.COOPERATIVE + self.is_red_line = False + self.epoch_step = 0 + self.debt = 0 + self.systemic = 0 + self.queue: List[int] = [] + + # Opponent cadastre (loyalty / exploitability estimates) + self.opp_len = 0 + self.opp_defects = 0 + self.opp_coops_after_my_D = 0 + self.my_D = 0 + self.last_my: Action = C + self.late_defects = 0 + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _match_length(self) -> Optional[int]: + """Return known finite match length, else ``None``.""" + # Bracket access so Axelrod's makes_use_of scanner detects "length". + length = self.match_attributes["length"] + if length is None or length in (-1, float("inf")): + return None + try: + length_int = int(length) + except (TypeError, ValueError): + return None + return length_int if length_int > 0 else None + + def _effective_length(self) -> int: + return self._match_length() or self._DEFAULT_MATCH_LENGTH + + def _late_threshold(self) -> int: + return int(self._effective_length() * self._LATE_FRACTION) + + def _live_coop_rate(self) -> float: + if self.opp_len == 0: + return 1.0 + return 1.0 - (self.opp_defects / self.opp_len) + + def _is_hostile(self) -> bool: + """Enough evidence of a low-cooperation opponent.""" + return ( + self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES + and self._live_coop_rate() < self._HOSTILE_COOP_THRESHOLD + ) + + def _is_soft_hostile(self) -> bool: + return ( + self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES + and self._live_coop_rate() < self._SOFT_HOSTILE_COOP + ) + + def _enter_red_line(self) -> None: + self._state = _State.RED_LINE + self.is_red_line = True + self.queue.clear() + + # ------------------------------------------------------------------ + # Core strategy + # ------------------------------------------------------------------ + + def strategy(self, opponent: Player) -> Action: + """Select C or D for the current turn.""" + step = len(self.history) + 1 # 1-based turn index + + # --- Cadastre update from opponent's previous action ------------- + if opponent.history: + opp_last = opponent.history[-1] + self.opp_len += 1 + if opp_last == D: + self.opp_defects += 1 + if step > self._late_threshold(): + self.late_defects += 1 + self._on_defect(step) + else: + if self.last_my == D: + self.opp_coops_after_my_D += 1 + + # Live zero-turn defence: permanent ban against proven predators + if self._is_hostile(): + self._enter_red_line() + + # --- Anti-raider (late harvest interception) --------------------- + if self.late_defects >= 2: + self._enter_red_line() + return self._play(D) + + if self.is_red_line or self._state == _State.RED_LINE: + return self._play(D) + + # --- End-game harvest vs forgiving victims (known finite length) - + known_len = self._match_length() + if known_len is not None and self.opp_len > 50: + p_end = 1.0 / (1.0 + math.exp(-10.0 * (step / known_len - 0.85))) + forgiveness = self.opp_coops_after_my_D / max(1, self.my_D) + is_grim = self.opp_len > 50 and self.opp_defects == 0 + is_victim = forgiveness > 0.6 or ( + self.opp_defects / max(1, self.opp_len) < 0.03 + ) + if p_end > 0.75 and is_victim and not is_grim: + return self._play(D) + + # --- Queued delayed retaliation ---------------------------------- + if step in self.queue: + self.queue.remove(step) + self.debt = max(0, self.debt - 1) + self._close_epoch() + return self._play(D) + + # --- Default: cooperate & advance epoch -------------------------- + self.epoch_step += 1 + self._close_epoch() + return self._play(C) + + def _play(self, action: Action) -> Action: + self.last_my = action + if action == D: + self.my_D += 1 + return action + + def _on_defect(self, step: int) -> None: + """Record an opponent defection and schedule / escalate response.""" + if self.debt > 0 or self.queue or self._state == _State.EQUALIZING: + self.systemic += 1 + + self.debt += 1 + self._state = _State.EQUALIZING + + # Dynamic red-line threshold (tighter under hostility). + # Default: 3 systemic defects; after the first systemic event (or + # soft hostility / late defects) the threshold tightens to 2. + threshold = 3 + if ( + self.systemic >= 1 + or self._is_soft_hostile() + or self.late_defects > 0 + ): + threshold = 2 + + if self.systemic >= threshold: + self._enter_red_line() + return + + # Adaptive buffer: near-immediate under pressure, else stochastic + if self._is_hostile() or self.late_defects > 0: + delay = 1 + else: + # numpy RandomState.randint is high-exclusive → use (1, 11) + delay = 5 + int(self._random.randint(1, 11)) + + self.queue.append(step + delay) + + def _close_epoch(self) -> None: + """Reset systemic counters when a clean epoch completes.""" + if self.is_red_line or self._state == _State.RED_LINE: + return + if ( + self.epoch_step >= self.base_epoch + and self.debt <= 0 + and not self.queue + ): + self.epoch_step = 0 + self.systemic = 0 + self._state = _State.COOPERATIVE diff --git a/axelrod/strategies/zeroresp_v2.py b/axelrod/strategies/zeroresp_v2.py new file mode 100644 index 000000000..3a649f668 --- /dev/null +++ b/axelrod/strategies/zeroresp_v2.py @@ -0,0 +1,324 @@ +""" +ZeroResp v2 (revision 2.2): adaptive IPD strategy. + +Tournament-tuned state machine with short adaptive retaliation, noise-aware +forgiveness, deadlock recovery, red-line ban, and smart end-game logic. +""" + +from __future__ import annotations + +from enum import Enum, auto +from typing import List, Optional, Tuple + +from axelrod.action import Action +from axelrod.player import Player + +C, D = Action.C, Action.D + + +class _State(Enum): + """Internal finite-state labels.""" + + COOPERATIVE = auto() + EQUALIZING = auto() + RED_LINE = auto() + + +class ZeroRespV2(Player): + """ + ZeroResp v2 — tournament-tuned adaptive state machine (revision 2.2). + + Keeps the ZeroResp backbone (dynamic epochs, systemic red line, anti-raider) + and adds: + + 1. **Early sharp** — in the first few turns, retaliate with delay 1 so + probers that check early punishment are answered. + 2. **Short adaptive buffer** — mid-game delay is 1–2 turns (not a long + 6–16 window), collapsing to 1 under hostility or late pressure. + 3. **Contrition / echo-shield** — after a queued retaliatory D, ignore one + mirror D to avoid cascade wars with Suspicious TFT-style reciprocators. + 4. **Noise-aware one-shot forgive** — isolated mid-game D after a long clean + mutual-cooperation stretch is treated as noise (up to twice per match). + 5. **Deadlock break** — CD/DC alternating loops force a cooperative reset + (Omega-TFT inspired) without permanent softness. + 6. **Smart end-game harvest / grim probe** — only with known finite length + and a short remaining horizon; cautious against never-defectors. + + Names: + + - ZeroResp v2: Library display name + - ZeroResp v2.2: Implementation revision (this module) + - ZeroRespV2: Class identifier + """ + + name = "ZeroResp v2" + classifier = { + "memory_depth": float("inf"), + "stochastic": True, + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": False, + } + + _DEFAULT_MATCH_LENGTH = 200 + _LATE_FRACTION = 0.75 + _LIVE_INTEL_MIN_SAMPLES = 10 + _HOSTILE_COOP_THRESHOLD = 0.4 + _SOFT_HOSTILE_COOP = 0.7 + + EARLY_WINDOW = 5 + ONE_SHOT_PEACE = 12 + DEADLOCK_THRESHOLD = 3 + + # revision 2.2 tuning + HARVEST_WINDOW = 5 + HARVEST_FORGIVENESS = 0.8 + ONE_SHOT_MAX = 2 + GRIM_LAST_SAFE = 1 + PROBE_WINDOW = 4 + PROBE_PROB = 0.15 + + def __init__(self, base_epoch: int = 25) -> None: + """Initialise epoch, red-line, and v2 counters.""" + super().__init__() + self.base_epoch = int(base_epoch) + self._init_state() + + def _init_state(self) -> None: + self._state = _State.COOPERATIVE + self.is_red_line = False + self.epoch_step = 0 + self.debt = 0 + self.systemic = 0 + self.queue: List[int] = [] + + self.opp_len = 0 + self.opp_defects = 0 + self.opp_coops_after_my_D = 0 + self.my_D = 0 + self.last_my: Action = C + self.late_defects = 0 + + self.echo_forgive = 0 + self.one_shot_forgives = 0 + self.one_shot_used = False + self.clean_peace = 0 + self.deadlock = 0 + self._last_pair: Optional[Tuple[Action, Action]] = None + self.probe_fired = False + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _match_length(self) -> Optional[int]: + """Return known finite match length, else ``None``.""" + length = self.match_attributes["length"] + if length is None or length in (-1, float("inf")): + return None + try: + length_int = int(length) + except (TypeError, ValueError): + return None + return length_int if length_int > 0 else None + + def _effective_length(self) -> int: + return self._match_length() or self._DEFAULT_MATCH_LENGTH + + def _late_threshold(self) -> int: + return int(self._effective_length() * self._LATE_FRACTION) + + def _live_coop_rate(self) -> float: + if self.opp_len == 0: + return 1.0 + return 1.0 - (self.opp_defects / self.opp_len) + + def _is_hostile(self) -> bool: + return ( + self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES + and self._live_coop_rate() < self._HOSTILE_COOP_THRESHOLD + ) + + def _is_soft_hostile(self) -> bool: + return ( + self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES + and self._live_coop_rate() < self._SOFT_HOSTILE_COOP + ) + + def _enter_red_line(self) -> None: + self._state = _State.RED_LINE + self.is_red_line = True + self.queue.clear() + self.debt = 0 + self.echo_forgive = 0 + + # ------------------------------------------------------------------ + # Core strategy + # ------------------------------------------------------------------ + + def strategy(self, opponent: Player) -> Action: + """Select C or D for the current turn.""" + step = len(self.history) + 1 + + if opponent.history: + opp_last = opponent.history[-1] + my_prev = self.last_my + self.opp_len += 1 + + if opp_last == D: + self.opp_defects += 1 + if step > self._late_threshold(): + self.late_defects += 1 + self._on_defect(step) + self.clean_peace = 0 + else: + if my_prev == D: + self.opp_coops_after_my_D += 1 + if my_prev == C: + self.clean_peace += 1 + else: + self.clean_peace = 0 + + pair = (my_prev, opp_last) + if self._last_pair is not None: + a0, b0 = self._last_pair + if (a0, b0) == (C, D) and pair == (D, C): + self.deadlock += 1 + elif (a0, b0) == (D, C) and pair == (C, D): + self.deadlock += 1 + elif a0 == b0 == C: + self.deadlock = 0 + elif a0 == D and b0 == D: + self.deadlock = 0 + self._last_pair = pair + + if self._is_hostile(): + self._enter_red_line() + + if self.late_defects >= 2: + self._enter_red_line() + return self._play(D) + + if self.is_red_line or self._state == _State.RED_LINE: + return self._play(D) + + # Deadlock break: force C and soft-reset equalizing + if self.deadlock >= self.DEADLOCK_THRESHOLD and not self.is_red_line: + self.deadlock = 0 + self.queue.clear() + self.debt = 0 + self.echo_forgive = 0 + self._state = _State.COOPERATIVE + self.systemic = max(0, self.systemic - 1) + self.epoch_step += 1 + self._close_epoch() + return self._play(C) + + # Smart harvest + grim probe (known finite horizon only) + known_len = self._match_length() + if known_len is not None and self.opp_len > 30: + remaining = known_len - step + if remaining < self.HARVEST_WINDOW: + forgiveness = self.opp_coops_after_my_D / max(1, self.my_D) + low_defect = (self.opp_defects / max(1, self.opp_len)) < 0.02 + high_forgive = forgiveness > self.HARVEST_FORGIVENESS + is_victim = high_forgive or low_defect + is_grim = self.opp_len > 50 and self.opp_defects == 0 + + if is_grim: + if remaining <= self.GRIM_LAST_SAFE: + return self._play(D) + if self.probe_fired: + return self._play(D) + if remaining <= self.PROBE_WINDOW and self.my_D == 0: + if self._random.random() < self.PROBE_PROB: + self.probe_fired = True + return self._play(D) + else: + if self.probe_fired and remaining <= self.PROBE_WINDOW: + return self._play(D) + if is_victim: + return self._play(D) + + # Queued delayed retaliation + if step in self.queue: + self.queue.remove(step) + self.debt = max(0, self.debt - 1) + self.echo_forgive = max(self.echo_forgive, 1) + self._close_epoch() + return self._play(D) + + self.epoch_step += 1 + self._close_epoch() + return self._play(C) + + def _play(self, action: Action) -> Action: + self.last_my = action + if action == D: + self.my_D += 1 + return action + + def _on_defect(self, step: int) -> None: + """Record opponent defection; apply v2 shields then schedule / ban.""" + # Contrition: ignore mirror D after our retaliatory fire + if self.echo_forgive > 0: + self.echo_forgive -= 1 + return + + # Noise-aware forgive after long clean peace + if ( + self.one_shot_forgives < self.ONE_SHOT_MAX + and self.opp_defects <= 3 + and step > self.EARLY_WINDOW + and self.clean_peace >= self.ONE_SHOT_PEACE + and self.debt <= 0 + and not self.queue + and self._state == _State.COOPERATIVE + and self.late_defects == 0 + and not self._is_soft_hostile() + ): + self.one_shot_forgives += 1 + self.one_shot_used = True + return + + if self.debt > 0 or self.queue or self._state == _State.EQUALIZING: + self.systemic += 1 + + self.debt += 1 + self._state = _State.EQUALIZING + + threshold = 3 + if ( + self.systemic >= 1 + or self._is_soft_hostile() + or self.late_defects > 0 + ): + threshold = 2 + + if self.systemic >= threshold: + self._enter_red_line() + return + + # Early sharp / hostile: delay 1; else short stochastic delay 1–2 + if self._is_hostile() or self.late_defects > 0: + delay = 1 + elif step <= self.EARLY_WINDOW or self.opp_len <= 3: + delay = 1 + else: + delay = 1 + int(self._random.randint(0, 1)) + + self.queue.append(step + delay) + + def _close_epoch(self) -> None: + """Reset systemic counters when a clean epoch completes.""" + if self.is_red_line or self._state == _State.RED_LINE: + return + if ( + self.epoch_step >= self.base_epoch + and self.debt <= 0 + and not self.queue + ): + self.epoch_step = 0 + self.systemic = 0 + self._state = _State.COOPERATIVE diff --git a/axelrod/tests/strategies/test_zeroresp.py b/axelrod/tests/strategies/test_zeroresp.py new file mode 100644 index 000000000..20540bec1 --- /dev/null +++ b/axelrod/tests/strategies/test_zeroresp.py @@ -0,0 +1,340 @@ +"""Tests for the ZeroResp strategy.""" + +import axelrod as axl + +from .test_player import TestPlayer + +C, D = axl.Action.C, axl.Action.D + + +class TestZeroResp(TestPlayer): + + name = "ZeroResp" + player = axl.ZeroResp + expected_classifier = { + "memory_depth": float("inf"), + "stochastic": True, + "makes_use_of": {"length"}, + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": False, + } + + def test_initial_move_is_always_c(self): + """Initial move is always C against any opponent.""" + for opponent in ( + axl.Cooperator(), + axl.Defector(), + axl.TitForTat(), + axl.Alternator(), + axl.Random(), + ): + player = self.player() + player.set_seed(0) + self.assertEqual(player.strategy(opponent), C) + self.assertFalse(player.is_red_line) + self.assertEqual(player.queue, []) + + def test_vs_cooperator(self): + """Never-defectors are grim-safe: full cooperation on known length.""" + actions = [(C, C)] * 50 + self.versus_test( + axl.Cooperator(), + expected_actions=actions, + match_attributes={"length": 200}, + seed=1, + attrs={"is_red_line": False, "debt": 0, "queue": []}, + ) + + def test_single_defect_queues_buffer_retaliation(self): + """ + A single D does not trigger immediate retaliation; a delayed D is + queued (buffer = 5 + U{1..10}) and fires later. + """ + # seed=1 → first delay draw yields a known schedule: + # turn 1: (C, D); turn 2 processes D and queues step 2 + delay. + # Under non-hostile samples delay ∈ [6, 15], so turn 2 is still C. + player = self.player() + opponent = axl.MockPlayer(actions=[D] + [C] * 30) + match = axl.Match( + (player, opponent), + turns=2, + seed=1, + match_attributes={"length": 200}, + ) + result = match.play() + self.assertEqual(result[0], (C, D)) + self.assertEqual(result[1], (C, C)) # buffered — not immediate D + self.assertFalse(player.is_red_line) + self.assertEqual(player.debt, 1) + self.assertEqual(len(player.queue), 1) + scheduled = player.queue[0] + self.assertGreaterEqual(scheduled, 2 + 6) # 5 + min U{1..10} + self.assertLessEqual(scheduled, 2 + 15) # 5 + max U{1..10} + + # Continue until the queued strike fires: exactly one delayed D. + player2 = self.player() + opponent2 = axl.MockPlayer(actions=[D] + [C] * 40) + match2 = axl.Match( + (player2, opponent2), + turns=30, + seed=1, + match_attributes={"length": 200}, + ) + match2.play() + self.assertEqual(player2.history[0], C) + self.assertEqual(player2.defections, 1) + self.assertFalse(player2.is_red_line) + # After the single strike clears debt/queue, no permanent ban. + self.assertEqual(player2.queue, []) + + def test_three_systemic_defects_trigger_red_line(self): + """ + Three opponent defections while debt/queue is open raise the + systemic counter and set is_red_line permanently True. + """ + # Consecutive D keeps debt open between events: + # 1st D opens debt (not yet systemic), 2nd → systemic=1, + # 3rd → systemic=2 ≥ threshold 2 → RED_LINE. + player = self.player() + opponent = axl.MockPlayer(actions=[D] * 10) + match = axl.Match( + (player, opponent), + turns=10, + seed=7, + match_attributes={"length": 200}, + ) + match.play() + self.assertTrue(player.is_red_line) + # After red line, remaining replies are unconditional D. + # First move C; after third processed D (around turn 4) permanent D. + self.assertEqual(player.history[0], C) + self.assertGreaterEqual(player.defections, 5) + # Permanent: still red-lined at end of match. + self.assertTrue(player.is_red_line) + self.assertEqual(player.queue, []) + + # versus_test form with attrs check at end of match. + # Turns 1–3: C while debt accumulates; turn 4+: permanent D (red line). + self.versus_test( + axl.Defector(), + expected_actions=[(C, D)] * 3 + [(D, D)] * 7, + turns=10, + seed=0, + match_attributes={"length": 200}, + attrs={"is_red_line": True}, + ) + + def test_reset_cleans_state_for_multi_rep_tournaments(self): + """reset() restores a clean match state (multi-rep tournaments).""" + player = self.player() + clone = player.clone() + opponent = axl.Defector() + match = axl.Match( + (player, opponent), + turns=20, + seed=11, + match_attributes={"length": 200}, + ) + match.play() + self.assertGreater(len(player.history), 0) + self.assertTrue( + player.is_red_line or player.debt > 0 or player.defections > 0 + ) + + player.reset() + self.assertEqual(player, clone) + self.assertEqual(len(player.history), 0) + self.assertFalse(player.is_red_line) + self.assertEqual(player.debt, 0) + self.assertEqual(player.systemic, 0) + self.assertEqual(player.queue, []) + self.assertEqual(player.epoch_step, 0) + self.assertEqual(player.opp_len, 0) + self.assertEqual(player.opp_defects, 0) + self.assertEqual(player.my_D, 0) + self.assertEqual(player.late_defects, 0) + self.assertEqual(player.last_my, C) + + # Second match after reset still starts with C + match2 = axl.Match( + (player, axl.Cooperator()), + turns=5, + seed=3, + match_attributes={"length": 200}, + ) + result = match2.play() + self.assertEqual(result[0], (C, C)) + self.assertFalse(player.is_red_line) + + def test_vs_tit_for_tat_cooperates(self): + actions = [(C, C)] * 20 + self.versus_test( + axl.TitForTat(), + expected_actions=actions, + match_attributes={"length": 200}, + seed=2, + attrs={"is_red_line": False}, + ) + + def test_seed_reproducible(self): + actions = None + for _ in range(2): + player = self.player() + opponent = axl.Defector() + match = axl.Match( + (player, opponent), + turns=25, + seed=42, + match_attributes={"length": 200}, + ) + result = match.play() + if actions is None: + actions = result + else: + self.assertEqual(result, actions) + + def test_unknown_length_vs_cooperator(self): + """Unknown / infinite length: no end-game harvest of pure C.""" + actions = [(C, C)] * 40 + self.versus_test( + axl.Cooperator(), + expected_actions=actions, + match_attributes={"length": float("inf")}, + seed=5, + attrs={"is_red_line": False}, + ) + + def test_match_length_edge_cases(self): + """Cover _match_length branches: invalid, non-positive, None.""" + player = self.player() + player.set_match_attributes(length=None) + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length=-1) + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length=float("inf")) + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length=0) + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length="not-a-number") + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length=[200]) # TypeError on int() + self.assertIsNone(player._match_length()) + + player.set_match_attributes(length=200) + self.assertEqual(player._match_length(), 200) + self.assertEqual(player._effective_length(), 200) + + # Fallback when length unknown + player.set_match_attributes(length=-1) + self.assertEqual( + player._effective_length(), player._DEFAULT_MATCH_LENGTH + ) + + def test_live_coop_rate_empty_history(self): + """With no observations, live coop rate defaults to 1.0.""" + player = self.player() + self.assertEqual(player.opp_len, 0) + self.assertEqual(player._live_coop_rate(), 1.0) + + def test_close_epoch_while_red_lined(self): + """_close_epoch is a no-op once is_red_line is set.""" + from axelrod.strategies.zeroresp import _State + + player = self.player() + player.is_red_line = True + player._state = _State.RED_LINE + player.epoch_step = 100 + player.systemic = 5 + player._close_epoch() + # Counters must not reset under red line + self.assertEqual(player.epoch_step, 100) + self.assertEqual(player.systemic, 5) + self.assertTrue(player.is_red_line) + + def test_epoch_resets_after_clean_window(self): + """After base_epoch clean turns, systemic counter resets.""" + player = self.player() + player.set_seed(0) + # Force equalising history then clear debt/queue and advance epoch + player.systemic = 1 + player.debt = 0 + player.queue = [] + player.epoch_step = player.base_epoch + player._close_epoch() + self.assertEqual(player.epoch_step, 0) + self.assertEqual(player.systemic, 0) + + def test_endgame_harvest_vs_soft_victim(self): + """ + Near a known match end, defect against a near-pure cooperator that + defected once early (victim, not grim). + """ + length = 100 + # One early D, then all C — low defect rate, not grim. + opp_actions = [D] + [C] * (length - 1) + player = self.player() + opponent = axl.MockPlayer(actions=opp_actions) + match = axl.Match( + (player, opponent), + turns=length, + seed=3, + match_attributes={"length": length}, + ) + match.play() + # Late turns should include harvest D (p_end high, opp_len > 50) + late = list(player.history[-8:]) + self.assertIn(D, late) + + def test_anti_raider_late_defects(self): + """Two late-game opponent defects trigger immediate red line.""" + length = 40 + # All C until late phase, then two Ds past 75% mark (threshold=30) + opp_actions = [C] * 31 + [D, D] + [C] * 10 + player = self.player() + opponent = axl.MockPlayer(actions=opp_actions) + match = axl.Match( + (player, opponent), + turns=len(opp_actions), + seed=4, + match_attributes={"length": length}, + ) + match.play() + self.assertTrue(player.is_red_line) + self.assertGreaterEqual(player.late_defects, 2) + + def test_hostile_short_delay_and_live_ban(self): + """High defect rate → hostile path (delay=1 / red line).""" + player = self.player() + # 10+ samples with coop_rate < 0.4 → _is_hostile + opponent = axl.MockPlayer(actions=[D] * 15 + [C] * 5) + match = axl.Match( + (player, opponent), + turns=20, + seed=9, + match_attributes={"length": 200}, + ) + match.play() + self.assertTrue(player.is_red_line) + + def test_forgiveness_cadastre_after_our_defect(self): + """Opponent C after our D increments opp_coops_after_my_D.""" + player = self.player() + # One early D from them → we queue strike; then they C while we may D + opponent = axl.MockPlayer(actions=[D] + [C] * 40) + match = axl.Match( + (player, opponent), + turns=30, + seed=1, + match_attributes={"length": 200}, + ) + match.play() + # After our delayed D, further opponent C should raise forgiveness count + self.assertGreaterEqual(player.opp_coops_after_my_D, 0) + self.assertGreaterEqual(player.my_D, 1) diff --git a/axelrod/tests/strategies/test_zeroresp_v2.py b/axelrod/tests/strategies/test_zeroresp_v2.py new file mode 100644 index 000000000..f3a8e63ca --- /dev/null +++ b/axelrod/tests/strategies/test_zeroresp_v2.py @@ -0,0 +1,414 @@ +"""Tests for the ZeroResp v2 strategy.""" + +import axelrod as axl +from axelrod.strategies.zeroresp_v2 import _State + +from .test_player import TestPlayer + +C, D = axl.Action.C, axl.Action.D + + +class TestZeroRespV2(TestPlayer): + + # Player.__repr__ appends init kwargs → "ZeroResp v2: 25" (base_epoch) + name = "ZeroResp v2: 25" + player = axl.ZeroRespV2 + expected_classifier = { + "memory_depth": float("inf"), + "stochastic": True, + "makes_use_of": {"length"}, + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": False, + } + + def test_initial_move_is_always_c(self): + for opponent in ( + axl.Cooperator(), + axl.Defector(), + axl.TitForTat(), + axl.Alternator(), + ): + player = self.player() + player.set_seed(0) + self.assertEqual(player.strategy(opponent), C) + self.assertFalse(player.is_red_line) + + def test_vs_cooperator(self): + actions = [(C, C)] * 40 + self.versus_test( + axl.Cooperator(), + expected_actions=actions, + match_attributes={"length": 200}, + seed=1, + attrs={"is_red_line": False, "debt": 0, "queue": []}, + ) + + def test_early_sharp_queues_delay_one(self): + """First-window D schedules retaliation for the next turn (delay=1).""" + player = self.player() + opponent = axl.MockPlayer(actions=[D] + [C] * 10) + match = axl.Match( + (player, opponent), + turns=3, + seed=1, + match_attributes={"length": 200}, + ) + result = match.play() + self.assertEqual(result[0], (C, D)) + # Turn 2: process D at step=2 (still early) → queue step+1=3, play C + self.assertEqual(result[1], (C, C)) + # Turn 3: fire queued D (delay=1 early sharp) + self.assertEqual(result[2], (D, C)) + self.assertEqual(player.echo_forgive, 1) + self.assertEqual(player.queue, []) + + def test_echo_shield_ignores_mirror_after_retaliation(self): + """After our queued D, one mirror D is absorbed (no new debt).""" + player = self.player() + # Opp: D, C, D — we retaliate on turn 3; their D on turn 4 is echo + opponent = axl.MockPlayer(actions=[D, C, D, C, C, C, C, C]) + match = axl.Match( + (player, opponent), + turns=6, + seed=2, + match_attributes={"length": 200}, + ) + match.play() + # echo_forgive was set when we fired; mirror should not red-line + self.assertFalse(player.is_red_line) + + def test_one_shot_forgive_after_long_peace(self): + """First isolated mid-game D after long mutual C is forgiven once.""" + player = self.player() + # 15 mutual C, then one D, then C — one-shot should apply + opp_actions = [C] * 15 + [D] + [C] * 10 + opponent = axl.MockPlayer(actions=opp_actions) + match = axl.Match( + (player, opponent), + turns=len(opp_actions), + seed=3, + match_attributes={"length": 200}, + ) + match.play() + self.assertTrue(player.one_shot_used) + # Single noise D after peace: mostly still cooperate + self.assertGreaterEqual(player.cooperations, 20) + + def test_three_systemic_defects_trigger_red_line(self): + player = self.player() + opponent = axl.MockPlayer(actions=[D] * 12) + match = axl.Match( + (player, opponent), + turns=12, + seed=7, + match_attributes={"length": 200}, + ) + match.play() + self.assertTrue(player.is_red_line) + self.assertEqual(player.history[0], C) + self.assertGreaterEqual(player.defections, 5) + + # Early sharp: C,C then D-heavy; permanent red line by end + self.versus_test( + axl.Defector(), + expected_actions=[ + (C, D), + (C, D), + (D, D), + (D, D), + (C, D), + (D, D), + (D, D), + (D, D), + (D, D), + (D, D), + ], + turns=10, + seed=0, + match_attributes={"length": 200}, + attrs={"is_red_line": True}, + ) + + def test_reset_cleans_state(self): + player = self.player() + clone = player.clone() + axl.Match( + (player, axl.Defector()), + turns=15, + seed=5, + match_attributes={"length": 200}, + ).play() + player.reset() + self.assertEqual(player, clone) + self.assertFalse(player.is_red_line) + self.assertEqual(player.debt, 0) + self.assertEqual(player.queue, []) + self.assertEqual(player.echo_forgive, 0) + self.assertFalse(player.one_shot_used) + self.assertEqual(player.deadlock, 0) + self.assertEqual(player.clean_peace, 0) + + def test_vs_tit_for_tat_cooperates(self): + actions = [(C, C)] * 20 + self.versus_test( + axl.TitForTat(), + expected_actions=actions, + match_attributes={"length": 200}, + seed=2, + attrs={"is_red_line": False}, + ) + + def test_seed_reproducible(self): + r1 = axl.Match( + (self.player(), axl.Defector()), + turns=25, + seed=42, + match_attributes={"length": 200}, + ).play() + r2 = axl.Match( + (self.player(), axl.Defector()), + turns=25, + seed=42, + match_attributes={"length": 200}, + ).play() + self.assertEqual(r1, r2) + + def test_match_length_edge_cases(self): + player = self.player() + player.set_match_attributes(length=None) + self.assertIsNone(player._match_length()) + player.set_match_attributes(length=-1) + self.assertIsNone(player._match_length()) + player.set_match_attributes(length=float("inf")) + self.assertIsNone(player._match_length()) + player.set_match_attributes(length=0) + self.assertIsNone(player._match_length()) + player.set_match_attributes(length="bad") + self.assertIsNone(player._match_length()) + player.set_match_attributes(length=[200]) + self.assertIsNone(player._match_length()) + player.set_match_attributes(length=200) + self.assertEqual(player._match_length(), 200) + player.set_match_attributes(length=-1) + self.assertEqual( + player._effective_length(), player._DEFAULT_MATCH_LENGTH + ) + + def test_live_coop_rate_empty(self): + self.assertEqual(self.player()._live_coop_rate(), 1.0) + + def test_close_epoch_while_red_lined(self): + player = self.player() + player.is_red_line = True + player._state = _State.RED_LINE + player.epoch_step = 100 + player.systemic = 5 + player._close_epoch() + self.assertEqual(player.epoch_step, 100) + self.assertEqual(player.systemic, 5) + + def test_epoch_resets_after_clean_window(self): + player = self.player() + player.systemic = 1 + player.debt = 0 + player.queue = [] + player.epoch_step = player.base_epoch + player._close_epoch() + self.assertEqual(player.epoch_step, 0) + self.assertEqual(player.systemic, 0) + + def test_endgame_harvest_vs_soft_victim(self): + length = 100 + opp_actions = [D] + [C] * (length - 1) + player = self.player() + match = axl.Match( + (player, axl.MockPlayer(actions=opp_actions)), + turns=length, + seed=3, + match_attributes={"length": length}, + ) + match.play() + self.assertIn(D, list(player.history[-8:])) + + def test_anti_raider_late_defects(self): + length = 40 + opp_actions = [C] * 31 + [D, D] + [C] * 10 + player = self.player() + axl.Match( + (player, axl.MockPlayer(actions=opp_actions)), + turns=len(opp_actions), + seed=4, + match_attributes={"length": length}, + ).play() + self.assertTrue(player.is_red_line) + self.assertGreaterEqual(player.late_defects, 2) + + def test_hostile_live_ban(self): + player = self.player() + axl.Match( + (player, axl.MockPlayer(actions=[D] * 15 + [C] * 5)), + turns=20, + seed=9, + match_attributes={"length": 200}, + ).play() + self.assertTrue(player.is_red_line) + + def test_deadlock_break_forces_cooperate(self): + """CD/DC alternation raises deadlock and forces a cooperative reset.""" + player = self.player() + player.set_seed(0) + player.set_match_attributes(length=200) + # Manually drive deadlock counter then call strategy + player.deadlock = player.DEADLOCK_THRESHOLD + player.debt = 2 + player.queue = [99] + player.echo_forgive = 1 + player.systemic = 2 + opp = axl.Cooperator() + # empty histories → first move path with high deadlock + action = player.strategy(opp) + self.assertEqual(action, C) + self.assertEqual(player.deadlock, 0) + self.assertEqual(player.debt, 0) + self.assertEqual(player.queue, []) + self.assertEqual(player.echo_forgive, 0) + + def test_deadlock_meter_from_alternating_pairs(self): + """Alternating exploitation pairs increment deadlock.""" + player = self.player() + # Sequence that produces (C,D) then (D,C) patterns via TFT-like fight + # After early D we retaliate sharp; then CD/DC loops with Alternator-ish + opp_actions = [D, C, D, C, D, C, D, C, D, C] + axl.Match( + (player, axl.MockPlayer(actions=opp_actions)), + turns=len(opp_actions), + seed=1, + match_attributes={"length": 200}, + ).play() + # Either broke deadlock with C or still tracking — just exercise path + self.assertGreaterEqual(len(player.history), 10) + + def test_unknown_length_vs_cooperator(self): + actions = [(C, C)] * 40 + self.versus_test( + axl.Cooperator(), + expected_actions=actions, + match_attributes={"length": float("inf")}, + seed=5, + attrs={"is_red_line": False}, + ) + + def test_midgame_noise_forgive_and_buffer(self): + """After long clean peace, isolated D is forgiven; later D is buffered.""" + player = self.player() + # First D after peace → one-shot forgive; second → short schedule + opp = [C] * 15 + [D, C, C, D] + [C] * 20 + axl.Match( + (player, axl.MockPlayer(actions=opp)), + turns=len(opp), + seed=11, + match_attributes={"length": 200}, + ).play() + self.assertTrue(player.one_shot_used) + self.assertGreaterEqual(player.one_shot_forgives, 1) + + def test_dd_and_cc_reset_deadlock(self): + player = self.player() + player._last_pair = (D, D) + player.deadlock = 2 + player.set_seed(0) + player.set_match_attributes(length=200) + # Simulate observe path via strategy against Cooperator after fake hist + # Use internal path: set history-like state + player.last_my = C + player._last_pair = (C, C) + player.deadlock = 2 + opp = axl.MockPlayer(actions=[C]) + # play one turn + axl.Match( + (player, opp), turns=1, seed=0, match_attributes={"length": 200} + ).play() + # After CC pair, deadlock should reset in observe + self.assertEqual(player.deadlock, 0) + + def _endgame_player( + self, + *, + hist_len: int, + length: int, + opp_len: int, + opp_defects: int, + my_D: int, + probe_fired: bool, + seed: int = 0, + opp_coops_after_my_D: int = 0, + ): + """Build a player mid end-game harvest window (known finite length).""" + player = self.player() + player.set_seed(seed) + player.set_match_attributes(length=length) + for _ in range(hist_len): + player.history.append(C, C) + player.opp_len = opp_len + player.opp_defects = opp_defects + player.my_D = my_D + player.opp_coops_after_my_D = opp_coops_after_my_D + player.probe_fired = probe_fired + player.last_my = C + return player + + def test_grim_last_safe_defects(self): + """vs never-defector: last-turn safe harvest is D (rev 2.2 grim).""" + # remaining = length - step = 60 - 59 = 1 <= GRIM_LAST_SAFE + player = self._endgame_player( + hist_len=58, + length=60, + opp_len=55, + opp_defects=0, + my_D=0, + probe_fired=False, + ) + self.assertEqual(player.strategy(axl.Cooperator()), D) + + def test_grim_after_probe_keeps_defecting(self): + """Once probe_fired against pure cooperator, stay D in harvest window.""" + player = self._endgame_player( + hist_len=56, + length=60, + opp_len=55, + opp_defects=0, + my_D=1, + probe_fired=True, + ) + self.assertEqual(player.strategy(axl.Cooperator()), D) + + def test_grim_stochastic_probe_fires(self): + """With remaining in PROBE_WINDOW and my_D==0, probe can fire (seeded).""" + player = self._endgame_player( + hist_len=56, + length=60, + opp_len=55, + opp_defects=0, + my_D=0, + probe_fired=False, + seed=7, + ) + action = player.strategy(axl.Cooperator()) + self.assertEqual(action, D) + self.assertTrue(player.probe_fired) + + def test_non_grim_probe_fired_stays_d_in_window(self): + """Non-grim end-game: if probe already fired, defect in PROBE_WINDOW.""" + # opp_len <= 50 → not is_grim; high defect rate → not is_victim + player = self._endgame_player( + hist_len=56, + length=60, + opp_len=40, + opp_defects=10, + my_D=5, + probe_fired=True, + opp_coops_after_my_D=0, + ) + self.assertEqual(player.strategy(axl.Cooperator()), D) diff --git a/docs/how-to/classify_strategies.rst b/docs/how-to/classify_strategies.rst index c529ebc67..12881cf92 100644 --- a/docs/how-to/classify_strategies.rst +++ b/docs/how-to/classify_strategies.rst @@ -57,7 +57,7 @@ strategies:: ... } >>> strategies = axl.filtered_strategies(filterset) >>> len(strategies) - 88 + 90 Or, to find out how many strategies only use 1 turn worth of memory to make a decision:: @@ -90,7 +90,7 @@ length of each match of the tournament:: ... } >>> strategies = axl.filtered_strategies(filterset) >>> len(strategies) - 22 + 24 Note that in the filterset dictionary, the value for the 'makes_use_of' key must be a list. Here is how we might identify the number of strategies that use diff --git a/docs/index.rst b/docs/index.rst index 82b9f41b5..602ca2356 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -53,7 +53,7 @@ Count the number of available players:: >>> import axelrod as axl >>> len(axl.strategies) - 243 + 245 Create matches between two players:: diff --git a/docs/reference/strategy_index.rst b/docs/reference/strategy_index.rst index 9764d3082..5d306e826 100644 --- a/docs/reference/strategy_index.rst +++ b/docs/reference/strategy_index.rst @@ -118,3 +118,7 @@ Here are the docstrings of all the strategies in the library. :members: .. automodule:: axelrod.strategies.zero_determinant :members: +.. automodule:: axelrod.strategies.zeroresp + :members: +.. automodule:: axelrod.strategies.zeroresp_v2 + :members: