SirStone/SirRoboGarage
1
0
Fork 0
Code Issues 28 Pull Requests Actions Projects Releases Wiki Activity
Files
47ce244740b847c19db384866a18e7e6a950bb3b
SirRoboGarage/common_libs/tests/state_window_gate.py
T
SirStone d85ff53d34 State-window gate: a temporal window of wave-relative states does NOT beat a single state 
Re-runs the SBC coincidence premise as a cheap veto test on the 70-battle
live-vs-DrussGT corpus (/tmp/tfil_ab2).  Defines the wave-relative state
(lat/vlat/toa/room/turn, 5/7.9/10 bits at Q=2/3/4), quantises the miss offset
at the bullet's arrival into 7 bins, and sweeps window length K in
{1,4,8,16,32,48} with an interpolated suffix-backoff model under a BY-BATTLE
70/30 split (3 seeds).

Result: NO.  On the pre-fire frame the window is worse than the single
fire-tick state at every K/Q/A (e.g. K=8 costs +0.35..+0.46 bits).  On the
during-flight frame the entire apparent gain is the later decision tick, not
the window; the single state alone drops 2.70 -> 1.31 bits as K goes 1 -> 32.
The shuffle-order control confirms recency matters but the windows do not:
by Q=4/K=8 they average ~1 observation and never recur.  The single state
survives as a strong predictor (log-loss 2.346 vs 2.698 majority; bin
accuracy 0.409 vs 0.235).

Gate only: no gun, no live-win claim.
2026-09-25 08:43:36 +02:00

