gplan-2/tools/balance_verifier.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""数值验证工具（判定端）——自动化评判 + 元进度爬升测量，替代「亲手试玩再反馈」循环。

两段采集（数据源 = sim/verifier_batch.gd，本文件驱动它并做全部判定）：
  adventure 段 全冒险连探基线（零永久强化）——逐关条件通关率派生 + 止步分布 + 败退节点归因（层/节点/敌人）；
               adventure 段支持 inject 注入档（永久强化 flat OFAT——与白板同种子流配对，行带
               inject_tag/override，标定面 = 逐关条件通关率差）
  campaign 段  营地循环战役（隔离档真实元进度账本：注入→冒险→收束入账→买锻造）——爬升曲线；
               行带 spend 花费序标签（多轨迹合并采集按序分组判定）

评判标准（溯源）：
  ① 逐关条件通关率（冒险口径派生——打通 k-1 关才到达第 k 关，通关数/到达数）：单关直入口口径
     （难度曲线与Boss v0.16）满血零上下文进关，与真实冒险的条件通关率脱节、会误导曲线调参，
     已随单关试玩入口一并退役；旧带值锚定直入口读数不可比，
     带重锚归曲线调参——重锚前本表只测不判。
  ② 元循环目标（数字待确认）：累计通过 5 次第 k 层的冒险后，第 k+1 层的
     条件通关率（滑窗）应进入该层验收带——判定同①随口径退役暂停（重锚前只给滑窗读数与触发点）。
  ③ 魔法道具面：道具工坊六件按通关次数解锁（阈值 1/1/2/2/3/3），局内效果未实装（归 Alpha 魔法道具）——
     本工具只报解锁时点，不入判定。
  ④ 换算率标定：OFAT 注入档 vs 白板的同种子配对逐关条件通关率差
     → 每级带平移量点估计±SE（配对 bootstrap）→ 与 meta.tres 现行表值对照、给建议投值。
     护甲行强条目：读数必须按解锁后参数实测（本段即重测面），不得用基线外推。

表值溯源（单一事实源 = meta.tres，本文件启动时解析——解析失败即 FAIL，不做静默兜底）：
  · 锻造三档 flat（forge_products.flat_value）· 换算率三行（conversion.per_level_shift_pct）
  · 钥石入账（adventure.tres + meta.tres）：每层通关 10 + Boss 首杀/重复杀 5/1 · 8/2 · 12/3
  · 工坊阈值（meta.tres workshop_items）：1/1/2/2/3/3（本文件只做中文名映射，不消费数值）

用法：
  python tools/balance_verifier.py                 # 全段默认规模
  python tools/balance_verifier.py --quick         # 小规模快跑（读数精度低）
  python tools/balance_verifier.py --na 60 --k 60 --seed0 9000 --spend atk_first
  python tools/balance_verifier.py --only adventure --inject max_hp:3 --na 100
  python tools/balance_verifier.py --from-raw <raw.jsonl>   # 免重跑，对既有采集重新判定/出报告
输出：控制台判定结论 + sim-data/balance-verifier-<时间戳>.md（报告）+ 同名 .raw.jsonl（原始逐局数据）。
      两件均落 sim-data/（整目录不入 git，读数以最新一份为准）。
