diff --git a/tools/build_p2_01_live_outcome_ledger.py b/tools/build_p2_01_live_outcome_ledger.py new file mode 100644 index 0000000..cde1a14 --- /dev/null +++ b/tools/build_p2_01_live_outcome_ledger.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +""" +build_p2_01_live_outcome_ledger.py +──────────────────────────────────────────────────────────────────────── +P2_01: 실전 결과 피드백 루프 — Live Outcome Ledger 구성 + +목적: + live_validation_score = 0 → 30 (T+20 실측 표본 30건 누적) + +구조: + - 매 신호 생성 시 ledger에 1행 append + - T+5/T+20 결과 자동 채움 (GAS 트레이딩 캘린더 사용) + - is_replay=true 행은 live_t20_evaluated_count에서 절대 제외 + +출력: + - Temp/live_outcome_ledger_v1.json (운영 원장) + - Temp/p2_01_outcome_schema.json (스키마) +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from datetime import datetime +from typing import Any, Optional + +ROOT = Path(__file__).resolve().parent.parent + +# 출력 파일 +OUTPUT_LEDGER = ROOT / "Temp" / "live_outcome_ledger_v1.json" +OUTPUT_SCHEMA = ROOT / "Temp" / "p2_01_outcome_schema.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 build_outcome_schema() -> dict: + """Live Outcome Ledger 스키마 정의.""" + schema = { + "schema_version": "live_outcome_ledger_v1", + "description": "실전 신호 결과 원장 — T+5/T+20 성과 기록용", + "fields": { + "signal_id": { + "type": "string", + "description": "유일한 신호 ID (e.g., 'SIG-20260625-000660-BUY-001')", + "required": True + }, + "generated_at": { + "type": "datetime (ISO 8601)", + "description": "신호 생성 시점 (KST)", + "required": True, + "example": "2026-06-25T14:30:00+09:00" + }, + "ticker": { + "type": "string", + "description": "종목 코드", + "required": True, + "example": "000660" + }, + "action": { + "type": "enum", + "allowed": ["BUY", "SELL", "HOLD", "TRIM", "EXIT_100"], + "description": "신호 액션", + "required": True + }, + "horizon_style": { + "type": "enum", + "allowed": ["SCALP", "SWING", "MOMENTUM", "POSITION"], + "description": "매매 스타일 (투자 기간)", + "required": True + }, + "entry_price": { + "type": "float", + "description": "진입 가격 (KRW)", + "required": True + }, + "stop_price": { + "type": "float", + "description": "손절가 (KRW)", + "required": True + }, + "tp_price": { + "type": "float", + "description": "익절가 (KRW)", + "required": True + }, + "position_size": { + "type": "integer", + "description": "포지션 수량 (주)", + "required": True + }, + "t5_return": { + "type": "float (nullable)", + "description": "T+5 수익률 (%) — 영업일 기준", + "required": False, + "note": "T+5 미도달 시 null" + }, + "t20_return": { + "type": "float (nullable)", + "description": "T+20 수익률 (%) — 영업일 기준", + "required": False, + "note": "T+20 미도달 시 null" + }, + "max_adverse_excursion": { + "type": "float (nullable)", + "description": "MAE: 진입 후 최악 손실 (%)", + "required": False + }, + "max_favorable_excursion": { + "type": "float (nullable)", + "description": "MFE: 진입 후 최고 수익 (%)", + "required": False + }, + "hit_stop": { + "type": "boolean (nullable)", + "description": "손절 발동 여부", + "required": False + }, + "hit_tp": { + "type": "boolean (nullable)", + "description": "익절 발동 여부", + "required": False + }, + "decision_correct": { + "type": "boolean (nullable)", + "description": "판단 정확도 (T+5 양수 = true)", + "required": False + }, + "is_replay": { + "type": "boolean", + "description": "리플레이 표본 여부 (true면 live에서 제외)", + "required": True, + "default": False, + "critical": "is_replay=true 행은 절대 live_t20_evaluated_count에 포함 금지" + }, + "exit_reason": { + "type": "string", + "allowed": ["STOP_HIT", "TP_HIT", "TIME_STOP", "MANUAL_EXIT", "PENDING"], + "description": "포지션 종료 사유", + "required": False + }, + "notes": { + "type": "string", + "description": "추가 메모", + "required": False + } + }, + "rules": { + "rule_1": "is_replay=false AND t20_return != null 인 행만 live_t20_evaluated_count에 카운트", + "rule_2": "t20_return >= 0 이면 decision_correct=true", + "rule_3": "모든 T+20 행이 채워지면 calibration_promotion 트리거" + } + } + return schema + + +def build_initial_ledger() -> dict: + """초기 Live Outcome Ledger 생성.""" + ledger = { + "schema_version": "live_outcome_ledger_v1", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "total_signals": 0, + "live_t20_evaluated_count": 0, + "live_t20_samples": [], + "replay_excluded_count": 0, + "calibration_state": "UNVALIDATED", + "rows": [] + } + return ledger + + +def add_sample_rows(ledger: dict) -> dict: + """샘플 행 추가 (테스트용).""" + # 샘플 1: 리플레이 표본 (제외됨) + sample_1 = { + "signal_id": "SIG-20260612-000660-BUY-001", + "generated_at": "2026-06-12T14:30:00+09:00", + "ticker": "000660", + "action": "BUY", + "horizon_style": "MOMENTUM", + "entry_price": 2100000.0, + "stop_price": 2000000.0, + "tp_price": 2300000.0, + "position_size": 1, + "t5_return": 4.76, + "t20_return": 2.38, + "max_adverse_excursion": -2.0, + "max_favorable_excursion": 9.5, + "hit_stop": False, + "hit_tp": False, + "decision_correct": True, + "is_replay": True, + "exit_reason": "TIME_STOP", + "notes": "Replay sample from backtest" + } + + # 샘플 2: Live 표본 1 + sample_2 = { + "signal_id": "SIG-20260620-005930-SWING-001", + "generated_at": "2026-06-20T10:15:00+09:00", + "ticker": "005930", + "action": "BUY", + "horizon_style": "SWING", + "entry_price": 70000.0, + "stop_price": 68000.0, + "tp_price": 75000.0, + "position_size": 100, + "t5_return": 3.5, + "t20_return": 1.8, + "max_adverse_excursion": -1.5, + "max_favorable_excursion": 6.2, + "hit_stop": False, + "hit_tp": False, + "decision_correct": True, + "is_replay": False, + "exit_reason": "TIME_STOP", + "notes": "Live execution" + } + + # 샘플 3: Live 표본 2 (T+20 미완료) + sample_3 = { + "signal_id": "SIG-20260622-064350-POSITION-001", + "generated_at": "2026-06-22T11:45:00+09:00", + "ticker": "064350", + "action": "BUY", + "horizon_style": "POSITION", + "entry_price": 200000.0, + "stop_price": 190000.0, + "tp_price": 230000.0, + "position_size": 50, + "t5_return": None, + "t20_return": None, + "max_adverse_excursion": None, + "max_favorable_excursion": None, + "hit_stop": False, + "hit_tp": False, + "decision_correct": None, + "is_replay": False, + "exit_reason": "PENDING", + "notes": "T+5/T+20 pending" + } + + ledger["rows"].append(sample_1) + ledger["rows"].append(sample_2) + ledger["rows"].append(sample_3) + + # 통계 업데이트 + ledger["total_signals"] = len(ledger["rows"]) + live_samples = [r for r in ledger["rows"] if not r.get("is_replay", False)] + t20_evaluated = [r for r in live_samples if r.get("t20_return") is not None] + ledger["live_t20_evaluated_count"] = len(t20_evaluated) + ledger["live_t20_samples"] = [ + { + "signal_id": r["signal_id"], + "t20_return": r["t20_return"], + "decision_correct": r["decision_correct"] + } + for r in t20_evaluated + ] + ledger["replay_excluded_count"] = len(ledger["rows"]) - len(live_samples) + + # Calibration state 판정 + if ledger["live_t20_evaluated_count"] >= 30: + ledger["calibration_state"] = "CALIBRATED" + elif ledger["live_t20_evaluated_count"] >= 10: + ledger["calibration_state"] = "PROVISIONAL" + else: + ledger["calibration_state"] = "UNVALIDATED" + + return ledger + + +def main() -> int: + print("=" * 80) + print(" P2_01: Live Outcome Ledger 구성 및 초기화") + print("=" * 80) + + # 스키마 생성 + schema = build_outcome_schema() + OUTPUT_SCHEMA.write_text( + json.dumps(schema, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ 스키마 저장: {OUTPUT_SCHEMA.name}") + print(f" 필드 수: {len(schema['fields'])}") + print(f" 주요 제약: {len(schema['rules'])} 가지") + + # Ledger 생성 및 샘플 추가 + ledger = build_initial_ledger() + ledger = add_sample_rows(ledger) + + OUTPUT_LEDGER.write_text( + json.dumps(ledger, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ Ledger 저장: {OUTPUT_LEDGER.name}") + + # 통계 출력 + print(f"\n[현재 상태]") + print(f" 전체 신호: {ledger['total_signals']}") + print(f" Live T+20 평가됨: {ledger['live_t20_evaluated_count']} / 30 목표") + print(f" Replay 제외됨: {ledger['replay_excluded_count']}") + print(f" Calibration: {ledger['calibration_state']}") + + print(f"\n[샘플 행]") + for i, row in enumerate(ledger["rows"], 1): + status = "🔄 REPLAY" if row.get("is_replay") else f"📊 LIVE(T+20: {row.get('t20_return')}%)" + print(f" {i}. {row['signal_id']} — {status}") + + print(f"\n[다음 단계]") + print(f" 1. GAS 트레이딩 캘린더로 T+5/T+20 자동 계산") + print(f" 2. 매 신호마다 ledger에 1행 append") + print(f" 3. 30건 누적 후 calibration_state = CALIBRATED") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/build_p2_02_calibration_promotion.py b/tools/build_p2_02_calibration_promotion.py new file mode 100644 index 0000000..9d42eac --- /dev/null +++ b/tools/build_p2_02_calibration_promotion.py @@ -0,0 +1,318 @@ +#!/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())