QuantEngineByItz/tools/build_calibration_decision_draft_v1.py

#!/usr/bin/env python3
"""
build_calibration_decision_draft_v1.py
───────────────────────────────────────────────────────────────────────────────
calibration_review_report_v1.json / calibration_approval_list_v1.json을 바탕으로
운영 승인 초안(APPROVE / HOLD / REJECT)을 만든다.

목적:
  - 사람 검토 전 단계에서 결정 초안을 자동 생성
  - source=PROVISIONAL은 원칙적으로 APPROVE
  - PROVISIONAL_CANDIDATE는 HOLD
  - 나머지는 REJECT 또는 HOLD로 사유를 명시

출력:
  Temp/calibration_decision_draft_v1.json
  Temp/calibration_decision_draft_v1.md

사용법:
  python tools/build_calibration_decision_draft_v1.py
"""

from __future__ import annotations

import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
REVIEW = ROOT / "Temp" / "calibration_review_report_v1.json"
APPROVAL = ROOT / "Temp" / "calibration_approval_list_v1.json"
OUT_JSON = ROOT / "Temp" / "calibration_decision_draft_v1.json"
OUT_MD = ROOT / "Temp" / "calibration_decision_draft_v1.md"

if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
    sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)


def _load_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def _table(rows: list[dict[str, Any]], keys: list[str], max_rows: int = 25) -> str:
    if not rows:
        return "_데이터 없음_"
    header = "| " + " | ".join(keys) + " |"
    sep = "| " + " | ".join(["---"] * len(keys)) + " |"
    body = []
    for row in rows[:max_rows]:
        body.append("| " + " | ".join(str(row.get(k, "")).replace("|", "ㅣ") for k in keys) + " |")
    suffix = f"\n\n_...총 {len(rows)}행 중 {max_rows}행 표시_" if len(rows) > max_rows else ""
    return "\n".join([header, sep, *body]) + suffix


def _decide(row: dict[str, Any]) -> tuple[str, str]:
    source = str(row.get("source") or "")
    readiness = str(row.get("readiness") or "")
    sample_n = int(row.get("sample_n") or 0)
    if source == "PROVISIONAL" and sample_n >= 30:
        return "APPROVE", "source=PROVISIONAL and sample_n>=30"
    if source == "PROVISIONAL":
        return "APPROVE", "source=PROVISIONAL"
    if readiness == "PROVISIONAL_CANDIDATE":
        return "HOLD", "Needs provisional review"
    if sample_n >= 10:
        return "HOLD", "Sample present but not provisional"
    return "REJECT", "Insufficient evidence"


def main() -> int:
    review = _load_json(REVIEW)
    approval = _load_json(APPROVAL)

    review_rows = review.get("review_rows") if isinstance(review.get("review_rows"), list) else []
    decisions: list[dict[str, Any]] = []
    summary = {"APPROVE": 0, "HOLD": 0, "REJECT": 0}

    for row in review_rows:
        if not isinstance(row, dict):
            continue
        decision, reason = _decide(row)
        item = {
            "id": row.get("id", ""),
            "source": row.get("source", ""),
            "sample_n": int(row.get("sample_n") or 0),
            "value": row.get("value"),
            "unit": row.get("unit", ""),
            "owner_formula": row.get("owner_formula", ""),
            "readiness": row.get("readiness", ""),
            "decision": decision,
            "reason": reason,
        }
        decisions.append(item)
        summary[decision] += 1

    decisions.sort(key=lambda item: ({"APPROVE": 0, "HOLD": 1, "REJECT": 2}.get(str(item.get("decision") or ""), 3), -int(item.get("sample_n") or 0), str(item.get("id") or "")))

    report = {
        "formula_id": "CALIBRATION_DECISION_DRAFT_V1",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "review_report_path": str(REVIEW),
        "approval_list_path": str(APPROVAL),
        "summary": summary,
        "decision_count": len(decisions),
        "decisions": decisions,
        "approval_candidate_count": int(approval.get("approval_candidate_count") or 0),
    }

    OUT_JSON.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

    md_lines = [
        "# Calibration Decision Draft",
        "",
        "## Summary",
        "",
        f"- APPROVE: {summary['APPROVE']}",
        f"- HOLD: {summary['HOLD']}",
        f"- REJECT: {summary['REJECT']}",
        f"- decision_count: {len(decisions)}",
        "",
        "## Decision Table",
        "",
        _table(decisions, ["id", "source", "sample_n", "decision", "reason", "owner_formula", "readiness"]),
        "",
        "## Evidence",
        "",
        f"- review report: {REVIEW}",
        f"- approval list: {APPROVAL}",
    ]
    OUT_MD.write_text("\n".join(md_lines), encoding="utf-8")

    print(json.dumps({
        "formula_id": report["formula_id"],
        "gate": "PASS" if summary["APPROVE"] else "WARN",
        "approve_count": summary["APPROVE"],
        "hold_count": summary["HOLD"],
        "reject_count": summary["REJECT"],
        "json_path": str(OUT_JSON),
        "md_path": str(OUT_MD),
    }, ensure_ascii=False, indent=2))
    return 0


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