兼容注：旧 raw.jsonl 含 level 段（直入口单关）行——判定自动忽略，
逐关读数一律由冒险行派生。
"""
import argparse
import json
import math
import random
import re
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SLICE = ROOT / "game"
GODOT = r"D:\code\godot\Godot_v4.7.2-stable_win64_console.exe"
SCRIPT = "res://sim/verifier_batch.gd"
READINGS_DIR = ROOT / "sim-data"
VJSON_PREFIX = "VJSON "
ERROR_MARKERS = ("SCRIPT ERROR", "Parse Error")

# ---- 判定标准区（溯源见文件头） ----
# 直入口验收带已随单关直入口口径退役——
# 逐关/检查点判定全部暂停（judge_band 传 band=None = 只测不判），带重锚归曲线调参。
# 带重锚后换算率平移随之启用：带按 band_shift_for(override) 平移后套用（表值驱动，④ 段供数）。
CAMPAIGN_CLEARS_TARGET = 5    # 元循环检查点：累计通过第 k 层的冒险次数（数字待确认）
CAMPAIGN_WINDOW = 10          # 检查点滑窗局数（条件通关率的窗口）
CAMPAIGN_MIN_ATTEMPTS = 5     # 滑窗内少于此局数 = 数据不足（不出判定）
LEVEL_CN = {1: "第一层", 2: "第二层", 3: "第三层"}
NODE_TYPE_CN = {"battle": "战斗", "boss": "Boss", "rest": "恢复", "reward": "奖励", "shop": "商店房"}
OVERRIDE_ATTRS = ["max_hp", "atk", "armor_class"]   # 注入契约三键（词表同 MetaTable.FORGE_ATTRS）
ATTR_CN = {"max_hp": "生命上限", "atk": "基础攻击", "armor_class": "护甲等级"}
SPEND_CN = {"hp_first": "生命优先", "atk_first": "攻击优先", "armor_first": "护甲优先"}
WORKSHOP_CN = {"probe_dust": "显影尘", "sapper_order": "工兵令", "loot_compass": "探宝罗盘",
               "nest_pendulum": "探巢摆钟", "catacomb_scroll": "地穴全图卷", "blast_kit": "爆破工兵包"}

# ---- meta.tres 表值（启动时解析——表值锚配置推导化：判定阈值随配置表走，零硬编码） ----
META_TRES = SLICE / "data" / "tables" / "meta.tres"
FORGE_FLAT = {}    # attr → flat_value（锻造三档每级 flat）
CONVERSION = {}    # attr → per_level_shift_pct（换算率三行）


def load_meta_table():
    """解析 meta.tres 的 forge_products/conversion 行族（; 注释行剥除 + 括号配平 + 键值扫描）。

    解析失败直接 FAIL——静默兜底会复生「过期硬编码常量」bug 类（表值锚配置推导化禁）。
    """
    try:
        text = META_TRES.read_text(encoding="utf-8")
    except OSError as e:
        die(f"meta.tres 不可读：{e}")
    text = "\n".join(ln for ln in text.splitlines() if not ln.strip().startswith(";"))
    for family in ("forge_products", "conversion"):
        rows = _parse_tres_rows(text, family)
        if not rows:
            die(f"meta.tres 行族解析为空：{family}（表结构或解析器失配——人工排查）")
        for row in rows:
            attr = str(row.get("attr", ""))
            if attr not in OVERRIDE_ATTRS:
                die(f"meta.tres {family} attr 出域 {OVERRIDE_ATTRS}: {attr}")
            if family == "forge_products":
                FORGE_FLAT[attr] = int(row.get("flat_value", 0))
            else:
                CONVERSION[attr] = int(row.get("per_level_shift_pct", 0))
    if sorted(FORGE_FLAT) != sorted(OVERRIDE_ATTRS) or sorted(CONVERSION) != sorted(OVERRIDE_ATTRS):
        die(f"meta.tres 三键覆盖不全：forge={FORGE_FLAT} conversion={CONVERSION}")


def _parse_tres_rows(text, family):
    """取 family = [ {...}, ... ] 的行字典列表（.tres 人类可读格式——非严格 YAML，同仓惯例不走上解析器）。"""
    marker = family + " = ["
    start = text.find(marker)
    if start < 0:
        return []
    i = start + len(marker) - 1          # 指向 '['
    depth, j = 0, i
    while j < len(text):
        if text[j] == "[":
            depth += 1
        elif text[j] == "]":
            depth -= 1
            if depth == 0:
                break
        j += 1
    body = text[i:j + 1]
    rows, depth, row_start = [], 0, None
    for idx, ch in enumerate(body):
        if ch == "{":
            if depth == 0:
                row_start = idx
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0 and row_start is not None:
                block = body[row_start:idx + 1]
                row = {}
                for key, val in re.findall(r'"(\w+)":\s*(-?\d+|"[^"]*")', block):
                    row[key] = int(val) if val.lstrip("-").isdigit() else val.strip('"')
                rows.append(row)
    return rows


def band_shift_for(override):
    """注入/锻造 flat 状态 → 通关率带平移量（百分点）= Σ flat÷每级flat×每级平移（表值驱动）。"""
    shift = 0.0
    for attr in OVERRIDE_ATTRS:
        flat = int((override or {}).get(attr, 0))
        per_flat = FORGE_FLAT.get(attr, 1)
        if per_flat > 0:
            shift += flat / per_flat * CONVERSION.get(attr, 0)
    return shift


def shifted_band(band, shift):
    """验收带随平移量平移（clamp [0,100]；band=None = 带未定不平移——重锚前恒此态）。"""
    if band is None:
        return None
    return (max(0.0, band[0] + shift), min(100.0, band[1] + shift))


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"na={args.na}", f"k={args.k}", f"seed0={args.seed0}",
           f"only={args.only}", f"spend={args.spend}", f"inject={args.inject}"]
    print(f"[数值验证] 采集启动（{' · '.join(segs)}，约 {args.est_min:.0f} 分钟）…")
    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:g},{hi:g}] ✓", 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 valid_adventure_runs(rows, policy, baseline_only=False):
    """冒险段有效局（summary 在场——levels_cleared >= 0）；编排异常局单列。
    baseline_only=True 只留白板档（①② 基线判定面——注入档行归 ④ 标定段）。"""
    runs = [r for r in rows if r.get("seg") == "adventure" and r.get("policy") == policy
            and int(r.get("levels_cleared", -1)) >= 0
            and (not baseline_only or r.get("inject_tag", "白板") == "白板")]
    invalid = sum(1 for r in rows if r.get("seg") == "adventure" and r.get("policy") == policy
                  and int(r.get("levels_cleared", -1)) < 0)
    return runs, invalid


def analyze_adventure_levels(rows):
    """逐关条件通关率（冒险口径派生——直入口口径已退役；白板档）：
    到达第 k 关 ⇔ levels_cleared >= k-1；通关第 k 关 ⇔ levels_cleared >= k。
    条件通关率 = 通关数 / 到达数（L1 的到达数 = 全部有效局——冒险即从第一关开始）。"""
    out = {}
    for policy in ("normal", "perfect"):
        runs, invalid = valid_adventure_runs(rows, policy, baseline_only=True)
        levels = {}
        for k in (1, 2, 3):
            reached = sum(1 for r in runs if int(r.get("levels_cleared", 0)) >= k - 1)
            cleared = sum(1 for r in runs if int(r.get("levels_cleared", 0)) >= k)
            levels[k] = {"reached": reached, "cleared": cleared}
        out[policy] = {"n": len(runs), "invalid": invalid, "levels": levels}
    return out


def analyze_adventure(rows):
    out = {}
    for policy in ("normal", "perfect"):
        runs, invalid = valid_adventure_runs(rows, policy, baseline_only=True)
        stopped, death_nodes = {}, {}
        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
        out[policy] = {"n": len(runs), "invalid": invalid, "stopped": stopped, "death_nodes": death_nodes}
    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 campaign_spend_tags(rows):
    """战役段花费序标签清单（按出现序；旧采集无 spend 字段 = 单策略，归 hp_first 桶）。"""
    tags = []
    for r in rows:
        if r.get("seg") != "campaign":
            continue
        t = r.get("spend", "hp_first")
        if t not in tags:
            tags.append(t)
    return tags


def analyze_campaign(rows, spend_tag="hp_first"):
    out = {}
    for policy in ("normal", "perfect"):
        runs = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
                and r.get("spend", "hp_first") == spend_tag and not r.get("anomaly")]
        anomalies = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
                     and r.get("spend", "hp_first") == spend_tag 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 层滑窗条件通关率——带待重锚（直入口口径退役）只报读数
        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)
                    ov = chunk[0][2].get("override", {})
                    cp["forge_at"] = ov
                    cp["band_shift"] = band_shift_for(ov)
                    cp["verdict"], cp["pass"] = judge_band(rate, None)   # 带重锚归曲线调参批
            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 ofat_cells(rows):
    """OFAT 分组（冒险行）：白板基线 {policy: [rows]} 与单轴注入档 {attr: {flat: {policy: [rows]}}}。"""
    base, cells = {}, {}
    for r in rows:
        if r.get("seg") != "adventure" or int(r.get("levels_cleared", -1)) < 0:
            continue
        ov = r.get("override") or {}
        nonzero = [a for a in OVERRIDE_ATTRS if int(ov.get(a, 0)) != 0]
        policy = r["policy"]
        if not nonzero:
            base.setdefault(policy, []).append(r)
        elif len(nonzero) == 1:
            a = nonzero[0]
            cells.setdefault(a, {}).setdefault(int(ov[a]), {}).setdefault(policy, []).append(r)
    return base, cells


def cond_rate(rows, k):
    """第 k 关条件通关率（%）：通关数/到达数；到达数 0 = None。"""
    reached = sum(1 for r in rows if int(r.get("levels_cleared", 0)) >= k - 1)
    cleared = sum(1 for r in rows if int(r.get("levels_cleared", 0)) >= k)
    if reached <= 0:
        return None, reached
    return 100.0 * cleared / reached, reached


def paired_cond_delta(base_rows, cell_rows, k, reps=2000):
    """同种子配对第 k 关条件通关率差（点）与配对 bootstrap SE。

    点估计 = 全配对样本上 两档条件率之差；SE = 配对 bootstrap（同一次重采样同时进两档——
    保种子级相关）。bootsrap 确定性种子固定，复判可重现。
    """
    bmap = {int(r["seed"]): r for r in base_rows}
    pairs = [(bmap[int(r["seed"])], r) for r in cell_rows if int(r["seed"]) in bmap]
    n = len(pairs)
    if n == 0:
        return None
    rate_b, reached_b = cond_rate([p[0] for p in pairs], k)
    rate_c, reached_c = cond_rate([p[1] for p in pairs], k)
    if rate_b is None or rate_c is None:
        return None
    rng = random.Random(20260923)
    deltas = []
    for _ in range(reps):
        idx = [rng.randrange(n) for _ in range(n)]
        rb, _ = cond_rate([pairs[i][0] for i in idx], k)
        rc, _ = cond_rate([pairs[i][1] for i in idx], k)
        if rb is not None and rc is not None:
            deltas.append(rc - rb)
    if not deltas:
        return None
    se = math.sqrt(sum((d - sum(deltas) / len(deltas)) ** 2 for d in deltas) / (len(deltas) - 1)) if len(deltas) > 1 else 0.0
    return {"n": n, "reached_b": reached_b, "reached_c": reached_c,
            "rate_b": rate_b, "rate_c": rate_c, "delta": rate_c - rate_b, "se": se}


def calibration_estimate(rows):
    """标定汇总：每属性 → 逐档明细 + 合并每级平移量（逆方差加权）+ 现行表值对照 + 建议投值。"""
    base, cells = ofat_cells(rows)
    out = {}
    for attr, flats in cells.items():
        per_flat = max(1, FORGE_FLAT.get(attr, 1))
        detail, pts = [], []
        for flat in sorted(flats):
            lv = flat / per_flat
            for policy in sorted(flats[flat]):
                for k in (1, 2, 3):
                    d = paired_cond_delta(base.get(policy, []), flats[flat][policy], k)
                    if d is None:
                        continue
                    per_lvl = d["delta"] / lv
                    per_lvl_se = d["se"] / lv if d["se"] > 0 else None
                    detail.append({"flat": flat, "policy": policy, "k": k, "n": d["n"],
                                   "rate_b": d["rate_b"], "rate_c": d["rate_c"],
                                   "delta": d["delta"], "se": d["se"],
                                   "per_lvl": per_lvl, "per_lvl_se": per_lvl_se})
                    if per_lvl_se is not None and per_lvl_se > 0:
                        pts.append((per_lvl, 1.0 / per_lvl_se ** 2))
        if pts:
            pooled = sum(p * w for p, w in pts) / sum(w for _, w in pts)
            pooled_se = math.sqrt(1.0 / sum(w for _, w in pts))
        elif detail:
            pooled = sum(d["per_lvl"] for d in detail) / len(detail)
            pooled_se = None
        else:
            pooled, pooled_se = None, None
        sug = None if pooled is None else max(0, min(100, round(pooled)))
        out[attr] = {"detail": detail, "pooled": pooled, "pooled_se": pooled_se,
                     "table_val": CONVERSION.get(attr, 0), "suggest": sug}
    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.na} 局/画像 · 战役 {args.k} 局/画像"
      f" · 种子 {args.seed0} 起逐局独立 · 锻造花费策略 = {args.spend}"
      + (f" · 注入档 = {args.inject}" if args.inject else ""))
    A(f"- 切片版本：{meta['commit']} · 口径：逐关读数 = 冒险条件口径（单关直入口口径已退役 2026-09-23，"
      f"带待重锚只测不判——重锚归曲线调参批） · 换算率表值（meta.tres 启动解析）："
      + " · ".join(f"{ATTR_CN[a]}每级平移{CONVERSION.get(a, 0)}点" for a in OVERRIDE_ATTRS))
    A("")

    # ① 逐关条件通关率（冒险口径派生，白板档）
    A("## ① 逐关条件通关率（冒险口径——打通 k-1 关才到达第 k 关；带待重锚只测不判）")
    A("")
    alv = analyze_adventure_levels(rows)
    tbl = []
    for policy in ("normal", "perfect"):
        d = alv[policy]
        cn = "普通档" if policy == "normal" else "完美档"
        for k in (1, 2, 3):
            lv = d["levels"][k]
            if lv["reached"] <= 0:
                tbl.append([LEVEL_CN[k], cn, f"—/{d['n']}", "—（未到达）", "未定"])
                continue
            rate = 100.0 * lv["cleared"] / lv["reached"]
            ci = binom_ci(lv["cleared"], lv["reached"]) * 100
            tbl.append([LEVEL_CN[k], cn, f"{lv['cleared']}/{lv['reached']}",
                        f"{rate:.1f}% ±{ci:.1f}", "未定"])
    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 '—'}")
        A("- 败退节点分布（失败局归因）：")
        for lb, c in sorted(d["death_nodes"].items(), key=lambda x: -x[1]):
            A(f"  - {lb} ×{c}")
        A("")

    # ③ 元循环爬升（多花费序轨迹分组——campaign 行带 spend 标签）
    A("## ③ 营地循环爬升（用户判据④：5 次通关 → 下一层可过）")
    A("")
    for spend in campaign_spend_tags(rows):
        camp = analyze_campaign(rows, spend)
        spend_cn = SPEND_CN.get(spend, spend)
        for policy in ("normal", "perfect"):
            d = camp[policy]
            cn = "普通档" if policy == "normal" else "完美档"
            A(f"### {cn} · {spend_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.get('max_hp', 1)}"
              f"/攻击+{d['forge_final'].get('atk', 0) * FORGE_FLAT.get('atk', 1)}"
              f"/护甲+{d['forge_final'].get('armor_class', 0) * FORGE_FLAT.get('armor_class', 1)}")
            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)}"
                     f"，表值平移 {cp.get('band_shift', 0.0):+.1f} 点——带重锚后随平移判）"
                     if "forge_at" in cp else ""))
            A("")
            tbl = []
            series = d.get("series", {})
            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("")

    # ④ 换算率标定（OFAT 注入档配对读数——护甲行强条目重测面；冒险条件口径）
    A("## ④ 换算率标定（OFAT 注入档 vs 白板，同种子配对逐关条件通关率差）")
    A("")
    est = calibration_estimate(rows)
    if not any(e["detail"] for e in est.values()):
        A("- 无注入档数据（本采集未带 inject 档——标定面跳过）")
        A("")
    else:
        for attr in OVERRIDE_ATTRS:
            e = est.get(attr)
            if not e or not e["detail"]:
                continue
            per_flat = max(1, FORGE_FLAT.get(attr, 1))
            A(f"### {ATTR_CN[attr]}（每级 flat +{per_flat}）")
            A("")
            tbl = []
            for d in e["detail"]:
                cn = "普通档" if d["policy"] == "normal" else "完美档"
                tbl.append([
                    f"+{d['flat']}（{d['flat'] // per_flat} 级）", cn, LEVEL_CN[d["k"]],
                    str(d["n"]), f"{d['rate_b']:.1f}→{d['rate_c']:.1f}",
                    f"{d['delta']:+.1f} ±{d['se']:.1f}",
                    f"{d['per_lvl']:+.1f}" + (f" ±{d['per_lvl_se']:.1f}" if d["per_lvl_se"] is not None else ""),
                ])
            A(fmt_table(["注入档", "画像", "层", "配对局数", "条件率 白板→档", "差(点)", "折每级(点/级)"], tbl))
            A("")
        A("### 三行汇总与建议投值")
        A("")
        tbl = []
        for attr in OVERRIDE_ATTRS:
            e = est.get(attr)
            if not e or (not e["detail"] and e["pooled"] is None):
                tbl.append([ATTR_CN[attr], str(CONVERSION.get(attr, 0)), "—", "—", "—"])
                continue
            pooled_s = "—" if e["pooled"] is None else f"{e['pooled']:+.1f}"
            if e["pooled"] is not None and e["pooled_se"] is not None:
                pooled_s += f" ±{e['pooled_se']:.1f}"
            if e["suggest"] is None:
                state = "无读数"
            elif e["table_val"] == 0:
                state = "表值未投值（现行 0）"
            elif e["table_val"] == e["suggest"]:
                state = "与表值一致 ✓"
            else:
                state = f"与表值偏差 {e['suggest'] - e['table_val']:+d}"
            tbl.append([ATTR_CN[attr], str(e["table_val"]), pooled_s,
                        "—" if e["suggest"] is None else str(e["suggest"]), state])
        A(fmt_table(["属性", "现行表值(点/级)", "实测每级(点/级, 合并)", "建议投值", "状态"], tbl))
        A("- 折算语义：每级该属性强化 ≈ 通关率带平移量（百分点）——配对同种子读数，护甲行即解锁后实测（强条目）")
        A("- 合并口径：逆方差加权跨（档位×画像×层，配对 bootstrap SE）；带平移为线性单值近似——近带顶/带底的读数受天花板压缩，判读看 ④ 明细表")
        A("")
    return "\n".join(lines)


def console_verdict(args, rows):
    """控制台判定结论（先读数后细节）。"""
    alv = analyze_adventure_levels(rows)
    print("\n判定结论（detail 见报告文件）")
    for policy in ("normal", "perfect"):
        d = alv[policy]
        cn = "普通档" if policy == "normal" else "完美档"
        cells = []
        for k in (1, 2, 3):
            lv = d["levels"][k]
            if lv["reached"] > 0:
                rate = 100.0 * lv["cleared"] / lv["reached"]
                cells.append(f"{LEVEL_CN[k]} {rate:.1f}%（{lv['cleared']}/{lv['reached']}）")
        print(f"  [·] 逐关条件通关率·{cn}：{' · '.join(cells) or '—'}（冒险口径——带待重锚只测不判）")
    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 ""))
    for spend in campaign_spend_tags(rows):
        camp = analyze_campaign(rows, spend)
        spend_cn = SPEND_CN.get(spend, spend)
        for policy in ("normal", "perfect"):
            d = camp[policy]
            cn = "普通档" if policy == "normal" else "完美档"
            if not d["runs"]:
                continue
            for cp in d.get("checkpoints", []):
                mark = {True: "✓", False: "✗", None: "·"}.get(cp.get("pass"), "·")
                print(f"  [{mark}] 营地循环·{cn}·{spend_cn}·{LEVEL_CN[cp['level']]}→{LEVEL_CN[cp['level'] + 1]}：{cp['verdict']}")
    est = calibration_estimate(rows)
    for attr in OVERRIDE_ATTRS:
        e = est.get(attr)
        if not e or (not e["detail"] and e["pooled"] is None):
            continue
        pooled_s = "—" if e["pooled"] is None else f"{e['pooled']:+.1f}"
        if e["pooled"] is not None and e["pooled_se"] is not None:
            pooled_s += f" ±{e['pooled_se']:.1f}"
        print(f"  [·] 换算率标定·{ATTR_CN[attr]}：实测每级 {pooled_s} 点 vs 表值 {e['table_val']}"
              + (f" → 建议投 {e['suggest']}" if e["suggest"] is not None else ""))
    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("--na", type=int, default=60, help="基线冒险局数/画像（±95%CI 约 ±9 点——判读须看 CI）")
    ap.add_argument("--k", type=int, default=60, help="战役局数/画像")
    ap.add_argument("--seed0", type=int, default=8000)
    ap.add_argument("--only", default="adventure,campaign")
    ap.add_argument("--spend", default="hp_first", choices=["hp_first", "atk_first", "armor_first"])
    ap.add_argument("--inject", default="", help="冒险段注入档（flat：max_hp:3,atk:1 式——OFAT 标定用）")
    ap.add_argument("--timeout", type=int, default=5400)
    ap.add_argument("--quick", action="store_true", help="小规模快跑（8/10）")
    ap.add_argument("--from-raw", default=None, help="跳过采集，对既有 raw.jsonl 重新判定出报告")
    args = ap.parse_args()
    if args.quick:
        args.na, args.k = 8, 10
    args.est_min = (2 * args.na * 4.0 + 2 * args.k * 4.0) / 60

    load_meta_table()   # 表值锚配置推导化——解析失败即 FAIL，零硬编码兜底

    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)
        READINGS_DIR.mkdir(parents=True, exist_ok=True)
        raw_path = READINGS_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 = READINGS_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())