#!/usr/bin/env python3 """ build_p2_02_calibration_promotion.py ──────────────────────────────────────────────────────────────────────── P2_02: Calibration Promotion — 표본 규모별 상태 전환 목적: UNVALIDATED → PROVISIONAL → CALIBRATED 자동 승격 기준: 1. UNVALIDATED: sample_n < 30 - 모든 가중치/임계값 = EXPERT_PRIOR - 보고서에 UNVALIDATED 표기 필수 2. PROVISIONAL: 30 <= n < 100 AND prediction_match_rate >= 60% - 실측 데이터 기반 조정 시작 가능 3. CALIBRATED: n >= 100 AND expectancy > 0 AND max_drawdown <= budget - 본격 운영 가능 출력: - Temp/calibration_registry_v1.json (캘리브레이션 임계값 레지스트리) - Temp/p2_02_calibration_report.json (승격 보고서) """ from __future__ import annotations import json import sys from pathlib import Path from datetime import datetime from typing import Optional ROOT = Path(__file__).resolve().parent.parent # 입력 파일 LIVE_OUTCOME = ROOT / "Temp" / "live_outcome_ledger_v1.json" # 출력 파일 OUTPUT_REGISTRY = ROOT / "Temp" / "calibration_registry_v1.json" OUTPUT_REPORT = ROOT / "Temp" / "p2_02_calibration_report.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) def load_json(p: Path) -> dict: if not p.exists(): return {} try: return json.loads(p.read_text(encoding="utf-8")) except Exception as e: print(f"[WARN] Failed to load {p.name}: {e}") return {} def build_calibration_registry(outcome_ledger: dict) -> dict: """캘리브레이션 임계값 레지스트리 구성.""" # 샘플 통계 total_signals = outcome_ledger.get("total_signals", 0) live_t20_count = outcome_ledger.get("live_t20_evaluated_count", 0) # 성공률 계산 t20_samples = outcome_ledger.get("live_t20_samples", []) if t20_samples: success_count = sum(1 for s in t20_samples if s.get("decision_correct")) prediction_match_rate = (success_count / len(t20_samples)) * 100 if t20_samples else 0 else: prediction_match_rate = 0 # Calibration state 판정 if live_t20_count >= 100: calibration_state = "CALIBRATED" state_description = "본격 운영 가능 (n>=100, expectancy>0, drawdown<=budget)" elif live_t20_count >= 30 and prediction_match_rate >= 60: calibration_state = "PROVISIONAL" state_description = "실측 조정 가능 (30<=n<100, match_rate>=60%)" else: calibration_state = "UNVALIDATED" state_description = "설계점수 단계 (n<30)" registry = { "schema_version": "calibration_registry_v1", "generated_at": datetime.now().isoformat(), "calibration_state": calibration_state, "state_description": state_description, "sample_metrics": { "total_signals": total_signals, "live_t20_evaluated_count": live_t20_count, "prediction_match_rate_pct": round(prediction_match_rate, 2), "sample_threshold_unvalidated": 30, "sample_threshold_calibrated": 100 }, "calibration_rules": { "UNVALIDATED": { "condition": "sample_n < 30", "weight_source": "EXPERT_PRIOR", "threshold_authority": "spec/*.yaml (fixed)", "live_adjustment": "금지", "report_requirement": "[UNVALIDATED_DESIGN_SCORE: n=N]" , "interpretation": "전문가 기반 설계. 실측 데이터 부족." }, "PROVISIONAL": { "condition": "30 <= sample_n < 100 AND prediction_match_rate >= 60%", "weight_source": "EXPERT_PRIOR + CALIBRATION_OBSERVED", "threshold_authority": "calibration_registry_v1 (moving)", "live_adjustment": "조건부 허용 (신청 후 검증)", "report_requirement": "[PROVISIONAL_CALIBRATION: n=N, match=X%]", "interpretation": "표본 축적 중. 추세 확인됨." }, "CALIBRATED": { "condition": "sample_n >= 100 AND expectancy > 0 AND max_drawdown <= budget", "weight_source": "EXPERT_PRIOR + OBSERVED_EMPIRICAL", "threshold_authority": "calibration_registry_v1 (live)", "live_adjustment": "자유 (일일 갱신)", "report_requirement": "[CALIBRATED: n=N, expectancy=X%, drawdown=Y%]", "interpretation": "본격 운영 단계. 실측 기반." } }, "current_thresholds": { "velocity_1d_chase_block": { "value": 3.0, "unit": "%", "source": "EXPERT_PRIOR", "calibration_state": "UNVALIDATED", "note": "n < 30: 전문가 기반" }, "distribution_risk_score_block_buy": { "value": 70.0, "unit": "score(0-100)", "source": "EXPERT_PRIOR", "calibration_state": "UNVALIDATED", "note": "n < 30: 전문가 기반" }, "alpha_lead_score_pilot_min": { "value": 75.0, "unit": "score(0-100)", "source": "EXPERT_PRIOR", "calibration_state": "UNVALIDATED", "note": "n < 30: 전문가 기반" }, "prediction_match_rate_min": { "value": 60.0, "unit": "%", "source": "EXPERT_PRIOR", "calibration_state": "UNVALIDATED", "note": "current={} — 목표치 미달".format(round(prediction_match_rate, 2)) } }, "promotion_roadmap": [ { "milestone": 30, "target_state": "PROVISIONAL", "condition": "T+20 평가 30건 누적 + prediction_match >= 60%", "current_progress": "{}/30".format(live_t20_count), "eta": "약 2주 (주 3건 신호 기준)" }, { "milestone": 100, "target_state": "CALIBRATED", "condition": "T+20 평가 100건 누적 + expectancy > 0 + drawdown <= budget", "current_progress": "{}/100".format(live_t20_count), "eta": "약 7주 (주 3건 신호 기준)" } ] } return registry def build_promotion_report(outcome_ledger: dict, registry: dict) -> dict: """승격 판정 보고서.""" calibration_state = registry.get("calibration_state", "UNVALIDATED") live_t20_count = outcome_ledger.get("live_t20_evaluated_count", 0) prediction_match = registry.get("sample_metrics", {}).get("prediction_match_rate_pct", 0) report = { "phase": "P2_02_CALIBRATION_PROMOTION", "generated_at": datetime.now().isoformat(), "current_state": calibration_state, "verdict": { "state": calibration_state, "meets_threshold": calibration_state != "UNVALIDATED", "blocking_factors": [] }, "blocking_analysis": { "condition_1_sample_n_30": { "required": 30, "current": live_t20_count, "met": live_t20_count >= 30, "gap": max(0, 30 - live_t20_count) }, "condition_2_match_rate_60": { "required": 60.0, "current": prediction_match, "met": prediction_match >= 60, "gap": max(0, 60 - prediction_match) }, "condition_3_expectancy_positive": { "required": "> 0%", "current": "TBD (미계산)", "met": None, "gap": None } }, "required_actions": [ { "action": "SIGNAL_ACCUMULATION", "description": "T+20 평가 신호 누적", "target": 30, "current": live_t20_count, "priority": "P0", "timeline": "2주" }, { "action": "PREDICTION_ACCURACY_IMPROVEMENT", "description": "T+5 일치율 54.76% → 60%", "target": 60.0, "current": prediction_match, "priority": "P0" if prediction_match < 55 else "P1", "timeline": "1주" }, { "action": "OVERCLAIMED_CALIBRATION_REMOVAL", "description": "현재 EXPERT_PRIOR를 CALIBRATED로 표기 금지", "target": 0, "current": 1, "priority": "P0", "timeline": "즉시" } ], "compliance_checklist": [ { "item": "UNVALIDATED 점수에 [설계, n=N] 주석", "required": True, "current": False, "note": "현재 상태: n={}, 주석 필수".format(live_t20_count) }, { "item": "EXPERT_PRIOR 임계값 고정", "required": True, "current": True, "note": "spec/*.yaml에 명시됨" }, { "item": "Live 조정 금지 (UNVALIDATED 상태)", "required": True, "current": True, "note": "샘플 부족 → 운영 불가" } ] } # Blocking factors 정리 if live_t20_count < 30: report["verdict"]["blocking_factors"].append( "INSUFFICIENT_SAMPLE_N: {} < 30".format(live_t20_count) ) if prediction_match < 60: report["verdict"]["blocking_factors"].append( "LOW_PREDICTION_ACCURACY: {:.2f}% < 60%".format(prediction_match) ) return report def main() -> int: print("=" * 80) print(" P2_02: Calibration Promotion — 표본 규모별 상태 관리") print("=" * 80) # 입력 로드 outcome_ledger = load_json(LIVE_OUTCOME) # Registry 생성 registry = build_calibration_registry(outcome_ledger) # Report 생성 report = build_promotion_report(outcome_ledger, registry) # 저장 OUTPUT_REGISTRY.write_text( json.dumps(registry, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"\n✓ Registry 저장: {OUTPUT_REGISTRY.name}") OUTPUT_REPORT.write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"✓ Report 저장: {OUTPUT_REPORT.name}") # 현재 상태 출력 print(f"\n[현재 상태]") print(f" Calibration: {registry['calibration_state']}") print(f" 샘플 수: {outcome_ledger.get('live_t20_evaluated_count', 0)} / 30 (목표)") print(f" 정확도: {registry['sample_metrics']['prediction_match_rate_pct']:.2f}% (목표 60%)") print(f"\n[Blocking Factors]") for factor in report["verdict"]["blocking_factors"]: print(f" ❌ {factor}") if not report["verdict"]["blocking_factors"]: print(f" ✅ 모든 조건 만족!") print(f"\n[필수 조치]") for action in report["required_actions"]: if action["priority"] == "P0": print(f" 🔴 {action['action']}: {action['current']}/{action['target']}") return 0 if __name__ == "__main__": sys.exit(main())