learned movement: danger-map alignment diagnostic (corr(P(arrival bin), P(hit|bin)) = -0.342) in the gate + ledger

common_libs/tests/learned_surfer_gate.py

@@ -256,6 +256,52 @@ class CountedModel:
                 for k in range(NBINS)]
 
 
+def danger_alignment(runs):
+    """Is the map the mover MINIMISES (P of the arrival bin) aligned with the
+    map it SHOULD minimise (P(hit | arrival bin))?
+
+    The module picks the lowest-predicted-mass bin. If the mass and the hit
+    probability are anti-correlated, minimising the mass steers INTO the
+    bullets, and no amount of state conditioning can fix that.
+    """
+    got, hit = [0] * NBINS, [0] * NBINS
+    n = 0
+    for run in runs:
+        for sh in run.shots():
+            if sh["perp_arr"] is None:
+                continue
+            t0, rnd = sh["tick"], sh["rnd"]
+            rs = run.start[rnd]
+            r0 = run.by_tick.get(t0)
+            if r0 is None or t0 <= rs:
+                continue
+            p0x, p0y, speed = sh["_x"], sh["_y"], sh["speed"]
+            tb0 = math.atan2(r0["ey"] - p0y, r0["ex"] - p0x)
+            maxea = math.asin(min(8.0 / speed, 1.0))
+            kf = max(1, int(math.ceil(
+                math.hypot(r0["ex"] - p0x, r0["ey"] - p0y) / speed)))
+            r = run.by_tick.get(t0 + kf)
+            if r is None:
+                continue
+            off = wrap180(math.atan2(r["ey"] - p0y, r["ex"] - p0x) - tb0)
+            b = gf_to_bin(max(-1.0, min(1.0, off / maxea)))
+            got[b] += 1
+            hit[b] += sh["hit"]
+            n += 1
+    mass = [g / n for g in got]
+    rate = [(hit[b] / got[b]) if got[b] else float("nan") for b in range(NBINS)]
+    used = [b for b in range(NBINS) if got[b] > 0]
+    corr = statistics.correlation([mass[b] for b in used],
+                                  [rate[b] for b in used])
+    base = sum(hit) / n
+    safest_mass = min(used, key=lambda b: mass[b])
+    safest_hit = min(used, key=lambda b: rate[b])
+    return dict(n=n, base_hit=base, corr=corr,
+                safest_mass=(safest_mass, mass[safest_mass], rate[safest_mass]),
+                safest_hit=(safest_hit, mass[safest_hit], rate[safest_hit]),
+                rate=[(b, rate[b], mass[b]) for b in used])
+
+
 def log2(x):
     return math.log2(max(x, 1e-12))
 
@@ -569,6 +615,26 @@ def main():
         f"top-1 {fixed_agg['state_top1']:.4f}")
     out()
 
+    out("## F. is the DANGER MAP the module minimises the RIGHT one? "
+        "(MEASURED)")
+    out()
+    al = danger_alignment(runs)
+    out(f"shots {al['n']}, base hit rate {al['base_hit']*100:.2f}%")
+    out(f"corr( P(arrival bin) , P(hit | arrival bin) ) over the 31 bins = "
+        f"{al['corr']:+.3f}")
+    b, m, r = al["safest_mass"]
+    out(f"safest bin by the histogram MASS the mover minimises: bin {b} "
+        f"(mass {m*100:.1f}%, hit rate {r*100:.1f}%)")
+    b, m, r = al["safest_hit"]
+    out(f"safest bin by the ACTUAL hit rate: bin {b} "
+        f"(mass {m*100:.1f}%, hit rate {r*100:.1f}%)")
+    out("")
+    out("| bin | P(hit) | P(arrival bin) |")
+    out("|---:|---:|---:|")
+    for b, r, m in al["rate"]:
+        out(f"| {b} | {r*100:.1f}% | {m*100:.1f}% |")
+    out()
+
     out("## MEASURED vs INFERRED")
     out()
     out("* MEASURED: every number in this file, produced by the command in the")