"""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())