SirStone
1adefaba26
Answers the user's hypothesis that DrussGT dodges low-power shots better. Measured on 70 live battles / 490 rounds / 54939 real shots vs real DrussGT (/tmp/tfil_ab2) and replicated on 35 more battles / 24280 shots (/tmp/powtest). Power is not randomly assigned - our policy caps it by RANGE (TR_POWER_FAR_DIST=200 -> 1.0) and by OUR OWN ENERGY (the slope), so inside a range band power is almost a deterministic function of our energy and a naive low-vs-high comparison is secretly a losing-vs-healthy comparison. Everything is stratified by range band and backed by a within-band shuffled-label null (arrival re-derived, so the null keeps the kinematic channel), a round-cluster bootstrap, and a within-shot CONTROL window 40 ticks later when the bullet is long gone. RESULT: no behavioural response. In band 450+ the raw miss distance at arrival is +8.25 px [+5.39,+11.25] for HIGH power - but per flight tick it is 4.52 vs 4.51 px/tick (delta -0.01 [-0.12,+0.10]), i.e. entirely the 2.13-tick longer flight window of the slower bullet. Fixed-12-tick lateral displacement is flat (55.63 vs 55.47, -0.15 [-1.15,+0.84]) and turn rate / speed are flat. The whole difference is already present 5 ticks after the trigger pull (+4.2 px) and is just as large in the bullet-free control window (+5.6 px), so it is a property of the low-energy situation, not of the shot. Hit rate is flat (0.10 vs 0.09). Corpus/attribution notes: e*=DrussGT (subject), s*=ModularBot, per TrBattleCapture.java; the Tank-Royale owner id is NOT stable across runs and is recovered per battle from fire geometry + the energy decrement, cross-checked on 496/496 death events. Geometry validated on the server's own hits (mean miss 11.6 px, 80.6% inside the 18 px radius).
796 lines
34 KiB
Python
796 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Does DrussGT's dodge quality change with the bullet power WE fire?
|
|
|
|
Reads REAL live-vs-real-DrussGT captures (per-tick worldstate jsonl + a
|
|
fire/hit event sidecar + a per-round tick index) produced by
|
|
``tools/robocode_shim/run_bridge_battle.sh``, and measures DrussGT's movement
|
|
response per shot, conditioned on the power ModularBot fired. No offline
|
|
fixture replay, no simulator: these are the recorded live battles.
|
|
|
|
--------------------------------------------------------------------------------
|
|
WHY THE OBVIOUS ANALYSIS IS WRONG
|
|
--------------------------------------------------------------------------------
|
|
Power is NOT randomly assigned. Our power policy (common_libs/gun_harness/
|
|
virtual_bullets.nim) only ever *caps* the gun's preference, by:
|
|
* RANGE -- TR_POWER_FAR_DIST=200 -> cap 1.0 beyond it (prFar);
|
|
* OUR ENERGY -- a linear slope from TR_POWER_ENERGY_MIN=0.5 at <=20 energy to
|
|
the cap at >=80 (prEnergySlope);
|
|
* the finishing rule (prFinishKill) and the sub-average-chances rule.
|
|
So "shots we fired at low power" are disproportionately long-range shots and
|
|
late/low-energy shots. Every comparison below is therefore stratified by the
|
|
fire-time range band, the headline is the stratified one, and the between-group
|
|
difference is compared against a NULL that permutes the power labels inside the
|
|
band (re-deriving the arrival tick so the kinematic channel survives the null).
|
|
We also report the confounder itself (our energy, enemy energy) per cell.
|
|
|
|
--------------------------------------------------------------------------------
|
|
WHAT IS MEASURED
|
|
--------------------------------------------------------------------------------
|
|
Per shot fired by US (ModularBot) at DrussGT. u = (cos dir, sin dir) is the
|
|
direction the bullet was fired; the bullet flies from the fire position along u
|
|
at v = 20 - 3p px/tick.
|
|
(a) MISS AT ARRIVAL -- perpendicular px between DrussGT's position on the
|
|
tick our bullet reaches its along-track plane and our bullet's intended
|
|
line. = |perp(perpendicular offset)| at that tick. Bot radius 18 px.
|
|
(b) LATERAL DISPLACEMENT -- perp offset change over the flight window, and the
|
|
SAME over a FIXED 12-tick window after the shot (power-neutral).
|
|
(c) RESPONSE LATENCY -- ticks from our fire until DrussGT's heading has turned
|
|
> LAT_DEG from its fire-tick heading.
|
|
(d) MEAN TURN RATE and MEAN SPEED over the flight window and the fixed window.
|
|
(e) DODGE DIRECTION -- did it move FURTHER off the aim line (|perp| grows).
|
|
(f) FLIGHT WINDOW LENGTH -- low power is a FASTER bullet, so a SHORTER window.
|
|
Outcome: hit rate per cell, resolved from the real hit/hitwall events.
|
|
|
|
--------------------------------------------------------------------------------
|
|
THE CONTROL
|
|
--------------------------------------------------------------------------------
|
|
A real bullet response must appear in the window the bullet is in the air. A
|
|
phase/distance artifact would appear just as strongly in a LATER window of the
|
|
same shot, when the bullet is long gone. So every metric is also computed over
|
|
a CONTROL window at [fire+CTRL_OFF, fire+CTRL_OFF+12] and the two are compared.
|
|
Anything that shows up in both is a property of the situation, not of the shot.
|
|
|
|
Run:
|
|
python3 common_libs/tests/analyze_drussgt_dodge_vs_power.py \
|
|
--tfil /tmp/tfil_ab2/out --powtest /tmp/powtest \
|
|
--json common_libs/tests/fixtures/dodge_vs_power_results.json
|
|
|
|
MEASURED = the numbers in the tables below (real recorded battles).
|
|
INFERRED = the attributions and the causal reading; both are flagged in the doc.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import collections
|
|
import json
|
|
import math
|
|
import os
|
|
import random
|
|
import statistics
|
|
|
|
BOT_RADIUS = 18.0
|
|
SPEED_A, SPEED_B = 20.0, 3.0 # bullet speed = 20 - 3p
|
|
LAT_DEG = 15.0 # heading turn that counts as a response
|
|
FIXED_WIN = 12 # power-neutral movement window (ticks)
|
|
CTRL_OFF = 40 # control window starts here after the fire
|
|
RANGE_MATCH_PX = 50 # range sub-bin used to match on range as well
|
|
MAX_FLIGHT = 220
|
|
BANDS = [(0, 100), (100, 200), (200, 300), (300, 450), (450, 10000)]
|
|
BAND_LABELS = ["0-100", "100-200", "200-300", "300-450", "450+"]
|
|
PWR_EDGES = [0.0, 0.5, 0.75, 1.0, 1.5, 4.0]
|
|
PWR_LABELS = ["<0.50", "0.50-0.75", "0.75-1.00", "1.00-1.50", ">=1.50"]
|
|
|
|
|
|
def pbin(p):
|
|
for i in range(len(PWR_EDGES) - 1):
|
|
if PWR_EDGES[i] <= p < PWR_EDGES[i + 1]:
|
|
return i
|
|
return None
|
|
|
|
|
|
def band_of(r):
|
|
for i, (lo, hi) in enumerate(BANDS):
|
|
if lo <= r < hi:
|
|
return BAND_LABELS[i]
|
|
return BAND_LABELS[-1]
|
|
|
|
|
|
def wrap180(x):
|
|
return abs(((x + 180.0) % 360.0) - 180.0)
|
|
|
|
|
|
def mean(xs):
|
|
return statistics.fmean(xs) if xs else float("nan")
|
|
|
|
|
|
def pct(xs, q):
|
|
if not xs:
|
|
return float("nan")
|
|
xs = sorted(xs)
|
|
return xs[min(len(xs) - 1, int(q * len(xs)))]
|
|
|
|
|
|
# --------------------------------------------------------------------- loader
|
|
def _rows(path):
|
|
out = []
|
|
with open(path) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
o = json.loads(line)
|
|
if "tick" in o:
|
|
out.append(o)
|
|
return out
|
|
|
|
|
|
class Run:
|
|
"""One live battle: capture rows, event sidecar, per-round tick index."""
|
|
|
|
def __init__(self, cap_path, events_path, rounds_path):
|
|
self.cap_path = cap_path
|
|
self.rows = _rows(cap_path)
|
|
self.by_tick = {r["tick"]: r for r in self.rows}
|
|
self.events = [json.loads(l) for l in open(events_path) if l.strip()]
|
|
self.rounds = json.load(open(rounds_path))["rounds"]
|
|
self.start = {r["round"]: r["startTick"] for r in self.rounds}
|
|
self.count = {r["round"]: r["count"] for r in self.rounds}
|
|
self.owner_side = self._resolve_owner_side()
|
|
self.resolution = {}
|
|
for ev in self.events:
|
|
if ev.get("type") in ("hit", "hitwall", "hitbullet"):
|
|
self.resolution[(ev["round"], ev["owner"], ev["bullet"])] = ev["type"]
|
|
|
|
# -- attribution ------------------------------------------------------
|
|
def _match(self, t, side, ev):
|
|
r, r1 = self.by_tick.get(t), self.by_tick.get(t + 1)
|
|
if r is None or r1 is None:
|
|
return False
|
|
if abs(r[side + "x"] - ev["x"]) > 0.02 or abs(r[side + "y"] - ev["y"]) > 0.02:
|
|
return False
|
|
ef = "ee" if side == "e" else "se"
|
|
return abs((r[ef] - r1[ef]) - ev["power"]) < 0.02
|
|
|
|
def _resolve_owner_side(self):
|
|
"""Tank-Royale owner id -> 'e' (DrussGT, the subject) or 's' (us).
|
|
|
|
The id is NOT stable across runs (bot start order varies), so it is
|
|
recovered per battle: a fire event's (x,y) is the firing bot's own
|
|
position and that bot's energy drops by exactly the power one tick
|
|
later. Cross-checked against the death events (see main())."""
|
|
votes = collections.defaultdict(collections.Counter)
|
|
for ev in self.events:
|
|
if ev.get("type") != "fire":
|
|
continue
|
|
guess = self.start.get(ev["round"], 0) + ev["tick"]
|
|
for t in range(guess - 8, guess + 9):
|
|
for side in ("e", "s"):
|
|
if self._match(t, side, ev):
|
|
votes[ev["owner"]][side] += 1
|
|
return {o: c.most_common(1)[0][0] for o, c in votes.items() if c}
|
|
|
|
def align(self, ev):
|
|
guess = self.start.get(ev["round"], 0) + ev["tick"]
|
|
side = self.owner_side.get(ev["owner"])
|
|
best = None
|
|
for t in range(guess - 8, guess + 9):
|
|
for sd in (("e", "s") if side is None else (side,)):
|
|
if self._match(t, sd, ev):
|
|
d = abs(t - guess)
|
|
if best is None or d < best[0]:
|
|
best = (d, t, sd)
|
|
return None if best is None else best[1]
|
|
|
|
# -- per-shot extraction ---------------------------------------------
|
|
def shots(self):
|
|
for ev in self.events:
|
|
if ev.get("type") != "fire" or self.owner_side.get(ev["owner"]) != "s":
|
|
continue
|
|
t0 = self.align(ev)
|
|
if t0 is None:
|
|
continue
|
|
s = self._shot(ev, t0)
|
|
if s is not None:
|
|
yield s
|
|
|
|
def _shot(self, ev, t0):
|
|
r0 = self.by_tick[t0]
|
|
p = ev["power"]
|
|
th = math.radians(ev["dir"])
|
|
ux, uy = math.cos(th), math.sin(th)
|
|
x0, y0 = ev["x"], ev["y"]
|
|
rnd = ev["round"]
|
|
end = self.start[rnd] + self.count[rnd]
|
|
|
|
rng = math.hypot(r0["ex"] - x0, r0["ey"] - y0)
|
|
# per-tick geometry along/perpendicular to the aim line, up to the round
|
|
# end (capped); kept so the permutation null can re-derive the arrival.
|
|
along, perp, rows = [], [], []
|
|
karr = None
|
|
perp_arr = None
|
|
MIN_V = SPEED_A - SPEED_B * PWR_EDGES[-2] # slowest bullet the null can make
|
|
for k in range(1, MAX_FLIGHT + 1):
|
|
if t0 + k >= end:
|
|
break
|
|
r = self.by_tick.get(t0 + k)
|
|
if r is None:
|
|
break
|
|
dx, dy = r["ex"] - x0, r["ey"] - y0
|
|
along.append(dx * ux + dy * uy)
|
|
perp.append(dx * uy - dy * ux)
|
|
rows.append(r)
|
|
if karr is None and along[-1] > 0 and (SPEED_A - SPEED_B * p) * k >= along[-1]:
|
|
karr, perp_arr = k, perp[-1]
|
|
# stop once the real arrival is known, the null's slower arrival could
|
|
# still land, and the control window has been collected
|
|
if karr is not None and k >= CTRL_OFF + FIXED_WIN and MIN_V * k > along[-1]:
|
|
break
|
|
if karr is None:
|
|
return None
|
|
|
|
kmax = len(along)
|
|
g0 = (r0["ex"] - x0) * uy - (r0["ey"] - y0) * ux
|
|
|
|
# (b) fixed 12-tick window
|
|
lat_fixed = None
|
|
if kmax >= FIXED_WIN:
|
|
lat_fixed = perp[FIXED_WIN - 1] - g0
|
|
# control window (bullet long gone, same phase and range)
|
|
lat_ctrl = None
|
|
if kmax >= CTRL_OFF + FIXED_WIN:
|
|
lat_ctrl = perp[CTRL_OFF + FIXED_WIN - 1] - perp[CTRL_OFF - 1]
|
|
|
|
# (c) response latency
|
|
h0 = r0["eh"]
|
|
lat = None
|
|
for i in range(karr):
|
|
if wrap180(rows[i]["eh"] - h0) > LAT_DEG:
|
|
lat = i + 1
|
|
break
|
|
|
|
def kin(lo, n):
|
|
"""mean |turn rate| and mean |speed| over ticks [lo+1, lo+n]."""
|
|
if lo > len(rows):
|
|
return None, None
|
|
tt, hh = [], []
|
|
prev = r0 if lo == 0 else rows[lo - 1]
|
|
for j in range(lo, min(lo + n, kmax)):
|
|
r = rows[j]
|
|
tt.append(wrap180(r["eh"] - prev["eh"]))
|
|
hh.append(abs(r["es"]))
|
|
prev = r
|
|
if not tt:
|
|
return None, None
|
|
return mean(tt), mean(hh)
|
|
|
|
tr_f, sp_f = kin(0, karr)
|
|
tr_x, sp_x = kin(0, FIXED_WIN)
|
|
tr_c, sp_c = kin(CTRL_OFF, FIXED_WIN)
|
|
kind = self.resolution.get((rnd, ev["owner"], ev["bullet"]))
|
|
|
|
return dict(
|
|
run=self.cap_path, rnd=rnd, tick=t0, power=p, speed=SPEED_A - SPEED_B * p,
|
|
range=rng, band=band_of(rng), flight=karr,
|
|
perp_fire=g0, perp_arr=perp_arr, miss=abs(perp_arr),
|
|
miss_per_tick=abs(perp_arr) / karr,
|
|
lat_disp=perp_arr - g0, lat_disp_abs=abs(perp_arr - g0),
|
|
lat_disp_per_tick=abs(perp_arr - g0) / karr,
|
|
lat_fixed=lat_fixed, lat_fixed_abs=None if lat_fixed is None else abs(lat_fixed),
|
|
lat_ctrl=lat_ctrl, lat_ctrl_abs=None if lat_ctrl is None else abs(lat_ctrl),
|
|
dodge_further=1.0 if abs(perp_arr) > abs(g0) else 0.0,
|
|
sign_consist=1.0 if (perp_arr - g0) * perp_arr > 0 else 0.0,
|
|
latency=lat, turn_flight=tr_f, speed_flight=sp_f,
|
|
turn_fixed=tr_x, speed_fixed=sp_x,
|
|
turn_ctrl=tr_c, speed_ctrl=sp_c,
|
|
resolution=kind, hit=1.0 if kind == "hit" else 0.0,
|
|
our_energy=r0["se"], enemy_energy=r0["ee"],
|
|
_dir=ev["dir"], _x=x0, _y=y0, _along=along, _perp=perp,
|
|
)
|
|
|
|
@staticmethod
|
|
def _arrival(along, perp, v):
|
|
for k in range(1, len(along) + 1):
|
|
if along[k - 1] > 0 and v * k >= along[k - 1]:
|
|
return k, perp[k - 1]
|
|
return None
|
|
|
|
|
|
def discover_tfil(root):
|
|
runs = []
|
|
for sub in sorted(os.listdir(root)):
|
|
d = os.path.join(root, sub)
|
|
if not os.path.isdir(d):
|
|
continue
|
|
for fn in sorted(os.listdir(d)):
|
|
if not fn.endswith(".jsonl") or fn.endswith(".events.jsonl"):
|
|
continue
|
|
cap = os.path.join(d, fn)
|
|
ev, rj = cap[:-6] + ".events.jsonl", cap + ".rounds.json"
|
|
if os.path.exists(ev) and os.path.exists(rj):
|
|
runs.append(Run(cap, ev, rj))
|
|
return runs
|
|
|
|
|
|
def discover_powtest(root, arms=None):
|
|
runs = []
|
|
for fn in sorted(os.listdir(root)):
|
|
if not (fn.startswith("cap_") and fn.endswith(".jsonl")):
|
|
continue
|
|
arm = fn[4:].split("_r")[0]
|
|
if arms and arm not in arms:
|
|
continue
|
|
cap = os.path.join(root, fn)
|
|
ev = os.path.join(root, "events_" + fn[4:].replace(".jsonl", ".json"))
|
|
rj = cap + ".rounds.json"
|
|
if os.path.exists(ev) and os.path.exists(rj):
|
|
runs.append(Run(cap, ev, rj))
|
|
return runs
|
|
|
|
|
|
# --------------------------------------------------------------- statistics
|
|
def cell_stats(shots, key, only=None):
|
|
"""{(band, pbin): (n, mean, sd)}"""
|
|
acc = collections.defaultdict(list)
|
|
for s in shots:
|
|
if only and not only(s):
|
|
continue
|
|
v = s.get(key)
|
|
if v is None:
|
|
continue
|
|
i = pbin(s["power"])
|
|
if i is None:
|
|
continue
|
|
acc[(s["band"], i)].append(v)
|
|
return {k: (len(v), mean(v), statistics.pstdev(v) if len(v) > 1 else 0.0)
|
|
for k, v in acc.items()}
|
|
|
|
|
|
def perm_key(s, p):
|
|
"""Re-derive the arrival-dependent metrics for a substituted power."""
|
|
v = SPEED_A - SPEED_B * p
|
|
arr = Run._arrival(s["_along"], s["_perp"], v)
|
|
if arr is None:
|
|
return None
|
|
k, pe = arr
|
|
return {"power": p, "flight": k, "miss": abs(pe),
|
|
"miss_per_tick": abs(pe) / k,
|
|
"lat_disp_abs": abs(pe - s["perp_fire"]),
|
|
"lat_disp_per_tick": abs(pe - s["perp_fire"]) / k}
|
|
|
|
|
|
ARRIVAL_METRICS = ("miss", "lat_disp_abs", "flight", "miss_per_tick",
|
|
"lat_disp_per_tick")
|
|
|
|
|
|
def perm_contrast_null(shots, key, reps, seed=4242):
|
|
"""Per range band, the permutation null of delta = mean(HIGH) - mean(LOW).
|
|
|
|
For the arrival-dependent metrics each shot carries TWO precomputed values:
|
|
the metric it would have had at a typical LOW power and at a typical HIGH
|
|
power of that band (arrival tick re-derived, so the kinematic channel is
|
|
preserved). The null then draws a random subset of n_high shots to be the
|
|
HIGH arm -- which is exactly a permutation of the power values among the
|
|
shots of the band. Metrics that do not depend mechanically on power get the
|
|
same value in both arms, i.e. a plain label permutation.
|
|
|
|
Returns {band: (observed delta, null sd, two-sided p)}.
|
|
"""
|
|
rng = random.Random(seed)
|
|
by_band = collections.defaultdict(list)
|
|
for s in shots:
|
|
if grp(s) is not None and s.get(key) is not None:
|
|
by_band[s["band"]].append(s)
|
|
out = {}
|
|
for b, ss in by_band.items():
|
|
lo = [s for s in ss if grp(s) == "LOW"]
|
|
hi = [s for s in ss if grp(s) == "HIGH"]
|
|
if len(lo) < 15 or len(hi) < 15:
|
|
continue
|
|
p_lo = mean([s["power"] for s in lo])
|
|
p_hi = mean([s["power"] for s in hi])
|
|
if key in ARRIVAL_METRICS:
|
|
def val(s, p):
|
|
d = perm_key(s, p)
|
|
return None if d is None else d[key]
|
|
lows = [val(s, p_lo) for s in ss]
|
|
highs = [val(s, p_hi) for s in ss]
|
|
keep = [i for i in range(len(ss)) if lows[i] is not None and highs[i] is not None]
|
|
lows = [lows[i] for i in keep]
|
|
highs = [highs[i] for i in keep]
|
|
tag = [1 if grp(ss[i]) == "HIGH" else 0 for i in keep]
|
|
obs = (mean([highs[j] for j in range(len(keep)) if tag[j] == 1])
|
|
- mean([lows[j] for j in range(len(keep)) if tag[j] == 0]))
|
|
else:
|
|
vals = [s[key] for s in ss]
|
|
lows = highs = vals
|
|
tag = [1 if grp(s) == "HIGH" else 0 for s in ss]
|
|
obs = mean([v for v, t in zip(vals, tag) if t == 1]) - \
|
|
mean([v for v, t in zip(vals, tag) if t == 0])
|
|
n = len(keep) if key in ARRIVAL_METRICS else len(ss)
|
|
k = sum(tag)
|
|
deltas = []
|
|
for _ in range(reps):
|
|
idx = list(range(n))
|
|
rng.shuffle(idx)
|
|
sel = set(idx[:k])
|
|
if key in ARRIVAL_METRICS:
|
|
deltas.append(mean([highs[j] for j in sel]) -
|
|
mean([lows[j] for j in range(n) if j not in sel]))
|
|
else:
|
|
deltas.append(mean([lows[j] for j in sel]) -
|
|
mean([lows[j] for j in range(n) if j not in sel]))
|
|
p_hi_ = (sum(1 for x in deltas if x >= obs) + 1) / (reps + 1)
|
|
p_lo_ = (sum(1 for x in deltas if x <= obs) + 1) / (reps + 1)
|
|
out[b] = (obs, statistics.pstdev(deltas), min(1.0, 2 * min(p_hi_, p_lo_)))
|
|
return out
|
|
|
|
|
|
def fixed_horizon(shots, ks=(5, 10, 15, 20, 25, 30)):
|
|
"""The confound-free test.
|
|
|
|
Instead of the arrival tick (which power itself moves, and which also depends
|
|
on DrussGT's radial velocity), measure |perpendicular offset from the aim
|
|
line| at a FIXED number of ticks after the fire, and the same quantity in the
|
|
CONTROL window CTRL_OFF ticks later (bullet long gone). A response to the
|
|
shot would be ~0 at small k and grow with k. A phase/distance difference is
|
|
present already at k=5 and identical in the bullet-free control window.
|
|
Returns {band: {k: (low, high, nlow, nhigh, ctrl_low, ctrl_high)}}."""
|
|
out = {}
|
|
for b in BAND_LABELS:
|
|
ss = [s for s in shots if s["band"] == b and grp(s) is not None]
|
|
if len(ss) < 200:
|
|
continue
|
|
for k in ks:
|
|
lo = [abs(s["_perp"][k - 1]) for s in ss
|
|
if grp(s) == "LOW" and len(s["_perp"]) >= k]
|
|
hi = [abs(s["_perp"][k - 1]) for s in ss
|
|
if grp(s) == "HIGH" and len(s["_perp"]) >= k]
|
|
cl = [abs(s["_perp"][k + CTRL_OFF - 1]) for s in ss
|
|
if grp(s) == "LOW" and len(s["_perp"]) >= k + CTRL_OFF]
|
|
ch = [abs(s["_perp"][k + CTRL_OFF - 1]) for s in ss
|
|
if grp(s) == "HIGH" and len(s["_perp"]) >= k + CTRL_OFF]
|
|
out.setdefault(b, {})[k] = (mean(lo), mean(hi), len(lo), len(hi),
|
|
mean(cl), mean(ch))
|
|
return out
|
|
|
|
|
|
def flight_matched(shots):
|
|
_doc_extra = None
|
|
"""The kinematic channel removed outright.
|
|
|
|
Two matching variables at once, because they are both mechanical and both
|
|
differ between the arms: (i) the flight window itself (v = 20-3p, so a slower
|
|
bullet flies longer and DrussGT drifts further off the line) and (ii) the fire
|
|
RANGE inside the band (at a fixed flight window, a slower bullet implies a
|
|
shorter range, and the gun's aim error grows with range). Cells are
|
|
(integer flight window, 50 px range sub-bin) and only cells with >= MIN_CELL
|
|
shots in BOTH arms contribute; the per-cell differences are pooled with n
|
|
weights. Anything left here is not a flight-time or a range artefact.
|
|
Returns {band: (delta, n_used, cells, null_sd, p)}.
|
|
"""
|
|
MIN_CELL = 8
|
|
out = {}
|
|
for b in BAND_LABELS:
|
|
ss = [s for s in shots if s["band"] == b and grp(s) is not None]
|
|
by = collections.defaultdict(lambda: {"LOW": [], "HIGH": []})
|
|
for s in ss:
|
|
by[(s["flight"], int(s["range"] // RANGE_MATCH_PX))][grp(s)].append(s)
|
|
pairs = [(fl, d) for fl, d in by.items()
|
|
if len(d["LOW"]) >= MIN_CELL and len(d["HIGH"]) >= MIN_CELL]
|
|
if not pairs:
|
|
continue
|
|
num = den = 0.0
|
|
for fl, d in pairs:
|
|
w = len(d["LOW"]) + len(d["HIGH"])
|
|
num += w * (mean([s["miss"] for s in d["HIGH"]])
|
|
- mean([s["miss"] for s in d["LOW"]]))
|
|
den += w
|
|
cells = [(fl, d["LOW"], d["HIGH"]) for fl, d in pairs]
|
|
out[b] = (num / den, int(den), len(pairs), cells)
|
|
return out
|
|
|
|
|
|
def flight_matched_null(shots, reps=400, seed=808):
|
|
"""Permutation null of the flight-matched contrast (labels shuffled inside
|
|
each band, so the flight bins and their sizes are preserved)."""
|
|
rng = random.Random(seed)
|
|
res = {}
|
|
for b in BAND_LABELS:
|
|
ss = [s for s in shots if s["band"] == b and grp(s) is not None]
|
|
if not ss:
|
|
continue
|
|
vals = [s["miss"] for s in ss]
|
|
fls = [s["flight"] for s in ss]
|
|
tag = [grp(s) == "HIGH" for s in ss]
|
|
n = len(ss)
|
|
k = sum(tag)
|
|
if k < 20 or n - k < 20:
|
|
continue
|
|
rgs = [s["range"] for s in ss]
|
|
bins = collections.defaultdict(list)
|
|
for i in range(n):
|
|
bins[(fls[i], int(rgs[i] // RANGE_MATCH_PX))].append(i)
|
|
pool = [ix for ix in bins.values() if len(ix) >= 16]
|
|
if not pool:
|
|
continue
|
|
ds = []
|
|
for _ in range(reps):
|
|
perm = tag[:]
|
|
# shuffle labels within flight bin (preserves the bin sizes and the
|
|
# flight distribution of both arms)
|
|
for ix in pool:
|
|
sub = [perm[i] for i in ix]
|
|
rng.shuffle(sub)
|
|
for i, t in zip(ix, sub):
|
|
perm[i] = t
|
|
num = den = 0.0
|
|
for ix in pool:
|
|
a = [vals[i] for i in ix if not perm[i]]
|
|
c = [vals[i] for i in ix if perm[i]]
|
|
if len(a) >= 8 and len(c) >= 8:
|
|
num += (len(a) + len(c)) * (mean(c) - mean(a))
|
|
den += len(a) + len(c)
|
|
if den:
|
|
ds.append(num / den)
|
|
obs_num = obs_den = 0.0
|
|
for ix in pool:
|
|
a = [vals[i] for i in ix if not tag[i]]
|
|
c = [vals[i] for i in ix if tag[i]]
|
|
if len(a) >= 8 and len(c) >= 8:
|
|
obs_num += (len(a) + len(c)) * (mean(c) - mean(a))
|
|
obs_den += len(a) + len(c)
|
|
if ds and obs_den:
|
|
obs = obs_num / obs_den
|
|
p_hi = (sum(1 for x in ds if x >= obs) + 1) / (len(ds) + 1)
|
|
p_lo = (sum(1 for x in ds if x <= obs) + 1) / (len(ds) + 1)
|
|
res[b] = (obs, obs_den, statistics.pstdev(ds),
|
|
min(1.0, 2 * min(p_hi, p_lo)))
|
|
return res
|
|
|
|
|
|
def grp(s):
|
|
return "LOW" if 0.5 <= s["power"] < 0.75 else (
|
|
"HIGH" if 1.0 <= s["power"] < 1.5 else None)
|
|
|
|
|
|
def cluster_boot(shots, key, group_fn, reps=2000, seed=99):
|
|
"""Round-cluster bootstrap over the per-run aggregates of `key` by group."""
|
|
by_run = collections.defaultdict(list)
|
|
for s in shots:
|
|
if s.get(key) is not None:
|
|
by_run[s["run"]].append(s)
|
|
runs = list(by_run)
|
|
# pre-aggregate per run per group
|
|
agg = {}
|
|
for r in runs:
|
|
g = collections.defaultdict(lambda: [0.0, 0])
|
|
for s in by_run[r]:
|
|
grp = group_fn(s)
|
|
if grp is None:
|
|
continue
|
|
g[grp][0] += s[key]
|
|
g[grp][1] += 1
|
|
agg[r] = {k: (v[0], v[1]) for k, v in g.items()}
|
|
rng = random.Random(seed)
|
|
out = collections.defaultdict(list)
|
|
for _ in range(reps):
|
|
tot = collections.defaultdict(lambda: [0.0, 0])
|
|
for _ in runs:
|
|
r = rng.choice(runs)
|
|
for k, (s_, c_) in agg[r].items():
|
|
tot[k][0] += s_
|
|
tot[k][1] += c_
|
|
for k, (s_, c_) in tot.items():
|
|
if c_:
|
|
out[k].append(s_ / c_)
|
|
return {k: (pct(v, .025), pct(v, .975)) for k, v in out.items()}
|
|
|
|
|
|
# --------------------------------------------------------------------- report
|
|
def report(name, runs, shots, reps, out):
|
|
print("=" * 104)
|
|
print("CORPUS %-10s %d battles %d rounds %d usable shots by US (ModularBot)"
|
|
% (name, len(runs), sum(len(r.rounds) for r in runs), len(shots)))
|
|
|
|
# ---- attribution cross-check against the death events
|
|
bad = tot = 0
|
|
for r in runs:
|
|
for ev in r.events:
|
|
if ev.get("type") != "death":
|
|
continue
|
|
end = r.start[ev["round"]] + r.count[ev["round"]] - 1
|
|
row = r.by_tick.get(end)
|
|
if row is None:
|
|
continue
|
|
tot += 1
|
|
vs = r.owner_side.get(ev["victim"])
|
|
if (vs == "e" and row["ee"] > 1.0) or (vs == "s" and row["se"] > 1.0):
|
|
bad += 1
|
|
print(" attribution: owner id -> {e=DrussGT, s=us} recovered per battle from fire"
|
|
" geometry+energy;\n cross-checked on the death events: %d/%d"
|
|
" deaths have the mapped victim at ~0 energy" % (tot - bad, tot))
|
|
print(" (attribution is per-battle because the Tank Royale owner id is not stable"
|
|
" across runs)")
|
|
|
|
# ---- the confounder, made visible
|
|
print("\n >> THE CONFOUNDER IS REAL: our fired power is a range + energy cap")
|
|
print(" %-9s %-11s %7s %8s %9s %9s %9s" %
|
|
("band", "power", "shots", "firePx", "ourE", "enemyE", "flight"))
|
|
for b in BAND_LABELS:
|
|
for i in range(len(PWR_EDGES) - 1):
|
|
ss = [s for s in shots if s["band"] == b and pbin(s["power"]) == i]
|
|
if len(ss) < 5:
|
|
continue
|
|
print(" %-9s %-11s %7d %8.1f %9.1f %9.1f %9.2f" % (
|
|
b, PWR_LABELS[i], len(ss), mean([s["range"] for s in ss]),
|
|
mean([s["our_energy"] for s in ss]),
|
|
mean([s["enemy_energy"] for s in ss]),
|
|
mean([s["flight"] for s in ss])))
|
|
|
|
RES = out
|
|
for key, title in (
|
|
("miss", "(a) MISS AT ARRIVAL px - perpendicular offset of DrussGT from our aim line"),
|
|
("miss_per_tick", "(a') MISS AT ARRIVAL / flight ticks (px per tick - power-neutral)"),
|
|
("lat_disp_abs", "(b) LATERAL DISPLACEMENT px over the flight window"),
|
|
("lat_disp_per_tick", "(b/ ) LATERAL DISPLACEMENT / flight ticks"
|
|
" (px per tick - power-neutral)"),
|
|
("lat_fixed_abs", "(b') LATERAL DISPLACEMENT px over a FIXED 12-tick window"
|
|
" (power-neutral)"),
|
|
("lat_ctrl_abs", "(b'') CONTROL: lateral displacement px, SAME shot, 40 ticks later"
|
|
" (bullet gone)"),
|
|
("flight", "(f) FLIGHT WINDOW LENGTH ticks (low power = FASTER bullet = shorter)"),
|
|
("latency", "(c) RESPONSE LATENCY ticks to a >%g deg heading turn" % LAT_DEG),
|
|
("turn_flight", "(d) MEAN TURN RATE deg/tick, flight window"),
|
|
("turn_fixed", "(d') MEAN TURN RATE deg/tick, fixed 12-tick window"),
|
|
("turn_ctrl", "(d'') CONTROL: mean turn rate deg/tick, 40 ticks later"),
|
|
("speed_flight", "(d) MEAN SPEED px/tick, flight window"),
|
|
("speed_ctrl", "(d'') CONTROL: mean speed px/tick, 40 ticks later"),
|
|
("dodge_further", "(e) fraction where |perp| GREW (moved further off the line)"),
|
|
("hit", "OUTCOME hit rate (real server events)"),
|
|
):
|
|
cs = cell_stats(shots, key)
|
|
print("\n %s" % title)
|
|
print(" %-9s" % "band" + "".join("%22s" % l for l in PWR_LABELS))
|
|
for b in BAND_LABELS:
|
|
cells, ns = [], []
|
|
for i in range(len(PWR_EDGES) - 1):
|
|
n, m, sd = cs.get((b, i), (0, float("nan"), 0.0))
|
|
cells.append(m)
|
|
ns.append(n)
|
|
RES.setdefault("tables", {}).setdefault(key, {}).setdefault(b, {})[
|
|
PWR_LABELS[i]] = {"n": n, "mean": None if n == 0 else m, "sd": sd}
|
|
print(" %-9s" % b + "".join(
|
|
"%22s" % ("%.2f" % c if c == c else "-") for c in cells))
|
|
print(" %-9s" % "" + "".join("%22s" % ("n=%d" % n) for n in ns))
|
|
|
|
# ---- shuffled-label null, per band, on the LOW-vs-HIGH contrast
|
|
print("\n" + "-" * 104)
|
|
print("SHUFFLED-LABEL NULL (labels permuted WITHIN each range band, %d reps)" % reps)
|
|
print(" For the arrival-dependent metrics the arrival tick is RE-DERIVED from the"
|
|
" permuted power,\n so the null keeps the kinematic channel (a slower bullet"
|
|
" really does arrive later) and destroys\n only the response. p is the two-sided"
|
|
" permutation p on delta = mean(HIGH) - mean(LOW) inside that band.")
|
|
RES["perm"] = {}
|
|
for key in ("miss", "miss_per_tick", "lat_disp_abs", "flight",
|
|
"lat_fixed_abs", "lat_ctrl_abs", "latency", "turn_fixed", "turn_ctrl",
|
|
"speed_fixed", "hit"):
|
|
print(" %-16s" % key, end="")
|
|
cells = perm_contrast_null(shots, key, reps)
|
|
RES["perm"][key] = {}
|
|
for b in BAND_LABELS:
|
|
if b in cells:
|
|
d, sd, p2 = cells[b]
|
|
print(" %s d=%+.3f(null sd=%.3f,p=%.3f)" % (b, d, sd, p2), end="")
|
|
RES["perm"][key][b] = {"delta": d, "null_sd": sd, "p_two": p2}
|
|
else:
|
|
print(" %s n/a" % b, end="")
|
|
print()
|
|
|
|
# ---- the decisive fixed-horizon / control-window test
|
|
print("\n" + "-" * 104)
|
|
print("FIXED-HORIZON TEST (no arrival tick, no flight-window confound)")
|
|
print(" |perpendicular offset of DrussGT from our aim line| at k ticks after the fire"
|
|
" (flight window),\n and the same quantity in the CONTROL window at k+%d ticks,"
|
|
" when the bullet is long gone.\n A response to our shot would be ~0 at k=5 and"
|
|
" grow with k; a phase/geometry difference is\n present at k=5 and identical in"
|
|
" the control." % CTRL_OFF)
|
|
fh = fixed_horizon(shots)
|
|
RES["fixed_horizon"] = {}
|
|
for b in BAND_LABELS:
|
|
if b not in fh:
|
|
continue
|
|
print(" band %s (nLOW=%d nHIGH=%d per k)" % (b, fh[b][5][2], fh[b][5][3]))
|
|
for k, (lo, hi, nlo, nhi, cl, ch) in sorted(fh[b].items()):
|
|
print(" k=%2d flight: LOW=%7.1f HIGH=%7.1f d=%+6.2f | control(+%d):"
|
|
" LOW=%7.1f HIGH=%7.1f d=%+6.2f"
|
|
% (k, lo, hi, hi - lo, CTRL_OFF, cl, ch, ch - cl))
|
|
RES["fixed_horizon"].setdefault(b, {})[k] = {
|
|
"low": lo, "high": hi, "delta": hi - lo, "n_low": nlo, "n_high": nhi,
|
|
"ctrl_low": cl, "ctrl_high": ch, "ctrl_delta": ch - cl}
|
|
|
|
# ---- flight-matched contrast: the kinematic channel removed
|
|
print("\n" + "-" * 104)
|
|
print("FLIGHT- *AND* RANGE-MATCHED CONTRAST (kinematics and range both removed)")
|
|
print(" Within each range band we compare LOW vs HIGH power only inside the SAME cell"
|
|
" = (integer\n flight window, 50 px range sub-bin), pooling the per-cell differences"
|
|
" with n weights. A longer\n window lets DrussGT drift further off the line; at a"
|
|
" fixed window a slower bullet implies a\n shorter range and the gun's aim error"
|
|
" grows with range. Anything left here is neither.")
|
|
fm = flight_matched(shots)
|
|
fnull = flight_matched_null(shots, reps)
|
|
print(" %-9s %8s %8s %8s | %10s %8s %8s" %
|
|
("band", "obs d", "n used", "bins", "null sd", "p_two", ""))
|
|
RES["flight_matched"] = {}
|
|
for b in BAND_LABELS:
|
|
if b in fnull:
|
|
obs, nused, sd, p2 = fnull[b]
|
|
print(" %-9s %+8.2f %8d %8d | %10.2f %8.3f"
|
|
% (b, obs, nused, fm.get(b, (0, 0, 0, []))[2], sd, p2))
|
|
RES["flight_matched"][b] = {"delta": obs, "n": nused, "null_sd": sd,
|
|
"p_two": p2}
|
|
|
|
# ---- the headline within-band contrast: LOW (0.50-0.75) vs HIGH (1.00-1.50)
|
|
print("\n" + "-" * 104)
|
|
print("HEADLINE WITHIN-BAND CONTRAST LOW p in [0.50,0.75) vs HIGH p in [1.00,1.50)")
|
|
print(" (HELD fixed inside each range band; 95%% CI from a ROUND-cluster bootstrap)"
|
|
"\n")
|
|
|
|
hdr = (" %-14s %-9s %7s %7s | %8s %8s %8s %8s" %
|
|
("metric", "band", "nLOW", "nHIGH", "LOW", "HIGH", "delta", "95% CI"))
|
|
for key in ("miss", "miss_per_tick", "lat_disp_abs", "lat_disp_per_tick",
|
|
"lat_fixed_abs", "lat_ctrl_abs", "flight", "latency", "turn_flight",
|
|
"turn_fixed", "turn_ctrl", "speed_flight", "speed_ctrl", "hit"):
|
|
print(hdr)
|
|
for b in BAND_LABELS:
|
|
ss = [s for s in shots if s["band"] == b and grp(s) and s.get(key) is not None]
|
|
lo = [s[key] for s in ss if grp(s) == "LOW"]
|
|
hi = [s[key] for s in ss if grp(s) == "HIGH"]
|
|
if len(lo) < 15 or len(hi) < 15:
|
|
print(" %-14s %-9s %7d %7d | (too few for a contrast)" %
|
|
(key, b, len(lo), len(hi)))
|
|
RES.setdefault("contrast", {}).setdefault(key, {})[b] = {
|
|
"n_low": len(lo), "n_high": len(hi)}
|
|
continue
|
|
ci = cluster_boot(ss, key, grp)
|
|
d = mean(hi) - mean(lo)
|
|
print(" %-14s %-9s %7d %7d | %8.2f %8.2f %+8.2f [%+.2f,%+.2f]" % (
|
|
key, b, len(lo), len(hi), mean(lo), mean(hi), d,
|
|
ci.get("HIGH", (float("nan"),) * 2)[0] - ci.get("LOW", (0, 0))[1],
|
|
ci.get("HIGH", (0, 0))[1] - ci.get("LOW", (float("nan"),) * 2)[0]))
|
|
RES.setdefault("contrast", {}).setdefault(key, {})[b] = {
|
|
"n_low": len(lo), "n_high": len(hi), "low": mean(lo), "high": mean(hi),
|
|
"delta": d,
|
|
"ci_lo": ci.get("HIGH", (float("nan"),) * 2)[0] - ci.get("LOW", (0, 0))[1],
|
|
"ci_hi": ci.get("HIGH", (0, 0))[1] - ci.get("LOW", (float("nan"),) * 2)[0]}
|
|
print()
|
|
|
|
# ---- power histogram of OUR shots, for the record
|
|
h = collections.Counter(round(s["power"], 2) for s in shots)
|
|
print(" OUR fired-power histogram (exact values, top 12):")
|
|
print(" " + " ".join("%.2f:%d" % (p, n) for p, n in h.most_common(12)))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--tfil", default=None)
|
|
ap.add_argument("--powtest", default=None)
|
|
ap.add_argument("--json", default=None)
|
|
ap.add_argument("--reps", type=int, default=800)
|
|
a = ap.parse_args()
|
|
out = {}
|
|
for name, runs in (("tfil_ab2", discover_tfil(a.tfil) if a.tfil else []),
|
|
("powtest", discover_powtest(a.powtest) if a.powtest else [])):
|
|
if not runs:
|
|
continue
|
|
shots = [s for r in runs for s in r.shots()]
|
|
out[name] = {"battles": len(runs), "shots": len(shots)}
|
|
report(name, runs, shots, a.reps, out[name])
|
|
if a.json:
|
|
with open(a.json, "w") as f:
|
|
json.dump(out, f, indent=1, sort_keys=True)
|
|
return out
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|