1.3 策略空间的规模控制

#!/usr/bin/env python3
# file: sim/gd_play_01_scale.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/gd_play_01_scale.py
"""卷二 1.3 策略空间的规模控制（S 层）：剪枝、去重、采样三道闸。

策略空间 |A|^D 会爆炸（|A|=6，D=25 时 6^25≈2.8e19）。本节把「怎么把它压回可算」拆成
三道**可量化**的闸，并给出「何时改用采样」的整数判据：

  ① 剪枝：只生成合法动作（非法指令在生成时丢弃，游戏结束后扇出为 0）。
  ② 去重：按 digest 合并到达同一状态的不同序列（序列 ≠ 状态）。
  ③ 采样：枚举撞上上限后，用自实现可播种 PRNG 抽合法序列，估一个代理指标。

采样器先在**能枚举**的深度（D=6）上校准：同一深度，枚举均值 vs 采样均值之差即
采样误差；再推到**不能枚举**的深度（D=12，6^12≈2.2e9）给出估计与分辨率。

判据（全整数）：设 NODE_CAP 为枚举上限，D* = 满足「合法序列数 ≤ NODE_CAP」的最大深度。
D ≤ D* → 枚举；D > D* → 采样，分辨率 = ceil(1000 / 采样数) ‰。

全整数、无浮点、无 random。产出 out/play01_scale.{json,csv,svg}，并报告总耗时。
"""
from __future__ import annotations

import json
import os
import sys
import time
import xml.etree.ElementTree as ET

import ashlib

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
CONFIG = os.path.join(ROOT, "config/gd_play_01_space.json")


def alphabet(spec: dict) -> list[str]:
    return ["p" + str(i) for i in range(spec["params"]["handSize"])] + ["e"]


def cost(spec: dict, cid: str) -> int:
    return next((c["cost"] for c in spec["cards"] if c["id"] == cid), 999999)


def legal(spec: dict, s: dict) -> list[str]:
    if s["over"]:
        return []
    a = ["p" + str(i) for i, cid in enumerate(s["hand"]) if cost(spec, cid) <= s["energy"]]
    a.append("e")
    return a


def seq_bfs(spec: dict, depth: int, cap: int) -> tuple:
    """按**序列**枚举：返回 (逐深度 legal 数, 累计, 撞上限的深度或 None)。"""
    frontier: list[list[str]] = [[]]
    edges, rows, hit = 0, [], None
    for d in range(1, depth + 1):
        nxt = []
        for seq in frontier:
            for a in legal(spec, ashlib.run_spec(spec, seq)):
                if edges >= cap:
                    hit = d
                    break
                edges += 1
                nxt.append(seq + [a])
            if hit:
                break
        rows.append((d, len(nxt), edges))
        if hit or not nxt:
            break
        frontier = nxt
    return rows, hit


def state_bfs(spec: dict, depth: int) -> list:
    """按 digest 去重枚举：返回 (深度, 该深度新增状态, 累计可达状态)。"""
    seen = {ashlib.digest(ashlib.run_spec(spec, []))}
    frontier, rows = [[]], []
    for d in range(1, depth + 1):
        nxt, new = [], 0
        for seq in frontier:
            for a in legal(spec, ashlib.run_spec(spec, seq)):
                g = ashlib.digest(ashlib.run_spec(spec, seq + [a]))
                if g in seen:
                    continue
                seen.add(g)
                new += 1
                nxt.append(seq + [a])
        rows.append((d, new, len(seen)))
        if not nxt:
            break
        frontier = nxt
    return rows


def sample(spec: dict, depth: int, count: int, seed: int) -> tuple:
    """自实现 PRNG 抽合法序列，返回 (样本快照列表, 末态 PRNG state)。"""
    snaps = []
    for _ in range(count):
        seq, state = [], seed % ashlib.RNG_MOD or 1
        for _ in range(depth):
            acts = legal(spec, ashlib.run_spec(spec, seq))
            if not acts:
                break
            state = ashlib.rng_next(state)
            seq.append(acts[state % len(acts)])
        snaps.append(ashlib.run_spec(spec, seq))
        seed = state
    return snaps, seed


def mean_dealt(snaps: list) -> int:
    return sum(s["dealt"] for s in snaps) // len(snaps)


def win_permille(snaps: list) -> int:
    return 1000 * sum(1 for s in snaps if s["won"]) // len(snaps)


def scale_svg(path: str, rows: list, srows: list) -> int:
    root = ashlib.svg_begin(680, 300)
    ashlib.svg_text(root, 20, 30, "1.3 剪枝与去重把空间压回可算（占 |A|^d 的 ‰）", 13)
    for i, (d, legal_n, _cum) in enumerate(rows):
        raw = 6 ** d
        y = 50 + i * 28
        ashlib.svg_text(root, 20, y + 15, f"d={d}", 12)
        ashlib.svg_bar(root, 60, y, 250, 11, legal_n, raw, "#38bdf8")
        st = next((s[2] for s in srows if s[0] == d), 0)
        ashlib.svg_bar(root, 60, y + 12, 250, 11, st, raw, "#f59e0b")
        ashlib.svg_text(root, 320, y + 15, f"合法 {legal_n} / 去重 {st} / 上界 {raw}", 11)
    ashlib.svg_text(root, 20, 288, "蓝 = 合法序列  橙 = 去重状态；两者都远小于 |A|^d", 11, "#94a3b8")
    return ashlib.svg_save(root, path)


