163 lines
7 KiB
Python
163 lines
7 KiB
Python
#!/usr/bin/env python3
|
|
"""A/B two simulation configurations on identical decks and seeds.
|
|
|
|
Win rate is a poor instrument at low game counts (+/-15.5pp at n=30), so this
|
|
compares POWER metrics -- engine assembly, pressure, cast rates -- which move on
|
|
far fewer games because they score every game rather than only the ones we won.
|
|
|
|
# validate the harness first: two identical arms must show NO difference
|
|
python3 tools/ab_test.py --deck decks/reaper-king-no-more --aa --games 12
|
|
|
|
# then a real comparison
|
|
python3 tools/ab_test.py --deck decks/reaper-king-no-more --games 20 --arm-b-profile Combo --arm-b-overlay
|
|
|
|
Arms differ ONLY in the named setting; deck, seeds and opponent pod are held fixed.
|
|
"""
|
|
import sys, os, re, json, math, subprocess, argparse, tempfile
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import power
|
|
from deckctx import DeckContext
|
|
|
|
FORGE = os.path.expanduser("~/forge")
|
|
POD = ["Faceless Menace [C19] [2019]", "Merciless Rage [C19] [2019]",
|
|
"Arcane Maelstrom [C20] [2020]"]
|
|
|
|
|
|
def run_arm(ctx, label, variant, games, seed, profile, overlay, clock, workdir):
|
|
"""One arm: build the deck, install/clear the overlay, run, return games."""
|
|
data = os.path.join(workdir, f"data_{label}")
|
|
prog = os.path.join(workdir, f"prog_{label}")
|
|
os.makedirs(os.path.join(data, "decks", "commander"), exist_ok=True)
|
|
os.makedirs(prog, exist_ok=True)
|
|
for entry in os.listdir(FORGE):
|
|
link = os.path.join(prog, entry)
|
|
if entry != "forge.profile.properties" and not os.path.lexists(link):
|
|
os.symlink(os.path.join(FORGE, entry), link)
|
|
with open(os.path.join(prog, "forge.profile.properties"), "w") as f:
|
|
f.write(f"userDir={data}\ncacheDir={os.path.join(data,'cache')}\n")
|
|
|
|
custom = os.path.join(data, "custom", "cards")
|
|
os.makedirs(custom, exist_ok=True)
|
|
for f in os.listdir(custom):
|
|
os.remove(os.path.join(custom, f))
|
|
if overlay:
|
|
import shutil
|
|
src = os.path.join(ctx.dir, "forge-overrides", "cardsfolder")
|
|
for r, _, files in os.walk(src):
|
|
for fn in files:
|
|
if fn.endswith(".txt"):
|
|
shutil.copy(os.path.join(r, fn), custom)
|
|
|
|
dst = os.path.join(data, "decks", "commander")
|
|
dck_name = ctx.dck_filename()
|
|
tools_dir = os.path.dirname(os.path.abspath(__file__))
|
|
subprocess.run([sys.executable, os.path.join(tools_dir, "to_dck.py"),
|
|
"--deck", ctx.dir,
|
|
os.path.join(ctx.dir, "decks", variant, "decklist.mtg"),
|
|
"-o", os.path.join(dst, dck_name)], capture_output=True)
|
|
import shutil as sh
|
|
for p in POD:
|
|
s = os.path.join(FORGE, "res", "quest", "commanderprecons", p + ".dck")
|
|
if os.path.exists(s):
|
|
sh.copy(s, dst)
|
|
|
|
jar = next(f for f in os.listdir(FORGE)
|
|
if f.startswith("forge-gui-desktop") and f.endswith(".jar"))
|
|
cmd = ["java", "-Xmx2g", "-jar", jar, "sim", "-d", dck_name] + \
|
|
[p + ".dck" for p in POD] + \
|
|
["-f", "Commander", "-n", str(games), "-s", str(seed),
|
|
"-c", str(clock), "-a", profile, "Default", "Default", "Default"]
|
|
log = os.path.join(workdir, f"{label}.log")
|
|
with open(log, "w") as fh:
|
|
proc = subprocess.run(cmd, cwd=prog, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, text=True)
|
|
for line in proc.stdout.splitlines():
|
|
if not re.match(r"^(Phase:|Warning:|WARNING)", line) and \
|
|
"not assigned to any set" not in line and "Upcoming set" not in line:
|
|
fh.write(line + "\n")
|
|
return power.parse(log), log
|
|
|
|
|
|
def stat(vals):
|
|
n = len(vals)
|
|
if n == 0:
|
|
return 0.0, 0.0
|
|
m = sum(vals) / n
|
|
if n < 2:
|
|
return m, 0.0
|
|
var = sum((v - m) ** 2 for v in vals) / (n - 1)
|
|
return m, math.sqrt(var / n) # mean, standard error
|
|
|
|
|
|
def compare(a, b, label_a, label_b):
|
|
metrics = {
|
|
"power score": lambda g: power.score(g),
|
|
"engine assembled": lambda g: 1.0 if g["assembled_turn"] else 0.0,
|
|
"commander deployed": lambda g: 1.0 if g["commander_turn"] else 0.0,
|
|
"life taken": lambda g: g["life_taken"],
|
|
"spells cast": lambda g: g["casts"],
|
|
"rounds survived": lambda g: g["survived"],
|
|
"timeout draw": lambda g: 1.0 if g["timeout"] else 0.0,
|
|
"won": lambda g: 1.0 if g["won"] else 0.0,
|
|
}
|
|
print("=" * 76)
|
|
print(f"A/B RESULT arm A = {label_a} arm B = {label_b}")
|
|
print("=" * 76)
|
|
print(f"games: {len(a)} vs {len(b)}\n")
|
|
print(f" {'metric':<20}{'A':>10}{'B':>10}{'delta':>10} verdict")
|
|
print(" " + "-" * 72)
|
|
for name, fn in metrics.items():
|
|
va, vb = [fn(g) for g in a], [fn(g) for g in b]
|
|
ma, sa = stat(va)
|
|
mb, sb = stat(vb)
|
|
d = mb - ma
|
|
se = math.sqrt(sa ** 2 + sb ** 2)
|
|
# 2 standard errors ~ 95%; below that we cannot distinguish the arms
|
|
sig = abs(d) > 2 * se and se > 0
|
|
verdict = "MOVED" if sig else "noise"
|
|
print(f" {name:<20}{ma:>10.2f}{mb:>10.2f}{d:>+10.2f} {verdict}"
|
|
f" (+/-{2*se:.2f})")
|
|
print()
|
|
print(" 'noise' means the arms are indistinguishable at this sample size --")
|
|
print(" NOT that the change did nothing. Raise --games to resolve smaller effects.")
|
|
print("=" * 76)
|
|
|
|
|
|
def main():
|
|
ctx = DeckContext.from_args() # pops --deck/-d out of sys.argv first
|
|
power.configure(ctx.forge_metadata_name(), ctx.commander)
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--variant", default="no-mikaeus")
|
|
ap.add_argument("--games", type=int, default=20)
|
|
ap.add_argument("--seed", type=int, default=20260907)
|
|
ap.add_argument("--clock", type=int, default=600)
|
|
ap.add_argument("--aa", action="store_true",
|
|
help="null test: run two IDENTICAL arms. Any 'MOVED' result "
|
|
"means the harness or seeding is broken.")
|
|
ap.add_argument("--arm-b-profile", default="Combo")
|
|
ap.add_argument("--arm-b-overlay", action="store_true")
|
|
a = ap.parse_args()
|
|
|
|
b_profile = "Default" if a.aa else a.arm_b_profile
|
|
b_overlay = False if a.aa else a.arm_b_overlay
|
|
label_a = "stock (Default, no overlay)"
|
|
label_b = ("stock (identical) [A/A NULL TEST]" if a.aa
|
|
else f"{b_profile}{' + overlay' if b_overlay else ''}")
|
|
|
|
wd = tempfile.mkdtemp(prefix="abtest_")
|
|
print(f"arm A: {label_a}\narm B: {label_b}\nseed {a.seed}, "
|
|
f"{a.games} games each, workdir {wd}\n", flush=True)
|
|
|
|
ga, la = run_arm(ctx, "A", a.variant, a.games, a.seed, "Default", False, a.clock, wd)
|
|
print(f"arm A done ({len(ga)} games)", flush=True)
|
|
gb, lb = run_arm(ctx, "B", a.variant, a.games, a.seed, b_profile, b_overlay, a.clock, wd)
|
|
print(f"arm B done ({len(gb)} games)\n", flush=True)
|
|
|
|
if not ga or not gb:
|
|
sys.exit("no games parsed from one or both arms")
|
|
compare(ga, gb, label_a, label_b)
|
|
print(f"logs: {la}\n {lb}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|