File size: 5,821 Bytes
a1bae02 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | #!/usr/bin/env python3
"""E20: whole-game score aggregation study (true (p, fired) pairs).
Pre-registered BEFORE computation (README E20):
split exploration = apac + own; confirmation = eu + na
candidates (over can-fire samples of each ship):
C1 meanP(actual) — the E17 aggregation (known to invert)
C2 medianP(actual)
C3 top-quartile meanP(actual)
C4 recall@p>=0.5 — fired in / (fired in + held) model-endorsed windows
C5 valueRecall — sum(p*fired) / sum(p) (captured model-value share)
C6 valuePerShot — sum(p*fired) / sum(fired)
C7 firedRate — raw volume control
C8 moveMedian — move-agreement control
pick rule best positive Spearman rho vs damage on exploration; ties by
semantic order C5 > C4 > C6 > C3 > C2 > C7 > C8
acceptance the PICK, run ONCE on confirmation: rho(damage) >= 0.30
(the unchanged E17 line) + within-replay pooled rank rho
No re-picking after the confirmation run.
"""
from __future__ import annotations
import json
import math
from pathlib import Path
import numpy as np
BASE = Path(__file__).resolve().parent / "out" / "score_validation"
EXPLORE = ["apac", "own"]
CONFIRM = ["eu", "na"]
TIE = ["C5", "C4", "C6", "C3", "C2", "C7", "C8"]
def spearman(x, y):
x, y = np.asarray(x, float), np.asarray(y, float)
good = np.isfinite(x) & np.isfinite(y)
x, y = x[good], y[good]
n = len(x)
if n < 10:
return float("nan"), n
rx = np.argsort(np.argsort(x)).astype(float)
ry = np.argsort(np.argsort(y)).astype(float)
rx = (rx - rx.mean()) / (rx.std() or 1.0)
ry = (ry - ry.mean()) / (ry.std() or 1.0)
return float((rx * ry).mean()), n
def load(shards):
ships = []
for s in shards:
with open(BASE / f"scores_e20_{s}.jsonl", encoding="utf-8") as f:
for line in f:
rep = json.loads(line)
for ps in rep["perShip"]:
oc = ps.get("outcome") or {}
if oc.get("status") != "ok":
continue
samples = ps.get("samples") or []
p = np.array(
[x["fireProb"] for x in samples if x.get("fireProb") is not None]
)
fired = np.array(
[x["fired"] for x in samples if x.get("fired") is not None], float
)
moves = [x["move"] for x in samples if x.get("move") is not None]
p_act = np.where(fired > 0.5, p, 1.0 - p)
endorsed = p >= 0.5
sum_p = p.sum()
ships.append(
{
"replay": rep["replayId"],
"damage": float(oc.get("damage") or 0.0),
"base_exp": float(oc.get("baseExp") or 0.0),
"C1": float(p_act.mean()) if len(p_act) else np.nan,
"C2": float(np.median(p_act)) if len(p_act) else np.nan,
"C3": (
float(np.sort(p_act)[int(0.75 * len(p_act)) :].mean())
if len(p_act)
else np.nan
),
"C4": (
float(fired[endorsed].mean()) if endorsed.any() else np.nan
),
"C5": float((p * fired).sum() / sum_p) if sum_p > 0 else np.nan,
"C6": (
float((p * fired).sum() / fired.sum()) if fired.sum() > 0 else np.nan
),
"C7": float(fired.mean()) if len(fired) else np.nan,
"C8": float(np.median(moves)) if moves else np.nan,
}
)
return ships
def within_rho(ships, key, min_ships=4):
by = {}
for s in ships:
by.setdefault(s["replay"], []).append(s)
rs, ns = [], []
for rows in by.values():
if len(rows) < min_ships:
continue
r, n = spearman([x[key] for x in rows], [x["damage"] for x in rows])
if math.isfinite(r):
rs.append(r)
ns.append(n)
return float(np.average(rs, weights=ns)) if rs else float("nan")
def report(tag, ships):
dmg = [s["damage"] for s in ships]
exp = [s["base_exp"] for s in ships]
out = {}
for k in TIE + ["C1"]:
r_d, n = spearman([s[k] for s in ships], dmg)
r_x, _ = spearman([s[k] for s in ships], exp)
out[k] = r_d
print(
f"[e20:{tag}] {k}: rho(damage)={r_d:+.4f} rho(baseExp)={r_x:+.4f} n={n}"
)
return out
def main():
explore = load(EXPLORE)
print(f"[e20] exploration: {len(explore)} ships ({EXPLORE})")
res = report("explore", explore)
pos = {k: v for k, v in res.items() if math.isfinite(v) and v > 0}
if not pos:
print("[e20] NO positive candidate — honest stop")
return
best = max(pos.values())
tied = [k for k, v in pos.items() if v >= best - 1e-9]
pick = min(tied, key=lambda k: TIE.index(k) if k in TIE else 99)
print(f"[e20] PICK {pick} (explore rho {pos[pick]:+.4f}); confirming on {CONFIRM}")
confirm = load(CONFIRM)
print(f"[e20] confirmation: {len(confirm)} ships")
r, n = spearman([s[pick] for s in confirm], [s["damage"] for s in confirm])
wr = within_rho(confirm, pick)
print(
f"[e20] CONFIRM {pick}: rho(damage)={r:+.4f} n={n}; within-replay pooled={wr:+.4f}"
f" -> {'PASS (rho>=0.30)' if r >= 0.30 else 'FAIL (honest report)'}"
)
if __name__ == "__main__":
main()
|