def main() -> int:
    t0 = time.perf_counter_ns()
    spec = ashlib.load_spec()
    with open(CONFIG, encoding="utf-8") as f:
        cfg = json.load(f)
    A = len(alphabet(spec))
    D, cap = cfg["enumDepth"], cfg["maxNodes"]
    S, SD = cfg["sampleCount"], cfg["sampleDepth"]

    rows, hit = seq_bfs(spec, D + 1, cap)          # 多探一层，看上限在哪一深度撞上
    srows = state_bfs(spec, D + 1)
    d_star = (hit - 1) if hit else rows[-1][0]     # 撞上限的上一层才是可完整枚举的最大深度

    # 采样器校准：在可枚举的深度 6 上，枚举真值 vs 采样估计
    cal_depth = 6
    enum_snaps = []
    frontier = [[]]
    for _ in range(cal_depth):
        nxt = []
        for seq in frontier:
            nxt += [seq + [a] for a in legal(spec, ashlib.run_spec(spec, seq))]
        frontier = nxt
    enum_snaps = [ashlib.run_spec(spec, seq) for seq in frontier]
    samp_snaps, _ = sample(spec, cal_depth, S, cfg["seeds"][0])
    cal = {"depth": cal_depth, "enumerated": len(enum_snaps), "sampled": S,
           "enumMeanDealt": mean_dealt(enum_snaps), "sampMeanDealt": mean_dealt(samp_snaps),
           "deltaMeanDealt": mean_dealt(samp_snaps) - mean_dealt(enum_snaps),
           "deltaPermille": abs(mean_dealt(samp_snaps) - mean_dealt(enum_snaps)) * 1000
           // max(1, mean_dealt(enum_snaps)),
           "enumWinPermille": win_permille(enum_snaps),
           "sampWinPermille": win_permille(samp_snaps)}

    # 采样器外推：在不可枚举的深度 12 上给估计
    deep_snaps, _ = sample(spec, SD, S, cfg["seeds"][1])
    deep = {"depth": SD, "rawSpace": A ** SD, "meanDealt": mean_dealt(deep_snaps),
            "winPermille": win_permille(deep_snaps),
            "resolutionPermille": -(-1000 // S)}

    n = ashlib.write_csv(os.path.join(OUT, "play01_scale.csv"),
                         ["depth", "legalSeq", "cumLegal", "distinctState"],
                         [[d, lg, cum, next((s[2] for s in srows if s[0] == d), 0)]
                          for d, lg, cum in rows])
    b = scale_svg(os.path.join(OUT, "play01_scale.svg"), rows, srows)

    elapsed = (time.perf_counter_ns() - t0) // 1000000
    report = {"id": "gd_play_01_scale", "alphabetSize": A, "maxNodes": cap,
              "dStar": d_star, "capHitAtDepth": hit, "seqRows": rows, "stateRows": srows,
              "calibration": cal, "deepSample": deep, "elapsedMs": elapsed}
    with open(os.path.join(OUT, "play01_scale.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, sort_keys=True)
        f.write("\n")

    print(f"|A|={A} maxNodes={cap} 判据 D*={d_star}（合法序列 ≤ 上限的最大深度）")
    print("depth legalAtDepth cumLegal distinctStates |A|^depth")
    for d, lg, cum in rows:
        st = next((s[2] for s in srows if s[0] == d), 0)
        print(f"{d} {lg} {cum} {st} {A ** d}")
    print(f"capHitAtDepth={hit}（第 {hit} 层撞上上限，故枚举只到 D*={d_star}）")
    print(f"calibration@{cal_depth}: enumerated={cal['enumerated']} sampled={cal['sampled']} "
          f"meanDealt {cal['enumMeanDealt']} vs {cal['sampMeanDealt']} "
          f"(Δ={cal['deltaMeanDealt']:+d}, {cal['deltaPermille']}‰) "
          f"winPermille {cal['enumWinPermille']} vs {cal['sampWinPermille']}")
    print(f"deepSample@{SD}: rawSpace={deep['rawSpace']} meanDealt={deep['meanDealt']} "
          f"winPermille={deep['winPermille']} resolution={deep['resolutionPermille']}‰")
    print(f"wrote out/play01_scale.csv rows={n}")
    print(f"wrote out/play01_scale.svg bytes={b}")
    print(f"elapsed={elapsed} ms")
    ET.parse(os.path.join(OUT, "play01_scale.svg"))
    return 0


if __name__ == "__main__":
    sys.exit(main())

|A|=6 maxNodes=200000 判据 D*=8（合法序列 ≤ 上限的最大深度）
depth legalAtDepth cumLegal distinctStates |A|^depth
1 6 6 6 6
2 31 37 22 36
3 123 160 67 216
4 464 624 177 1296
5 1975 2599 385 7776
6 8285 10884 711 46656
7 33746 44630 1188 279936
8 136535 181165 1877 1679616
9 18835 200000 2827 10077696
capHitAtDepth=9（第 9 层撞上上限，故枚举只到 D*=8）
calibration@6: enumerated=8285 sampled=4096 meanDealt 18 vs 20 (Δ=+2, 111‰) winPermille 0 vs 0
deepSample@12: rawSpace=2176782336 meanDealt=44 winPermille=294 resolution=1‰
wrote out/play01_scale.csv rows=9
wrote out/play01_scale.svg bytes=5662
elapsed=1106 ms