Files
gplan-2/tools/balance_verifier.py
T
DKChan e8d8d6051e feat(tools): 数值验证工具批——自动化评判三段采集(直入口基线带判定+零强化止步归因+营地循环爬升战役)
- 冒险摘要增败退明细 defeat{level_index,node_id,node_type}(仅失败局附加,消费方零感知)
- level_autoplay 冒险驱动增 hero_override 尾参透传(永久强化 flat 注入)+末场战斗敌方名采集
- 新增采集端 smoke/verifier_batch.gd(VJSON 行契约,隔离档真实元进度账本,绝不碰 user://save.json)
- 新增判定端 tools/balance_verifier.py(带判定/止步归因/滑窗检查点判定/报告落 production/qa)
- 冒烟 552 项 0 失败(新增败退明细断言 2 项)
2026-09-22 20:12:04 +08:00

477 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""数值验证工具(判定端)——自动化评判 + 元进度爬升测量,替代「亲手试玩再反馈」循环。
三段采集(数据源 = smoke/verifier_batch.gd,本文件驱动它并做全部判定):
level 段 直入口单关基线(零加成)——验收带判定
adventure 段 全冒险连探基线(零永久强化)——止步分布 + 败退节点归因(层/节点/敌人)
campaign 段 营地循环战役(隔离档真实元进度账本:注入→冒险→收束入账→买锻造)——爬升曲线
评判标准(溯源):
① 直入口基线通关率带(分带批 2026-09-20 用户拍板,难度曲线与Boss v0.16):
第一层 普通 [35,55]% · 第二层 普通 [25,45]% · 完美档 [35,65]%(两关同带);第三层带未定——只测不判。
② 元循环目标(用户提案 2026-09-22,数字待确认):累计通过 5 次第 k 层的冒险后,第 k+1 层的
条件通关率(滑窗)应进入该层验收带——「5 次通过第一层的冒险后能过第二层;5 次通过第二层后能过第三层」。
第三层带未定,只给爬升曲线与首次读数,不出判定。
③ 魔法道具面:道具工坊六件按通关次数解锁(阈值 1/1/2/2/3/3),局内效果未实装(归 Alpha 魔法道具批)——
本工具只报解锁时点,不入判定。
表值硬编码溯源(单一事实源仍在 .tres / GDD,变更须同步):
· 锻造三档 flat(meta.tres):生命 +3 / 攻击 +1 / 护甲 +1
· 钥石入账(adventure.tres + meta.tres):每层通关 10 + Boss 首杀/重复杀 5/1 · 8/2 · 12/3
· 工坊阈值(meta.tres):显影尘1 · 工兵令1 · 探宝罗盘2 · 探巢摆钟2 · 地穴全图卷3 · 爆破工兵包3
用法:
python tools/balance_verifier.py # 全段默认规模(约 20–35 分钟)
python tools/balance_verifier.py --quick # 小规模快跑(约 5–10 分钟,读数精度低)
python tools/balance_verifier.py --n 100 --na 60 --k 60 --seed0 9000 --spend atk_first
python tools/balance_verifier.py --from-raw <raw.jsonl> # 免重跑,对既有采集重新判定/出报告
输出:控制台判定结论 + production/qa/balance-verifier-<时间戳>.md(报告)+ 同名 .raw.jsonl(原始逐局数据)。
"""
import argparse
import json
import math
import subprocess
import sys
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
SLICE = ROOT / "prototypes" / "dungeon-sweeper-vertical-slice"
GODOT = r"D:\code\godot\Godot_v4.7.2-stable_win64_console.exe"
SCRIPT = "res://smoke/verifier_batch.gd"
QA_DIR = ROOT / "production" / "qa"
VJSON_PREFIX = "VJSON "
ERROR_MARKERS = ("SCRIPT ERROR", "Parse Error")
# ---- 判定标准区(溯源见文件头) ----
LEVEL_BANDS = {
("normal", 1): (35, 55),
("normal", 2): (25, 45),
("perfect", 1): (35, 65),
("perfect", 2): (35, 65),
}
CAMPAIGN_CLEARS_TARGET = 5 # 元循环检查点:累计通过第 k 层的冒险次数(用户提案 2026-09-22,待确认)
CAMPAIGN_WINDOW = 10 # 检查点滑窗局数(条件通关率的窗口)
CAMPAIGN_MIN_ATTEMPTS = 5 # 滑窗内少于此局数 = 数据不足(不出判定)
LEVEL_CN = {1: "第一层", 2: "第二层", 3: "第三层"}
NODE_TYPE_CN = {"battle": "战斗", "boss": "Boss", "rest": "恢复", "reward": "奖励", "shop": "商店房"}
FORGE_FLAT = {"max_hp": 3, "atk": 1, "armor_class": 1} # meta.tres 锻造三档每级 flat 值
WORKSHOP_CN = {"probe_dust": "显影尘", "sapper_order": "工兵令", "loot_compass": "探宝罗盘",
"nest_pendulum": "探巢摆钟", "catacomb_scroll": "地穴全图卷", "blast_kit": "爆破工兵包"}
def die(msg):
print(f"[数值验证] FAIL:{msg}")
sys.exit(1)
def disp_w(s):
return sum(2 if ord(c) > 0x2E7F else 1 for c in str(s))
def pad(s, w, right=False):
s = str(s)
gap = w - disp_w(s)
return (" " * gap + s) if right else (s + " " * gap)
# ---------------------------------------------------------------- 采集
def collect(args):
segs = args.only.split(",")
cmd = [args.godot, "--headless", "--path", str(SLICE), "--script", SCRIPT, "--",
f"n={args.n}", f"na={args.na}", f"k={args.k}", f"seed0={args.seed0}",
f"only={args.only}", f"spend={args.spend}"]
print(f"[数值验证] 采集启动({' · '.join(segs)},约 {args.est_min} 分钟)…")
t0 = time.time()
try:
proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=args.timeout)
except subprocess.TimeoutExpired:
die(f"外部超时(>{args.timeout}s)——可加大 --timeout 或缩小规模")
out = (proc.stdout or "") + (proc.stderr or "")
if proc.returncode != 0:
print(out[-3000:])
die(f"采集进程退出码 {proc.returncode}")
hits = [ln.strip() for ln in out.splitlines() if any(k in ln for k in ERROR_MARKERS)]
if hits:
print("\n".join(hits[:5]))
die(f"脚本错误标记命中 {len(hits)} 行(脚本错误静默中断断言/采集,必须人工排查)")
rows, bad = [], 0
for ln in out.splitlines():
if ln.startswith(VJSON_PREFIX):
try:
rows.append(json.loads(ln[len(VJSON_PREFIX):]))
except json.JSONDecodeError:
bad += 1
if bad:
die(f"VJSON 行解析失败 {bad} 行(采集端输出契约被破坏)")
if not rows:
die("采集零数据行(verifier_batch.gd 未按契约输出)")
print(f"[数值验证] 采集完成:{len(rows)} 局,耗时 {(time.time() - t0) / 60:.1f} 分钟")
return rows, out
def load_raw(path):
rows = []
with open(path, encoding="utf-8") as f:
for ln in f:
ln = ln.strip()
if ln:
rows.append(json.loads(ln))
return rows
# ---------------------------------------------------------------- 判定件
def binom_ci(wins, n):
if n == 0:
return 0.0
p = wins / n
return 1.96 * math.sqrt(p * (1 - p) / n)
def judge_band(rate_pct, band):
"""落带判定:返回 (结论, 是否达标)。band=None = 带未定,只报读数。"""
if band is None:
return "带未定(只测不判)", None
lo, hi = band
if rate_pct < lo:
return f"越下界(低于 {lo}% 达 {lo - rate_pct:.1f} 点)", False
if rate_pct > hi:
return f"越上界(高于 {hi}% 达 {rate_pct - hi:.1f} 点)", False
return f"落带 [{lo},{hi}] ✓", True
def node_label(defeat):
"""败退节点人读名:第X层·节点N·类型·敌人(敌人 = 死亡遭遇显示名,最多 3 个)。"""
if not defeat:
return "—"
t = NODE_TYPE_CN.get(defeat.get("node_type", ""), defeat.get("node_type", "?"))
s = f"{LEVEL_CN.get(defeat.get('level_index', -1), '?层')}·节点{defeat.get('node_id', '?')}·{t}"
enemies = defeat.get("enemies") or []
if enemies:
seen, names = set(), []
for e in enemies:
if e not in seen:
seen.add(e)
names.append(e)
s += "·" + "、".join(names[:3])
return s
# ---------------------------------------------------------------- 分段分析
def analyze_level(rows):
"""直入口基线:按(画像, 层)聚合计数。"""
agg = {}
for r in rows:
if r.get("seg") != "level":
continue
key = (r["policy"], int(r["level"]))
a = agg.setdefault(key, {"n": 0, "wins": 0, "deaths": {}})
a["n"] += 1
if r.get("result") == "victory":
a["wins"] += 1
elif r.get("result") == "defeat":
k = (int(r.get("defeat_node_id", -1)), r.get("defeat_node_type", ""))
a["deaths"][k] = a["deaths"].get(k, 0) + 1
return agg
def analyze_adventure(rows):
out = {}
for policy in ("normal", "perfect"):
runs = [r for r in rows if r.get("seg") == "adventure" and r.get("policy") == policy]
stopped, death_nodes, l1 = {}, {}, {"n": 0, "wins": 0}
for r in runs:
sa = int(r.get("stopped_at", -1))
if sa == 0:
stopped["通关"] = stopped.get("通关", 0) + 1
elif sa > 0:
stopped[f"止步第{sa}层"] = stopped.get(f"止步第{sa}层", 0) + 1
d = r.get("defeat") or {}
if d:
lb = node_label(d)
death_nodes[lb] = death_nodes.get(lb, 0) + 1
if int(r.get("levels_cleared", -1)) >= 0:
l1["n"] += 1
if int(r.get("levels_cleared", 0)) >= 1:
l1["wins"] += 1
out[policy] = {"n": len(runs), "stopped": stopped, "death_nodes": death_nodes, "l1": l1}
return out
def campaign_series(runs):
"""逐局爬升序列:条件通关尝试流(第 k 层的尝试 = levels_cleared ≥ k-1 的局,按局序)。"""
return {
k: [(r["run_no"], int(r.get("levels_cleared", -1)) >= k, r)
for r in sorted(runs, key=lambda x: x["run_no"])
if int(r.get("levels_cleared", -1)) >= k - 1]
for k in (1, 2, 3)
}
def window_rate(series, end_idx, window):
"""尝试流截至 end_idx(含)往前 window 次尝试的通关率(%)。"""
chunk = series[max(0, end_idx - window + 1): end_idx + 1]
if not chunk:
return None, 0
return 100.0 * sum(1 for _, ok, _ in chunk if ok) / len(chunk), len(chunk)
def analyze_campaign(rows):
out = {}
for policy in ("normal", "perfect"):
runs = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
and not r.get("anomaly")]
anomalies = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
and r.get("anomaly")]
res = {"n": len(runs), "anomalies": len(anomalies), "runs": runs}
if not runs:
out[policy] = res
continue
series = campaign_series(runs)
res["series"] = series
# 检查点判定:累计通过第 k 层达目标后,第 k+1 层滑窗条件通关率 vs 验收带
checkpoints = []
for k in (1, 2):
clears = 0
trigger_run = None
for r in sorted(runs, key=lambda x: x["run_no"]):
if int(r.get("levels_cleared", 0)) >= k:
clears += 1
if clears >= CAMPAIGN_CLEARS_TARGET and trigger_run is None:
trigger_run = r["run_no"]
cp = {"level": k, "trigger_run": trigger_run, "clears": clears}
if trigger_run is None:
cp["verdict"] = f"数据不足:全程通过{LEVEL_CN[k]}仅 {clears} 次(目标 {CAMPAIGN_CLEARS_TARGET})——加大 --k 重跑"
else:
attempts = [(rn, ok, r) for rn, ok, r in series.get(k + 1, []) if rn > trigger_run]
chunk = attempts[:CAMPAIGN_WINDOW]
if len(chunk) < CAMPAIGN_MIN_ATTEMPTS:
cp["verdict"] = (f"触发于第 {trigger_run} 局,但其后通过{LEVEL_CN[k + 1]}的尝试仅 "
f"{len(chunk)} 局(不足 {CAMPAIGN_MIN_ATTEMPTS})——加大 --k 重跑")
else:
rate = 100.0 * sum(1 for _, ok, _ in chunk if ok) / len(chunk)
cp["rate"] = rate
cp["attempts"] = len(chunk)
band = LEVEL_BANDS.get((policy, k + 1))
cp["verdict"], cp["pass"] = judge_band(rate, band)
cp["forge_at"] = chunk[0][2].get("override", {})
checkpoints.append(cp)
res["checkpoints"] = checkpoints
# 工坊解锁时点(魔法道具面——只报不判)
unlocks = {}
for r in runs:
for iid in r.get("new_unlocks", []):
unlocks[WORKSHOP_CN.get(iid, iid)] = r["run_no"]
res["unlocks"] = unlocks
# 累计钥石与锻造终态
last = max(runs, key=lambda x: x["run_no"])
res["forge_final"] = last.get("forge", {})
res["keystone_final"] = int(last.get("keystone_after", -1))
res["income_total"] = sum(int(r.get("keystone_income", 0)) for r in runs)
out[policy] = res
return out
# ---------------------------------------------------------------- 报告
def fmt_table(header, rows):
widths = [max(disp_w(h), max((disp_w(r[i]) for r in rows), default=0)) for i, h in enumerate(header)]
def line(cells):
return "| " + " | ".join(pad(c, w) for c, w in zip(cells, widths)) + " |"
sep = "|" + "|".join("-" * (w + 2) for w in widths) + "|"
out = [line(header), sep]
out += [line(r) for r in rows]
return "\n".join(out)
def report(args, rows, meta):
lines = []
A = lines.append
A(f"# 数值验证报告({meta['ts']})")
A("")
A(f"- 采集规模:直入口每格 {args.n} 局 · 基线冒险 {args.na} 局/画像 · 战役 {args.k} 局/画像"
f" · 种子 {args.seed0} 起逐局独立 · 锻造花费策略 = {args.spend}")
A(f"- 切片版本:{meta['commit']} · 表状态:判定带与表值溯源见 tools/balance_verifier.py 文件头")
A("")
# ① 直入口基线
A("## ① 直入口基线通关率(验收带判定)")
A("")
lvl = analyze_level(rows)
tbl = []
for (policy, li), a in sorted(lvl.items()):
rate = 100.0 * a["wins"] / max(1, a["n"])
ci = 1.96 * math.sqrt((rate / 100) * (1 - rate / 100) / max(1, a["n"])) * 100
band = LEVEL_BANDS.get((policy, li))
verdict, _ = judge_band(rate, band)
tbl.append([f"{LEVEL_CN[li]}", "普通档" if policy == "normal" else "完美档",
f"{a['wins']}/{a['n']}", f"{rate:.1f}% ±{ci:.1f}",
f"[{band[0]},{band[1]}]" if band else "未定", verdict])
A(fmt_table(["层", "画像", "通关/局数", "通关率±95%CI", "验收带", "判定"], tbl))
A("")
# ② 冒险基线:止步分布与败退节点
A("## ② 零永久强化冒险:止步分布与败退节点(用户判据③)")
A("")
adv = analyze_adventure(rows)
for policy in ("normal", "perfect"):
d = adv[policy]
cn = "普通档" if policy == "normal" else "完美档"
A(f"### {cn}({d['n']} 局)")
A("")
stop_line = "、".join(f"{k} {v}({100.0 * v / max(1, d['n']):.0f}%)" for k, v in sorted(d["stopped"].items()))
A(f"- 止步分布:{stop_line or '—'}")
l1r = 100.0 * d["l1"]["wins"] / max(1, d["l1"]["n"])
A(f"- 第一层通过率(冒险口径交叉核对):{l1r:.1f}%")
A("- 败退节点分布(失败局归因):")
for lb, c in sorted(d["death_nodes"].items(), key=lambda x: -x[1]):
A(f" - {lb} ×{c}")
A("")
# ③ 元循环爬升
A("## ③ 营地循环爬升(用户判据④:5 次通关 → 下一层可过)")
A("")
camp = analyze_campaign(rows)
for policy in ("normal", "perfect"):
d = camp[policy]
cn = "普通档" if policy == "normal" else "完美档"
A(f"### {cn}({d['n']} 局战役)")
A("")
if not d["runs"]:
A("- 无数据")
A("")
continue
A(f"- 钥石:全程入账 {d['income_total']} · 终态余额 {d['keystone_final']}"
f" · 锻造终态 = 生命+{d['forge_final'].get('max_hp', 0) * FORGE_FLAT['max_hp']}"
f"/攻击+{d['forge_final'].get('atk', 0) * FORGE_FLAT['atk']}"
f"/护甲+{d['forge_final'].get('armor_class', 0) * FORGE_FLAT['armor_class']}")
if d["unlocks"]:
A("- 道具工坊解锁时点(魔法道具面,局内效果未实装只报不判):" +
"、".join(f"{k} 第 {v} 局" for k, v in sorted(d["unlocks"].items(), key=lambda x: x[1])))
else:
A("- 道具工坊解锁时点:无(未达阈值)")
if d.get("anomalies"):
A(f"- ⚠ 编排异常局 {d['anomalies']} 局(未结算零入账——timeout/advance_failed,须排查)")
A("")
for cp in d.get("checkpoints", []):
A(f"- 判据:累计通过{LEVEL_CN[cp['level']]} {CAMPAIGN_CLEARS_TARGET} 次后,{LEVEL_CN[cp['level'] + 1]}"
f"条件通关率(滑窗 {CAMPAIGN_WINDOW} 局)——{cp['verdict']}"
+ (f"(触发时注入 = 生命+{cp['forge_at'].get('max_hp', 0)}/攻击+{cp['forge_at'].get('atk', 0)}"
f"/护甲+{cp['forge_at'].get('armor_class', 0)})" if "forge_at" in cp else ""))
A("")
tbl = []
series = d.get("series", {})
runs_by_no = {r["run_no"]: r for r in d["runs"]}
for r in d["runs"][::max(1, len(d["runs"]) // 20)]:
cells = [f"第{r['run_no']}局",
f"血{r['override'].get('max_hp', 0)}/攻{r['override'].get('atk', 0)}/甲{r['override'].get('armor_class', 0)}",
str(r.get("keystone_after", "-"))]
for k in (1, 2, 3):
s = series.get(k, [])
idx = next((i for i, (rn, _, _) in enumerate(s) if rn == r["run_no"]), None)
if idx is None:
cells.append("—")
else:
rate, cnt = window_rate(s, idx, CAMPAIGN_WINDOW)
cells.append(f"{rate:.0f}%(n={cnt})" if rate is not None else "—")
sa = int(r.get("stopped_at", -1))
cells.append("通关" if sa == 0 else (f"止步第{sa}层" if sa > 0 else str(r.get("result", "?"))))
tbl.append(cells)
A(fmt_table(["局", "注入(flat)", "钥石", "第1层率(窗)", "第2层率(窗)", "第3层率(窗)", "终局"], tbl))
A("")
return "\n".join(lines)
def console_verdict(args, rows):
"""控制台判定结论(先读数后细节)。"""
lvl = analyze_level(rows)
print("\n判定结论(detail 见报告文件)")
for (policy, li), a in sorted(lvl.items()):
rate = 100.0 * a["wins"] / max(1, a["n"])
band = LEVEL_BANDS.get((policy, li))
verdict, ok = judge_band(rate, band)
mark = {True: "✓", False: "✗", None: "·"}[ok]
print(f" [{mark}] 直入口{LEVEL_CN[li]}{'普通档' if policy == 'normal' else '完美档'}:"
f"{rate:.1f}%({a['wins']}/{a['n']})→ {verdict}")
adv = analyze_adventure(rows)
for policy in ("normal", "perfect"):
d = adv[policy]
cn = "普通档" if policy == "normal" else "完美档"
top = sorted(d["death_nodes"].items(), key=lambda x: -x[1])[:2]
print(f" [·] 零强化冒险·{cn}:止步分布 {'、'.join(f'{k}{v}' for k, v in sorted(d['stopped'].items()))}"
+ (f";主要败退点 = {';'.join(f'{lb} ×{c}' for lb, c in top)}" if top else ""))
camp = analyze_campaign(rows)
for policy in ("normal", "perfect"):
d = camp[policy]
cn = "普通档" if policy == "normal" else "完美档"
if not d["runs"]:
print(f" [·] 营地循环·{cn}:无数据")
continue
for cp in d.get("checkpoints", []):
mark = {True: "✓", False: "✗", None: "·"}.get(cp.get("pass"), "·")
print(f" [{mark}] 营地循环·{cn}·{LEVEL_CN[cp['level']]}→{LEVEL_CN[cp['level'] + 1]}:{cp['verdict']}")
print()
# ---------------------------------------------------------------- 入口
def main():
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
ap = argparse.ArgumentParser(description="数值验证工具(判定端)")
ap.add_argument("--godot", default=GODOT)
ap.add_argument("--n", type=int, default=100, help="直入口每格局数(±95%CI 约 ±9 点——带边判读须看 CI)")
ap.add_argument("--na", type=int, default=60, help="基线冒险局数/画像")
ap.add_argument("--k", type=int, default=60, help="战役局数/画像")
ap.add_argument("--seed0", type=int, default=8000)
ap.add_argument("--only", default="level,adventure,campaign")
ap.add_argument("--spend", default="hp_first", choices=["hp_first", "atk_first"])
ap.add_argument("--timeout", type=int, default=5400)
ap.add_argument("--quick", action="store_true", help="小规模快跑(12/8/10)")
ap.add_argument("--from-raw", default=None, help="跳过采集,对既有 raw.jsonl 重新判定出报告")
args = ap.parse_args()
if args.quick:
args.n, args.na, args.k = 12, 8, 10
args.est_min = (3 * 2 * args.n * 1.5 + 2 * args.na * 3.5 + 2 * args.k * 3.5) / 60
ts = time.strftime("%Y%m%d-%H%M%S")
if args.from_raw:
rows = load_raw(args.from_raw)
raw_path = Path(args.from_raw)
print(f"[数值验证] 已载入既有采集 {len(rows)} 局:{raw_path}")
else:
rows, _ = collect(args)
QA_DIR.mkdir(parents=True, exist_ok=True)
raw_path = QA_DIR / f"balance-verifier-{ts}.raw.jsonl"
with open(raw_path, "w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
commit = "?"
try:
commit = subprocess.run(["git", "-C", str(SLICE), "rev-parse", "--short", "HEAD"],
capture_output=True, text=True).stdout.strip()
except OSError:
pass
meta = {"ts": ts, "commit": commit}
console_verdict(args, rows)
md = report(args, rows, meta)
if args.from_raw:
md_path = raw_path.with_suffix(".md")
else:
md_path = QA_DIR / f"balance-verifier-{ts}.md"
md_path.write_text(md, encoding="utf-8")
print(f"[数值验证] 报告:{md_path}")
print(f"[数值验证] 原始逐局数据:{raw_path}(--from-raw 可免重跑重判)")
return 0
if __name__ == "__main__":
sys.exit(main())