580 lines
23 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""STATE-WINDOW GATE: does a temporal WINDOW of wave-relative states predict
DrussGT's future lateral position better than a SINGLE state?
This is the cheap veto test for the "feed SBC a temporal list of states" design
(docs/state_window_gate.md). It is NOT a gun and makes no live-win claim.
DATA (real live battles, never regenerated here)
/tmp/tfil_ab2/out/<A..E>/runN.jsonl + .events.jsonl + .rounds.json
70 battles / 490 rounds / ~55k shots fired by ModularBot at the real
unmodified DrussGT, recorded by tools/robocode_shim/run_bridge_battle.sh.
In the capture rows `e*` is DrussGT (the subject) and `s*` is ModularBot (us);
the per-shot geometry is re-derived by the validated instrument in
common_libs/tests/analyze_drussgt_dodge_vs_power.py, which this file imports.
STATE (a design artifact -- see docs/state_window_gate.md for the rationale)
One state at absolute tick t, in the frame of the bullet fired at t0 along
direction u = (cos dir, sin dir):
lat = (D(t) - P0) x u lateral offset from the bullet line, px
vlat = lat(t) - lat(t-1) lateral velocity, px/tick (crossing/returning)
toa = (t0 - t) + karr ticks until the bullet reaches arrival
room = ray distance from D(t) along sign(vlat)*n until the arena wall, px
turn = wrap180(eh(t) - eh(t-1)) signed turn rate, deg/tick
Each field is quantised into Q in {2,3,4} bins (the numerosity dial). A
window is K consecutive states; K=1 is the single-state baseline.
Two frames are tested, and BOTH compare a window against the single state at
the SAME decision tick D (otherwise a longer window would win only because its
decision tick is later):
pre D = t0 (the fire tick); the window is the K pre-fire states ending at D.
fly D = t0+K-1; the window is the first K states of the flight, and the
baseline is the single state at D. Only shots with karr > 31 are used,
so the whole window is strictly before the bullet's arrival.
TARGET
perp_arr = DrussGT's SIGNED perpendicular offset from the bullet line at the
tick our bullet reaches its along-track plane (the miss offset that decides
the hit). Quantised into 7 bins (edges +-120, +-60, +-18 px); the central
+-18 px bin is the hit window.
MODEL (deliberately dull: the question is about INFORMATION, not modelling)
An interpolated (Jelinek-Mercer) suffix-backoff table over the quantised
window: P = global; for j=1..K, P <- (count(suffix_j) + A*P)/(total+A). This
is the direct analogue of the SBC "count coincidences" idea, it is
order-sensitive, and it can never do much worse than the shorter context, so
the sweep isolates information rather than overfitting. The context depth is
capped at 12 (orders above that are never observed often enough to matter).
A in {1,5,20} is swept as a robustness check (A=5 is primary).
SPLIT
BY BATTLE, never by tick. All rounds of a battle go to one side. 70 %/30 %
battle split, repeated over 3 seeds; model hyper-parameters (bin edges) are
derived from the TRAIN side only.
CONTROLS
1. shuffle -- random permutation of the K states inside each window (the
multiset is preserved, order destroyed). MANDATORY.
2. reverse -- deterministic reversal (preserves recurrence, reverses time).
3. majority -- no-window floor.
4. recurrence -- distinct windows, and how often a window repeats.
Run:
python3 common_libs/tests/state_window_gate.py --tfil /tmp/tfil_ab2/out \
--json common_libs/tests/fixtures/state_window_gate.json \
> common_libs/tests/fixtures/state_window_gate_report.txt
"""
from __future__ import annotations
import argparse
import bisect
import collections
import json
import math
import os
import random
import statistics
import sys
from array import array
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import analyze_drussgt_dodge_vs_power as dodge # noqa: E402
ARENA_W, ARENA_H = 800.0, 600.0
BOT_R = 18.0
MAXK_PRE = 48 # pre-fire history depth (all K <= 48 are feasible)
MAXK_FLY = 32 # during-flight window depth (max real flight is 43)
FIELDS = ("lat", "vlat", "toa", "room", "turn")
D_CAP = 12 # model context depth cap (see docs; orders > cap never observed)
TARGET_EDGES = [-120.0, -60.0, -18.0, 18.0, 60.0, 120.0]
HIT_BIN = 3 # bin [ -18, 18 ) -> the bot radius
K_PRE = (1, 4, 8, 16, 32, 48)
K_FLY = (1, 4, 8, 16, 32)
def wrap180(a):
return ((a + 180.0) % 360.0) - 180.0
def room_to_wall(px, py, dx, dy):
"""Distance from (px,py) along unit (dx,dy) until leaving the arena
(accounting for the 18 px bot radius)."""
t = float("inf")
for p, d, lo, hi in ((px, dx, BOT_R, ARENA_W - BOT_R),
(py, dy, BOT_R, ARENA_H - BOT_R)):
if abs(d) > 1e-9:
cand = (hi - p) / d if d > 0 else (lo - p) / d
if cand < t:
t = cand
if t == float("inf"):
return 0.0
return max(0.0, t)
def tbin(v):
return bisect.bisect_right(TARGET_EDGES, v)
def raw_state(run, t0, karr, P0, ux, uy, t, rnd):
rs = run.start[rnd]
re = rs + run.count[rnd] - 1
tc = min(max(t, rs), re)
tp = max(tc - 1, rs)
r = run.by_tick.get(tc)
rp = run.by_tick.get(tp)
if r is None or rp is None:
return None
lat = (r["ex"] - P0[0]) * uy - (r["ey"] - P0[1]) * ux
latp = (rp["ex"] - P0[0]) * uy - (rp["ey"] - P0[1]) * ux
vlat = lat - latp
toa = (t0 - tc) + karr
turn = wrap180(r["eh"] - rp["eh"])
sg = 1.0 if vlat >= 0 else -1.0
room = room_to_wall(r["ex"], r["ey"], -uy * sg, ux * sg)
return (lat, vlat, toa, room, turn)
def extract_run(run):
out = []
for s in run.shots():
t0 = s["tick"]
karr = s["flight"]
if karr is None:
continue
th = math.radians(s["_dir"])
ux, uy = math.cos(th), math.sin(th)
P0 = (s["_x"], s["_y"])
rnd = s["rnd"]
pre = array("f")
ok = True
for off in range(-(MAXK_PRE - 1), 1):
rs = raw_state(run, t0, karr, P0, ux, uy, t0 + off, rnd)
if rs is None:
ok = False
break
pre.extend(rs)
if not ok:
continue
fly = []
for off in range(MAXK_FLY):
rs = raw_state(run, t0, karr, P0, ux, uy, t0 + off, rnd)
fly.append(None if rs is None else rs)
out.append(dict(
battle=os.path.basename(os.path.dirname(run.cap_path)) + "/" +
os.path.basename(run.cap_path),
rnd=rnd, t0=t0, karr=karr, perp_arr=s["perp_arr"],
perp_fire=s["perp_fire"], range=s["range"], power=s["power"],
hit=s["hit"], tbin=tbin(s["perp_arr"]), pre=pre, fly=fly,
))
return out
# ------------------------------------------------------------------ quantise
def edges_for(vals, q):
xs = sorted(vals)
n = len(xs)
return [xs[min(n - 1, int((qi / q) * n))] for qi in range(1, q)]
def frame_edges(samples, frame, Q):
"""Quantile bin edges per field, derived from the TRAIN side only."""
edges = []
for f in range(5):
vals = []
if frame == "pre":
for s in samples:
if s["_split"] != "train":
continue
a = s["pre"]
for i in range(MAXK_PRE):
vals.append(a[i * 5 + f])
else:
for s in samples:
if s["_split"] != "train":
continue
for st in s["fly"]:
if st is not None:
vals.append(st[f])
edges.append(edges_for(vals, Q))
return edges
def apply_codes(samples, frame, edges, Q):
res = []
for s in samples:
if frame == "pre":
a = s["pre"]
n = MAXK_PRE
codes = [0] * n
for i in range(n):
base = i * 5
c = 0
for f in range(5):
c += bisect.bisect_right(edges[f], a[base + f]) * (Q ** f)
codes[i] = c
else:
a = s["fly"]
n = MAXK_FLY
codes = [0] * n
for i in range(n):
if a[i] is None:
codes[i] = -1
continue
c = 0
for f in range(5):
c += bisect.bisect_right(edges[f], a[i][f]) * (Q ** f)
codes[i] = c
res.append(codes)
return res
# ------------------------------------------------------------------ the model
def order_window(w, mode, rng):
if mode == "temporal":
return w
if mode == "reverse":
return w[::-1]
if mode == "shuffled":
return [w[i] for i in rng.sample(range(len(w)), len(w))]
raise ValueError(mode)
def build_counts(windows, ys, mode, Dcap, rng):
"""cbyorder[j][context] = [ {target_bin: count}, total ].
The model is a standard interpolated (Jelinek-Mercer) suffix backoff: for a
window w the prediction for the next-state target is
P = global
for j = 1..K: P = (count(suffix_j) + A*P) / (total(suffix_j) + A)
so a longer context is only believed as far as the data supports it, and the
model can never do much worse than the shorter one. This keeps the MODEL
uninteresting, which is what the gate needs."""
cbyorder = [dict() for _ in range(Dcap + 1)]
for w, y in zip(windows, ys):
wo = order_window(w, mode, rng)
ctx = ()
for j in range(1, min(Dcap, len(wo)) + 1):
ctx = (wo[-j],) + ctx
e = cbyorder[j].get(ctx)
if e is None:
e = [{}, 0]
cbyorder[j][ctx] = e
e[0][y] = e[0].get(y, 0) + 1
e[1] += 1
return cbyorder
NBINS = len(TARGET_EDGES) + 1
def glob_vec(glob, nbins=NBINS):
tot = sum(glob.values())
return [(glob.get(b, 0)) / tot for b in range(nbins)]
def predict(w, cbyorder, P0, K, A):
P = list(P0)
for j in range(1, min(K, len(w), len(cbyorder) - 1) + 1):
ctx = tuple(w[-j:])
e = cbyorder[j].get(ctx)
if e is None:
continue
cnt, tot = e
P = [(cnt.get(b, 0) + A * P[b]) / (tot + A) for b in range(len(P))]
return P
def evaluate(windows, ys, cbyorder, P0, mode, Kdepth, A, rng):
acc = 0
n = 0
ll = 0.0
hitp = 0.0
for w, y in zip(windows, ys):
wo = order_window(w, mode, rng)
P = predict(wo, cbyorder, P0, Kdepth, A)
best = max(range(len(P)), key=lambda b: (P[b], -b))
if best == y:
acc += 1
ll += -math.log2(max(P[y], 1e-12))
hitp += P[HIT_BIN]
n += 1
return {"n": n, "acc": acc / n, "logloss": ll / n,
"implied_hit": hitp / n}
def majority_floor(samples):
train = [s for s in samples if s["_split"] == "train"]
glob = collections.Counter(s["tbin"] for s in train)
top = max(glob, key=lambda b: glob[b])
acc = ll = 0.0
n = 0
test_hit = 0
for s in samples:
if s["_split"] != "test":
continue
n += 1
if s["tbin"] == top:
acc += 1
ll += -math.log2(glob[s["tbin"]] / sum(glob.values()))
test_hit += 1 if s["tbin"] == HIT_BIN else 0
return {"top": top, "n": n, "acc": acc / n, "logloss": ll / n,
"emp_hit": test_hit / n,
"dist": {str(k): v / sum(glob.values()) for k, v in sorted(glob.items())}}
def recurrence(windows, K):
"""Distinct windows and repeat rate over the WHOLE corpus (no split)."""
seen = collections.Counter()
for w in windows:
k = min(K, len(w))
seen[tuple(w[len(w) - k:])] += 1
total = len(windows)
rep2 = sum(c for c in seen.values() if c >= 2)
return {"distinct": len(seen), "total": total, "repeat_frac": rep2 / total,
"mean_count": total / len(seen)}
# ------------------------------------------------------------------ driver
def split_battles(samples, seed):
battles = sorted({s["battle"] for s in samples})
rng = random.Random(seed)
rng.shuffle(battles)
ntr = int(round(0.70 * len(battles)))
train = set(battles[:ntr])
for s in samples:
s["_split"] = "train" if s["battle"] in train else "test"
def window_list(codes, frame, K):
"""The ordered window for one sample.
pre frame: the full 48-state pre-fire history ENDING at the fire tick; the
model's depth parameter selects how many of the most recent
states it may use, so K=1 is the fire-tick state itself.
fly frame: the first K states of the flight, i.e. ending at t0+K-1; the
single-state baseline is then the state at that same tick.
"""
if frame == "pre":
return list(codes)
return list(codes[:K])
def run_frame(samples, frame, Ks, seeds, As):
res = {"frame": frame, "seeds": {}}
for seed in seeds:
split_battles(samples, seed)
tr = [i for i, s in enumerate(samples) if s["_split"] == "train"]
te = [i for i, s in enumerate(samples) if s["_split"] == "test"]
ytr = [samples[i]["tbin"] for i in tr]
yte = [samples[i]["tbin"] for i in te]
seed_res = {"majority": majority_floor(samples), "Q": {}}
for Q in (2, 3, 4):
edges = frame_edges(samples, frame, Q)
codes = apply_codes(samples, frame, edges, Q)
glob = {}
for i in tr:
glob[samples[i]["tbin"]] = glob.get(samples[i]["tbin"], 0) + 1
P0 = glob_vec(glob)
qres = {}
for mode in ("temporal", "shuffled", "reverse"):
mres = {str(A): {} for A in As}
cache = {}
for K in Ks:
wid = "full" if frame == "pre" else K
if wid not in cache:
wl = [window_list(codes[i], frame, K) for i in range(len(samples))]
cby = build_counts([wl[i] for i in tr], ytr, mode, D_CAP,
random.Random(1000 + seed))
cache[wid] = (cby, wl)
cby, wl = cache[wid]
wte = [wl[i] for i in te]
kd = min(K, D_CAP)
for A in As:
rnge = random.Random(4000 + seed * 13 + K)
mres[str(A)][str(K)] = {
"window": evaluate(wte, yte, cby, P0, mode, kd, A, rnge)}
qres[mode] = mres
# single-state-at-the-same-decision-tick baseline (true recent state)
sres = {str(A): {} for A in As}
for K in Ks:
wid = "full" if frame == "pre" else K
wl = [window_list(codes[i], frame, K) for i in range(len(samples))]
cby = build_counts([wl[i] for i in tr], ytr, "temporal", D_CAP,
random.Random(1000 + seed))
wte = [wl[i] for i in te]
for A in As:
sres[str(A)][str(K)] = evaluate(
wte, yte, cby, P0, "temporal", 1, A,
random.Random(5000 + seed * 17 + K))
qres["single"] = sres
qres["recurrence"] = {str(K): recurrence(
[window_list(codes[i], frame, K) for i in range(len(samples))], K)
for K in Ks}
seed_res["Q"][str(Q)] = qres
res["seeds"][str(seed)] = seed_res
return res
def mean_over(xs):
return statistics.fmean(xs) if xs else float("nan")
def cell(res, frame, Q, mode, A, K, key="window"):
if key == "single":
q = lambda sd: res[frame]["seeds"][sd]["Q"][str(Q)]["single"][str(A)][str(K)]
else:
q = lambda sd: res[frame]["seeds"][sd]["Q"][str(Q)][mode][str(A)][str(K)][key]
a = [q(sd) for sd in res[frame]["seeds"]]
return (mean_over([c["acc"] for c in a]),
mean_over([c["logloss"] for c in a]),
mean_over([c["implied_hit"] for c in a]))
def render(res, out, A_primary):
lines = []
def p(s=""):
lines.append(s)
print(s)
for frame in res:
Ks = K_PRE if frame == "pre" else K_FLY
p("=" * 100)
p("FRAME %s%s" % (frame, " (window ENDS at the fire tick, looks BACK)"
if frame == "pre" else
" (window STARTS at the fire tick, ends at t0+K-1)"))
p("=" * 100)
maj = [res[frame]["seeds"][sd]["majority"] for sd in res[frame]["seeds"]]
p("MAJORITY / NO-WINDOW FLOOR (test acc %.4f, log-loss %.4f bits,"
" empirical hit %.4f)" % (mean_over([m["acc"] for m in maj]),
mean_over([m["logloss"] for m in maj]),
mean_over([m["emp_hit"] for m in maj])))
p(" target-bin edges %s px -> central +-18 px hit bin. At the median"
% TARGET_EDGES)
p(" fire range (487 px) the 36 px hit window subtends %.2f deg; at 450 px"
% math.degrees(2 * math.atan(18.0 / 487.0)))
p(" it is %.2f deg; at 100 px %.2f deg. (atan(18/range).)"
% (math.degrees(2 * math.atan(18.0 / 450.0)),
math.degrees(2 * math.atan(18.0 / 100.0))))
p(" target-bin distribution on train: %s" %
", ".join("%s:%.3f" % (k, v) for k, v in sorted(maj[0]["dist"].items())))
p()
for Q in (2, 3, 4):
nstates = Q ** 5
p("-" * 100)
p("COARSENESS Q=%d -> %d distinct single states (%.1f bits)"
% (Q, nstates, math.log2(nstates)))
p(" %-4s | %-24s | %-24s | %-13s | %-13s" %
("K", "single@D (temporal)", "window (temporal)", "window=SHUFFLED",
"window=REVERSE"))
p(" %-4s | %-24s | %-24s | %-13s | %-13s" %
("", "logloss acc hitP", "logloss acc hitP", "logloss",
"logloss"))
for K in Ks:
sc = cell(res, frame, Q, "temporal", A_primary, K, "single")
tc = cell(res, frame, Q, "temporal", A_primary, K, "window")
sh = cell(res, frame, Q, "shuffled", A_primary, K, "window")
rv = cell(res, frame, Q, "reverse", A_primary, K, "window")
p(" %-4d | %.4f %.4f %.4f | %.4f %.4f %.4f | %.4f | %.4f"
% (K, sc[1], sc[0], sc[2], tc[1], tc[0], tc[2],
sh[1], rv[1]))
rec = res[frame]["seeds"][sorted(res[frame]["seeds"])[0]]["Q"][str(Q)]["recurrence"]
p(" recurrence, distinct ordered window tuples over the whole corpus:")
for K in Ks:
r = rec[str(K)]
p(" K=%-2d distinct=%-8d of %-6d (mean count %.2f, repeat_frac %.3f)"
% (K, r["distinct"], r["total"], r["mean_count"], r["repeat_frac"]))
p()
p("=" * 100)
p("HEADLINE (mean over %d battle-split seeds; A=%g)" %
(len(res[frame]["seeds"]), A_primary))
p(" held-out log-loss. delta_window = window - single@D (NEGATIVE = the")
p(" window beats the single state at the same decision tick)")
for Q in (2, 3, 4):
srow = [cell(res, frame, Q, "temporal", A_primary, K, "single")[1] for K in Ks]
trow = [cell(res, frame, Q, "temporal", A_primary, K, "window")[1] for K in Ks]
p(" Q=%d single@D " % Q + " ".join("K=%d %.4f" % (K, v)
for K, v in zip(Ks, srow)))
p(" window " + " ".join("K=%d %.4f(%+.4f)" % (K, t, t - s)
for K, t, s in zip(Ks, trow, srow)))
p(" shuffle control: window(shuffled) - window(temporal) (must be >>0)")
for Q in (2, 3, 4):
row = [cell(res, frame, Q, "shuffled", A_primary, K, "window")[1] -
cell(res, frame, Q, "temporal", A_primary, K, "window")[1]
for K in Ks]
p(" Q=%d " % Q + " ".join("K=%d %+.4f" % (K, v) for K, v in zip(Ks, row)))
p(" reverse control: window(reverse) - window(temporal)")
for Q in (2, 3, 4):
row = [cell(res, frame, Q, "reverse", A_primary, K, "window")[1] -
cell(res, frame, Q, "temporal", A_primary, K, "window")[1]
for K in Ks]
p(" Q=%d " % Q + " ".join("K=%d %+.4f" % (K, v) for K, v in zip(Ks, row)))
p(" robustness in the interpolation strength A (window temporal log-loss):")
for Q in (2, 3, 4):
for A in out.get("A_all", []):
row = [cell(res, frame, Q, "temporal", A, K, "window")[1] for K in Ks]
p(" Q=%d A=%-4g " % (Q, A) +
" ".join("K=%d %.4f" % (K, v) for K, v in zip(Ks, row)))
p("=" * 100)
out["report"] = lines
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--tfil", default="/tmp/tfil_ab2/out")
ap.add_argument("--json", default=None)
ap.add_argument("--limit-runs", type=int, default=0)
ap.add_argument("--seed", type=int, default=1)
ap.add_argument("--seeds", type=int, default=3)
ap.add_argument("--frames", default="pre")
ap.add_argument("--alphas", default="1,5,20")
a = ap.parse_args()
runs = dodge.discover_tfil(a.tfil)
if a.limit_runs:
runs = runs[:a.limit_runs]
print("[gate] %d battles" % len(runs), file=sys.stderr)
samples = []
for k, run in enumerate(runs):
samples.extend(extract_run(run))
if (k + 1) % 10 == 0:
print("[gate] %d/%d battles, %d shots" % (k + 1, len(runs),
len(samples)), file=sys.stderr)
print("[gate] %d samples" % len(samples), file=sys.stderr)
hit = sum(1 for s in samples if s["tbin"] == HIT_BIN)
print("[gate] empirical arrival hit bin (|perp|<18): %.4f" % (hit / len(samples)),
file=sys.stderr)
seeds = [a.seed + i for i in range(a.seeds)]
As = [float(x) for x in a.alphas.split(",")]
A_primary = 5.0
res = {}
nfly = 0
for frame in a.frames.split(","):
if frame == "pre":
fs = samples
else:
kmin = max(K_FLY) + 1 # keep the whole window strictly before arrival
fs = [s for s in samples if s["karr"] >= kmin]
nfly = len(fs)
print("[gate] frame %s: %d shots with karr >= %d" %
(frame, len(fs), kmin), file=sys.stderr)
res[frame] = run_frame(fs, frame, K_PRE if frame == "pre" else K_FLY,
seeds, As)
out = {"samples": len(samples), "battles": len(runs), "seeds": seeds,
"fly_samples": nfly,
"alphas": As, "A_primary": A_primary, "A_all": As,
"D_cap": D_CAP, "target_edges": TARGET_EDGES, "hit_bin": HIT_BIN,
"k_pre": list(K_PRE), "k_fly": list(K_FLY)}
render(res, out, A_primary)
out["raw"] = res
if a.json:
with open(a.json, "w") as f:
json.dump(out, f, indent=1, default=str)
return out
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink