QuantEngineByItz/tools/validate_predictive_alpha_dialectic_v2.py

"""validate_predictive_alpha_dialectic_v2.py — P1-014: Dialectical Predictive Alpha V2 Calibration

Checks:
  1. bearish_buy_violation_count: BEARISH 판정 종목에 BUY 제안 0건 검증
  2. synthesis_decile_monotonicity: T+20 표본 부족 시 INSUFFICIENT_DATA
  3. alpha_calibration_gate: violations=0 → BEARISH_SAFE / T+20>=30 + monoton → CALIBRATED

T+20 표본이 30건 미만이면 monotonicity 검증 불가 → gate는 BEARISH_SAFE_PENDING_CALIBRATION.
"""
from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

from v7_hardening_common import ROOT, TEMP, load_json, save_json

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)

DEFAULT_ALPHA  = TEMP / "predictive_alpha_engine_v2.json"
DEFAULT_HIST   = ROOT / "Temp" / "proposal_evaluation_history.json"
DEFAULT_OUT    = TEMP / "predictive_alpha_calibration_v2.json"

DECILE_MIN_SAMPLES = 30   # decile monotonicity 검증에 필요한 최소 T+20 표본 수
DECILE_COUNT       = 5    # synthesis_score 5분위

_BEARISH_VERDICTS = {"BEARISH", "STRONG_BEARISH", "AVOID"}
_BUY_ACTIONS      = {"BUY", "STRONG_BUY", "LIMIT_BUY"}


def _check_bearish_violations(alpha_rows: list[dict]) -> list[dict]:
    """현재 alpha 행에서 BEARISH 판정이지만 allow_execution=True인 경우."""
    violations = []
    for row in alpha_rows:
        ticker = row.get("ticker", "")
        if ticker == "DATA_MISSING":
            continue
        verdict = str(row.get("synthesis_verdict") or "").upper()
        dc = float(row.get("direction_confidence") or 0)
        if verdict in _BEARISH_VERDICTS and row.get("allow_execution"):
            violations.append({
                "ticker": ticker,
                "synthesis_verdict": verdict,
                "direction_confidence": dc,
                "violation_type": "BEARISH_EXECUTION_ALLOWED",
            })
    return violations


def _check_history_violations(records: list[dict]) -> list[dict]:
    """과거 이력에서 BEARISH 판정+BUY 제안 케이스."""
    violations = []
    for r in records:
        if r.get("data_origin") == "REPLAY_FROM_KRX_EOD":
            continue  # REPLAY 제외
        action = str(r.get("action") or "").upper()
        pa_json = r.get("predictive_alpha_json") or {}
        if isinstance(pa_json, str):
            try:
                pa_json = json.loads(pa_json)
            except Exception:
                continue
        verdict = str(pa_json.get("synthesis_verdict") or "").upper() if pa_json else ""
        if action in _BUY_ACTIONS and verdict in _BEARISH_VERDICTS:
            violations.append({
                "proposal_id": r.get("proposal_id"),
                "ticker": r.get("ticker"),
                "action": action,
                "synthesis_verdict": verdict,
                "violation_type": "HISTORY_BEARISH_BUY",
            })
    return violations


def _check_decile_monotonicity(records: list[dict]) -> dict:
    """T+20 표본이 있는 경우 synthesis_score decile별 수익률 단조 증가 검증."""
    t20_with_score = [
        r for r in records
        if r.get("t20_evaluation_status") == "EVALUATED_T20"
        and r.get("t20_return_pct") is not None
        and r.get("data_origin") != "REPLAY_FROM_KRX_EOD"
    ]
    if len(t20_with_score) < DECILE_MIN_SAMPLES:
        return {
            "status": "INSUFFICIENT_DATA",
            "note": f"T+20 LIVE 표본 {len(t20_with_score)}/{DECILE_MIN_SAMPLES} — 단조성 검증 불가",
            "pass": False,
        }
    # (실측 데이터가 있을 때의 로직: 향후 자동 실행)
    return {
        "status": "INSUFFICIENT_DATA",
        "note": f"T+20 LIVE 표본 {len(t20_with_score)}/{DECILE_MIN_SAMPLES} — 검증 준비 중",
        "pass": False,
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--alpha", default=str(DEFAULT_ALPHA))
    ap.add_argument("--hist",  default=str(DEFAULT_HIST))
    ap.add_argument("--out",   default=str(DEFAULT_OUT))
    args = ap.parse_args()

    alpha = load_json(Path(args.alpha))
    hist_raw = load_json(Path(args.hist))
    records: list[dict] = hist_raw.get("records", []) if isinstance(hist_raw, dict) else (hist_raw if isinstance(hist_raw, list) else [])

    alpha_rows: list[dict] = alpha.get("rows", []) if isinstance(alpha, dict) else []

    # ── 검증 1: bearish buy violations ──────────────────────────────────────
    live_violations = _check_bearish_violations(alpha_rows)
    history_violations = _check_history_violations(records)
    all_violations = live_violations + history_violations
    bearish_buy_violation_count = len(all_violations)

    # ── 검증 2: synthesis decile monotonicity ──────────────────────────────
    monotonicity = _check_decile_monotonicity(records)
    synthesis_decile_monotonicity = "PASS" if monotonicity.get("pass") else monotonicity["status"]

    # ── gate 판정 ──────────────────────────────────────────────────────────
    if bearish_buy_violation_count > 0:
        gate = "BLOCKED_BEARISH_VIOLATION"
        alpha_calibration_gate = "BLOCKED_BEARISH_VIOLATION"
    elif synthesis_decile_monotonicity == "PASS":
        gate = "CALIBRATED"
        alpha_calibration_gate = "CALIBRATED"
    else:
        gate = "BEARISH_SAFE_PENDING_CALIBRATION"
        alpha_calibration_gate = "BEARISH_SAFE_PENDING_CALIBRATION"

    result = {
        "formula_id": "PREDICTIVE_ALPHA_CALIBRATION_V2",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "gate": gate,
        "alpha_calibration_gate": alpha_calibration_gate,
        "bearish_buy_violation_count": bearish_buy_violation_count,
        "synthesis_decile_monotonicity": synthesis_decile_monotonicity,
        "violations": all_violations,
        "monotonicity_detail": monotonicity,
        "targets": {
            "alpha_calibration_gate": "CALIBRATED",
            "synthesis_decile_monotonicity": "PASS",
            "bearish_buy_violation_count": "==0",
        },
        "gate_path": {
            "CALIBRATED": "bearish_violations==0 AND decile_monotonicity==PASS",
            "BEARISH_SAFE_PENDING_CALIBRATION": "bearish_violations==0 AND T+20<30 (현재 상태)",
            "BLOCKED_BEARISH_VIOLATION": "bearish BUY 감지됨 → 즉시 차단",
        },
    }
    save_json(args.out, result)
    print(json.dumps({k: v for k, v in result.items() if k != "violations"}, ensure_ascii=False, indent=2))
    return 1 if gate == "BLOCKED_BEARISH_VIOLATION" else 0


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