QuantEngineByItz/tools/validate_live_data_activation_gate_v1.py

"""validate_live_data_activation_gate_v1.py — P1-2: live data 자동 전환 게이트

live_t20_count >= 30 도달 시 아래 상태 전환이 자동으로 이루어졌는지 검증한다:
1. continuous_evaluation_dashboard: gate INSUFFICIENT_DATA → PASS
2. prediction_accuracy_harness: calibration_state = CALIBRATED
3. algorithm_guidance_proof: honest_proof_score >= 70
4. pass_100_criteria: RELEASE_GATE_TRUTH_V1 PASS → hts_order_mode 해제

live_t20 < 30이면 PENDING 상태를 보고하고 현재 축적 진행률을 출력한다.

formula_id: VALIDATE_LIVE_DATA_ACTIVATION_GATE_V1
"""
from __future__ import annotations

import json
import sys
from datetime import date
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
TEMP = ROOT / "Temp"

OUTPUT_PATH = TEMP / "live_data_activation_gate_v1.json"

LIVE_T20_THRESHOLD = 30
HONEST_SCORE_THRESHOLD = 70.0
ACTIVATION_DATE_TARGET = "2026-07-15"


def _load_json(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _get_live_t20_count(dashboard: dict, replay_sep: dict) -> int:
    """Extract live T+20 count from available sources."""
    # Try continuous_evaluation_dashboard
    count = dashboard.get("live_t20_count")
    if count is not None:
        return int(count)
    weekly = dashboard.get("weekly_scorecard") or []
    if isinstance(weekly, list) and weekly:
        total = sum(w.get("trade_count", 0) for w in weekly)
        return total

    # Try live_replay_separation
    live_section = replay_sep.get("live") or {}
    return int(live_section.get("t20_count") or live_section.get("evaluated_count") or 0)


def _check_transition_state(dashboard: dict, pred_acc: dict, agp: dict, pass100: dict) -> list[dict]:
    """Check each auto-transition condition."""
    checks = []

    # Check 1: dashboard gate
    d_gate = str(dashboard.get("gate") or "INSUFFICIENT_DATA")
    checks.append({
        "condition": "evaluation_dashboard_gate",
        "target": "PASS",
        "actual": d_gate,
        "passed": d_gate == "PASS",
        "source": "Temp/continuous_evaluation_dashboard_v1.json",
    })

    # Check 2: calibration state
    cal_state = str(pred_acc.get("calibration_state") or "")
    checks.append({
        "condition": "prediction_calibration_state",
        "target": "CALIBRATED",
        "actual": cal_state,
        "passed": cal_state == "CALIBRATED",
        "source": "Temp/prediction_accuracy_harness_v2.json",
    })

    # Check 3: honest_proof_score
    honest = float(agp.get("honest_proof_score") or 0)
    checks.append({
        "condition": "honest_proof_score",
        "target": f">= {HONEST_SCORE_THRESHOLD}",
        "actual": honest,
        "passed": honest >= HONEST_SCORE_THRESHOLD,
        "source": "Temp/algorithm_guidance_proof_v1.json",
    })

    # Check 4: hts_order_mode
    hts_mode = str(pass100.get("hts_order_mode") or "THEORETICAL_ONLY")
    release_gate = str(pass100.get("effective_release_gate") or "FAIL")
    checks.append({
        "condition": "release_gate_truth",
        "target": "PASS",
        "actual": release_gate,
        "passed": release_gate == "PASS",
        "source": "Temp/pass_100_criteria_v3.json",
    })

    return checks


def run() -> dict:
    dashboard  = _load_json(TEMP / "continuous_evaluation_dashboard_v1.json")
    pred_acc   = _load_json(TEMP / "prediction_accuracy_harness_v2.json")
    agp        = _load_json(TEMP / "algorithm_guidance_proof_v1.json")
    pass100    = _load_json(TEMP / "pass_100_criteria_v3.json")
    replay_sep = _load_json(TEMP / "live_replay_separation_v3.json")

    live_t20_count = _get_live_t20_count(dashboard, replay_sep)
    progress_pct   = round(100.0 * live_t20_count / LIVE_T20_THRESHOLD, 1)
    days_remaining = (
        (date.fromisoformat(ACTIVATION_DATE_TARGET) - date.today()).days
        if date.today() < date.fromisoformat(ACTIVATION_DATE_TARGET) else 0
    )

    if live_t20_count < LIVE_T20_THRESHOLD:
        result = {
            "gate": "PENDING",
            "live_t20_count": live_t20_count,
            "live_t20_threshold": LIVE_T20_THRESHOLD,
            "progress_pct": progress_pct,
            "days_to_target": days_remaining,
            "target_date": ACTIVATION_DATE_TARGET,
            "message": (
                f"live_t20={live_t20_count}/{LIVE_T20_THRESHOLD} "
                f"({progress_pct}%) — 목표일 {ACTIVATION_DATE_TARGET}까지 "
                f"{days_remaining}일 잔여"
            ),
            "transition_checks": [],
            "all_transitions_complete": False,
        }
    else:
        transition_checks = _check_transition_state(dashboard, pred_acc, agp, pass100)
        all_pass = all(c["passed"] for c in transition_checks)
        gate = "PASS" if all_pass else "FAIL"
        result = {
            "gate": gate,
            "live_t20_count": live_t20_count,
            "live_t20_threshold": LIVE_T20_THRESHOLD,
            "progress_pct": 100.0,
            "message": (
                f"live_t20 임계값 도달({live_t20_count}건). "
                f"전환 조건 {'전부 PASS' if all_pass else '일부 FAIL — 재빌드 필요'}."
            ),
            "transition_checks": transition_checks,
            "all_transitions_complete": all_pass,
        }

    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    return result


def main() -> None:
    result = run()
    gate = result.get("gate", "PENDING")
    print(f"[VALIDATE_LIVE_DATA_ACTIVATION_GATE_V1] gate={gate}")
    print(f"  {result.get('message')}")
    if gate == "FAIL":
        failed = [c for c in result.get("transition_checks", []) if not c["passed"]]
        for f in failed:
            print(f"  FAIL: {f['condition']} — actual={f['actual']} target={f['target']}")
        sys.exit(1)
    if gate == "PENDING":
        sys.exit(0)  # PENDING은 오류가 아님, 정상 진행 중


if __name__ == "__main__":
    main()