Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions axelrod/data/all_classifiers.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2040,3 +2040,12 @@ ZD-SET-2:
manipulates_state: false
memory_depth: 1
stochastic: true
ZeroResp:
inspects_source: false
long_run_time: false
makes_use_of: !!set
length: null
manipulates_source: false
manipulates_state: false
memory_depth: .inf
stochastic: true
3 changes: 2 additions & 1 deletion axelrod/fingerprint.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,8 @@ 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):
points.append(Point(x, y))
# Cast to Python float so probe names/repr stay stable across NumPy versions
points.append(Point(float(x), float(y)))

if progress_bar:
p_bar.update()
Expand Down
2 changes: 2 additions & 0 deletions axelrod/strategies/_strategies.py
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,7 @@
ZDMischief,
ZDSet2,
)
from .zeroresp import ZeroResp

# Note: Meta* strategies are handled in .__init__.py

Expand Down Expand Up @@ -509,5 +510,6 @@
ZDMem2,
ZDMischief,
ZDSet2,
ZeroResp,
e,
]
2 changes: 1 addition & 1 deletion axelrod/strategies/ann.py
Original file line number Diff line number Diff line change
Expand Up @@ -210,7 +210,7 @@ def __init__(

def _process_weights(self, weights, num_features, num_hidden):
self.weights = list(weights)
(i2h, h2o, bias) = split_weights(weights, num_features, num_hidden)
i2h, h2o, bias = split_weights(weights, num_features, num_hidden)
self.input_to_hidden_layer_weights = np.array(i2h)
self.hidden_to_output_layer_weights = np.array(h2o)
self.bias_weights = np.array(bias)
Expand Down
2 changes: 1 addition & 1 deletion axelrod/strategies/memoryone.py
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,7 @@ def set_initial_four_vector(self, four_vector):
pass

def receive_match_attributes(self):
(R, P, S, T) = self.match_attributes["game"].RPST()
R, P, S, T = self.match_attributes["game"].RPST()
if self.p is None:
self.p = min(1 - (T - R) / (R - S), (R - P) / (T - P))
four_vector = [1, self.p, 1, self.p]
Expand Down
2 changes: 1 addition & 1 deletion axelrod/strategies/qlearner.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ def __init__(self) -> None:
self.prev_state = ""

def receive_match_attributes(self):
(R, P, S, T) = self.match_attributes["game"].RPST()
R, P, S, T = self.match_attributes["game"].RPST()
self.payoff_matrix = {C: {C: R, D: S}, D: {C: T, D: P}}

def strategy(self, opponent: Player) -> Action:
Expand Down
4 changes: 2 additions & 2 deletions axelrod/strategies/zero_determinant.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,7 @@ def __init__(self, phi: float = 1 / 9, s: float = 0.5) -> None:
super().__init__(phi, s, None)

def receive_match_attributes(self):
(R, P, S, T) = self.match_attributes["game"].RPST()
R, P, S, T = self.match_attributes["game"].RPST()
self.l = P
super().receive_match_attributes()

Expand Down Expand Up @@ -228,7 +228,7 @@ def __init__(self, phi: float = 0.25, s: float = 0.5) -> None:
super().__init__(phi, s, None)

def receive_match_attributes(self):
(R, P, S, T) = self.match_attributes["game"].RPST()
R, P, S, T = self.match_attributes["game"].RPST()
self.l = R
super().receive_match_attributes()

Expand Down
257 changes: 257 additions & 0 deletions axelrod/strategies/zeroresp.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,257 @@
"""
ZeroResp: adaptive state-machine strategy for the Iterated Prisoner's Dilemma.

Designed to resist both heuristic exploiters and simple RL / tabular Q-learners
via delayed stochastic retaliation, epoch-based debt accounting, and a permanent
red-line ban after systemic abuse.
"""

from __future__ import annotations

import math
from enum import Enum, auto
from typing import List, Optional

from axelrod.action import Action
from axelrod.player import Player

C, D = Action.C, Action.D


class _State(Enum):
"""Internal finite-state labels."""

COOPERATIVE = auto()
EQUALIZING = auto()
RED_LINE = auto()


class ZeroResp(Player):
"""
An adaptive state machine that balances cooperation with delayed,
randomised retaliation and a permanent ban against systemic defectors.

Architecture
------------
1. **Dynamic epochs** — interaction is partitioned into epochs of length
``base_epoch`` (default 25). While a retaliation debt or queued strike
is outstanding the epoch is extended; once cleared the systemic-abuse
counter resets and the bot returns to cooperative mode.

2. **Stochastic retaliation buffer** — a defection does not trigger an
immediate mirror response. Instead a retaliatory ``D`` is scheduled
``5 + U{1..10}`` turns later. The random delay breaks short-horizon
Markov estimates used by tabular Q-learners and reduces cascade wars
against tit-for-tat family strategies.

3. **Red line (ban list)** — systemic defections (defects that arrive while
debt/queue is still open, or while already equalising) raise a counter.
After a dynamic threshold (2 or 3 depending on observed hostility) the
strategy enters permanent red line (``is_red_line = True``) and defects
unconditionally for the rest of the match.

4. **Anti-raider** — two or more late-game defections (past ~75% of the
known match length) are treated as end-game harvest and trigger red
line immediately.

5. **End-game harvest** — against highly forgiving / near-pure cooperators
(and never against grim-trigger types that never defected), ZeroResp may
defect near the known end of a finite match. This is disabled when
match length is unknown.

Names:

- ZeroResp: Original name by EpochRedLine / SovereignStabilizer authors
- EpochRedLine: Earlier development name
- SmartTitForTat: Legacy sandbox name
"""

name = "ZeroResp"
classifier = {
"memory_depth": float("inf"),
"stochastic": True,
"long_run_time": False,
"inspects_source": False,
"manipulates_source": False,
"manipulates_state": False,
}

# Fallback when match length is unknown / infinite.
_DEFAULT_MATCH_LENGTH = 200
_LATE_FRACTION = 0.75
_LIVE_INTEL_MIN_SAMPLES = 10
_HOSTILE_COOP_THRESHOLD = 0.4
_SOFT_HOSTILE_COOP = 0.7

def __init__(self) -> None:
"""Initialise epoch accounting and red-line state."""
super().__init__()
self.base_epoch = 25

self._state = _State.COOPERATIVE
self.is_red_line = False
self.epoch_step = 0
self.debt = 0
self.systemic = 0
self.queue: List[int] = []

# Opponent cadastre (loyalty / exploitability estimates)
self.opp_len = 0
self.opp_defects = 0
self.opp_coops_after_my_D = 0
self.my_D = 0
self.last_my: Action = C
self.late_defects = 0

# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------

def _match_length(self) -> Optional[int]:
"""Return known finite match length, else ``None``."""
# Bracket access so Axelrod's makes_use_of scanner detects "length".
length = self.match_attributes["length"]
if length is None or length in (-1, float("inf")):
return None
try:
length_int = int(length)
except (TypeError, ValueError):
return None
return length_int if length_int > 0 else None

def _effective_length(self) -> int:
return self._match_length() or self._DEFAULT_MATCH_LENGTH

def _late_threshold(self) -> int:
return int(self._effective_length() * self._LATE_FRACTION)

def _live_coop_rate(self) -> float:
if self.opp_len == 0:
return 1.0
return 1.0 - (self.opp_defects / self.opp_len)

def _is_hostile(self) -> bool:
"""Enough evidence of a low-cooperation opponent."""
return (
self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES
and self._live_coop_rate() < self._HOSTILE_COOP_THRESHOLD
)

def _is_soft_hostile(self) -> bool:
return (
self.opp_len >= self._LIVE_INTEL_MIN_SAMPLES
and self._live_coop_rate() < self._SOFT_HOSTILE_COOP
)

def _enter_red_line(self) -> None:
self._state = _State.RED_LINE
self.is_red_line = True
self.queue.clear()

# ------------------------------------------------------------------
# Core strategy
# ------------------------------------------------------------------

def strategy(self, opponent: Player) -> Action:
"""Select C or D for the current turn."""
step = len(self.history) + 1 # 1-based turn index

# --- Cadastre update from opponent's previous action -------------
if opponent.history:
opp_last = opponent.history[-1]
self.opp_len += 1
if opp_last == D:
self.opp_defects += 1
if step > self._late_threshold():
self.late_defects += 1
self._on_defect(step)
else:
if self.last_my == D:
self.opp_coops_after_my_D += 1

# Live zero-turn defence: permanent ban against proven predators
if self._is_hostile():
self._enter_red_line()

# --- Anti-raider (late harvest interception) ---------------------
if self.late_defects >= 2:
self._enter_red_line()
return self._play(D)

if self.is_red_line or self._state == _State.RED_LINE:
return self._play(D)

# --- End-game harvest vs forgiving victims (known finite length) -
known_len = self._match_length()
if known_len is not None and self.opp_len > 50:
p_end = 1.0 / (1.0 + math.exp(-10.0 * (step / known_len - 0.85)))
forgiveness = self.opp_coops_after_my_D / max(1, self.my_D)
is_grim = self.opp_len > 50 and self.opp_defects == 0
is_victim = forgiveness > 0.6 or (
self.opp_defects / max(1, self.opp_len) < 0.03
)
if p_end > 0.75 and is_victim and not is_grim:
return self._play(D)

# --- Queued delayed retaliation ----------------------------------
if step in self.queue:
self.queue.remove(step)
self.debt = max(0, self.debt - 1)
self._close_epoch()
return self._play(D)

# --- Default: cooperate & advance epoch --------------------------
self.epoch_step += 1
self._close_epoch()
return self._play(C)

def _play(self, action: Action) -> Action:
self.last_my = action
if action == D:
self.my_D += 1
return action

def _on_defect(self, step: int) -> None:
"""Record an opponent defection and schedule / escalate response."""
if self.debt > 0 or self.queue or self._state == _State.EQUALIZING:
self.systemic += 1

self.debt += 1
self._state = _State.EQUALIZING

# Dynamic red-line threshold (tighter under hostility).
# Default: 3 systemic defects; after the first systemic event (or
# soft hostility / late defects) the threshold tightens to 2.
threshold = 3
if (
self.systemic >= 1
or self._is_soft_hostile()
or self.late_defects > 0
):
threshold = 2

if self.systemic >= threshold:
self._enter_red_line()
return

# Adaptive buffer: near-immediate under pressure, else stochastic
if self._is_hostile() or self.late_defects > 0:
delay = 1
else:
# numpy RandomState.randint is high-exclusive → use (1, 11)
delay = 5 + int(self._random.randint(1, 11))

self.queue.append(step + delay)

def _close_epoch(self) -> None:
"""Reset systemic counters when a clean epoch completes."""
if self.is_red_line or self._state == _State.RED_LINE:
return
if (
self.epoch_step >= self.base_epoch
and self.debt <= 0
and not self.queue
):
self.epoch_step = 0
self.systemic = 0
self._state = _State.COOPERATIVE
2 changes: 1 addition & 1 deletion axelrod/tests/strategies/test_memoryone.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ def test_strategy2(self):
)

def test_four_vector(self):
(R, P, S, T) = axl.Game().RPST()
R, P, S, T = axl.Game().RPST()
p = min(1 - (T - R) / (R - S), (R - P) / (T - P))
expected_dictionary = {(C, C): 1.0, (C, D): p, (D, C): 1.0, (D, D): p}
test_four_vector(self, expected_dictionary)
Expand Down
2 changes: 1 addition & 1 deletion axelrod/tests/strategies/test_qlearner.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ class TestRiskyQLearner(TestPlayer):
}

def test_payoff_matrix(self):
(R, P, S, T) = axl.Game().RPST()
R, P, S, T = axl.Game().RPST()
payoff_matrix = {C: {C: R, D: S}, D: {C: T, D: P}}
player = self.player()
self.assertEqual(player.payoff_matrix, payoff_matrix)
Expand Down
Loading
Loading