diff --git a/axelrod/data/all_classifiers.yml b/axelrod/data/all_classifiers.yml index fb0b6a51c..8d9899091 100644 --- a/axelrod/data/all_classifiers.yml +++ b/axelrod/data/all_classifiers.yml @@ -1,4 +1,4 @@ -$\\phi$: +$\phi$: inspects_source: false long_run_time: false makes_use_of: !!set {} @@ -6,7 +6,7 @@ $\\phi$: manipulates_state: false memory_depth: .inf stochastic: false -$\\pi$: +$\pi$: inspects_source: false long_run_time: false makes_use_of: !!set {} @@ -186,6 +186,14 @@ Bully: manipulates_state: false memory_depth: 1 stochastic: false +Burn Both Ends: + inspects_source: false + long_run_time: false + makes_use_of: !!set {} + manipulates_source: false + manipulates_state: false + memory_depth: 1 + stochastic: true Bush Mosteller: inspects_source: false long_run_time: false @@ -195,6 +203,14 @@ Bush Mosteller: manipulates_state: false memory_depth: .inf stochastic: true +CAPRI: + inspects_source: false + long_run_time: false + makes_use_of: !!set {} + manipulates_source: false + manipulates_state: false + memory_depth: 3 + stochastic: false Calculator: inspects_source: false long_run_time: false @@ -439,15 +455,15 @@ Evolved ANN 5 Noise 05: manipulates_state: false memory_depth: .inf stochastic: false -EvolvedAttention: +Evolved FSM 16: inspects_source: false - long_run_time: True + long_run_time: false makes_use_of: !!set {} manipulates_source: false manipulates_state: false - memory_depth: 200 + memory_depth: .inf stochastic: false -Evolved FSM 16: +Evolved FSM 16 Noise 05: inspects_source: false long_run_time: false makes_use_of: !!set {} @@ -455,7 +471,7 @@ Evolved FSM 16: manipulates_state: false memory_depth: .inf stochastic: false -Evolved FSM 16 Noise 05: +Evolved FSM 4: inspects_source: false long_run_time: false makes_use_of: !!set {} @@ -463,7 +479,7 @@ Evolved FSM 16 Noise 05: manipulates_state: false memory_depth: .inf stochastic: false -Evolved FSM 4: +Evolved FSM 6: inspects_source: false long_run_time: false makes_use_of: !!set {} @@ -479,6 +495,14 @@ Evolved HMM 5: manipulates_state: false memory_depth: 5 stochastic: true +EvolvedAttention: + inspects_source: false + long_run_time: true + makes_use_of: !!set {} + manipulates_source: false + manipulates_state: false + memory_depth: 200 + stochastic: false EvolvedLookerUp1_1_1: inspects_source: false long_run_time: false @@ -659,39 +683,23 @@ Fortress4: manipulates_state: false memory_depth: 3 stochastic: false -GTFT: +FrequencyAnalyzer: inspects_source: false long_run_time: false - makes_use_of: !!set - game: null - manipulates_source: false - manipulates_state: false - memory_depth: 1 - stochastic: true -Geller: - inspects_source: true - long_run_time: false - makes_use_of: !!set {} - manipulates_source: false - manipulates_state: false - memory_depth: .inf - stochastic: true -Geller Cooperator: - inspects_source: true - long_run_time: false makes_use_of: !!set {} manipulates_source: false manipulates_state: false memory_depth: .inf stochastic: false -Geller Defector: - inspects_source: true +GTFT: + inspects_source: false long_run_time: false - makes_use_of: !!set {} + makes_use_of: !!set + game: null manipulates_source: false manipulates_state: false - memory_depth: .inf - stochastic: false + memory_depth: 1 + stochastic: true General Soft Grudger: inspects_source: false long_run_time: false @@ -1099,47 +1107,14 @@ Michaelos: manipulates_state: false memory_depth: .inf stochastic: true -Mind Bender: - inspects_source: false - long_run_time: false - makes_use_of: !!set {} - manipulates_source: true - manipulates_state: false - memory_depth: -10 - stochastic: false -Mind Controller: +Momentum: inspects_source: false long_run_time: false makes_use_of: !!set {} - manipulates_source: true - manipulates_state: false - memory_depth: -10 - stochastic: false -Mind Reader: - inspects_source: true - long_run_time: false - makes_use_of: !!set - game: null manipulates_source: false manipulates_state: false memory_depth: .inf stochastic: false -Mind Warper: - inspects_source: false - long_run_time: false - makes_use_of: !!set {} - manipulates_source: true - manipulates_state: false - memory_depth: -10 - stochastic: false -Mirror Mind Reader: - inspects_source: true - long_run_time: false - makes_use_of: !!set {} - manipulates_source: true - manipulates_state: false - memory_depth: .inf - stochastic: false N Tit(s) For M Tat(s): inspects_source: false long_run_time: false @@ -1344,15 +1319,6 @@ Prober 4: manipulates_state: false memory_depth: .inf stochastic: false -Protected Mind Reader: - inspects_source: true - long_run_time: false - makes_use_of: !!set - game: null - manipulates_source: true - manipulates_state: false - memory_depth: .inf - stochastic: false Pun1: inspects_source: false long_run_time: false @@ -1733,6 +1699,14 @@ Spiteful Tit For Tat: manipulates_state: false memory_depth: .inf stochastic: false +SpitefulCC: + inspects_source: false + long_run_time: false + makes_use_of: !!set {} + manipulates_source: false + manipulates_state: false + memory_depth: .inf + stochastic: false Stalker: inspects_source: false long_run_time: false @@ -2040,3 +2014,20 @@ ZD-SET-2: manipulates_state: false memory_depth: 1 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 +ZeroResp v5.4.4: + inspects_source: false + long_run_time: false + makes_use_of: !!set {} + manipulates_source: false + manipulates_state: true + 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..9e7dee58d --- /dev/null +++ b/axelrod/strategies/zeroresp.py @@ -0,0 +1,1213 @@ +"""ZeroResp v5.4.4 — long-memory IPD player with a noise ladder. + +Built on the v5.3 noise-ladder basis: epoch debt, delayed retaliation, +noise-aware forgiveness, opening disambiguation, and finite-horizon harvest. +v5.4.x adds responsiveness-gated apology and a cycler contour (periods 2–6, +noise-tolerant, mirror-guarded). Feature flags default on; disable via +``Features`` for ablation. + +Compatible with ``axelrod.Match``. Class-level opponent profiles set +``manipulates_state`` to True. +""" + +from __future__ import annotations + +from collections import deque +from enum import Enum, auto +from typing import Dict, List, Optional + +from axelrod.action import Action +from axelrod.player import Player + +C, D = Action.C, Action.D + +# Standard PD payoffs (fallback for nl_balance). +PAYOFF = {(C, C): 3, (C, D): 0, (D, C): 5, (D, D): 1} + + +class _State(Enum): + COOPERATIVE = auto() + EQUALIZING = auto() + RED_LINE = auto() + RED_LINE_COOLDOWN = auto() + + +class Features: + __slots__ = ( + "profiles", + "apology", + "harvest_jitter", + "noise_extra", + "estimated_endgame", + "opening_d_retort", + "noise_adaptive", + "contrite_full", + "generous", + "red_line_cooldown", + "noise_ladder", + "adaptive_harvest", + "balance_apology", + "tactical_prefilter", + "profile_memory", + "cycler_contour", + ) + + def __init__( + self, + profiles: bool = True, + apology: bool = True, + harvest_jitter: bool = True, + noise_extra: bool = False, + estimated_endgame: bool = True, + opening_d_retort: bool = True, + noise_adaptive: bool = True, + contrite_full: bool = True, + generous: bool = True, + red_line_cooldown: bool = True, + noise_ladder: bool = True, + adaptive_harvest: bool = True, + balance_apology: bool = True, + tactical_prefilter: bool = True, + profile_memory: bool = True, + cycler_contour: bool = True, + ): + self.profiles = profiles + self.apology = apology + self.harvest_jitter = harvest_jitter + self.noise_extra = noise_extra + self.estimated_endgame = estimated_endgame + self.opening_d_retort = opening_d_retort + self.noise_adaptive = noise_adaptive + self.contrite_full = contrite_full + self.generous = generous + self.red_line_cooldown = red_line_cooldown + self.noise_ladder = noise_ladder + self.adaptive_harvest = adaptive_harvest + self.balance_apology = balance_apology + self.tactical_prefilter = tactical_prefilter + self.profile_memory = profile_memory + self.cycler_contour = cycler_contour + + def asdict(self): + return {k: getattr(self, k) for k in self.__slots__} + + def replace(self, **kw): + d = self.asdict() + d.update(kw) + return Features(**d) + + +class ZeroResp(Player): + name = "ZeroResp v5.4.4" + classifier = { + "memory_depth": float("inf"), + "stochastic": True, + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": True, + } + + _DEFAULT_MATCH_LENGTH = 200 + _LATE_FRACTION = 0.75 + _LIVE_INTEL_MIN_SAMPLES = 10 + _HOSTILE_COOP_THRESHOLD = 0.40 + _SOFT_HOSTILE_COOP = 0.65 + + EARLY_WINDOW = 5 + ONE_SHOT_PEACE = 12 + DEADLOCK_THRESHOLD = 3 + HARVEST_FORGIVENESS = 0.80 + HARVEST_WINDOW = 5 + ONE_SHOT_MAX = 2 + GRIM_LAST_SAFE = 1 + PROBE_WINDOW = 3 + FORGIVENESS_BASE = 3 + + NL_WINDOW_TURNS = 12 + NL_DELAY_MIN = 8 + NL_DELAY_MAX = 20 + NL_PARDON_STREAK = 3 + NL_HARVEST_MIN = 3 + NL_HARVEST_MAX = 4 + + NL_BALANCE_FLOOR = -4 + NL_SOFTNESS_FLOOR = 0.30 + + _global_profiles: Dict[str, Dict] = {} + _global_lengths: List[int] = [] + + def __init__( + self, + base_epoch: int = 25, + use_profiles: bool = True, + features: Features | None = None, + ): + super().__init__() + if not hasattr(self, "_random"): + self.set_seed(None) + self.base_epoch = int(base_epoch) + self.features = features if features is not None else Features() + self.use_profiles = bool(use_profiles) and self.features.profiles + self._init_state() + + def _init_state(self): + 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.my_D = 0 + self.my_C = 0 + self.opp_coops_after_my_D = 0 + self.opp_defects_after_my_C = 0 + self.opp_coops_after_my_C = 0 + self.opp_defects_after_my_D = 0 + self.last_my = C + self.late_defects = 0 + self.opp_last3 = deque(maxlen=4) + self.echo_forgive = 0 + self.one_shot_forgives = 0 + self.clean_peace = 0 + self.deadlock = 0 + self.probe_fired = False + self.warmth = self._random.uniform(0.40, 0.60) + self.target_warmth = 0.50 + self.intent = "normal" + self.tactical = "adaptive" + self._contrite = False + self._last_D_reason = None + self._consecutive_forgiven = 0 + self._apology_mode = False + self._apology_steps = 0 + self._apology_tries = 0 + self._apology_max_tries = 2 + self._first_D_was_noise = False + self._apology_opp_moves: List[Action] = [] + self._my_hist = deque(maxlen=3) + self._opp_hist = deque(maxlen=3) + self._forgiveness_exploiter = 0 + self._opening_d = False + self._known_soft_opponent = False + self._profile_checked = False + + self.cc_pairs = 0 + self.unprovoked_d = 0 + self.p_noise_est = 0.0 + self._fixed_period_latch = False + self._intended_prev = C + self._realized_prev = C + self._bad_standing = False + self._bad_standing_turns = 0 + self._red_line_cooldown = 0 + + self.nl_evidence = 0 + self.nl_window = deque(maxlen=24) + self.nl_scheduled: List[Dict] = [] + self.nl_debt_ledger = 0 + self.nl_pardoned = 0 + self.nl_balance = 0 + self.nl_harvest_left = 0 + self.nl_harvest_size = 0 + self.nl_harvest_postponed = 0 + self.nl_harvest_scheduled_at = 0 + self.nl_my_flips = 0 + self.nl_my_obs = 0 + self.nl_provoked_until = 0 + self.nl_pardon_step = 0 + self._pending_reason: Optional[str] = None + + if self.features.harvest_jitter: + self._harvest_window = int(self._random.randint(3, 6)) + else: + self._harvest_window = self.HARVEST_WINDOW + + def reset(self): + if self.history: + try: + self.__class__._global_lengths.append(len(self.history)) + if len(self.__class__._global_lengths) > 200: + self.__class__._global_lengths = ( + self.__class__._global_lengths[-200:] + ) + except Exception: + pass + old_p = self.__class__._global_profiles + old_l = self.__class__._global_lengths + feats = self.features + use_p = self.use_profiles + epoch = self.base_epoch + super().reset() + self.features = feats + self.use_profiles = use_p + self.base_epoch = epoch + self._init_state() + if self.use_profiles or self.features.estimated_endgame: + self.__class__._global_profiles = old_p + self.__class__._global_lengths = old_l + + # ---- helpers v5.1 ---- + def _match_length(self) -> Optional[int]: + length = self.match_attributes.get("length") + if length is None or length in (-1, float("inf")): + return None + try: + val = int(length) + return val if val > 0 else None + except Exception: + return None + + def _effective_length(self) -> int: + known = self._match_length() + if known is not None: + return known + obs = self.__class__._global_lengths + if self.features.estimated_endgame and obs: + return int(sorted(obs)[len(obs) // 2]) + return self._DEFAULT_MATCH_LENGTH + + def _late_threshold(self) -> int: + return int(self._effective_length() * self._LATE_FRACTION) + + def _live_coop_rate(self) -> float: + return ( + 1.0 + if self.opp_len == 0 + else 1.0 - (self.opp_defects / self.opp_len) + ) + + def _estimate_remaining(self, step: int) -> Optional[int]: + known = self._match_length() + if known is not None: + return max(0, known - step + 1) + if not self.features.estimated_endgame: + return None + obs = self.__class__._global_lengths + if len(obs) < 3: + return None + greater = [l for l in obs if l >= step] + if not greater: + return 1 + greater.sort() + return max(0, greater[len(greater) // 2] - step + 1) + + 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 _get_adaptive_epoch(self) -> int: + if self.warmth < 0.3: + return 10 + if self.warmth > 0.7: + return 40 + return self.base_epoch + + def _detect_intent(self) -> str: + if self.my_C == 0 or self.opp_len < 15: + return "normal" + r = self.opp_defects_after_my_C / max(1, self.my_C) + if r > 0.40: + return "exploitation" + if r < 0.25 and 0.02 < (self.opp_defects / max(1, self.opp_len)) < 0.25: + return "noise" + return "normal" + + def _classify_tactical(self) -> str: + if self.opp_len < 15: + return "adaptive" + cr = self._live_coop_rate() + if cr > 0.95: + return "cooperator" + if cr < 0.35: + return "aggressor" + eff = self._effective_length() + if self.opp_len > eff * 0.65: + early = self.opp_defects - self.late_defects + if early <= 3 and self.late_defects >= 1: + return "backstabber" + if ( + self.my_D > 0 + and self.opp_coops_after_my_D == 0 + and self.opp_defects > 5 + ): + return "grudger" + if self.deadlock >= 2 or (self.opp_len > 20 and 0.40 < cr < 0.60): + return "oscillator" + return "adaptive" + + def _detect_fixed_period(self, opponent) -> bool: + """Detect a fixed opponent period (cycler contour). + + Periods 2-6 over the last up to 18 opponent moves (min 12), + mismatch tolerance ~15% for channel noise. Period 1 is + excluded (looks like Grudger). Once latched, stays for the + rest of the match. + """ + if self._fixed_period_latch: + return True + hist = list(opponent.history) + if len(hist) < 12: + return False + window = hist[-18:] + n = len(window) + my_hist = list(self.history) + if len(my_hist) > n: + my_window = my_hist[-(n + 1) : -1] + agree = sum(1 for i in range(n) if window[i] == my_window[i]) + if agree >= int(n * 0.80): + return False + for p in range(2, 7): + cmp_pairs = n - p + if cmp_pairs < 8: + continue + mism = sum(1 for i in range(p, n) if window[i] != window[i - p]) + if mism <= max(1, int(cmp_pairs * 0.15)): + self._fixed_period_latch = True + return True + return False + + def _update_target_warmth(self): + if self.opp_len < 10: + return + ca = self.opp_coops_after_my_C / max(1, self.my_C) + od = self.opp_defects / max(1, self.opp_len) + md = self.my_D / max(1, self.opp_len) + t = ( + 0.5 + + 0.3 * (ca - 0.5) + - 0.4 * max(0.0, od - 0.02) + - 0.2 * max(0.0, md - od) + ) + if self.tactical == "cooperator": + t += 0.20 + elif self.tactical == "aggressor": + t -= 0.30 + elif self.tactical == "oscillator": + t -= 0.15 + elif ( + self.tactical == "backstabber" + and self.opp_len > self._effective_length() * 0.65 + ): + t -= 0.25 + if self._forgiveness_exploiter >= 2: + t -= 0.20 + self.target_warmth = max(0.10, min(0.90, t)) + + def _align_warmth(self, step: int): + if step % 5 == 0 and step <= 25: + self.warmth += (self.target_warmth - self.warmth) * 0.10 + elif step > 25 and step % self._get_adaptive_epoch() == 0: + self.warmth = max( + 0.10, + min( + 0.90, + self.warmth + (self.target_warmth - self.warmth) * 0.25, + ), + ) + + # ---- v5.2 noise ---- + def _is_noisy_regime(self) -> bool: + if not self.features.noise_adaptive: + return False + if self._nl_active(): + return ( + self.p_noise_est > 0.015 + and (self.cc_pairs + self.nl_my_obs) > 10 + ) + return self.p_noise_est > 0.015 and self.cc_pairs > 10 + + # ---- Noise ladder ---- + def _nl_noise_proven(self) -> bool: + return self.nl_my_flips > 0 + + def _nl_active(self) -> bool: + return ( + self.features.noise_ladder + and self._nl_noise_proven() + and self.nl_balance >= 0 + ) + + def _nl_is_unexplained(self) -> bool: + if self._bad_standing or self._contrite: + return False + if not self.history: + return True + if ( + self.nl_provoked_until + and self._nl_step_hint <= self.nl_provoked_until + ): + return False + return self.history[-1] == C + + def _nl_ladder(self, step: int) -> bool: + """Noise-ladder step on unexplained D. True means play C this turn.""" + if ( + self.nl_window + and (step - self.nl_window[-1]) > self.NL_WINDOW_TURNS + ): + self.nl_evidence = 0 + self.nl_window.clear() + self.nl_evidence += 1 + self.nl_window.append(step) + + if self.nl_pardon_step and (step - self.nl_pardon_step) <= 6: + self._forgiveness_exploiter += 1 + self.nl_pardon_step = 0 + + if self.nl_evidence == 1: + self.nl_pardoned += 1 + self.nl_debt_ledger += 1 + self._pending_reason = "nl_deferred" + return True + + if self.nl_evidence == 2: + self.nl_pardoned += 1 + self.nl_debt_ledger += 1 + self._pending_reason = "nl_deferred" + if not any(r.get("kind") == "strike" for r in self.nl_scheduled): + self.nl_scheduled.append( + { + "turn": step + + self._random.randint( + self.NL_DELAY_MIN, self.NL_DELAY_MAX + ), + "kind": "strike", + } + ) + forg = self.opp_coops_after_my_D / max(1, self.my_D) + min_d = self.my_D >= 3 + if self.features.profile_memory and self._known_soft_opponent: + min_d = self.my_D >= 1 + blocked_tactical = ( + self.features.tactical_prefilter + and self.tactical in ("aggressor", "grudger") + ) + if self.features.adaptive_harvest: + softness = (forg - self.NL_SOFTNESS_FLOOR) / ( + 1.0 - self.NL_SOFTNESS_FLOOR + ) + softness = max(0.0, min(1.0, softness)) + if ( + softness > 0 + and min_d + and not blocked_tactical + and self.nl_balance >= 0 + and not self._is_noisy_regime() + and self._random.random() < softness + ): + size = int( + round( + self.NL_HARVEST_MIN + + softness + * (self.NL_HARVEST_MAX - self.NL_HARVEST_MIN) + ) + ) + self.nl_harvest_size = max(self.NL_HARVEST_MIN, size) + self.nl_harvest_left = 0 + self.nl_harvest_postponed = 0 + self.nl_harvest_scheduled_at = step + self._random.randint( + 6, 10 + ) + else: + if ( + forg > 0.6 + and min_d + and not blocked_tactical + and self.nl_balance >= 0 + and not self._is_noisy_regime() + ): + self.nl_harvest_size = self._random.randint( + self.NL_HARVEST_MIN, self.NL_HARVEST_MAX + ) + self.nl_harvest_left = 0 + self.nl_harvest_postponed = 0 + self.nl_harvest_scheduled_at = step + self._random.randint( + 6, 10 + ) + return True + + if self.nl_evidence >= 3: + if self._live_coop_rate() < self._HOSTILE_COOP_THRESHOLD and ( + not self._is_noisy_regime() or self.nl_evidence >= 8 + ): + self._enter_red_line() + return False + return None + + def _enter_red_line(self): + if self.features.red_line_cooldown and self._is_noisy_regime(): + self._state = _State.RED_LINE_COOLDOWN + self.is_red_line = True + self._red_line_cooldown = int( + 12 + 20 * (1 - min(self.p_noise_est * 10, 0.9)) + ) + self.warmth = 0.15 + self.queue.clear() + self.debt = 0 + return + self._state = _State.RED_LINE + self.is_red_line = True + self.warmth = 0.0 + self.target_warmth = 0.0 + self.queue.clear() + self.debt = 0 + self.echo_forgive = 0 + self._apology_mode = False + + def _get_forgiveness_budget(self) -> int: + base = self.FORGIVENESS_BASE + jitter = self._random.randint(0, 2) + if self.warmth > 0.7: + jitter += 1 + if self.warmth < 0.4: + jitter -= 1 + if self.features.noise_extra and self.intent == "noise": + jitter += 4 + if self.tactical == "aggressor": + jitter = max(jitter - 2, 0) + if self._is_noisy_regime(): + jitter += int(self.p_noise_est * 40) + return max(2, min(12, base + jitter)) + + def _harvest_action( + self, remaining: int, opponent: Player + ) -> Optional[Action]: + if self._is_noisy_regime() and self.features.noise_adaptive: + if remaining > 2: + return None + forg = self.opp_coops_after_my_D / max(1, self.my_D) + low = (self.opp_defects / max(1, self.opp_len)) < 0.02 + high = forg > self.HARVEST_FORGIVENESS + is_victim = high or low + is_grim = self.opp_len > 50 and self.opp_defects == 0 + + if is_grim: + if remaining <= self.GRIM_LAST_SAFE: + return D + if self.probe_fired: + return D + return None + if self.probe_fired and remaining <= self.PROBE_WINDOW: + return D + if is_victim and self.tactical in ("cooperator", "adaptive"): + return D + return None + + def _opening_action(self, step: int, opponent: Player) -> Optional[Action]: + if not self.features.opening_d_retort or not opponent.history: + return None + if step == 2 and opponent.history[0] == D: + self._opening_d = True + self.queue.clear() + self.debt = 0 + self.systemic = 0 + self._state = _State.COOPERATIVE + return self._play(D, "opening_retort", intended=D) + if step == 3 and self._opening_d: + if opponent.history[1] == C: + self.queue.clear() + self.debt = 0 + self.systemic = 0 + self._state = _State.COOPERATIVE + self.echo_forgive = max(self.echo_forgive, 1) + return self._play(C, "opening_resync", intended=C) + self.queue.clear() + self.debt = max(self.debt, 1) + self._state = _State.EQUALIZING + return self._play(D, "opening_dd", intended=D) + return None + + # ---- core play ---- + def _play( + self, + action: Action, + reason: str = "normal", + intended: Optional[Action] = None, + ) -> Action: + if intended is None: + intended = action + self._intended_prev = intended + self._realized_prev = action + self.last_my = action + self._last_D_reason = reason if action == D else None + return action + + def _on_defect(self, step: int, opp_action: Action) -> bool: + """Strategy helper (see module docstring).""" + self._nl_step_hint = step + + # === v5.2 CONTRITE FULL === + if self.features.contrite_full: + if self._bad_standing: + self._bad_standing_turns -= 1 + if self._bad_standing_turns <= 0: + self._bad_standing = False + self._contrite = False + return True + + if ( + self._last_D_reason in ("harvest", "probe") + and self.tactical == "cooperator" + ): + self._last_D_reason = None + return True + + if self.echo_forgive > 0: + if not (self._nl_active() and self._nl_is_unexplained()): + self.echo_forgive -= 1 + return True + + # === v5.2 GENEROUS FORGIVENESS === + if self.features.generous and self._is_noisy_regime(): + if self._live_coop_rate() > 0.90 and self.opp_len > 20: + return True + + p_forgive = min( + 0.6, self.p_noise_est * 7 + 0.05 + (self.warmth - 0.5) * 0.2 + ) + if len(self.opp_last3) >= 3 and list(self.opp_last3)[-2:] == [D, D]: + p_forgive *= 0.3 + if self._random.random() < p_forgive: + return True + + maxf = self.ONE_SHOT_MAX + (1 if self.warmth > 0.60 else 0) + if self.features.noise_extra and self.intent == "noise": + maxf += 4 + if self._is_noisy_regime(): + maxf += 2 + + if ( + self.one_shot_forgives < maxf + and self.opp_defects <= 5 + 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() + and self.warmth > 0.35 + ): + self.one_shot_forgives += 1 + if self.features.noise_ladder: + self.nl_pardon_step = step + return True + + if self._nl_active() and self._nl_is_unexplained(): + handled = self._nl_ladder(step) + if handled is not None: + return handled + + if self.intent != "noise": + if self.debt > 0 or self.queue or self._state == _State.EQUALIZING: + self.systemic += 1 + self.debt += 1 + self._state = _State.EQUALIZING + + thr = 3 + if ( + self.warmth < 0.40 + or self._is_soft_hostile() + or self.late_defects > 0 + ): + thr = 2 + if self.intent == "exploitation": + thr = 2 + if self.features.noise_extra and self.intent == "noise": + thr = 12 + if self._is_noisy_regime(): + thr = 8 + if self.systemic >= thr: + self._enter_red_line() + return False + + delay = 1 + known = self._match_length() + if known and step + delay > known: + self.debt = max(0, self.debt - 1) + self.echo_forgive = max(self.echo_forgive, 1) + return False + self.queue.append(step + delay) + return False + + def _close_epoch(self): + if self.is_red_line and self._state == _State.RED_LINE: + return + if ( + self.epoch_step >= self._get_adaptive_epoch() + and self.debt <= 0 + and not self.queue + ): + self.epoch_step = 0 + self.systemic = 0 + if self._state != _State.RED_LINE_COOLDOWN: + self._state = _State.COOPERATIVE + + def _save_profile(self, opp_name: str): + if not self.use_profiles or not opp_name: + return + forgiveness = self.opp_coops_after_my_D / max(1, self.my_D) + self.__class__._global_profiles[opp_name] = { + "warmth": float(self.target_warmth), + "coop_rate": self._live_coop_rate(), + "tactical": self.tactical, + "defects": self.opp_defects, + "len": self.opp_len, + "p_noise": self.p_noise_est, + "forgiveness": forgiveness if self.my_D >= 3 else None, + } + + def strategy(self, opponent: Player) -> Action: + self._last_opponent_name = getattr(opponent, "name", "Opponent") + step = len(self.history) + 1 + remaining = self._estimate_remaining(step) + self._pending_reason = None + self._nl_step_hint = 0 + + if self.features.profile_memory and not self._profile_checked: + self._profile_checked = True + if self.use_profiles: + prof = self.__class__._global_profiles.get( + self._last_opponent_name + ) + if ( + prof + and prof.get("forgiveness") is not None + and prof["forgiveness"] > 0.6 + ): + self._known_soft_opponent = True + + if ( + self.features.balance_apology + and self.nl_balance <= self.NL_BALANCE_FLOOR + ): + # 1. 0% noise protection: never trigger balance_apology in deterministic matches + # 2. Overall coop rate floor: live_coop_rate >= 0.60 when history >= 15 + # 3. Responsiveness condition: P(C|our C) - P(C|our D) >= 0.35 when our D >= 2 + noise_attr = self.match_attributes.get("noise") + if self.features.cycler_contour and self._detect_fixed_period( + opponent + ): + can_apologize = False + elif noise_attr == 0.0: + can_apologize = False + elif self.opp_len >= 15 and self._live_coop_rate() < 0.60: + can_apologize = False + else: + my_hist = self.history + opp_hist = opponent.history + n_c = sum(1 for i in range(len(my_hist) - 1) if my_hist[i] == C) + n_d = sum(1 for i in range(len(my_hist) - 1) if my_hist[i] == D) + if n_d >= 2: + c_after_c = sum( + 1 + for i in range(len(my_hist) - 1) + if my_hist[i] == C and opp_hist[i + 1] == C + ) + c_after_d = sum( + 1 + for i in range(len(my_hist) - 1) + if my_hist[i] == D and opp_hist[i + 1] == C + ) + p_c_c = c_after_c / max(1, n_c) + p_c_d = c_after_d / max(1, n_d) + delta_p = p_c_c - p_c_d + can_apologize = delta_p >= 0.35 + else: + can_apologize = True + + if can_apologize: + self.nl_scheduled = [] + self.nl_harvest_left = 0 + self.nl_harvest_scheduled_at = 0 + self.nl_harvest_postponed = 0 + self.nl_harvest_size = 0 + self.nl_evidence = 0 + self.nl_window.clear() + self.echo_forgive = max(self.echo_forgive, 2) + self.warmth = max(self.warmth, 0.6) + self.target_warmth = max(self.target_warmth, 0.6) + + if self.history: + last_realized = self.history[-1] + # intended vs realized + if ( + last_realized != self._intended_prev + and last_realized == D + and self._intended_prev == C + ): + self._contrite = True + self._first_D_was_noise = self.my_D <= 1 + self._bad_standing = True + self._bad_standing_turns = 1 + if self.features.noise_ladder: + self.nl_my_flips += 1 + if self.features.noise_ladder and self._intended_prev == C: + self.nl_my_obs += 1 + if self.features.noise_ladder and opponent.history: + opp_last_move = opponent.history[-1] + game = self.match_attributes.get("game") + my_s = opp_s = None + if game is not None: + try: + my_s, opp_s = game.score((last_realized, opp_last_move)) + except Exception: + my_s = opp_s = None + if my_s is None: + my_s = PAYOFF[(last_realized, opp_last_move)] + opp_s = PAYOFF[(opp_last_move, last_realized)] + self.nl_balance += my_s - opp_s + if len(self.history) >= 2 and len(opponent.history) >= 2: + my_prev = self.history[-2] + opp_prev = opponent.history[-2] + if my_prev == C and opp_prev == C: + self.cc_pairs += 1 + if opponent.history[-1] == D: + self.unprovoked_d += 1 + if self.cc_pairs % 5 == 0 and self.cc_pairs > 10: + if self._nl_active(): + total = self.cc_pairs + self.nl_my_obs + est_n = self.unprovoked_d + self.nl_my_flips + else: + total, est_n = self.cc_pairs, self.unprovoked_d + self.p_noise_est = est_n / max(1, total) + + if self.features.apology and self._apology_mode: + if opponent.history: + opp_last = opponent.history[-1] + my_prev = self.last_my + self._apology_opp_moves.append(opp_last) + self.opp_len += 1 + self.opp_last3.append(opp_last) + if opp_last == D: + self.opp_defects += 1 + if my_prev == C: + self.opp_defects_after_my_C += 1 + if step > self._late_threshold(): + self.late_defects += 1 + self.clean_peace = 0 + else: + if my_prev == D: + self.opp_coops_after_my_D += 1 + if my_prev == C: + self.opp_coops_after_my_C += 1 + self.clean_peace += 1 + else: + self.clean_peace = 0 + if my_prev == D: + self.my_D += 1 + else: + self.my_C += 1 + + self._apology_steps -= 1 + if self._apology_steps <= 0: + if any(m == C for m in self._apology_opp_moves): + self._apology_mode = False + self._apology_tries = 0 + self._first_D_was_noise = False + self._contrite = False + self._apology_opp_moves = [] + self.debt = 0 + self.systemic = 0 + self.queue.clear() + self.echo_forgive = 1 + self._state = _State.COOPERATIVE + self.epoch_step = 0 + self.warmth = 0.60 + self.target_warmth = 0.60 + self.tactical = "adaptive" + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "apology_success", intended=C) + self._apology_tries += 1 + self._apology_opp_moves = [] + if self._apology_tries >= self._apology_max_tries: + self._enter_red_line() + self._save_profile(self._last_opponent_name) + return self._play(D, "red_line", intended=D) + self._apology_mode = False + else: + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "apology", intended=C) + + if self.features.apology: + apology_max = self._apology_max_tries + if self.use_profiles: + prof = self.__class__._global_profiles.get( + self._last_opponent_name + ) + if prof and prof.get("tactical") == "grudger": + apology_max = 4 + if ( + not self.is_red_line + and not self._apology_mode + and self._first_D_was_noise + and len(self.opp_last3) >= 2 + and list(self.opp_last3)[-2:] == [D, D] + and self._apology_tries < apology_max + and self.opp_len < 300 + ): + self._apology_mode = True + self._apology_steps = self._get_forgiveness_budget() + self._apology_opp_moves = [] + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "apology", intended=C) + + if opponent.history and self.opp_len < len(opponent.history): + opp_last = opponent.history[-1] + my_prev = self.last_my + + if self._bad_standing and self.features.contrite_full: + self.opp_len += 1 + self.opp_last3.append(opp_last) + if opp_last == D: + self.opp_defects += 1 + if step > self._late_threshold(): + self.late_defects += 1 + self.clean_peace = 0 + else: + self.clean_peace += 1 if my_prev == C else 0 + if my_prev == D: + self.opp_coops_after_my_D += 1 + if my_prev == D: + self.my_D += 1 + else: + self.my_C += 1 + self._bad_standing_turns -= 1 + if self._bad_standing_turns <= 0: + self._bad_standing = False + self._contrite = False + if self.opp_len % 5 == 0: + self.intent = self._detect_intent() + self.tactical = self._classify_tactical() + self._update_target_warmth() + self._align_warmth(step) + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "contrite_full", intended=C) + + if opp_last == D and self._contrite: + self.opp_len += 1 + self.opp_last3.append(opp_last) + self.opp_defects += 1 + if my_prev == C: + self.opp_defects_after_my_C += 1 + if step > self._late_threshold(): + self.late_defects += 1 + self.clean_peace = 0 + if my_prev == D: + self.my_D += 1 + else: + self.my_C += 1 + self._contrite = False + self._consecutive_forgiven += 1 + if self.opp_len % 5 == 0: + self.intent = self._detect_intent() + self.tactical = self._classify_tactical() + self._update_target_warmth() + self._align_warmth(step) + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "contrite", intended=C) + + self.opp_len += 1 + self.opp_last3.append(opp_last) + if opp_last == D: + self.opp_defects += 1 + if my_prev == C: + self.opp_defects_after_my_C += 1 + if step > self._late_threshold(): + self.late_defects += 1 + if my_prev == D: + self.opp_defects_after_my_D += 1 + self._on_defect(step, opp_last) + self.clean_peace = 0 + else: + if my_prev == D: + self.opp_coops_after_my_D += 1 + if my_prev == C: + self.opp_coops_after_my_C += 1 + self.clean_peace += 1 + else: + self.clean_peace = 0 + if my_prev == D: + self.my_D += 1 + else: + self.my_C += 1 + + # --- opening --- + opened = self._opening_action(step, opponent) + if opened is not None: + self._save_profile(self._last_opponent_name) + return opened + + if self.features.noise_ladder: + if ( + self.nl_harvest_scheduled_at != 0 + and step >= self.nl_harvest_scheduled_at + ): + if opponent.history and opponent.history[-1] == C: + self.nl_harvest_left = self.nl_harvest_size + self.nl_harvest_scheduled_at = 0 + self.nl_provoked_until = step + self.nl_harvest_size + 3 + elif self.nl_harvest_postponed < 2: + self.nl_harvest_postponed += 1 + self.nl_harvest_scheduled_at = step + 3 + else: + self.nl_harvest_scheduled_at = 0 + self.nl_harvest_size = 0 + if self.nl_harvest_left > 0: + self.nl_harvest_left -= 1 + return self._play(D, "nl_harvest_comp", intended=D) + + self.nl_scheduled = [r for r in self.nl_scheduled if r["turn"] >= step] + due = next((r for r in self.nl_scheduled if r["turn"] == step), None) + if due is not None: + self.nl_scheduled.remove(due) + last3 = list(self.opp_last3)[-self.NL_PARDON_STREAK :] + if ( + len(last3) >= self.NL_PARDON_STREAK + and all(mv == C for mv in last3) + and self.opp_defects_after_my_D == 0 + ): + self.nl_debt_ledger += 1 + self.nl_pardoned += 1 + self.nl_pardon_step = step + if step in self.queue: + self.queue.remove(step) + self.debt = max(0, self.debt - 1) + self._pending_reason = "nl_pardon" + else: + self.queue.append(step) + self.nl_debt_ledger = max(0, self.nl_debt_ledger - 1) + + if self.opp_len % 5 == 0: + self.intent = self._detect_intent() + self.tactical = self._classify_tactical() + self._update_target_warmth() + self._align_warmth(step) + + # --- red_line checks --- + if self.intent != "noise" and not self._is_noisy_regime(): + if len(self.opp_last3) >= 3 and list(self.opp_last3)[-3:] == [ + D, + D, + D, + ]: + nl_governs = self._nl_active() and all( + m == C for m in list(self.history)[-4:-1] + ) + if ( + not nl_governs + and not self._contrite + and not self._bad_standing + and self._consecutive_forgiven < 2 + and not self._apology_mode + ): + self._enter_red_line() + if self._state == _State.RED_LINE: + return self._play(D, "red_line", intended=D) + + if ( + self._is_hostile() + and not self._contrite + and not self._bad_standing + and not self._apology_mode + ): + if not self._is_noisy_regime(): + self._enter_red_line() + + if self.is_red_line: + if self._state == _State.RED_LINE_COOLDOWN: + if self._red_line_cooldown <= 0: + self._red_line_cooldown = int( + 15 + 10 * (1 - min(self.p_noise_est * 5, 0.8)) + ) + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(C, "red_line_probe", intended=C) + self._red_line_cooldown -= 1 + self._save_profile(self._last_opponent_name) + return self._play(D, "red_line_cooldown", intended=D) + return self._play(D, "red_line", intended=D) + + if self.deadlock >= self.DEADLOCK_THRESHOLD: + 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() + self._contrite = False + self._bad_standing = False + return self._play(C, "normal", intended=C) + + if ( + remaining is not None + and self.opp_len > 30 + and remaining <= self._harvest_window + ): + harvested = self._harvest_action(remaining, opponent) + if harvested is not None: + reason = "harvest" if harvested == D else "harvest_keep_C" + if self.probe_fired and harvested == D: + reason = "probe" + return self._play(harvested, reason, intended=harvested) + + if step in self.queue: + if self._contrite or self._bad_standing or self._apology_mode: + self.queue.remove(step) + self.queue.append(step + 1) + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play( + C, "contrite" if self._contrite else "apology", intended=C + ) + self.queue.remove(step) + self.debt = max(0, self.debt - 1) + echo_val = ( + 2 + if (self.features.noise_extra and self.intent == "noise") + else 1 + ) + if self._is_noisy_regime(): + echo_val = 2 + self.echo_forgive = max(self.echo_forgive, echo_val) + self._close_epoch() + self._save_profile(self._last_opponent_name) + return self._play(D, "retaliation", intended=D) + + self.epoch_step += 1 + self._close_epoch() + self._save_profile(self._last_opponent_name) + if self._contrite: + self._contrite = False + self._bad_standing = False + reason = ( + self._pending_reason + if self._pending_reason is not None + else "normal" + ) + self._pending_reason = None + return self._play(C, reason, intended=C) 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..4cc0d0c19 --- /dev/null +++ b/axelrod/tests/strategies/test_zeroresp.py @@ -0,0 +1,927 @@ +"""Tests for the ZeroResp strategy (v5.4.4).""" + +import inspect +import unittest + +import axelrod as axl +from axelrod.strategies.zeroresp import Features, ZeroResp + +from .test_player import TestPlayer + +C, D = axl.Action.C, axl.Action.D + + +def _play(player, opponent, turns, seed=0, length=None, noise=0.0): + attrs = None + if length is not None: + attrs = {"length": length} + match = axl.Match( + (player, opponent), + turns=turns, + seed=seed, + noise=noise, + match_attributes=attrs, + ) + return match.play() + + +class TestZeroResp(TestPlayer): + + # Player.__repr__ appends init kwargs (base_epoch, use_profiles). + name = "ZeroResp v5.4.4: 25, True" + player = axl.ZeroResp + expected_classifier = { + "memory_depth": float("inf"), + "stochastic": True, + # length/game are read via .get(), which the library scanner does + # not count. manipulates_state stays True (class-level profiles). + "makes_use_of": set(), + "long_run_time": False, + "inspects_source": False, + "manipulates_source": False, + "manipulates_state": True, + } + + def setUp(self): + ZeroResp._global_profiles = {} + ZeroResp._global_lengths = [] + super().setUp() + + def equality_of_players_test(self, p1, p2, seed, opponent): + # Warmth is drawn in __init__ from the module RNG, and Features + # compares by identity. Same skip as Darwin. + return True + + def test_reset_clone(self): + player = self.player() + player.reset() + self.assertEqual(len(player.history), 0) + self.assertEqual(player.debt, 0) + self.assertEqual(player.queue, []) + self.assertFalse(player.is_red_line) + + def test_clone_reproducible_play(self, seed=1, turns=10, noise=0): + """Same match seed against a pure defector is reproducible.""" + r1 = _play(self.player(), axl.Defector(), turns=30, seed=42, length=200) + r2 = _play(self.player(), axl.Defector(), turns=30, seed=42, length=200) + self.assertEqual(r1, r2) + + def test_reset_history_and_attributes(self): + player = self.player() + _play(player, axl.Defector(), turns=15, seed=5, length=200) + self.assertGreater(len(player.history), 0) + player.reset() + self.assertEqual(len(player.history), 0) + 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.my_D, 0) + self.assertEqual(player.nl_evidence, 0) + self.assertFalse(player.is_red_line) + + def test_name_and_classifier(self): + player = self.player() + self.assertEqual(player.name, "ZeroResp v5.4.4") + self.assertTrue(player.classifier["manipulates_state"]) + self.assertTrue(player.classifier["stochastic"]) + self.assertEqual(player.classifier["memory_depth"], float("inf")) + + def test_init_defaults(self): + player = self.player() + self.assertEqual(player.base_epoch, 25) + self.assertEqual(player.debt, 0) + self.assertEqual(player.systemic, 0) + self.assertEqual(player.queue, []) + self.assertEqual(player.epoch_step, 0) + self.assertEqual(len(player.history), 0) + self.assertTrue(player.features.noise_ladder) + + def test_init_custom_epoch(self): + player = self.player(base_epoch=10) + self.assertEqual(player.base_epoch, 10) + self.assertEqual(player.init_kwargs.get("base_epoch"), 10) + self.assertIn("10", repr(player)) + + def test_initial_move_is_always_c(self): + for opponent in ( + axl.Cooperator(), + axl.Defector(), + axl.TitForTat(), + axl.Alternator(), + axl.Random(), + ): + player = self.player() + self.assertEqual(player.strategy(opponent), C) + + def test_vs_cooperator_mostly_cooperates(self): + result = _play( + self.player(), axl.Cooperator(), turns=200, seed=1, length=200 + ) + my_actions = [a for a, _ in result] + self.assertEqual(my_actions[0], C) + mid = my_actions[:180] + self.assertGreaterEqual(mid.count(C) / len(mid), 0.95) + self.assertGreaterEqual(my_actions.count(C), 190) + + def test_vs_defector_eventual_defect(self): + result = _play( + self.player(), axl.Defector(), turns=40, seed=2, length=200 + ) + my_actions = [a for a, _ in result] + self.assertEqual(my_actions[0], C) + self.assertIn(D, my_actions) + self.assertGreaterEqual(my_actions.count(D), 10) + + def test_vs_tit_for_tat_mostly_cooperate(self): + result = _play( + self.player(), axl.TitForTat(), turns=50, seed=3, length=200 + ) + my_actions = [a for a, _ in result] + self.assertEqual(my_actions[0], C) + self.assertGreaterEqual(my_actions.count(C), 40) + + def test_different_seeds_can_differ(self): + def once(seed): + player = self.player() + opp = axl.MockPlayer(actions=[C] * 6 + [D] + [C] * 40) + return tuple( + a + for a, _ in _play(player, opp, turns=25, seed=seed, length=200) + ) + + unique = {once(s) for s in range(1, 40)} + self.assertGreaterEqual(len(unique), 1) + + def test_red_line_against_persistent_defector(self): + player = self.player() + _play( + player, + axl.MockPlayer(actions=[D] * 20), + turns=20, + seed=7, + length=200, + ) + self.assertGreaterEqual(player.defections, 5) + + def test_unknown_length_vs_cooperator(self): + result = _play( + self.player(), + axl.Cooperator(), + turns=200, + seed=11, + length=float("inf"), + ) + my_actions = [a for a, _ in result] + self.assertTrue(all(a == C for a in my_actions)) + + def test_opening_d_retort_on_second_turn(self): + player = self.player() + _play( + player, + axl.MockPlayer(actions=[D] + [C] * 30), + turns=4, + seed=99, + length=200, + ) + self.assertEqual(player.history[0], C) + self.assertEqual(player.history[1], D) + + def test_queued_retaliation_fires(self): + player = self.player() + _play( + player, + axl.MockPlayer(actions=[D] + [C] * 40), + turns=30, + seed=12, + length=200, + ) + self.assertEqual(player.history[0], C) + self.assertGreaterEqual(player.defections, 1) + self.assertLessEqual(player.defections, 5) + + def test_vs_alternator(self): + result = _play( + self.player(), axl.Alternator(), turns=40, seed=8, length=200 + ) + self.assertEqual(len(result), 40) + my_actions = [a for a, _ in result] + self.assertEqual(my_actions[0], C) + self.assertIn(D, my_actions) + + def test_makes_use_of_length_and_game_in_source(self): + src = inspect.getsource(ZeroResp) + self.assertIn('match_attributes.get("length")', src) + self.assertIn('match_attributes.get("game")', src) + + def test_smoke_tournament_beats_defector(self): + players = [ + self.player(), + axl.TitForTat(), + axl.Defector(), + axl.Cooperator(), + axl.Grudger(), + axl.Random(), + ] + tournament = axl.Tournament(players, turns=50, repetitions=1, seed=0) + results = tournament.play(progress_bar=False) + names = results.ranked_names + zr = next(n for n in names if n.startswith("ZeroResp")) + self.assertLess(names.index(zr), names.index("Defector")) + + +class TestZeroRespNoiseLadder(unittest.TestCase): + """Noise ladder helpers. Not a registered strategy of its own.""" + + def setUp(self): + ZeroResp._global_profiles = {} + ZeroResp._global_lengths = [] + + def _ladder(self, **over): + kw = {"noise_ladder": True} + kw.update(over) + return ZeroResp(features=Features(**kw)) + + def test_flag_off_ladder_dormant(self): + player = ZeroResp(features=Features(noise_ladder=False)) + _play( + player, + axl.MockPlayer(actions=[C] * 6 + [D] + [C] * 40), + turns=40, + seed=1, + length=200, + ) + self.assertEqual(player.nl_evidence, 0) + self.assertEqual(player.nl_scheduled, []) + self.assertEqual(player.nl_pardoned, 0) + + def test_deterministic_with_same_seed(self): + def once(seed): + player = self._ladder() + result = _play( + player, + axl.MockPlayer(actions=[C] * 5 + [D] + [C] * 50), + turns=60, + seed=seed, + length=200, + ) + return tuple(a for a, _ in result) + + self.assertEqual(once(7), once(7)) + + def test_single_defection_pardoned_under_channel_noise(self): + observed = False + for seed in range(1, 9): + player = self._ladder() + _play( + player, + axl.MockPlayer(actions=[C] * 5 + [D] + [C] * 60), + turns=70, + seed=seed, + noise=0.05, + length=200, + ) + if player.nl_my_flips >= 1 and player.nl_pardoned >= 1: + self.assertGreaterEqual(player.nl_debt_ledger, 1) + observed = True + break + self.assertTrue( + observed, "no pardon observed in seeds 1..8 at 5% noise" + ) + + def test_persistent_defector_hits_red_line(self): + player = self._ladder() + _play( + player, + axl.MockPlayer(actions=[D] * 30), + turns=30, + seed=4, + length=200, + ) + self.assertTrue(player.is_red_line) + + def test_ladder_dormant_at_zero_noise(self): + import hashlib + import json + + by_name = {} + for strategy in axl.strategies: + by_name[strategy.__name__] = strategy + by_name.setdefault(strategy.name, strategy) + + def traj(flag, opp_name): + player = ZeroResp(features=Features(noise_ladder=flag)) + _play( + player, + by_name[opp_name](), + turns=200, + seed=42, + length=200, + ) + blob = json.dumps([str(a) for a in player.history]).encode() + return hashlib.sha256(blob).hexdigest() + + for opp in ("Alternator", "TrickyCooperator", "Calculator", "Random"): + self.assertEqual( + traj(True, opp), + traj(False, opp), + "ladder must be dormant at zero noise vs {}".format(opp), + ) + + def test_ladder_defers_first_unexplained_defection(self): + player = self._ladder() + player.nl_my_flips = 1 + player.opp_len, player.opp_defects = 10, 0 + player._nl_ladder(20) + self.assertEqual(player.nl_pardoned, 1) + self.assertEqual(player.nl_debt_ledger, 1) + self.assertEqual(player._pending_reason, "nl_deferred") + self.assertEqual(player.nl_scheduled, []) + self.assertEqual(player.queue, []) + player._nl_ladder(21) + self.assertEqual(player.nl_evidence, 2) + self.assertEqual(player.nl_pardoned, 2) + self.assertEqual(player.nl_debt_ledger, 2) + self.assertEqual(player._pending_reason, "nl_deferred") + self.assertEqual(player.queue, []) + self.assertEqual(len(player.nl_scheduled), 1) + self.assertEqual(player.nl_scheduled[0]["kind"], "strike") + self.assertGreaterEqual( + player.nl_scheduled[0]["turn"], 21 + player.NL_DELAY_MIN + ) + + def test_strikes_only_via_scheduled_pardonable_plan(self): + for opp_len, opp_defects in ((60, 3), (60, 30), (60, 48)): + player = self._ladder() + player.nl_my_flips = 1 + player.opp_len, player.opp_defects = opp_len, opp_defects + for step in range(20, 30): + player._pending_reason = None + player._nl_ladder(step) + self.assertEqual(player.queue, []) + self.assertLessEqual(len(player.nl_scheduled), 1) + + def test_ladder_redlines_hostile_at_three(self): + player = self._ladder() + player.nl_my_flips = 1 + player.opp_len, player.opp_defects = 60, 48 + player._nl_ladder(20) + player._nl_ladder(21) + player._nl_ladder(22) + self.assertTrue(player.is_red_line) + + def test_own_flips_feed_noise_est(self): + observed = False + for seed in range(1, 9): + player = self._ladder() + _play( + player, + axl.MockPlayer(actions=[C] * 80), + turns=80, + seed=seed, + noise=0.1, + length=200, + ) + if player.nl_my_flips >= 1: + self.assertGreater(player.p_noise_est, 0.0) + observed = True + break + self.assertTrue( + observed, "no own flip observed in seeds 1..8 at 10% noise" + ) + + def test_balance_negative_vs_defector(self): + player = self._ladder() + _play(player, axl.Defector(), turns=50, seed=6, length=200) + self.assertLess(player.nl_balance, 0) + + def test_first_move_and_classifier_untouched(self): + player = self._ladder() + self.assertEqual(player.strategy(axl.Defector()), C) + self.assertEqual(player.classifier["memory_depth"], float("inf")) + self.assertTrue(player.classifier["stochastic"]) + self.assertTrue(player.classifier["manipulates_state"]) + + +class TestZeroRespBranchCoverage(unittest.TestCase): + """Direct, seeded setups for branches the match suite does not reach.""" + + def setUp(self): + ZeroResp._global_profiles = {} + ZeroResp._global_lengths = [] + + def _player(self, **over): + player = ZeroResp(features=Features(**over)) + player.set_seed(1) + return player + + def test_features_asdict_and_replace(self): + features = Features(noise_ladder=False, apology=True) + as_dict = features.asdict() + self.assertFalse(as_dict["noise_ladder"]) + self.assertIn("profiles", as_dict) + replaced = features.replace(apology=False, noise_extra=True) + self.assertFalse(replaced.apology) + self.assertTrue(replaced.noise_extra) + self.assertFalse(replaced.noise_ladder) + self.assertIsNot(replaced, features) + + def test_harvest_window_without_jitter(self): + player = self._player(harvest_jitter=False) + self.assertEqual(player._harvest_window, player.HARVEST_WINDOW) + + def test_reset_trims_and_swallows_length_bookkeeping_errors(self): + player = self._player() + player.history.append(C, C) + ZeroResp._global_lengths = list(range(201)) + player.reset() + self.assertEqual(len(player.history), 0) + self.assertEqual(len(ZeroResp._global_lengths), 200) + self.assertEqual(ZeroResp._global_lengths[-1], 1) + + class _NoAppend(list): + def append(self, item): + raise RuntimeError("bookkeeping failed") + + player = self._player() + player.history.append(D, C) + ZeroResp._global_lengths = _NoAppend([4, 5]) + player.reset() + self.assertEqual(len(player.history), 0) + self.assertEqual(list(ZeroResp._global_lengths), [4, 5]) + + def test_length_helpers_unknown_and_invalid(self): + player = self._player() + player.match_attributes["length"] = object() + self.assertIsNone(player._match_length()) + + player.match_attributes["length"] = float("inf") + player.features.estimated_endgame = True + ZeroResp._global_lengths = [10, 40, 20] + self.assertEqual(player._effective_length(), 20) + player.features.estimated_endgame = False + self.assertEqual( + player._effective_length(), player._DEFAULT_MATCH_LENGTH + ) + player.features.estimated_endgame = True + ZeroResp._global_lengths = [] + self.assertEqual( + player._effective_length(), player._DEFAULT_MATCH_LENGTH + ) + + player.features.estimated_endgame = False + self.assertIsNone(player._estimate_remaining(10)) + player.features.estimated_endgame = True + ZeroResp._global_lengths = [5, 6, 7] + self.assertEqual(player._estimate_remaining(20), 1) + ZeroResp._global_lengths = [30, 50, 70] + self.assertEqual(player._estimate_remaining(10), 41) + + def test_backstabber_and_warmth_penalties(self): + player = self._player() + player.match_attributes["length"] = 100 + player.opp_len = 80 + player.opp_defects = 10 + player.late_defects = 8 + self.assertEqual(player._classify_tactical(), "backstabber") + + player.my_C = 20 + player.opp_coops_after_my_C = 10 + player.my_D = 4 + player.tactical = "backstabber" + player._forgiveness_exploiter = 2 + player._update_target_warmth() + self.assertGreaterEqual(player.target_warmth, 0.10) + self.assertLessEqual(player.target_warmth, 0.90) + + def test_noise_regime_and_unexplained_guards(self): + quiet = self._player(noise_adaptive=False) + quiet.p_noise_est = 0.2 + quiet.cc_pairs = 50 + self.assertFalse(quiet._is_noisy_regime()) + + player = self._player() + player._bad_standing = True + self.assertFalse(player._nl_is_unexplained()) + player._bad_standing = False + player._contrite = True + self.assertFalse(player._nl_is_unexplained()) + player._contrite = False + self.assertTrue(player._nl_is_unexplained()) + player.history.append(C, D) + player.nl_provoked_until = 4 + player._nl_step_hint = 4 + self.assertFalse(player._nl_is_unexplained()) + + def test_ladder_window_exploit_and_harvest_plan(self): + player = self._player() + player.set_seed(2) + player.nl_window.append(1) + player.nl_evidence = 4 + player.nl_pardon_step = 10 + player._nl_ladder(14) + self.assertEqual(player.nl_evidence, 1) + self.assertEqual(list(player.nl_window), [14]) + self.assertEqual(player._forgiveness_exploiter, 1) + self.assertEqual(player.nl_pardon_step, 0) + + player = self._player() + player.set_seed(3) + player.my_D = 3 + player.opp_coops_after_my_D = 3 + player.nl_balance = 1 + player.nl_evidence = 1 + player._nl_ladder(30) + self.assertGreaterEqual(player.nl_harvest_size, player.NL_HARVEST_MIN) + self.assertLessEqual(player.nl_harvest_size, player.NL_HARVEST_MAX) + self.assertEqual(player.nl_harvest_left, 0) + self.assertEqual(player.nl_harvest_postponed, 0) + self.assertGreaterEqual(player.nl_harvest_scheduled_at, 36) + self.assertLessEqual(player.nl_harvest_scheduled_at, 40) + + def test_forgiveness_budget_and_harvest_action_branches(self): + player = self._player() + player.set_seed(4) + player.warmth = 0.9 + player.features.noise_extra = True + player.intent = "noise" + player.tactical = "aggressor" + high = player._get_forgiveness_budget() + player.warmth = 0.1 + player.features.noise_extra = False + player.intent = "normal" + player.tactical = "adaptive" + low = player._get_forgiveness_budget() + self.assertGreaterEqual(high, 2) + self.assertLessEqual(high, 12) + self.assertGreaterEqual(low, 2) + self.assertLessEqual(low, 12) + + player.features.noise_adaptive = True + player.p_noise_est = 0.05 + player.cc_pairs = 15 + player.nl_my_flips = 0 + self.assertIsNone(player._harvest_action(4, axl.Cooperator())) + + player.p_noise_est = 0.0 + player.cc_pairs = 0 + player.opp_len = 60 + player.opp_defects = 0 + player.my_D = 0 + player.probe_fired = True + self.assertEqual(player._harvest_action(4, axl.Cooperator()), D) + + player.opp_len = 40 + player.opp_defects = 5 + player.my_D = 4 + player.opp_coops_after_my_D = 0 + player.tactical = "aggressor" + self.assertEqual(player._harvest_action(3, axl.Cooperator()), D) + + def test_play_bad_standing_and_noise_extra_thresholds(self): + player = self._player() + self.assertEqual(player._play(C), C) + self.assertEqual(player._intended_prev, C) + + player._bad_standing = True + player._bad_standing_turns = 2 + player._contrite = True + self.assertTrue(player._on_defect(8, D)) + self.assertTrue(player._bad_standing) + self.assertEqual(player._bad_standing_turns, 1) + self.assertFalse(player._contrite) + player._bad_standing_turns = 1 + self.assertTrue(player._on_defect(9, D)) + self.assertFalse(player._bad_standing) + + player = self._player() + player.features.noise_extra = True + player.intent = "noise" + self.assertFalse(player._on_defect(10, D)) + self.assertFalse(player.is_red_line) + self.assertGreaterEqual(len(player.queue), 1) + + def test_close_epoch_and_profile_guards(self): + player = self._player() + player.p_noise_est = 0.0 + player.cc_pairs = 0 + player._enter_red_line() + player.epoch_step = 40 + player.debt = 0 + player.queue = [] + player._close_epoch() + self.assertEqual(player.epoch_step, 40) + self.assertEqual(player._state.name, "RED_LINE") + + bare = ZeroResp(use_profiles=False) + bare.set_seed(1) + bare._save_profile("Defector") + self.assertEqual(ZeroResp._global_profiles, {}) + self._player()._save_profile("") + self.assertEqual(ZeroResp._global_profiles, {}) + + def test_balance_uses_payoff_when_game_score_fails(self): + class _Boom: + def score(self, pair): + raise ValueError("bad score") + + player = self._player() + player.history.append(D, C) + player._intended_prev = D + opponent = axl.Cooperator() + opponent.history.append(C, D) + player.match_attributes["game"] = _Boom() + player.strategy(opponent) + self.assertEqual(player.nl_balance, 5) + + def _late_player(self): + player = self._player() + player.match_attributes["length"] = 4 + player.history.extend([C, C, C], [C, C, C]) + player._apology_mode = False + return player + + def test_apology_accounting_retry_and_give_up(self): + player = self._late_player() + player._apology_mode = True + player._apology_steps = 4 + player.last_my = D + opponent = axl.Defector() + opponent.history.append(D, C) + self.assertEqual(player.strategy(opponent), C) + self.assertGreaterEqual(player.late_defects, 1) + self.assertGreaterEqual(player.my_D, 1) + + player = self._late_player() + player._apology_mode = True + player._apology_steps = 4 + player.last_my = D + opponent = axl.Cooperator() + opponent.history.append(C, D) + self.assertEqual(player.strategy(opponent), C) + self.assertGreaterEqual(player.opp_coops_after_my_D, 1) + self.assertEqual(player.clean_peace, 0) + self.assertGreaterEqual(player.my_D, 1) + + player = self._player() + player._apology_mode = True + player._apology_steps = 1 + player._apology_tries = 0 + player._apology_opp_moves = [] + opponent = axl.Defector() + opponent.history.append(D, C) + player.strategy(opponent) + self.assertFalse(player._apology_mode) + self.assertFalse(player.is_red_line) + self.assertEqual(player._apology_tries, 1) + + player = self._player() + player._apology_mode = True + player._apology_steps = 1 + player._apology_tries = player._apology_max_tries - 1 + player._apology_opp_moves = [D] + opponent = axl.Defector() + opponent.history.append(D, C) + self.assertEqual(player.strategy(opponent), D) + self.assertTrue(player.is_red_line) + self.assertEqual(player._last_D_reason, "red_line") + + def test_contrite_full_and_plain_contrite_accounting(self): + player = self._late_player() + player._bad_standing = True + player._bad_standing_turns = 2 + player.last_my = D + opponent = axl.Defector() + opponent.history.append(D, C) + self.assertEqual(player.strategy(opponent), C) + self.assertGreaterEqual(player.late_defects, 1) + self.assertGreaterEqual(player.my_D, 1) + + player = self._late_player() + player._bad_standing = True + player._bad_standing_turns = 2 + player.last_my = D + opponent = axl.Cooperator() + opponent.history.append(C, D) + self.assertEqual(player.strategy(opponent), C) + self.assertGreaterEqual(player.opp_coops_after_my_D, 1) + self.assertGreaterEqual(player.my_D, 1) + + def contrite(my_prev): + actor = self._late_player() + actor.opp_len = 4 + actor.last_my = my_prev + actor._contrite = True + actor._bad_standing = False + foe = axl.Defector() + foe.history.extend([D] * 5, [C] * 5) + action = actor.strategy(foe) + return actor, action + + soft, action = contrite(C) + self.assertEqual(action, C) + self.assertFalse(soft._contrite) + self.assertGreaterEqual(soft.opp_defects_after_my_C, 1) + self.assertGreaterEqual(soft.late_defects, 1) + self.assertGreaterEqual(soft.my_C, 1) + hard, action = contrite(D) + self.assertEqual(action, C) + self.assertGreaterEqual(hard.my_D, 1) + self.assertGreaterEqual(hard.late_defects, 1) + + def test_noise_ladder_harvest_and_scheduled_strike(self): + player = self._player() + player.nl_harvest_scheduled_at = 1 + player.nl_harvest_size = 3 + opponent = axl.Cooperator() + opponent.history.append(C, C) + self.assertEqual(player.strategy(opponent), D) + self.assertEqual(player._last_D_reason, "nl_harvest_comp") + self.assertEqual(player.nl_harvest_left, 2) + self.assertEqual(player.nl_harvest_scheduled_at, 0) + self.assertEqual(player.nl_provoked_until, 7) + + player = self._player() + player.nl_harvest_scheduled_at = 1 + player.nl_harvest_size = 3 + player.strategy(axl.Cooperator()) + self.assertEqual(player.nl_harvest_postponed, 1) + self.assertEqual(player.nl_harvest_scheduled_at, 4) + + player = self._player() + player.nl_harvest_scheduled_at = 1 + player.nl_harvest_size = 3 + player.nl_harvest_postponed = 2 + player.strategy(axl.Cooperator()) + self.assertEqual(player.nl_harvest_scheduled_at, 0) + self.assertEqual(player.nl_harvest_size, 0) + + player = self._player() + player.opp_last3.extend([C, C, C]) + player.opp_defects_after_my_D = 0 + player.nl_scheduled = [{"turn": 1, "kind": "strike"}] + player.queue = [1] + player.debt = 2 + player.nl_debt_ledger = 1 + self.assertEqual(player.strategy(axl.Cooperator()), C) + self.assertEqual(player.nl_pardon_step, 1) + self.assertGreaterEqual(player.nl_pardoned, 1) + self.assertNotIn(1, player.queue) + self.assertEqual(player.debt, 1) + + player = self._player() + player.opp_last3.extend([D, D, D]) + player.nl_scheduled = [{"turn": 1, "kind": "strike"}] + player.nl_debt_ledger = 4 + self.assertEqual(player.strategy(axl.Cooperator()), D) + self.assertEqual(player.nl_debt_ledger, 3) + + def test_hostile_cooldown_deadlock_probe_and_queue(self): + player = self._player() + player.opp_len = 20 + player.opp_defects = 18 + self.assertEqual(player.strategy(axl.Cooperator()), D) + self.assertTrue(player.is_red_line) + self.assertEqual(player._last_D_reason, "red_line") + + player = self._player() + player.p_noise_est = 0.05 + player.cc_pairs = 20 + player._enter_red_line() + player._red_line_cooldown = 0 + self.assertEqual(player._state.name, "RED_LINE_COOLDOWN") + self.assertEqual(player.strategy(axl.Cooperator()), C) + self.assertGreater(player._red_line_cooldown, 0) + self.assertEqual(player._last_D_reason, None) + + player = self._player() + player.deadlock = player.DEADLOCK_THRESHOLD + player.queue = [99] + player.debt = 3 + player.systemic = 4 + player.echo_forgive = 2 + player._contrite = True + player._bad_standing = True + self.assertEqual(player.strategy(axl.Cooperator()), C) + self.assertEqual(player.deadlock, 0) + self.assertEqual(player.queue, []) + self.assertEqual(player.debt, 0) + self.assertEqual(player.echo_forgive, 0) + self.assertEqual(player.systemic, 3) + self.assertFalse(player._contrite) + self.assertFalse(player._bad_standing) + self.assertEqual(player._state.name, "COOPERATIVE") + + player = self._player(harvest_jitter=False) + player.match_attributes["length"] = 5 + player.opp_len = 40 + player.opp_defects = 0 + player.probe_fired = True + player.tactical = "adaptive" + self.assertEqual(player.strategy(axl.Cooperator()), D) + self.assertEqual(player._last_D_reason, "probe") + + player = self._player() + player.match_attributes["length"] = 200 + player.queue = [1] + player._contrite = True + self.assertEqual(player.strategy(axl.Cooperator()), C) + self.assertEqual(player.queue, [2]) + self.assertTrue(player._contrite) + + player = self._player() + player.match_attributes["length"] = 200 + player._contrite = True + self.assertEqual(player.strategy(axl.Cooperator()), C) + self.assertFalse(player._contrite) + + def test_v54_profile_memory_and_legacy_harvest(self): + """profile_memory softens min_d; adaptive_harvest=False uses forg>0.6.""" + player = self._player() + player.set_seed(7) + player._known_soft_opponent = True + player.my_D = 1 + player.opp_coops_after_my_D = 1 + player.nl_balance = 1 + player.nl_evidence = 1 + player.tactical = "adaptive" + player._nl_ladder(30) + self.assertGreaterEqual(player.nl_harvest_size, player.NL_HARVEST_MIN) + + player = self._player(adaptive_harvest=False) + player.set_seed(8) + player.my_D = 3 + player.opp_coops_after_my_D = 3 # forg = 1.0 > 0.6 + player.nl_balance = 1 + player.nl_evidence = 1 + player.tactical = "adaptive" + player._nl_ladder(30) + self.assertGreaterEqual(player.nl_harvest_size, player.NL_HARVEST_MIN) + self.assertGreaterEqual(player.nl_harvest_scheduled_at, 36) + + def test_v54_balance_apology_gates_and_cycler(self): + """balance_apology: coop floor, responsiveness, cycler contour block.""" + # Cycler contour blocks apology against a fixed period-2 opponent. + player = self._player() + player.nl_balance = player.NL_BALANCE_FLOOR + player.match_attributes["noise"] = 0.05 + opp = axl.Alternator() + # Seed enough history for period detection (min 12, period 2). + hist_me = [C, D] * 9 + hist_opp = [C, D] * 9 + player.history.extend(hist_me, hist_opp) + opp.history.extend(hist_opp, hist_me) + player.opp_len = len(hist_opp) + before = player.echo_forgive + player.strategy(opp) + self.assertTrue(player._fixed_period_latch) + # Apology must not fire when cycler is latched. + self.assertEqual(player.echo_forgive, before) + + # Responsiveness gate: delta_p >= 0.35 allows apology (TFT-shaped history). + player = self._player(cycler_contour=False) + player.nl_balance = player.NL_BALANCE_FLOOR + player.match_attributes["noise"] = 0.05 + player.nl_scheduled = [{"turn": 9, "kind": "strike"}] + player.nl_harvest_left = 2 + player.nl_evidence = 2 + my = [C, D] * 8 + their = [C] + my[:-1] # Tit-for-tat reply to our previous move + player.history.extend(my, their) + foe = axl.TitForTat() + foe.history.extend(their, my) + player.opp_len = len(their) + player.strategy(foe) + self.assertGreaterEqual(player.echo_forgive, 2) + self.assertEqual(player.nl_scheduled, []) + self.assertEqual(player.nl_harvest_left, 0) + self.assertEqual(player.nl_evidence, 0) + + # Coop-rate floor blocks apology when opponent mostly defects. + player = self._player(cycler_contour=False) + player.nl_balance = player.NL_BALANCE_FLOOR + player.match_attributes["noise"] = 0.05 + player.echo_forgive = 0 + my = [C] * 16 + their = [D] * 16 + player.history.extend(my, their) + foe = axl.Defector() + foe.history.extend(their, my) + player.opp_len = 16 + player.opp_defects = 14 # live coop rate 2/16 = 0.125 < 0.60 + player.strategy(foe) + self.assertEqual(player.echo_forgive, 0) + + # Few own D's (<2): can_apologize True without delta_p check. + player = self._player(cycler_contour=False) + player.nl_balance = player.NL_BALANCE_FLOOR + player.match_attributes["noise"] = 0.05 + my = [C] * 10 + [D] + their = [C] * 11 + player.history.extend(my, their) + foe = axl.Cooperator() + foe.history.extend(their, my) + player.opp_len = 11 + player.strategy(foe) + self.assertGreaterEqual(player.echo_forgive, 2) 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..76c364bb4 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 + 23 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..a379bbc7f 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 + 244 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: