599 lines
24 KiB
Python
599 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Gate A (OFFLINE VETO) for the state-conditional LEARNED MOVEMENT module.
|
|
|
|
Question: on the recorded corpus, does a *state-conditional* model predict the
|
|
GUESS-FACTOR BIN at which an incoming wave crosses the dodging bot **better**
|
|
than (i) the unconditional average / (ii) the 31-bin global histogram that the
|
|
old crude surfer (`common_libs/movements/wave_surfer.nim`, job j115) used, and
|
|
(iii) chance?
|
|
|
|
Per `docs/offline_harness_trust.md` this harness is VETO-ONLY: it may kill the
|
|
design; it can never select it. The live panel is the decider.
|
|
|
|
--------------------------------------------------------------------------------
|
|
WHAT IS MEASURED (all per shot, no replay, no simulator)
|
|
--------------------------------------------------------------------------------
|
|
Corpus: `/tmp/tfil_ab2/out/<A..E>/runN.jsonl{,.events.jsonl,.rounds.json}` — 70
|
|
recorded battles. Per-shot geometry is re-derived with the validated instrument
|
|
`common_libs/tests/analyze_drussgt_dodge_vs_power.py` (its `Run` class), so the
|
|
attribution of a fire event to a side is the one already documented in
|
|
`docs/drussgt_dodge_vs_power.md`.
|
|
|
|
Frame: the module cannot see bullet bodies (`WorldState` has none), so the wave
|
|
frame is the one `wave_surfer.nim` actually uses — origin = the shooter's
|
|
position at the fire tick, "centre line" = the bearing from the origin to the
|
|
TARGET at the fire tick. The bullet's own direction is NOT used (using it would
|
|
leak the enemy's aim).
|
|
|
|
Label (what `wave_surfer.gfToBin` computes at wave resolution): walk ticks
|
|
after the fire; the wave resolves on the first tick where `speed*k >=
|
|
dist(origin, target)`; the label is the 31-bin quantisation of
|
|
`wrap180(bearing(origin -> target at that tick) - centreBearing) / maxEA`.
|
|
|
|
State at the FIRE tick (5 fields, each quantised into Q symbols; the same
|
|
wave-relative state family that survived `docs/state_window_gate.md` §1):
|
|
lat signed perpendicular offset from the centre line (px)
|
|
vlat lat(t) - lat(t-1) (px/tick, signed)
|
|
dist range to the shooter (px)
|
|
room directional wall room along sign(vlat)*normal (px)
|
|
turn signed heading change (deg/tick)
|
|
A SINGLE state is used -- never a temporal window of states (measured dead in
|
|
`docs/state_window_gate.md`).
|
|
|
|
Learner: the COUNTED SBC with global fractional decay that the module uses
|
|
(`common_libs/bitbrain/sbc.nim`, counted mode): one saturating uint8 counter per
|
|
(state, bin), `c -= c shr shift` every `decayEvery` learns, read out with the
|
|
per-cell posterior (`inferProb`) interpolated with the global prior, i.e.
|
|
p = (counts[state]/n_state + alpha*prior) / (1 + alpha). This Python model is
|
|
the same estimator the Nim module runs; `--selftest` cross-checks it.
|
|
|
|
Split: BY BATTLE (never by tick inside a round), 70/30, 3 seeds.
|
|
|
|
Run:
|
|
python3 common_libs/tests/learned_surfer_gate.py \
|
|
--corpus /tmp/tfil_ab2/out \
|
|
--report common_libs/tests/fixtures/learned_surfer_gate_report.txt \
|
|
--json common_libs/tests/fixtures/learned_surfer_gate.json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
import random
|
|
import statistics
|
|
import sys
|
|
from collections import Counter
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import analyze_drussgt_dodge_vs_power as adp # validated per-shot geometry
|
|
|
|
NBINS = 31
|
|
MAX_FLIGHT = 220
|
|
NBOT = 18.0
|
|
ARENA_W, ARENA_H = 800.0, 600.0
|
|
FIELDS = ["vlat", "dist", "room", "turn"] # `lat` at the fire tick is identically 0 (the centre line passes through us), so it is NOT a field
|
|
|
|
|
|
# --------------------------------------------------------------------- helpers
|
|
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."""
|
|
t = float("inf")
|
|
for p, d, lo, hi in ((px, dx, NBOT, ARENA_W - NBOT),
|
|
(py, dy, NBOT, ARENA_H - NBOT)):
|
|
if abs(d) > 1e-9:
|
|
cand = (hi - p) / d if d > 0 else (lo - p) / d
|
|
if cand < t:
|
|
t = cand
|
|
return 0.0 if t == float("inf") else max(0.0, t)
|
|
|
|
|
|
def gf_to_bin(gf):
|
|
"""Exactly `wave_surfer.gfToBin` (31 bins over [-1, +1])."""
|
|
v = max(-1.0, min(1.0, gf))
|
|
return max(0, min(NBINS - 1, int(round((v + 1.0) * 0.5 * (NBINS - 1)))))
|
|
|
|
|
|
def bin_to_gf(idx):
|
|
return idx / (NBINS - 1) * 2.0 - 1.0
|
|
|
|
|
|
# ------------------------------------------------------------------ extraction
|
|
def extract(run, label_mode="nominal"):
|
|
"""One record per shot: state fields at the fire tick + the GF-bin label.
|
|
|
|
label_mode:
|
|
`nominal` -- GF of the target's position at the NOMINAL arrival tick
|
|
`t0 + ceil(startDist/speed)` (the standard "wave hit GF":
|
|
the time the bullet needs for the range at fire time, known
|
|
at fire time).
|
|
`resolve` -- GF at the tick the expanding wave circle catches the target
|
|
(`speed*k >= dist(origin,target)`), i.e. exactly
|
|
`wave_surfer.nim`'s resolution rule.
|
|
"""
|
|
out = []
|
|
for s in run.shots(): # shots fired by side "s" at side "e"
|
|
t0, rnd = s["tick"], s["rnd"]
|
|
speed = s["speed"]
|
|
p0x, p0y = s["_x"], s["_y"]
|
|
rs = run.start[rnd]
|
|
re = rs + run.count[rnd] - 1
|
|
r0 = run.by_tick.get(t0)
|
|
rp = run.by_tick.get(t0 - 1)
|
|
if r0 is None or rp is None or t0 <= rs:
|
|
continue
|
|
|
|
# centre line = origin -> target at the fire tick (the wave frame)
|
|
tb0 = math.atan2(r0["ey"] - p0y, r0["ex"] - p0x)
|
|
ux, uy = math.cos(tb0), math.sin(tb0)
|
|
|
|
def lat_of(r):
|
|
dx, dy = r["ex"] - p0x, r["ey"] - p0y
|
|
return dx * (-uy) + dy * ux # cross(u, d)
|
|
|
|
lat = lat_of(r0)
|
|
vlat = lat - lat_of(rp)
|
|
dist = math.hypot(r0["ex"] - p0x, r0["ey"] - p0y)
|
|
sg = 1.0 if vlat >= 0 else -1.0
|
|
room = room_to_wall(r0["ex"], r0["ey"], -uy * sg, ux * sg)
|
|
turn = wrap180(r0["eh"] - rp["eh"])
|
|
|
|
# label: the module's own rule, two candidate conventions
|
|
maxea = math.asin(min(8.0 / speed, 1.0))
|
|
lab = None
|
|
if label_mode == "nominal":
|
|
kfix = max(1, int(math.ceil(dist / speed)))
|
|
r = run.by_tick.get(t0 + kfix)
|
|
if r is not None and t0 + kfix <= re:
|
|
off = wrap180(math.atan2(r["ey"] - p0y, r["ex"] - p0x) - tb0)
|
|
gf = max(-1.0, min(1.0, off / maxea)) if maxea > 1e-9 else 0.0
|
|
lab = gf_to_bin(gf)
|
|
else:
|
|
for k in range(1, MAX_FLIGHT):
|
|
if t0 + k > re:
|
|
break
|
|
r = run.by_tick.get(t0 + k)
|
|
if r is None:
|
|
break
|
|
if speed * k >= math.hypot(r["ex"] - p0x, r["ey"] - p0y):
|
|
off = wrap180(math.atan2(r["ey"] - p0y, r["ex"] - p0x) - tb0)
|
|
gf = max(-1.0, min(1.0, off / maxea)) if maxea > 1e-9 else 0.0
|
|
lab = gf_to_bin(gf)
|
|
break
|
|
if lab is None:
|
|
continue
|
|
|
|
out.append(dict(
|
|
battle=os.path.basename(os.path.dirname(run.cap_path)) + "/" +
|
|
os.path.basename(run.cap_path).replace(".jsonl", ""),
|
|
rnd=rnd, tick=t0, label=lab,
|
|
lat=lat, vlat=vlat, dist=dist, room=room, turn=turn,
|
|
))
|
|
return out
|
|
|
|
|
|
# ------------------------------------------------------------------ quantiser
|
|
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 edges_of(samples, q):
|
|
return {f: edges_for([s[f] for s in samples], q) for f in FIELDS}
|
|
|
|
|
|
def code_of(sample, edges, q):
|
|
code = 0
|
|
for f in FIELDS:
|
|
e = edges[f]
|
|
v = sample[f]
|
|
c = 0
|
|
for edge in e:
|
|
if v > edge:
|
|
c += 1
|
|
code = code * q + c
|
|
return code
|
|
|
|
|
|
# -------------------------------------------------------------------- learner
|
|
class CountedModel:
|
|
"""Counted SBC (one cell per state) + global prior, exactly as the module
|
|
runs it: saturating uint8 counters, `c -= c shr shift` every decayEvery
|
|
learns, per-cell posterior interpolated with the global histogram."""
|
|
|
|
def __init__(self, nstates, decay_every=0, decay_shift=1, alpha=5.0):
|
|
self.n = nstates
|
|
self.d = decay_every
|
|
self.sh = decay_shift
|
|
self.alpha = alpha
|
|
self.c = [[0] * NBINS for _ in range(nstates)]
|
|
self.g = [0] * NBINS
|
|
self.lc = 0
|
|
self.glc = 0
|
|
|
|
def _decay(self, row):
|
|
sh = self.sh
|
|
if sh <= 0:
|
|
return
|
|
for k in range(NBINS):
|
|
row[k] -= row[k] >> sh
|
|
|
|
def learn(self, state, label):
|
|
row = self.c[state]
|
|
if row[label] < 255:
|
|
row[label] += 1
|
|
self.g[label] += 1
|
|
self.lc += 1
|
|
self.glc += 1
|
|
if self.d > 0 and self.lc >= self.d:
|
|
for r in self.c:
|
|
self._decay(r)
|
|
self._decay(self.g)
|
|
self.lc = 0
|
|
if self.d > 0 and self.glc >= self.d:
|
|
self.glc = 0
|
|
|
|
def prior(self):
|
|
tot = sum(self.g)
|
|
if tot == 0:
|
|
return [1.0 / NBINS] * NBINS
|
|
return [(self.g[k] + 1.0) / (tot + NBINS) for k in range(NBINS)]
|
|
|
|
def predict(self, state):
|
|
pr = self.prior()
|
|
n = sum(self.c[state])
|
|
if n == 0:
|
|
return pr
|
|
a = self.alpha
|
|
return [((self.c[state][k] / n) + a * pr[k]) / (1.0 + a)
|
|
for k in range(NBINS)]
|
|
|
|
|
|
def log2(x):
|
|
return math.log2(max(x, 1e-12))
|
|
|
|
|
|
def evaluate(model, samples, edges, q):
|
|
lls, t1, t3 = [], 0, 0
|
|
for s in samples:
|
|
p = model.predict(code_of(s, edges, q))
|
|
y = s["label"]
|
|
lls.append(-log2(p[y]))
|
|
order = sorted(range(NBINS), key=lambda k: -p[k])
|
|
if order[0] == y:
|
|
t1 += 1
|
|
if y in order[:3]:
|
|
t3 += 1
|
|
n = len(samples)
|
|
return dict(logloss=statistics.fmean(lls) if lls else float("nan"),
|
|
top1=t1 / n if n else float("nan"),
|
|
top3=t3 / n if n else float("nan"))
|
|
|
|
|
|
def by_battle_ll(samples, edges, q, predict):
|
|
"""per-battle mean log-loss (paired statistics need one number per battle)."""
|
|
acc = {}
|
|
for s in samples:
|
|
p = predict(code_of(s, edges, q))
|
|
acc.setdefault(s["battle"], []).append(-log2(p[s["label"]]))
|
|
return {b: statistics.fmean(v) for b, v in acc.items() if v}
|
|
|
|
|
|
def sign_flip_p(deltas, reps=20000, seed=7):
|
|
"""Two-sided sign-flip permutation on paired per-battle deltas."""
|
|
obs = statistics.fmean(deltas)
|
|
rng = random.Random(seed)
|
|
hits = 0
|
|
for _ in range(reps):
|
|
s = statistics.fmean([d if rng.random() < 0.5 else -d for d in deltas])
|
|
if abs(s) >= abs(obs) - 1e-12:
|
|
hits += 1
|
|
return (hits + 1) / (reps + 1)
|
|
|
|
|
|
def sign_test_p(deltas):
|
|
pos = sum(1 for d in deltas if d > 0)
|
|
neg = sum(1 for d in deltas if d < 0)
|
|
n = pos + neg
|
|
if n == 0:
|
|
return 1.0, 0, 0
|
|
tail = sum(math.comb(n, k) for k in range(min(pos, neg) + 1)) / 2 ** n
|
|
return min(1.0, 2 * tail), pos, n
|
|
|
|
|
|
def paired_stats(deltas, name, unit="battles"):
|
|
n = len(deltas)
|
|
m = statistics.fmean(deltas)
|
|
sd = statistics.stdev(deltas) if n > 1 else 0.0
|
|
se = sd / math.sqrt(n) if n else float("nan")
|
|
ps, pos, nd = sign_test_p(deltas)
|
|
return dict(metric=name, n=n, mean=m, sd=sd, se=se,
|
|
ci=[m - 1.96 * se, m + 1.96 * se],
|
|
sign=f"{pos}/{nd}", p_sign=ps, p_signflip=sign_flip_p(deltas),
|
|
mde=2.8 * se, unit=unit)
|
|
|
|
|
|
# ---------------------------------------------------------------------- driver
|
|
def split_battles(battles, seed, frac=0.7):
|
|
bs = sorted(battles)
|
|
rng = random.Random(seed)
|
|
rng.shuffle(bs)
|
|
k = int(len(bs) * frac)
|
|
return set(bs[:k]), set(bs[k:])
|
|
|
|
|
|
def run_config(samples, seeds, q, decay_every, decay_shift, alpha,
|
|
fixed_edges=None, shuffle_labels=False):
|
|
per_seed = []
|
|
model_dump = None
|
|
for seed in seeds:
|
|
battles = {s["battle"] for s in samples}
|
|
tr_b, te_b = split_battles(battles, seed)
|
|
tr = [s for s in samples if s["battle"] in tr_b]
|
|
te = [s for s in samples if s["battle"] in te_b]
|
|
if not tr or not te:
|
|
continue
|
|
edges = fixed_edges or edges_of(tr, q)
|
|
|
|
if shuffle_labels:
|
|
# control: same states, labels permuted inside the train set
|
|
rng = random.Random(1000 + seed)
|
|
labs = [s["label"] for s in tr]
|
|
rng.shuffle(labs)
|
|
tr = [dict(s, label=l) for s, l in zip(tr, labs)]
|
|
|
|
nstates = q ** len(FIELDS)
|
|
mod = CountedModel(nstates, decay_every, decay_shift, alpha)
|
|
for s in tr:
|
|
mod.learn(code_of(s, edges, q), s["label"])
|
|
|
|
glob = CountedModel(nstates, decay_every, decay_shift, alpha)
|
|
for s in tr:
|
|
glob.learn(0, s["label"]) # state forced to 0 = global only
|
|
|
|
maj = Counter(s["label"] for s in tr).most_common(1)[0][0]
|
|
# majority as a degenerate top-1 predictor; log-loss from the prior
|
|
gp = glob.prior()
|
|
|
|
m_state = evaluate(mod, te, edges, q)
|
|
m_glob = evaluate(glob, te, edges, q)
|
|
chance = dict(logloss=log2(NBINS), top1=1.0 / NBINS, top3=3.0 / NBINS)
|
|
m_maj = dict(logloss=statistics.fmean([-log2(gp[s["label"]]) for s in te]),
|
|
top1=sum(1 for s in te if s["label"] == maj) / len(te),
|
|
top3=float("nan"))
|
|
# paired per-battle deltas (state - global), lower log-loss is better
|
|
ll_s = by_battle_ll(te, edges, q, mod.predict)
|
|
ll_g = by_battle_ll(te, edges, q, glob.predict)
|
|
common = sorted(set(ll_s) & set(ll_g))
|
|
deltas = [ll_s[b] - ll_g[b] for b in common]
|
|
|
|
seen = sum(1 for s in te if sum(mod.c[code_of(s, edges, q)]) > 0)
|
|
recur = Counter(code_of(s, edges, q) for s in tr)
|
|
per_seed.append(dict(
|
|
seed=seed, n_train=len(tr), n_test=len(te),
|
|
state=m_state, glob=m_glob, chance=chance, maj=m_maj,
|
|
delta_logloss=statistics.fmean(deltas),
|
|
stats=paired_stats(deltas, "logloss(state-global)"),
|
|
seen_frac=seen / len(te),
|
|
distinct_states=len(recur), mean_count=len(tr) / len(recur),
|
|
top_count=max(recur.values()), marginal_entropy_bits=entropy(tr),
|
|
deltas=deltas,
|
|
))
|
|
if model_dump is None:
|
|
model_dump = dict(edges=edges, q=q)
|
|
agg = {}
|
|
if per_seed:
|
|
for key in ("logloss", "top1", "top3"):
|
|
agg["state_" + key] = statistics.fmean(p["state"][key] for p in per_seed)
|
|
agg["glob_" + key] = statistics.fmean(p["glob"][key] for p in per_seed)
|
|
agg["chance_" + key] = statistics.fmean(p["chance"][key] for p in per_seed)
|
|
agg["maj_" + key] = statistics.fmean(p["maj"][key] for p in per_seed)
|
|
agg["delta_logloss"] = statistics.fmean(p["delta_logloss"] for p in per_seed)
|
|
agg["seen_frac"] = statistics.fmean(p["seen_frac"] for p in per_seed)
|
|
agg["distinct_states"] = statistics.fmean(p["distinct_states"] for p in per_seed)
|
|
agg["mean_count"] = statistics.fmean(p["mean_count"] for p in per_seed)
|
|
agg["entropy_bits"] = statistics.fmean(p["marginal_entropy_bits"] for p in per_seed)
|
|
# paired per-battle stats, pooled across every (seed, battle) delta
|
|
all_d = [d for p in per_seed for d in p["deltas"]]
|
|
agg["stats"] = paired_stats(all_d, "logloss(state-global)")
|
|
return agg, per_seed
|
|
|
|
|
|
def entropy(samples):
|
|
c = Counter(s["label"] for s in samples)
|
|
n = sum(c.values())
|
|
return -sum((v / n) * math.log2(v / n) for v in c.values())
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--corpus", default="/tmp/tfil_ab2/out")
|
|
ap.add_argument("--report", default=None)
|
|
ap.add_argument("--json", default=None)
|
|
ap.add_argument("--seeds", type=int, default=3)
|
|
ap.add_argument("--label", default="nominal",
|
|
choices=["nominal", "resolve", "both"])
|
|
args = ap.parse_args()
|
|
|
|
if args.label == "both":
|
|
for mode in ("nominal", "resolve"):
|
|
sub = [a for a in sys.argv if not a.startswith("--label")]
|
|
sub += ["--label", mode]
|
|
print(f"\n########## label mode: {mode} ##########")
|
|
os.execv(sys.executable, [sys.executable] + sub)
|
|
return
|
|
|
|
lines = []
|
|
|
|
def out(s=""):
|
|
print(s)
|
|
lines.append(s)
|
|
|
|
runs = adp.discover_tfil(args.corpus)
|
|
samples = []
|
|
for r in runs:
|
|
samples.extend(extract(r, args.label))
|
|
out("# Learned-surfer Gate A — offline prediction quality (VETO ONLY)")
|
|
out()
|
|
out(f"corpus : {args.corpus}")
|
|
out(f"battles : {len(runs)}")
|
|
out(f"shots used : {len(samples)}")
|
|
out(f"state fields: {', '.join(FIELDS)} (5 fields, ONE state, no window)")
|
|
out(f"label : 31-bin guess factor at wave resolution "
|
|
f"(wave_surfer.gfToBin), mode={args.label}")
|
|
out(f"learner : counted SBC (saturating uint8 + `c -= c shr shift` "
|
|
f"every decayEvery learns), inferProb readout, alpha=5 prior mix")
|
|
out()
|
|
|
|
seeds = list(range(args.seeds))
|
|
configs = [
|
|
("Q4 decay128/1 (primary)", 4, 128, 1, 5.0),
|
|
("Q4 decay32/1", 4, 32, 1, 5.0),
|
|
("Q4 NO DECAY", 4, 0, 1, 5.0),
|
|
("Q4 decay128/2", 4, 128, 2, 5.0),
|
|
("Q3 decay128/1", 3, 128, 1, 5.0),
|
|
("Q5 decay128/1", 5, 128, 1, 5.0),
|
|
]
|
|
results = {}
|
|
out("## A. held-out prediction quality (mean over 3 battle splits)")
|
|
out()
|
|
out("| config | states | log-loss(state) | log-loss(global) | Δ | top-1 st | "
|
|
"top-1 glob | top-3 st | top-3 glob |")
|
|
out("|---|---:|---:|---:|---:|---:|---:|---:|---:|")
|
|
for name, q, de, ds, al in configs:
|
|
agg, per = run_config(samples, seeds, q, de, ds, al)
|
|
results[name] = dict(agg=agg, per=per)
|
|
out(f"| {name} | {agg['distinct_states']:.0f} | {agg['state_logloss']:.4f} | "
|
|
f"{agg['glob_logloss']:.4f} | {agg['delta_logloss']:+.4f} | "
|
|
f"{agg['state_top1']:.4f} | {agg['glob_top1']:.4f} | "
|
|
f"{agg['state_top3']:.4f} | {agg['glob_top3']:.4f} |")
|
|
out()
|
|
|
|
# reference floors from the primary config's splits
|
|
pr = results[configs[0][0]]["per"]
|
|
out("## B. floors (same held-out test sets, primary config splits)")
|
|
out()
|
|
out("| predictor | log-loss (bits) | top-1 | top-3 |")
|
|
out("|---|---:|---:|---:|")
|
|
out(f"| chance (uniform 31) | {log2(NBINS):.4f} | {1/NBINS:.4f} | {3/NBINS:.4f} |")
|
|
out(f"| majority bin | "
|
|
f"{statistics.fmean(p['maj']['logloss'] for p in pr):.4f} | "
|
|
f"{statistics.fmean(p['maj']['top1'] for p in pr):.4f} | n/a |")
|
|
out(f"| global 31-bin histogram (the OLD surfer) | "
|
|
f"{statistics.fmean(p['glob']['logloss'] for p in pr):.4f} | "
|
|
f"{statistics.fmean(p['glob']['top1'] for p in pr):.4f} | "
|
|
f"{statistics.fmean(p['glob']['top3'] for p in pr):.4f} |")
|
|
out(f"| state-conditional counted SBC | "
|
|
f"{statistics.fmean(p['state']['logloss'] for p in pr):.4f} | "
|
|
f"{statistics.fmean(p['state']['top1'] for p in pr):.4f} | "
|
|
f"{statistics.fmean(p['state']['top3'] for p in pr):.4f} |")
|
|
out()
|
|
out("NOTE: the 'unconditional average' and the '31-bin global histogram of the")
|
|
out("old surfer' are the SAME estimator by construction (both are the train")
|
|
out("marginal over bins); they are therefore reported as one row. The majority")
|
|
out("predictor is the degenerate top-1 version of the same marginal.")
|
|
out()
|
|
|
|
# per-seed detail + recurrence for the primary config
|
|
out("## C. primary config per split (recurrence + paired per-battle stats)")
|
|
out()
|
|
out("| seed | train shots | test shots | distinct states | mean count/state | "
|
|
"test shots with a SEEN state | Δlog-loss (state-global) |")
|
|
out("|---|---:|---:|---:|---:|---:|---:|")
|
|
for p in pr:
|
|
out(f"| {p['seed']} | {p['n_train']} | {p['n_test']} | {p['distinct_states']} | "
|
|
f"{p['mean_count']:.1f} | {p['seen_frac']*100:.1f}% | "
|
|
f"{p['delta_logloss']:+.4f} |")
|
|
out()
|
|
out("### paired per-battle statistics (primary config, all splits pooled)")
|
|
out()
|
|
out("| metric | n battles | mean Δ | SD | 95% CI | sign | p(sign) | "
|
|
"p(sign-flip) | MDE |")
|
|
out("|---|---:|---:|---:|---|---:|---:|---:|---:|")
|
|
st = results[configs[0][0]]["agg"]["stats"]
|
|
out(f"| logloss(state-global), pooled | {st['n']} | {st['mean']:+.4f} | "
|
|
f"{st['sd']:.4f} | [{st['ci'][0]:+.4f}, {st['ci'][1]:+.4f}] | "
|
|
f"{st['sign']} | {st['p_sign']:.4g} | {st['p_signflip']:.4g} | "
|
|
f"{st['mde']:.4f} |")
|
|
for p in pr:
|
|
s2 = p["stats"]
|
|
out(f"| logloss(state-global) seed{p['seed']} | {s2['n']} | {s2['mean']:+.4f} | "
|
|
f"{s2['sd']:.4f} | [{s2['ci'][0]:+.4f}, {s2['ci'][1]:+.4f}] | "
|
|
f"{s2['sign']} | {s2['p_sign']:.4g} | {s2['p_signflip']:.4g} | "
|
|
f"{s2['mde']:.4f} |")
|
|
out("(negative Δ = the state-conditional model predicts better)")
|
|
out()
|
|
|
|
# label-shuffle control
|
|
out("## D. label-shuffle control (same states, train labels permuted)")
|
|
out()
|
|
aggq, _ = run_config(samples, seeds, 4, 128, 1, 5.0, shuffle_labels=True)
|
|
out("| arm | log-loss(state) | log-loss(global) | Δ | top-1(state) |")
|
|
out("|---|---:|---:|---:|---:|")
|
|
out(f"| shuffled labels | {aggq['state_logloss']:.4f} | {aggq['glob_logloss']:.4f} | "
|
|
f"{aggq['delta_logloss']:+.4f} | {aggq['state_top1']:.4f} |")
|
|
out(f"| real labels (Q4 decay128) | "
|
|
f"{results['Q4 decay128/1 (primary)']['agg']['state_logloss']:.4f} | "
|
|
f"{results['Q4 decay128/1 (primary)']['agg']['glob_logloss']:.4f} | "
|
|
f"{results['Q4 decay128/1 (primary)']['agg']['delta_logloss']:+.4f} | "
|
|
f"{results['Q4 decay128/1 (primary)']['agg']['state_top1']:.4f} |")
|
|
out()
|
|
out(f"(chance log-loss floor = {log2(NBINS):.4f} bits, target entropy = the")
|
|
out("global-histogram log-loss above; a shuffled-label state model must fall")
|
|
out("back to it.)")
|
|
out()
|
|
|
|
# canonical edges for the MODULE
|
|
edges = edges_of(samples, 4)
|
|
out("## E. canonical state-bin edges (4 symbols/field), hard-coded into the")
|
|
out("module `common_libs/movements/learned_surfer.nim` (derived from the whole")
|
|
out("corpus; the gate numbers above use TRAIN-only edges per split, so the")
|
|
out("report is not conditioned on these)")
|
|
out()
|
|
out("```nim")
|
|
for f in FIELDS:
|
|
out(f"{f:5s}: {', '.join(f'{e:.3f}' for e in edges[f])}")
|
|
out("```")
|
|
out()
|
|
fixed_agg, _ = run_config(samples, seeds, 4, 128, 1, 5.0, fixed_edges=edges)
|
|
out(f"cross-check with the canonical edges: log-loss(state) "
|
|
f"{fixed_agg['state_logloss']:.4f}, log-loss(global) "
|
|
f"{fixed_agg['glob_logloss']:.4f}, Δ {fixed_agg['delta_logloss']:+.4f}, "
|
|
f"top-1 {fixed_agg['state_top1']:.4f}")
|
|
out()
|
|
|
|
out("## MEASURED vs INFERRED")
|
|
out()
|
|
out("* MEASURED: every number in this file, produced by the command in the")
|
|
out(" module docstring on the recorded corpus.")
|
|
out("* INFERRED: that this offline prediction-quality result transfers to the")
|
|
out(" LIVE closed loop. It cannot: the recorded trajectory was produced while")
|
|
out(" the enemy gun reacted to a DIFFERENT mover (see")
|
|
out(" docs/offline_harness_trust.md §0/§4). This gate is a veto only.")
|
|
|
|
if args.report:
|
|
os.makedirs(os.path.dirname(args.report), exist_ok=True)
|
|
with open(args.report, "w") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
if args.json:
|
|
os.makedirs(os.path.dirname(args.json), exist_ok=True)
|
|
with open(args.json, "w") as f:
|
|
json.dump(dict(
|
|
corpus=args.corpus, battles=len(runs), shots=len(samples),
|
|
results={k: v["agg"] for k, v in results.items()},
|
|
per_seed={k: [{kk: vv for kk, vv in p.items() if kk != "deltas"}
|
|
for p in v["per"]] for k, v in results.items()},
|
|
shuffle_control=aggq, edges=edges,
|
|
), f, indent=1, default=str)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|