SirStone
1ed7797cb6
- New guns: guess-factor (GF histogram), pattern-matcher (movement tape replay) - New modules: minimum-risk melee movement, spinning melee radar - New test bots: PatternMover, RandomMover, WaveSurfer - Fixed: FeedbackEvent now carries actualX/actualY for proper GF learning - Fixed: TM gun warmup gating + directional residuals - Fixed: circular gun integrated formula + multi-bin omega cache - Fixed: oscillator wall-bounce lockout - Fixed: phantom meteor perpendicular body orientation - 6/6 battle wins across all enemy types
653 lines
29 KiB
Python
653 lines
29 KiB
Python
"""
|
||
Alternative learning methods backtest — binary input, online learning, no gradients.
|
||
Stdlib only: csv, math, random, collections.
|
||
Baselines: linear extrapolation MAE≈12.24, WiSARD K=12 MAE≈9.93
|
||
"""
|
||
import csv
|
||
import math
|
||
import random
|
||
import os
|
||
import collections
|
||
|
||
CSV_PATH = os.path.join(os.path.dirname(__file__), "../data/target_battle_1_decimal.csv")
|
||
OUT_PATH = os.path.join(os.path.dirname(__file__), "alternatives_results.txt")
|
||
|
||
POWER_LEVELS = [0.10, 0.42, 0.74, 1.07, 1.39, 1.71, 2.03, 2.36, 2.68, 3.00]
|
||
POWER_STRS = ["p0.10","p0.42","p0.74","p1.07","p1.39","p1.71","p2.03","p2.36","p2.68","p3.00"]
|
||
MAX_DIST = 1414.0
|
||
HIT_THRESH = 18.0 # px
|
||
REP_POWER = 1.07
|
||
REP_PS = "p1.07"
|
||
|
||
# ── bit encoding (matches backtest_wisard.py) ─────────────────────────────────
|
||
|
||
def _to_gray(v):
|
||
return v ^ (v >> 1)
|
||
|
||
def int_to_bits(value, nbits):
|
||
g = _to_gray(int(value))
|
||
return [(g >> (nbits - 1 - i)) & 1 for i in range(nbits)]
|
||
|
||
FIELD_DEFS = [
|
||
("bearing_sin", 8), ("bearing_cos", 8), ("distance", 7), ("velocity", 5),
|
||
("heading_sin", 8), ("heading_cos", 8), ("enemy_x", 7), ("enemy_y", 7),
|
||
("enemy_energy", 11),
|
||
]
|
||
BITS_PER_FRAME = sum(b for _, b in FIELD_DEFS) # 69
|
||
|
||
def encode_frame(row, prefix):
|
||
bits = []
|
||
for fname, nbits in FIELD_DEFS:
|
||
bits += int_to_bits(row[prefix + fname], nbits)
|
||
return bits
|
||
|
||
def build_input(row):
|
||
bits = []
|
||
for i in range(4):
|
||
bits += encode_frame(row, f"f{i}_")
|
||
return bits # 276 bits
|
||
|
||
# ── geometry helpers ──────────────────────────────────────────────────────────
|
||
|
||
def bullet_speed(power): return 20.0 - 3.0 * power
|
||
def flight_ticks(dist_enc, power): return dist_enc / 99.0 * MAX_DIST / bullet_speed(power)
|
||
def euclid(ax, ay, bx, by): return math.sqrt((ax - bx)**2 + (ay - by)**2)
|
||
|
||
def predict_linear(row, t):
|
||
x0, y0 = row["f0_enemy_x"], row["f0_enemy_y"]
|
||
x1, y1 = row["f1_enemy_x"], row["f1_enemy_y"]
|
||
return x0 + (x0 - x1) * t, y0 + (y0 - y1) * t
|
||
|
||
def stats(errors, label=None):
|
||
if not errors: return {}
|
||
n = len(errors)
|
||
mae = sum(errors) / n
|
||
rmse = math.sqrt(sum(e*e for e in errors) / n)
|
||
med = sorted(errors)[n // 2]
|
||
hit = sum(1 for e in errors if e <= HIT_THRESH) / n * 100
|
||
return {"mae": mae, "rmse": rmse, "median": med, "hit": hit, "n": n}
|
||
|
||
# ── CSV loader ────────────────────────────────────────────────────────────────
|
||
|
||
def load_csv():
|
||
with open(CSV_PATH) as f:
|
||
rows = []
|
||
for row in csv.DictReader(f):
|
||
try:
|
||
rows.append({k: float(v) for k, v in row.items()})
|
||
except ValueError:
|
||
pass
|
||
return rows
|
||
|
||
# ── Generic online backtest harness ──────────────────────────────────────────
|
||
|
||
def run_method(rows, predict_fn, learn_fn):
|
||
"""
|
||
predict_fn(bits) -> (cx, cy) correction on top of linear
|
||
learn_fn(bits, rx, ry) update on true residual
|
||
Returns: (all_errors_by_ps, first50_p107, last50_p107)
|
||
"""
|
||
errors = {ps: [] for ps in POWER_STRS}
|
||
first50, last50 = [], []
|
||
n = len(rows)
|
||
for i, row in enumerate(rows):
|
||
bits = build_input(row)
|
||
cx, cy = predict_fn(bits)
|
||
for ps, power in zip(POWER_STRS, POWER_LEVELS):
|
||
t = flight_ticks(row["f0_distance"], power)
|
||
lx, ly = predict_linear(row, t)
|
||
e = euclid(lx + cx, ly + cy, row[f"{ps}_enemy_x"], row[f"{ps}_enemy_y"])
|
||
errors[ps].append(e)
|
||
if ps == REP_PS:
|
||
if i < 50: first50.append(e)
|
||
if i >= n - 50: last50.append(e)
|
||
# learn residual at representative power
|
||
t_r = flight_ticks(row["f0_distance"], REP_POWER)
|
||
lx_r, ly_r = predict_linear(row, t_r)
|
||
cx2, cy2 = predict_fn(bits) # use same prediction (already called above)
|
||
rx = row[f"{REP_PS}_enemy_x"] - (lx_r + cx)
|
||
ry = row[f"{REP_PS}_enemy_y"] - (ly_r + cy)
|
||
learn_fn(bits, rx, ry)
|
||
return errors, first50, last50
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 1. Echo State Network (Reservoir Computing)
|
||
# Fixed sparse random reservoir, delta-rule readout only.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class EchoStateNet:
|
||
def __init__(self, n_in=276, n_res=512, sparsity=0.1, seed=7):
|
||
rng = random.Random(seed)
|
||
self.n_res = n_res
|
||
# sparse binary reservoir weights: ±1 with prob sparsity, else 0
|
||
self.W_res = [[0.0]*n_res for _ in range(n_res)]
|
||
for i in range(n_res):
|
||
for j in range(n_res):
|
||
if rng.random() < sparsity:
|
||
self.W_res[i][j] = 1.0 if rng.random() < 0.5 else -1.0
|
||
# scale spectral radius to ~0.9 (approx: divide by expected density*n)
|
||
scale = sparsity * n_res * 0.9
|
||
if scale > 0:
|
||
for i in range(n_res):
|
||
for j in range(n_res):
|
||
self.W_res[i][j] /= scale
|
||
|
||
# input projection: random binary ±1
|
||
self.W_in = []
|
||
for i in range(n_res):
|
||
col = rng.randint(0, n_in - 1) # random input index
|
||
sign = 1 if rng.random() < 0.5 else -1
|
||
self.W_in.append((col, sign))
|
||
|
||
# reservoir state
|
||
self.state = [0.0] * n_res
|
||
# readout weights (x and y)
|
||
self.w_x = [0.0] * n_res
|
||
self.w_y = [0.0] * n_res
|
||
self.lr = 0.01
|
||
|
||
def _step(self, bits):
|
||
new_state = [0.0] * self.n_res
|
||
for i in range(self.n_res):
|
||
col, sign = self.W_in[i]
|
||
s = sign * bits[col]
|
||
for j in range(self.n_res):
|
||
s += self.W_res[i][j] * self.state[j]
|
||
new_state[i] = math.tanh(s)
|
||
self.state = new_state
|
||
|
||
def predict(self, bits):
|
||
self._step(bits)
|
||
cx = sum(self.w_x[i] * self.state[i] for i in range(self.n_res))
|
||
cy = sum(self.w_y[i] * self.state[i] for i in range(self.n_res))
|
||
return cx, cy
|
||
|
||
def learn(self, bits, rx, ry):
|
||
# delta rule on readout (reservoir already stepped in predict)
|
||
for i in range(self.n_res):
|
||
self.w_x[i] += self.lr * rx * self.state[i]
|
||
self.w_y[i] += self.lr * ry * self.state[i]
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 2. Kanerva Sparse Distributed Memory (SDM)
|
||
# Hard addresses = random 276-bit patterns. Read/write by Hamming proximity.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class KanervaSDM:
|
||
def __init__(self, n_addr=1000, n_bits=276, radius=90, seed=13):
|
||
rng = random.Random(seed)
|
||
self.n_bits = n_bits
|
||
self.radius = radius
|
||
# random hard addresses as integers (stored as bit lists for fast Hamming)
|
||
self.hard_addr = []
|
||
for _ in range(n_addr):
|
||
addr = [rng.randint(0, 1) for _ in range(n_bits)]
|
||
self.hard_addr.append(addr)
|
||
# storage cells: sum_x, sum_y, count
|
||
self.cells = [[0.0, 0.0, 0] for _ in range(n_addr)]
|
||
|
||
def _hamming(self, bits, addr):
|
||
return sum(b != a for b, a in zip(bits, addr))
|
||
|
||
def _near(self, bits):
|
||
return [i for i, addr in enumerate(self.hard_addr)
|
||
if self._hamming(bits, addr) <= self.radius]
|
||
|
||
def predict(self, bits):
|
||
near = self._near(bits)
|
||
hits = [(self.cells[i][0], self.cells[i][1], self.cells[i][2])
|
||
for i in near if self.cells[i][2] > 0]
|
||
if not hits:
|
||
return 0.0, 0.0
|
||
cx = sum(sx/c for sx, sy, c in hits) / len(hits)
|
||
cy = sum(sy/c for sx, sy, c in hits) / len(hits)
|
||
return cx, cy
|
||
|
||
def learn(self, bits, rx, ry):
|
||
for i in self._near(bits):
|
||
self.cells[i][0] += rx
|
||
self.cells[i][1] += ry
|
||
self.cells[i][2] += 1
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 3. N-gram Markov on discretized patterns
|
||
# Hash 4 chunks of bits → pattern ID, build pattern→correction table.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class NGramMarkov:
|
||
def __init__(self, chunk_size=69, n_chunks=4):
|
||
self.chunk_size = chunk_size
|
||
self.n_chunks = n_chunks
|
||
# table: pattern_key -> [sum_x, sum_y, count]
|
||
self.table = {}
|
||
|
||
def _key(self, bits):
|
||
chunks = []
|
||
for c in range(self.n_chunks):
|
||
start = c * self.chunk_size
|
||
end = min(start + self.chunk_size, len(bits))
|
||
val = 0
|
||
for b in bits[start:end]:
|
||
val = (val << 1) | b
|
||
chunks.append(val)
|
||
return tuple(chunks)
|
||
|
||
def predict(self, bits):
|
||
key = self._key(bits)
|
||
e = self.table.get(key)
|
||
if e is None or e[2] == 0:
|
||
return 0.0, 0.0
|
||
return e[0] / e[2], e[1] / e[2]
|
||
|
||
def learn(self, bits, rx, ry):
|
||
key = self._key(bits)
|
||
if key not in self.table:
|
||
self.table[key] = [0.0, 0.0, 0]
|
||
self.table[key][0] += rx
|
||
self.table[key][1] += ry
|
||
self.table[key][2] += 1
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 4. Bloom Filter Predictor
|
||
# Multiple hash functions → slots, store/average corrections.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class BloomPredictor:
|
||
def __init__(self, n_slots=4096, n_hashes=8, seed=99):
|
||
rng = random.Random(seed)
|
||
self.n_slots = n_slots
|
||
self.n_hashes = n_hashes
|
||
# random hash seeds
|
||
self.seeds = [rng.randint(1, 2**31) for _ in range(n_hashes)]
|
||
self.slots_x = [0.0] * n_slots
|
||
self.slots_y = [0.0] * n_slots
|
||
self.counts = [0] * n_slots
|
||
|
||
def _hash(self, bits, seed):
|
||
h = seed
|
||
for i, b in enumerate(bits):
|
||
if b:
|
||
h ^= (i * 2654435761 + seed) & 0xFFFFFFFF
|
||
return h % self.n_slots
|
||
|
||
def _addrs(self, bits):
|
||
return [self._hash(bits, s) for s in self.seeds]
|
||
|
||
def predict(self, bits):
|
||
addrs = self._addrs(bits)
|
||
hits = [(self.slots_x[a], self.slots_y[a], self.counts[a])
|
||
for a in addrs if self.counts[a] > 0]
|
||
if not hits:
|
||
return 0.0, 0.0
|
||
cx = sum(sx/c for sx, sy, c in hits) / len(hits)
|
||
cy = sum(sy/c for sx, sy, c in hits) / len(hits)
|
||
return cx, cy
|
||
|
||
def learn(self, bits, rx, ry):
|
||
for a in self._addrs(bits):
|
||
self.slots_x[a] += rx
|
||
self.slots_y[a] += ry
|
||
self.counts[a] += 1
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 5. Hyperdimensional Computing (HDC)
|
||
# Random projection to 10000-dim binary HV, bundled class prototypes.
|
||
# Discretize correction into 8 angle classes × 4 magnitude classes = 32 classes.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class HDC:
|
||
def __init__(self, n_in=276, n_hd=10000, n_classes=32, seed=17):
|
||
rng = random.Random(seed)
|
||
self.n_hd = n_hd
|
||
self.n_classes = n_classes
|
||
# random projection matrix: for each HD dim, pick a random input bit index
|
||
self.proj = [rng.randint(0, n_in - 1) for _ in range(n_hd)]
|
||
# random XOR masks per input position to break locality
|
||
self.masks = [rng.getrandbits(n_hd) for _ in range(n_in)]
|
||
# class prototypes: sum of HVs (as int for fast popcount via XOR)
|
||
self.proto_sum = [0] * (n_hd * n_classes) # flattened int sums
|
||
self.proto_count = [0] * n_classes
|
||
# store accumulated corrections per class
|
||
self.class_cx = [0.0] * n_classes
|
||
self.class_cy = [0.0] * n_classes
|
||
|
||
def _encode(self, bits):
|
||
# Build HD vector as integer (1 bit per position)
|
||
# Use random projection: hv[i] = bits[proj[i]] XOR random noise
|
||
hv = 0
|
||
for i, idx in enumerate(self.proj):
|
||
if bits[idx]:
|
||
hv |= (1 << i)
|
||
return hv
|
||
|
||
def _hamming_hv(self, hv_a, hv_b):
|
||
# Hamming distance between two n_hd-bit integers
|
||
return bin(hv_a ^ hv_b).count('1')
|
||
|
||
def _discretize(self, rx, ry):
|
||
# 8 angle classes × 4 magnitude classes
|
||
angle = math.atan2(ry, rx) # -pi..pi
|
||
angle_cls = int((angle + math.pi) / (2 * math.pi) * 8) % 8
|
||
mag = math.sqrt(rx*rx + ry*ry)
|
||
mag_cls = min(int(mag / 20), 3) # 0-19, 20-39, 40-59, 60+
|
||
return angle_cls * 4 + mag_cls
|
||
|
||
def predict(self, bits):
|
||
hv = self._encode(bits)
|
||
best_cls = -1
|
||
best_dist = self.n_hd + 1
|
||
for c in range(self.n_classes):
|
||
if self.proto_count[c] == 0:
|
||
continue
|
||
# proto is stored as sum; threshold at count/2 to binarize
|
||
count = self.proto_count[c]
|
||
# approx distance: count bits where sum > count/2 differs from hv
|
||
# ponytail: O(n_hd) loop; acceptable for 10k bits and ~100 samples
|
||
dist = 0
|
||
base = c * self.n_hd
|
||
for i in range(self.n_hd):
|
||
proto_bit = 1 if self.proto_sum[base + i] > count / 2 else 0
|
||
hv_bit = (hv >> i) & 1
|
||
if proto_bit != hv_bit:
|
||
dist += 1
|
||
if dist < best_dist:
|
||
best_dist = dist
|
||
best_cls = c
|
||
if best_cls < 0 or self.proto_count[best_cls] == 0:
|
||
return 0.0, 0.0
|
||
c = self.proto_count[best_cls]
|
||
return self.class_cx[best_cls] / c, self.class_cy[best_cls] / c
|
||
|
||
def learn(self, bits, rx, ry):
|
||
hv = self._encode(bits)
|
||
cls = self._discretize(rx, ry)
|
||
self.proto_count[cls] += 1
|
||
self.class_cx[cls] += rx
|
||
self.class_cy[cls] += ry
|
||
base = cls * self.n_hd
|
||
for i in range(self.n_hd):
|
||
self.proto_sum[base + i] += (hv >> i) & 1
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 6. Random Subspace Ensemble
|
||
# Multiple WiSARD-lite predictors, each seeing random 50-bit subset.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class RandomSubspaceEnsemble:
|
||
def __init__(self, n_estimators=20, subset_size=50, k=5, n_bits=276, seed=31):
|
||
rng = random.Random(seed)
|
||
self.estimators = []
|
||
for _ in range(n_estimators):
|
||
indices = random.sample(range(n_bits), subset_size)
|
||
tables = {} # addr -> [sum_x, sum_y, count]
|
||
self.estimators.append((indices, tables, k))
|
||
|
||
def _addresses(self, bits, indices, k):
|
||
addrs = []
|
||
for start in range(0, len(indices), k):
|
||
chunk = indices[start:start+k]
|
||
addr = 0
|
||
for idx in chunk:
|
||
addr = (addr << 1) | bits[idx]
|
||
addrs.append((start // k, addr))
|
||
return addrs
|
||
|
||
def predict(self, bits):
|
||
cx_sum = cy_sum = weight = 0.0
|
||
for indices, tables, k in self.estimators:
|
||
addrs = self._addresses(bits, indices, k)
|
||
est_cx = est_cy = 0.0
|
||
hit = 0
|
||
for node, addr in addrs:
|
||
key = (node, addr)
|
||
e = tables.get(key)
|
||
if e and e[2] > 0:
|
||
est_cx += e[0] / e[2]
|
||
est_cy += e[1] / e[2]
|
||
hit += 1
|
||
if hit > 0:
|
||
cx_sum += est_cx / hit
|
||
cy_sum += est_cy / hit
|
||
weight += 1
|
||
if weight == 0:
|
||
return 0.0, 0.0
|
||
return cx_sum / weight, cy_sum / weight
|
||
|
||
def learn(self, bits, rx, ry):
|
||
for indices, tables, k in self.estimators:
|
||
for node, addr in self._addresses(bits, indices, k):
|
||
key = (node, addr)
|
||
if key not in tables:
|
||
tables[key] = [0.0, 0.0, 0]
|
||
tables[key][0] += rx
|
||
tables[key][1] += ry
|
||
tables[key][2] += 1
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# 7. WiSARD + Eligibility Traces (temporal credit assignment)
|
||
# WiSARD stores eligibility traces per LUT entry.
|
||
# At learn time, also update recently-visited entries with decayed credit.
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
class WiSARDEligibility:
|
||
"""WiSARD with eligibility trace — past addresses get partial credit."""
|
||
|
||
def __init__(self, n_bits=276, k=12, trace_len=5, decay=0.7, seed=42):
|
||
self.k = k
|
||
n_pad = math.ceil(n_bits / k) * k
|
||
self.n_nodes = n_pad // k
|
||
rng = random.Random(seed)
|
||
indices = list(range(n_bits)) + [0] * (n_pad - n_bits)
|
||
self.perm = indices[:]
|
||
rng.shuffle(self.perm)
|
||
self.tables = [{} for _ in range(self.n_nodes)]
|
||
self.trace_len = trace_len
|
||
self.decay = decay
|
||
# history: deque of (addrs_list, weight)
|
||
self.history = collections.deque()
|
||
|
||
def _addresses(self, bits):
|
||
addrs = []
|
||
for node in range(self.n_nodes):
|
||
addr = 0
|
||
for bit_idx in range(self.k):
|
||
perm_idx = node * self.k + bit_idx
|
||
b = bits[self.perm[perm_idx]] if perm_idx < len(bits) else 0
|
||
addr = (addr << 1) | b
|
||
addrs.append(addr)
|
||
return addrs
|
||
|
||
def predict(self, bits):
|
||
addrs = self._addresses(bits)
|
||
cx_sum = cy_sum = 0.0
|
||
count = 0
|
||
for node, addr in enumerate(addrs):
|
||
e = self.tables[node].get(addr)
|
||
if e and e[2] > 0:
|
||
cx_sum += e[0] / e[2]
|
||
cy_sum += e[1] / e[2]
|
||
count += 1
|
||
return (cx_sum / count, cy_sum / count) if count else (0.0, 0.0)
|
||
|
||
def learn(self, bits, rx, ry):
|
||
addrs = self._addresses(bits)
|
||
# current step with full credit
|
||
self.history.appendleft((addrs, 1.0))
|
||
if len(self.history) > self.trace_len:
|
||
self.history.pop()
|
||
# update all traces
|
||
for past_addrs, weight in self.history:
|
||
for node, addr in enumerate(past_addrs):
|
||
e = self.tables[node].get(addr)
|
||
if e is None:
|
||
self.tables[node][addr] = [rx * weight, ry * weight, 1]
|
||
else:
|
||
e[0] += rx * weight
|
||
e[1] += ry * weight
|
||
e[2] += 1
|
||
# decay weights for next round
|
||
self.history = collections.deque(
|
||
[(a, w * self.decay) for a, w in self.history]
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
# Main
|
||
# ════════════════════════════════════════════════════════════════════════════════
|
||
|
||
def fmt_row(name, s_all, s_first, s_last, p1_mae):
|
||
mae = s_all.get("mae", float("nan"))
|
||
hit = s_all.get("hit", float("nan"))
|
||
f50 = s_first.get("mae", float("nan")) if s_first else float("nan")
|
||
l50 = s_last.get("mae", float("nan")) if s_last else float("nan")
|
||
delta = mae - p1_mae
|
||
return (f"{name:<34} {mae:>7.2f} {hit:>7.1f}% {f50:>8.2f} {l50:>8.2f} {delta:>+8.2f}")
|
||
|
||
def run_and_report(rows, name, predict_fn, learn_fn):
|
||
print(f" running {name}...")
|
||
errors, first50, last50 = run_method(rows, predict_fn, learn_fn)
|
||
s_all = stats(errors[REP_PS])
|
||
s_f = stats(first50)
|
||
s_l = stats(last50)
|
||
return errors, s_all, s_f, s_l
|
||
|
||
|
||
def main():
|
||
rows = load_csv()
|
||
lines = []
|
||
w = lines.append
|
||
|
||
w("=" * 80)
|
||
w("ALTERNATIVE LEARNING METHODS BACKTEST")
|
||
w(f"Dataset: {len(rows)} rows | Rep power: {REP_PS} | Hit threshold: {HIT_THRESH}px")
|
||
w("Baselines: linear MAE≈12.24 | WiSARD K=12 MAE≈9.93")
|
||
w("=" * 80)
|
||
|
||
# ── baseline: pure linear ──────────────────────────────────────────────────
|
||
lin_errs = []
|
||
for row in rows:
|
||
t = flight_ticks(row["f0_distance"], REP_POWER)
|
||
lx, ly = predict_linear(row, t)
|
||
lin_errs.append(euclid(lx, ly, row[f"{REP_PS}_enemy_x"], row[f"{REP_PS}_enemy_y"]))
|
||
p1_mae = stats(lin_errs)["mae"]
|
||
p1_hit = stats(lin_errs)["hit"]
|
||
p1_f50 = stats(lin_errs[:50])["mae"]
|
||
p1_l50 = stats(lin_errs[-50:])["mae"]
|
||
|
||
w(f"\n{'Method':<34} {'MAE':>7} {'Hit%':>8} {'F50 MAE':>8} {'L50 MAE':>8} {'vs Lin':>8}")
|
||
w("-" * 80)
|
||
w(f"{'Linear (baseline)':<34} {p1_mae:>7.2f} {p1_hit:>7.1f}% {p1_f50:>8.2f} {p1_l50:>8.2f} {'---':>8}")
|
||
w(f"{'WiSARD K=12 (reference)':<34} {'9.93':>7} {'~55%':>8} {'---':>8} {'---':>8} {'-2.31':>8}")
|
||
|
||
results = []
|
||
|
||
# ── 1. Echo State Network ─────────────────────────────────────────────────
|
||
esn = EchoStateNet(n_in=276, n_res=512, sparsity=0.05, seed=7)
|
||
_last_state = [None]
|
||
def esn_predict(bits):
|
||
cx, cy = esn.predict(bits)
|
||
return cx, cy
|
||
def esn_learn(bits, rx, ry):
|
||
esn.learn(bits, rx, ry)
|
||
|
||
e, sa, sf, sl = run_and_report(rows, "1. Echo State Network", esn_predict, esn_learn)
|
||
results.append(("1. Echo State Net (res=512)", sa, sf, sl))
|
||
w(fmt_row("1. Echo State Net (res=512)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 2. Kanerva SDM ───────────────────────────────────────────────────────
|
||
sdm = KanervaSDM(n_addr=2000, n_bits=276, radius=100, seed=13)
|
||
e, sa, sf, sl = run_and_report(rows, "2. Kanerva SDM", sdm.predict, sdm.learn)
|
||
results.append(("2. Kanerva SDM (addr=2000)", sa, sf, sl))
|
||
w(fmt_row("2. Kanerva SDM (addr=2000)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 3. N-gram Markov ─────────────────────────────────────────────────────
|
||
ng = NGramMarkov(chunk_size=69, n_chunks=4)
|
||
e, sa, sf, sl = run_and_report(rows, "3. N-gram Markov", ng.predict, ng.learn)
|
||
results.append(("3. N-gram Markov (4×69 chunks)", sa, sf, sl))
|
||
w(fmt_row("3. N-gram Markov (4×69 chunks)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 4. Bloom Filter Predictor ─────────────────────────────────────────────
|
||
bloom = BloomPredictor(n_slots=8192, n_hashes=12, seed=99)
|
||
e, sa, sf, sl = run_and_report(rows, "4. Bloom Filter", bloom.predict, bloom.learn)
|
||
results.append(("4. Bloom Filter (8192 slots)", sa, sf, sl))
|
||
w(fmt_row("4. Bloom Filter (8192 slots)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 5. HDC ───────────────────────────────────────────────────────────────
|
||
print(" running 5. HDC (slow, n_hd=2000)...")
|
||
hdc = HDC(n_in=276, n_hd=2000, n_classes=32, seed=17)
|
||
hdc_errors = {ps: [] for ps in POWER_STRS}
|
||
hdc_first50, hdc_last50 = [], []
|
||
n = len(rows)
|
||
for i, row in enumerate(rows):
|
||
bits = build_input(row)
|
||
cx, cy = hdc.predict(bits)
|
||
for ps, power in zip(POWER_STRS, POWER_LEVELS):
|
||
t = flight_ticks(row["f0_distance"], power)
|
||
lx, ly = predict_linear(row, t)
|
||
e_val = euclid(lx + cx, ly + cy, row[f"{ps}_enemy_x"], row[f"{ps}_enemy_y"])
|
||
hdc_errors[ps].append(e_val)
|
||
if ps == REP_PS:
|
||
if i < 50: hdc_first50.append(e_val)
|
||
if i >= n - 50: hdc_last50.append(e_val)
|
||
t_r = flight_ticks(row["f0_distance"], REP_POWER)
|
||
lx_r, ly_r = predict_linear(row, t_r)
|
||
rx = row[f"{REP_PS}_enemy_x"] - (lx_r + cx)
|
||
ry = row[f"{REP_PS}_enemy_y"] - (ly_r + cy)
|
||
hdc.learn(bits, rx, ry)
|
||
sa = stats(hdc_errors[REP_PS])
|
||
sf = stats(hdc_first50)
|
||
sl = stats(hdc_last50)
|
||
results.append(("5. HDC (n_hd=2000, 32 cls)", sa, sf, sl))
|
||
w(fmt_row("5. HDC (n_hd=2000, 32 cls)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 6. Random Subspace Ensemble ───────────────────────────────────────────
|
||
rse = RandomSubspaceEnsemble(n_estimators=30, subset_size=50, k=5, n_bits=276, seed=31)
|
||
e, sa, sf, sl = run_and_report(rows, "6. Random Subspace Ensemble", rse.predict, rse.learn)
|
||
results.append(("6. RandSubspace (30×50bits)", sa, sf, sl))
|
||
w(fmt_row("6. RandSubspace (30×50bits)", sa, sf, sl, p1_mae))
|
||
|
||
# ── 7. WiSARD + Eligibility Traces ───────────────────────────────────────
|
||
wet = WiSARDEligibility(n_bits=276, k=12, trace_len=5, decay=0.7, seed=42)
|
||
e, sa, sf, sl = run_and_report(rows, "7. WiSARD+Eligibility", wet.predict, wet.learn)
|
||
results.append(("7. WiSARD+Elig (k=12,tr=5)", sa, sf, sl))
|
||
w(fmt_row("7. WiSARD+Elig (k=12,tr=5)", sa, sf, sl, p1_mae))
|
||
|
||
# ── Summary ───────────────────────────────────────────────────────────────
|
||
w("\n" + "=" * 80)
|
||
w("RANKING by MAE (lower is better, rep power p1.07)")
|
||
w("=" * 80)
|
||
ranked = sorted(results, key=lambda x: x[1].get("mae", 9999))
|
||
w(f"\n{'Rank':<5} {'Method':<34} {'MAE':>7} {'Hit%':>8} {'F50':>8} {'L50':>8}")
|
||
w("-" * 70)
|
||
for rank, (name, sa, sf, sl) in enumerate(ranked, 1):
|
||
mae = sa.get("mae", float("nan"))
|
||
hit = sa.get("hit", float("nan"))
|
||
f50 = sf.get("mae", float("nan")) if sf else float("nan")
|
||
l50 = sl.get("mae", float("nan")) if sl else float("nan")
|
||
w(f"{rank:<5} {name:<34} {mae:>7.2f} {hit:>7.1f}% {f50:>8.2f} {l50:>8.2f}")
|
||
|
||
w(f"\n Linear baseline: MAE={p1_mae:.2f} Hit={p1_hit:.1f}%")
|
||
w(f" WiSARD K=12 ref: MAE=9.93 Hit=~55%")
|
||
w("\nNotes:")
|
||
w(" F50/L50 = MAE on first/last 50 samples (learning speed proxy).")
|
||
w(" Hit% = fraction within 18px (Robocode bullet half-width).")
|
||
w(" All methods: online, binary input, no gradients, no supervised labels.")
|
||
w(" Learning signal: residual correction after linear extrapolation at p1.07.")
|
||
|
||
report = "\n".join(lines)
|
||
print("\n" + report)
|
||
with open(OUT_PATH, "w") as f:
|
||
f.write(report + "\n")
|
||
print(f"\n[saved to {OUT_PATH}]")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|