DrussGT dodge vs fired power: no movement response once range is controlled
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).
This commit is contained in:
@@ -0,0 +1,795 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user