예제 소스

pass_study_report.py

#!/usr/bin/env python3
"""matches.sqlite 로 R5 PASS 원인·승패 리포트를 만든다.

  .venv/bin/python scripts/pass_study_report.py --db logs/<run>/matches.sqlite
"""

from __future__ import annotations

import argparse
import json
import sqlite3
from collections import Counter
from pathlib import Path

MIDGAME_EMPTY_MIN = 120


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--db", required=True)
    ap.add_argument("--out", default="", help="기본: db 옆 PASS_REPORT.md")
    a = ap.parse_args()
    db = Path(a.db)
    out = Path(a.out) if a.out else db.with_name("PASS_REPORT.md")
    con = sqlite3.connect(str(db))

    games = con.execute(
        "select game_id, end_reason, plies, winner, black_area, white_area, katago_lead_b, katago_winner "
        "from games order by game_id"
    ).fetchall()
    lines = [f"# R5 PASS 원인 리포트 — `{db}`", "", "## 대국 (흑=휴리스틱, 백=R5)", ""]
    lines.append("| game | end | plies | 모니터 계가 | B-W | KataGo 흑 기준 | KataGo 승자 |")
    lines.append("|---:|---|---:|---|---:|---:|---|")
    win_m: Counter = Counter()
    win_k: Counter = Counter()
    for gid, end, plies, w, ba, wa, kl, kw in games:
        win_m[w or "none"] += 1
        win_k[kw or "none"] += 1
        diff = "" if ba is None or wa is None else f"{ba - wa:+.1f}"
        lines.append(f"| {gid} | {end} | {plies} | {w} | {diff} | {'' if kl is None else f'{kl:+.1f}'} | {kw} |")
    lines += ["", f"- 모니터 승자: `{dict(win_m)}`", f"- KataGo 승자: `{dict(win_k)}`", ""]

    w_rows = con.execute(
        "select game_id, ply, attempt, kind, empty, pass_reason, pass_requested, late_endgame, "
        "unsettled_empty, legal_n, server_rec, row, col from moves where color='W'"
    ).fetchall()
    first = [r for r in w_rows if r[2] == "first"]
    linked = sum(1 for r in first if r[10])
    lines += [
        "## 백(R5) 응답",
        "",
        f"- 첫 요청 {len(first)}회, 서버 jsonl 연결 {linked}회 ({linked / max(1, len(first)):.1%})",
    ]
    follow = prefer_n = 0
    for r in first:
        if not r[10] or r[3] != "move":
            continue
        rec = json.loads(r[10])
        p = rec.get("vlm_prefer")
        if p:
            prefer_n += 1
            follow += int([r[11], r[12]] == list(p))
    lines.append(f"- 서버 최종 착수 = 모델 제안: {follow}/{prefer_n} ({follow / max(1, prefer_n):.1%})")

    passes = [r for r in w_rows if r[3] == "pass"]
    lines += ["", "## PASS", "", f"- 전체 PASS 응답: {len(passes)} (first {sum(1 for r in passes if r[2] == 'first')}, "
              f"recheck {sum(1 for r in passes if r[2] == 'recheck')}, illegal_retry {sum(1 for r in passes if r[2] == 'illegal_retry')})"]
    by_reason: Counter = Counter()
    by_phase: Counter = Counter()
    requested: Counter = Counter()
    for r in passes:
        if r[2] != "first":
            continue
        phase = "midgame" if (r[4] or 0) >= MIDGAME_EMPTY_MIN else "endgame"
        by_reason[f"{phase}:{r[5] or 'unlinked'}"] += 1
        by_phase[phase] += 1
        requested["모델이 PASS 출력" if r[6] else ("모델은 좌표 출력" if r[6] is not None else "unknown")] += 1
    lines.append(f"- 국면별: `{dict(by_phase)}`")
    lines.append(f"- 모델 출력: `{dict(requested)}`")
    lines += ["", "| phase:pass_reason | n |", "|---|---:|"]
    for k, v in by_reason.most_common():
        lines.append(f"| {k} | {v} |")

    lines += ["", "## 중반 PASS 상세 (빈점 ≥ 120)", ""]
    for r in passes:
        if r[2] != "first" or (r[4] or 0) < MIDGAME_EMPTY_MIN:
            continue
        rec = json.loads(r[10]) if r[10] else {}
        dbg = rec.get("pick_debug") or {}
        tops = {k: (v.get("top") or [])[:3] for k, v in dbg.items()}
        dmg = {k: v.get("damage") for k, v in dbg.items()}
        lines.append(
            f"- g{r[0]} ply{r[1]} empty={r[4]} reason={r[5]} requested={r[6]} late={r[7]} "
            f"unsettled={r[8]} legal={r[9]} top={tops} damage={dmg}"
        )
    out.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"wrote {out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())