scripts/summarize_math_ink_06_p_formula_seeds.py

"""P Formula adapter seed 17·31·47의 개별 gate와 distillation 허용 여부를 요약한다."""

from __future__ import annotations

import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import statistics
from typing import Any


REQUIRED_SEEDS06 = frozenset({17, 31, 47})


def summarize_p_formula_seeds06(reports: list[dict[str, Any]]) -> dict[str, Any]:
    """필요 변수: 세 seed report. 작동 원리: seed 누락·R 혼입·test 생략을 거부하고 metric 분산과 AND gate를 계산한다."""

    seeds = [int(report["seed"]) for report in reports]
    if len(reports) != 3 or set(seeds) != REQUIRED_SEEDS06 or len(set(seeds)) != 3:
        raise ValueError("P Formula 요약에는 seed 17·31·47 report가 정확히 하나씩 필요합니다.")
    if any(report.get("track") != "P_approved_formula_only" for report in reports):
        raise ValueError("R-track 또는 알 수 없는 formula adapter report를 distillation 요약에 섞을 수 없습니다.")
    if any(report.get("official_test") is None for report in reports):
        raise ValueError("official test를 생략한 seed는 3-seed gate에 사용할 수 없습니다.")
    fingerprints = {str(report.get("data_sha256") or "") for report in reports}
    if len(fingerprints) != 1 or "" in fingerprints:
        raise ValueError("세 seed는 동일한 비어 있지 않은 P Formula data_sha256을 가져야 합니다.")
    ordered = sorted(reports, key=lambda report: int(report["seed"]))

    def aggregate(path: tuple[str, ...]) -> dict[str, Any]:
        """필요 변수: nested metric 경로. 작동 원리: 세 seed 값·평균·표준편차·최솟값을 반환한다."""

        values = []
        for report in ordered:
            value: Any = report
            for key in path:
                value = value[key]
            values.append(float(value))
        return {
            "values": values,
            "mean": statistics.fmean(values),
            "std": statistics.pstdev(values),
            "minimum": min(values),
        }

    metrics = {
        "validation_exact_top1": aggregate(("selected_validation", "exact_top1")),
        "test_exact_top1": aggregate(("official_test", "exact_top1")),
        "test_exact_top5": aggregate(("official_test", "exact_top5")),
        "test_visual_family_top1": aggregate(("official_test", "visual_family_top1")),
        "test_macro_f1": aggregate(("official_test", "macro_f1")),
        "test_writer_floor": aggregate(("official_test", "writer", "floor")),
    }
    individual = {
        str(report["seed"]): {
            "passed": bool(report.get("seed_gate", {}).get("passed")),
            "checks": report.get("seed_gate", {}).get("checks"),
            "checkpoint": report.get("checkpoint"),
        }
        for report in ordered
    }
    all_passed = all(row["passed"] for row in individual.values())
    return {
        "experiment": "P-MATH-INK-06-FORMULA-ADAPTER-3SEED-001",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "seeds": [17, 31, 47],
        "data_sha256": next(iter(fingerprints)),
        "metrics": metrics,
        "individual_seed_gates": individual,
        "decision": {
            "all_individual_seed_gates_passed": all_passed,
            "student_distillation_allowed": all_passed,
            "teacher_ensemble_mobile_deployment_allowed": False,
            "product_validation": False,
            "next_gate": (
                "single mobile student distillation, LiteRT parity, Android 3-tier benchmark"
                if all_passed
                else "P formula data/recipe 개선 후 세 seed 전부 재검증"
            ),
        },
        "track": "P_approved_formula_only",
        "product_validation": False,
    }


def main() -> None:
    """필요 변수: seed별 report 경로·출력. 작동 원리: UTF-8 JSON summary를 생성한다."""

    parser = argparse.ArgumentParser(description="Summarize 3-seed P formula adapters")
    parser.add_argument("--report", type=Path, action="append", required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    reports = [
        json.loads(path.read_text(encoding="utf-8"))
        for path in args.report
    ]
    summary = summarize_p_formula_seeds06(reports)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()