QuantEngineByItz/tools/build_truthfulness_guard_v1.py

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]


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


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default=str(ROOT / "Temp" / "truthfulness_guard_v1.json"))
    args = ap.parse_args()

    oq = _load_json(ROOT / "Temp" / "outcome_quality_score_v1.json")
    agp = _load_json(ROOT / "Temp" / "algorithm_guidance_proof_v1.json")
    ehc = _load_json(ROOT / "Temp" / "evaluation_history_coverage_v1.json")
    tqt5 = _load_json(ROOT / "Temp" / "trade_quality_from_t5_v1.json")
    pah = _load_json(ROOT / "Temp" / "prediction_accuracy_harness_v2.json")

    oq_gate = str(oq.get("gate") or "")
    agp_score = float(agp.get("score") or 0.0)
    maturity = float(((ehc.get("metrics") or {}).get("maturity_pct")) or 0.0)
    oq_metrics = oq.get("metrics") if isinstance(oq.get("metrics"), dict) else {}
    t20_source = str(oq_metrics.get("t20_source") or "")
    t5_op_eval = int(oq_metrics.get("t5_operational_evaluated_count") or 0)
    trade_quality_basis = str(oq_metrics.get("trade_quality_basis") or "")
    tq5_gate = str(tqt5.get("gate") or "MISSING")
    tq5_count = int(tqt5.get("scored_count") or 0)

    contradictions: list[dict[str, Any]] = []

    # TG001: 거버넌스 점수 높은데 결과 평가 불충분
    if oq_gate == "INSUFFICIENT_EVAL" and agp_score >= 99.0:
        contradictions.append(
            {
                "id": "TG001",
                "message": "governance_score_high_but_outcome_insufficient_eval",
                "evidence": {
                    "algorithm_guidance_score": agp_score,
                    "outcome_gate": oq_gate,
                    "evaluation_maturity_pct": maturity,
                },
            }
        )

    # TG002: 운영 T5 표본이 충분한데 중립 fallback을 여전히 사용 (거짓 부풀림 재발 차단)
    _NEUTRAL_SOURCES = {"neutral_due_to_no_operational_t20", "neutral_due_to_insufficient_operational_samples"}
    if t5_op_eval >= 30 and t20_source in _NEUTRAL_SOURCES:
        contradictions.append(
            {
                "id": "TG002",
                "message": "neutral_fallback_used_despite_sufficient_t5_operational_samples",
                "evidence": {
                    "t20_source": t20_source,
                    "t5_operational_evaluated_count": t5_op_eval,
                    "required_min_samples": 30,
                    "action_required": "t20_source must be t5_operational_proxy when t5_op_eval >= 30",
                },
            }
        )

    # TG003: T5 거래품질 표본이 충분한데 trade_quality_basis가 NEUTRAL_MISSING
    if tq5_gate == "PASS" and tq5_count >= 30 and trade_quality_basis == "NEUTRAL_MISSING":
        contradictions.append(
            {
                "id": "TG003",
                "message": "trade_quality_neutral_used_despite_t5_data_available",
                "evidence": {
                    "trade_quality_basis": trade_quality_basis,
                    "trade_quality_from_t5_gate": tq5_gate,
                    "scored_count": tq5_count,
                    "action_required": "trade_quality_basis must be t5_operational when tq_t5 gate=PASS",
                },
            }
        )

    pass_guard = len(contradictions) == 0
    result = {
        "formula_id": "TRUTHFULNESS_GUARD_V1",
        "gate": "PASS" if pass_guard else "CAUTION",
        "contradiction_count": len(contradictions),
        "contradictions": contradictions,
        "policy": {
            "do_not_claim_prediction_100_when_outcome_insufficient_eval": True,
            "governance_score_and_prediction_score_must_be_reported_separately": True,
            "neutral_fallback_forbidden_when_operational_t5_samples_sufficient": True,
            "trade_quality_neutral_forbidden_when_t5_data_available": True,
        },
        "measured_values": {
            "t20_source": t20_source,
            "t5_operational_evaluated_count": t5_op_eval,
            "trade_quality_basis": trade_quality_basis,
            "tq5_gate": tq5_gate,
        },
    }

    out_path = Path(args.out)
    if not out_path.is_absolute():
        out_path = ROOT / out_path
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False))
    return 0


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