diff --git a/axelrod/data/all_classifiers.yml b/axelrod/data/all_classifiers.yml index fb0b6a51c..aa8bafa8e 100644 --- a/axelrod/data/all_classifiers.yml +++ b/axelrod/data/all_classifiers.yml @@ -2040,3 +2040,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.3: + 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..38cdc6ee4 --- /dev/null +++ b/axelrod/strategies/zeroresp.py @@ -0,0 +1,1149 @@ +""" +ZeroResp v5.3 - v5.2 + noise ladder (лестница эскалации по необъяснимым D) + +Наследует v5.2: +1. Full contrite: если мой intended=C но realized=D (шум), D оппонента не считается атакой +2. Generous forgiveness: под шумом прощение стохастическое p~f(p_noise_est), а не капами +3. Red line cooldown: под шумом не перманентный, а с ре-пробой C каждые N ходов +4. p_noise_est: unprovoked D + наши флипы / наблюдения. Gating: off при p_est<0.015 + +v5.3-variantA (features.noise_ladder, по умолчанию on): +5. Необъяснимый D (оппонент D при нашем реализованном C) идёт по лестнице evidence: + 1-2 - чистое прощение (долг копится в nl_debt_ledger, удара нет); на evidence 2 + при прощающем оппоненте может планироваться компенсационная акция (3-4 D); + >=3 при coop_rate<0.40 (под шумом >=8) - red line. Лестница сама не бьёт по + необъяснимым D - эскалацию ведут старый системный путь и красная линия. + +Совместим с axelrod.Match, сохраняет endgame логику v5.1 +""" + +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 + +# v5.3: стандартная матрица выигрышей (фолбэк для баланса 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() # v5.2: временный red_line под шумом + + +class Features: + __slots__ = ( + "profiles", + "apology", + "harvest_jitter", + "noise_extra", + "estimated_endgame", + "opening_d_retort", + # v5.2 new + "noise_adaptive", + "contrite_full", + "generous", + "red_line_cooldown", + # v5.3 new + "noise_ladder", + ) + + 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, + ): + 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 + + 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.3" + 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 + + # v5.3 noise ladder + NL_WINDOW_TURNS = 12 + # v5.3-I: отложенное возмездие (теория I) - удар планируется со случайной + # задержкой (антиадаптация), исполняется только если оппонент не очистился + NL_DELAY_MIN = 8 + NL_DELAY_MAX = 20 + NL_PARDON_STREAK = 3 + NL_HARVEST_MIN = 3 + NL_HARVEST_MAX = 4 + + _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 + + # v5.2 - noise adaptation + self.cc_pairs = 0 + self.unprovoked_d = 0 + self.p_noise_est = 0.0 + self._intended_prev = C + self._realized_prev = C + self._bad_standing = False + self._bad_standing_turns = 0 + self._red_line_cooldown = 0 + + # v5.3 - noise ladder + self.nl_evidence = 0 + self.nl_window = deque(maxlen=24) # ходы необъяснимых D + 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 # v5.3-G: их D после нашей акции спровоцированы + ) + 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 _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 + # v5.3: при доказанном шуме учитываем и наши наблюдения + 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 + + # ---- v5.3 noise ladder ---- + def _nl_noise_proven(self) -> bool: + # Канальный шум доказывается только НАШИМИ флипами C->D: D оппонента + # и наш шум не различить, а собственный флип — независимое доказательство. + # При 0% nl_my_flips == 0 и лестница уснёт (поведение как в v5.2 без потерь). + return self.nl_my_flips > 0 + + def _nl_active(self) -> bool: + # лестница работает только когда флаг on И доказан канальный шум + # v5.3-A+B: лестница активна только при доказанном шуме И пока мы + # не в минусе с этим оппонентом (в минусе работает строгий путь v5.2) + return ( + self.features.noise_ladder + and self._nl_noise_proven() + and self.nl_balance >= 0 + ) + + def _nl_is_unexplained(self) -> bool: + # необъяснимый D: наш прошлый реализованный ход был C, не в contrite/bad_standing + if self._bad_standing or self._contrite: + return False + if not self.history: + return True + # v5.3-G: их D сразу после нашей компенсационной акции спровоцированы + # нами же (урок SlowTF2T2 seed 44) - уликами не считаются + 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: + """Лестница эскалации по необъяснимым D. True = текущий ход C.""" + # сброс окна, если прошлый необъяснимый D был давно + 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) + + # exploit-детект: D вскоре после нашего помилования + 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: + # Первый в окне: чисто прощаем (текущий ход C). Это либо шум + # (оппонент вернётся к C — мир без потерь), либо укол — тогда + # долг копится в nl_debt_ledger, а эскалацию ведут старый + # системный путь и красная линия (evidence>=3). + self.nl_pardoned += 1 + self.nl_debt_ledger += 1 + self._pending_reason = "nl_deferred" + return True + + if self.nl_evidence == 2: + # Второй в окне: такое же чистое прощение, как и первый (вариант A: + # немедленный удар по необъяснимым D перезапускал войны 1-к-1 со + # строгими реципрокаторами). Долг снова растёт, очередь ударов не + # трогаем. Компенсационная акция планируется по прежним гейтам: + # оппонент прощает наши D, баланс не хуже и не шумный режим. + self.nl_pardoned += 1 + self.nl_debt_ledger += 1 + self._pending_reason = "nl_deferred" + # v5.3-I: вторая улика подтверждает повторность - планируем + # отложенное возмездие со случайной задержкой (один план за окно) + 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) + # v5.3-H: при my_D<3 дробь forg нестабильна и запускала акцию + # против жёстких наказателей (урок SlowTF2T2 seed 44) + if ( + forg > 0.6 + and self.my_D >= 3 + 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 + + # evidence >= 3: красная линия против агрессивного (низкий кооп). + # В шумном режиме — только при очень плотной серии (evidence>=8): + # шумовые D накручивают улики, и ранний краситель ломает ритмы + # строгих реципрокаторов (Punisher/TF2T). Осцилляторы (~0.5 кооп) + # не красим вовсе — их D периодичны, а не враждебны; умеренных и + # «шумных» ведёт старый системный путь (debt/queue). + 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 + # умеренный кооператив: контратака по старому пути (debt/queue, задержка 1) + return None + + def _enter_red_line(self): + # v5.2: под шумом не перманентный + 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 + # классический перманентный red_line как в v5.1 + 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) + # v5.2: под шумом больше бюджет + if self._is_noisy_regime(): + jitter += int(self.p_noise_est * 40) # +2 при 5% шуме + return max(2, min(12, base + jitter)) + + def _harvest_action( + self, remaining: int, opponent: Player + ) -> Optional[Action]: + # полностью как в v5.1, но отключаем под шумом если noisy + 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 + # realized будет равен action, если шум не выбьет - проверим на след. ходу по history + 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: + """Возвращает True если простить, False если наказывать. v5.2 логика.""" + self._nl_step_hint = step # v5.3-G: шаг для иммунитета улик + + # === 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: + # v5.3: echo не глотает необъяснимые D - их ведёт лестница + 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(): + # Forgiver: если coop_rate >0.9 - всегда прощать + if self._live_coop_rate() > 0.90 and self.opp_len > 20: + return True + + # стохастическое прощение калиброванное под p_noise_est + # при 5% шуме p≈0.4, при 2% p≈0.2 + p_forgive = min( + 0.6, self.p_noise_est * 7 + 0.05 + (self.warmth - 0.5) * 0.2 + ) + # не прощаем кластеры D (3 подряд) - это не шум + 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 + + # детерминированные капы как в v5.1 + 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 # v5.3: окно exploit-детекта + return True + + # === v5.3 NOISE LADDER: только необъяснимые D, после доказанного шума === + if self._nl_active() and self._nl_is_unexplained(): + handled = self._nl_ladder(step) + if handled is not None: + return handled + + # provoked D (ответ на наш удар) - старая механика как есть + 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 # v5.2: гораздо выше порог до red_line под шумом + 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 + 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, + } + + 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 # v5.3: одноразовый reason + self._nl_step_hint = 0 # v5.3-G: шаг для иммунитета улик + + # --- детект моего шума по истории --- + 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 + # v5.3: наблюдения наших intended-C ходов (после первого) + if self.features.noise_ladder and self._intended_prev == C: + self.nl_my_obs += 1 + # v5.3: баланс очков за последний разыгранный ход + 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 + # v5.2: обновление p_est каждые 5 пар; v5.3: при доказанном + # шуме в числитель/знаменатель входят и наши флипы + 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) + + # --- apology логика как в v5.1 --- + 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 + + # v5.2 contrite: если я был в bad standing - не считать его D атакой + 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 + + # --- v5.3 noise ladder: исполнение запланированного --- + 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 + # v5.3-G: окно иммунитета улик на время акции + ответ + 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: + # после 2 переносов отменяем + 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) + + # v5.3-I: исполнение отложенного возмездия. Чистому (3xC и ни разу + # не наказал наш 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 --- + # под шумом не входим в перманентный red_line по 3xD + 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, + ]: + # v5.3: серия необъяснимых 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(): # под шумом не уходим в hostile + self._enter_red_line() + + if self.is_red_line: + # v5.2 cooldown probe + 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)) + ) + # ре-проба стоит -2 против AllD, спасает +200 против TFT + 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 + # v5.3: одноразовый отложенный reason (nl_deferred / nl_pardon) + reason = ( + self._pending_reason + if self._pending_reason is not None + else "normal" + ) + self._pending_reason = None + return self._play(C, reason, intended=C) + + +ZeroRespV5 = ZeroResp +ZeroRespV51 = ZeroResp +ZeroRespV52 = ZeroResp 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..083372bd5 --- /dev/null +++ b/axelrod/tests/strategies/test_zeroresp.py @@ -0,0 +1,837 @@ +"""Tests for the ZeroResp strategy (v5.3).""" + +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.3: 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.3") + 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): + """v5.3 noise ladder. 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) 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: