QuantEngineByItz/tools/build_calibration_priority_v1.py

#!/usr/bin/env python3
"""
build_calibration_priority_v1.py
───────────────────────────────────────────────────────────────────────────────
P4 확장: alpha_feedback_loop_v2.json → calibration_registry.yaml 보정 제안 연결

alpha_feedback_loop_v2.json의 recommended_adjustments 를 읽어
calibration_registry.yaml의 해당 임계값과 연결한 보정 우선순위 리포트를 생성한다.

출력:
  Temp/calibration_priority_v1.json
    - 보정 우선순위 목록 (feedback 신호 기반)
    - 각 임계값의 현재 상태(EXPERT_PRIOR/샘플 수)와 권장 조치
    - alpha_feedback_loop 미포착(miss5_count) 신호와의 연결

사용법:
  python tools/build_calibration_priority_v1.py
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import yaml

ROOT   = Path(__file__).resolve().parent.parent
AFL    = ROOT / "Temp" / "alpha_feedback_loop_v2.json"
REG    = ROOT / "spec" / "calibration_registry.yaml"
OUTPUT = ROOT / "Temp" / "calibration_priority_v1.json"
PREDICTION_ACCURACY = ROOT / "Temp" / "prediction_accuracy_harness_v2.json"


def registry_source_breakdown(reg_index: dict[str, dict]) -> dict:
    """WBS-7.1(2026-06-21) — calibration_registry.yaml 전체의 source별 분포를 매 실행마다
    집계해 'CALIBRATED 비율이 실제로 몇 %인가'를 사람이 grep으로 직접 세지 않아도
    항상 최신 상태로 노출한다(2026-06-21 비판적 리뷰 0c절에서 0/190 발견 당시 수동 집계 필요했던 문제 해소)."""
    counts: dict[str, int] = {"SPEC_DERIVED": 0, "EXPERT_PRIOR": 0, "PROVISIONAL": 0, "CALIBRATED": 0}
    for entry in reg_index.values():
        source = str(entry.get("source", "")).upper()
        if source in counts:
            counts[source] += 1
    total = sum(counts.values())
    return {
        "total_thresholds": total,
        "counts": counts,
        "calibrated_pct": round(100.0 * counts["CALIBRATED"] / total, 2) if total else 0.0,
        "unvalidated_pct": round(100.0 * (counts["SPEC_DERIVED"] + counts["EXPERT_PRIOR"]) / total, 2) if total else 0.0,
    }


def live_t5_status() -> dict:
    """WBS-7.2/7.1(2026-06-21) — T+5 수치를 하드코딩하지 않고 항상 최신 산출물에서 읽는다.
    Temp/prediction_accuracy_harness_v2.json이 없거나 sample=0이면 정직하게 DATA_GATED로 보고한다."""
    if not PREDICTION_ACCURACY.exists():
        return {"status": "ARTIFACT_MISSING", "t5_sample": 0, "t5_match_rate_pct": None}
    data = load_json(PREDICTION_ACCURACY)
    t5_sample = int(data.get("t5_sample") or 0)
    t5_rate = data.get("t5_op_rate")
    return {
        "status": "DATA_GATED" if t5_sample == 0 else "OK",
        "as_of_date": data.get("as_of_date"),
        "t5_sample": t5_sample,
        "t5_match_rate_pct": t5_rate,
    }

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)


# alpha_feedback 요인 → 관련 calibration_registry ID 매핑
FACTOR_TO_REGISTRY: dict[str, list[str]] = {
    "antithesis_balance": [
        "ALEG_V2_GATE1_BLOCK_PCT",
        "ALEG_V2_GATE2_BLOCK_PCT",
        "DSD_V1_CONFIRMED_WS",
    ],
    "passive_signal_quality": [
        "ALEG_V2_GATE1_BLOCK_PCT",   # 뒷박 차단 임계 — timing=None 진입 허용 과다
        "ALEG_V2_GATE1_WAIT_PCT",    # PULLBACK_WAIT 경계
        "K2_REBOUND_TRIGGER_ATR_MULT",  # 반등 트리거 — 타이밍 조건
    ],
    "active_signal_confidence": [
        "ALEG_V2_GATE1_BLOCK_PCT",
        "ALEG_V2_GATE2_BLOCK_PCT",
        "DSD_V1_SIG1_WEIGHT",
        "DSD_V1_SIG2_WEIGHT",
    ],
    "k2_rebound_efficiency": [
        "K2_SPLIT_RATIO",
        "K2_REBOUND_TRIGGER_ATR_MULT",
        "K2_DEADLINE_DAYS",
        "SCR_V4_EFFICIENCY_DAMAGE_PENALTY_COEFF",
    ],
    # CAPITAL_STYLE_ALLOCATION_V1 — 투자성향별 가중치 보정
    # passive_signal_quality miss5_count=51 → 단타/단기 신호 가중치 재보정 필요
    "passive_signal_quality_style": [
        "CSA_SCALP_W_TECHNICAL",     # 단타에서 기술지표 과도 의존 여부 확인
        "CSA_SCALP_W_SMARTMONEY",
        "CSA_SWING_W_TECHNICAL",
        "CSA_SWING_W_SMARTMONEY",
        "CSA_TECH_RSI_OVERSOLD",     # RSI<35 임계 최적화
        "CSA_TECH_DISPARITY_PULLBACK",  # 눌림목 3% 임계 최적화
    ],
    "conviction_calibration": [
        "CSA_POSITION_PCT_HIGH_CONVICTION",  # 80점 임계 → 실측 분포 기반 조정
        "CSA_POSITION_PCT_STRONG",           # 65점 임계
        "CSA_POSITION_PCT_MODERATE",         # 50점 임계
        "CSA_POSITION_PCT_PILOT",            # 35점 임계
    ],
}


def load_json(p: Path) -> dict:
    if not p.exists():
        return {}
    return json.loads(p.read_text(encoding="utf-8"))


def load_registry(p: Path) -> dict[str, dict]:
    if not p.exists():
        return {}
    data = yaml.safe_load(p.read_text(encoding="utf-8"))
    return {t["id"]: t for t in data.get("thresholds", []) if "id" in t}


def _priority_from_registry_entry(entry: dict, source_tag: str, urgency_bias: int) -> dict:
    sample_n = int(entry.get("sample_n", 0) or 0)
    source = str(entry.get("source", "EXPERT_PRIOR"))
    threshold_class = str(entry.get("threshold_class", "standard"))
    urgency = urgency_bias
    if source == "EXPERT_PRIOR":
        urgency += 10
    if source == "PROVISIONAL":
        urgency += 20
    if threshold_class == "live_critical":
        urgency += 15
    if sample_n == 0:
        urgency += 5
    if sample_n > 0:
        urgency += max(0, 30 - sample_n)
    return {
        "calibration_id": entry.get("id", ""),
        "current_value": entry.get("value"),
        "owner_formula": entry.get("owner_formula", ""),
        "source": source,
        "sample_n": sample_n,
        "linked_factor": source_tag,
        "alpha_action": "registry_review",
        "urgency_score": urgency,
        "calibration_path": (
            (
                "표본 30건 이상 확보 후 PROVISIONAL 승격 → "
                if sample_n >= 30
                else f"표본 {30 - sample_n}건 추가 수집 후 PROVISIONAL 승격 → "
            )
            + "실측 T+5 승률 기반 최적값 backtest → CALIBRATED 확정"
        ),
        "rationale": f"source={source}, class={threshold_class}, sample_n={sample_n}",
    }


def main() -> int:
    afl_data  = load_json(AFL)
    reg_index = load_registry(REG)

    sep = "=" * 70
    print(sep)
    print("  알파 피드백 루프 → 보정 우선순위 연결기 (CALIB-PRIORITY-V1)")
    print(sep)

    adjustments = afl_data.get("recommended_adjustments", [])
    cases_analyzed = afl_data.get("cases_analyzed", 0)
    miss5_count    = 0
    for adj in adjustments:
        if adj.get("factor") == "passive_signal_quality":
            miss5_count = int(adj.get("miss5_count", 0))

    print(f"\n  [alpha_feedback_loop_v2] cases_analyzed={cases_analyzed}")
    print(f"  miss5_count(5%+ 급등 미포착)={miss5_count}건 → passive_signal_quality 개선 필요")

    priority_list: list[dict] = []

    for adj in adjustments:
        factor = str(adj.get("factor", ""))
        action = str(adj.get("action", ""))
        rationale = str(adj.get("rationale", ""))
        reg_ids = FACTOR_TO_REGISTRY.get(factor, [])

        for rid in reg_ids:
            reg_entry = reg_index.get(rid)
            if not reg_entry:
                continue
            item = _priority_from_registry_entry(reg_entry, factor, miss5_count if factor == "passive_signal_quality" else 0)
            item["alpha_action"] = action or "feedback_review"
            if rationale:
                item["rationale"] = rationale[:200]
            priority_list.append(item)

    if not priority_list:
        # alpha_feedback_loop가 비어 있어도 registry 자체의 보정 debt를 추적할 수 있게 한다.
        for reg_id, reg_entry in reg_index.items():
            source = str(reg_entry.get("source", "EXPERT_PRIOR"))
            if source not in {"EXPERT_PRIOR", "PROVISIONAL"}:
                continue
            tag = f"registry:{source.lower()}"
            item = _priority_from_registry_entry(reg_entry, tag, 0)
            if source == "PROVISIONAL":
                item["urgency_score"] += 5
            priority_list.append(item)

    # 중복 제거 (같은 rid, 높은 urgency 유지)
    seen: dict[str, dict] = {}
    for p in priority_list:
        rid = p["calibration_id"]
        if rid not in seen or p["urgency_score"] > seen[rid]["urgency_score"]:
            seen[rid] = p
    priority_list = sorted(seen.values(), key=lambda x: -x["urgency_score"])

    print(f"\n  [보정 우선순위 TOP-10]")
    print(f"  {'순위':<4} {'ID':<45} {'값':>7} {'샘플':>5} {'긴급도':>6}")
    print(f"  {'-'*4} {'-'*45} {'-'*7} {'-'*5} {'-'*6}")
    for rank, item in enumerate(priority_list[:10], 1):
        print(
            f"  {rank:<4} {item['calibration_id']:<45} "
            f"{str(item['current_value']):>7} {item['sample_n']:>5} {item['urgency_score']:>6}"
        )

    print(f"\n  [보정 로드맵]")
    print(f"  Step 1 (즉시): 표본 누적 — 매 거래일 T+5 결과 자동 수집")
    print(f"  Step 2 (30건 후): ALEG_V2_GATE1_BLOCK_PCT 3.0% → 실측 최적값으로 PROVISIONAL 승격")
    print(f"  Step 3 (50건 후): DSD_V1 가중치 logistic regression 최적화")
    print(f"  Step 4 (100건 후): K2_SPLIT_RATIO backtest 비교 → CALIBRATED 확정")
    registry_health = registry_source_breakdown(reg_index)
    t5_status = live_t5_status()

    print(f"\n  [캘리브레이션 레지스트리 건강도] (WBS-7.1)")
    print(f"  total={registry_health['total_thresholds']} {registry_health['counts']}")
    print(f"  CALIBRATED={registry_health['calibrated_pct']}%  미검증(SPEC_DERIVED+EXPERT_PRIOR)={registry_health['unvalidated_pct']}%")

    if t5_status["status"] == "DATA_GATED":
        print(f"  miss5_count={miss5_count}건 → T+5 현재 DATA_GATED(sample=0) — passive_signal_quality 개선 영향은 표본 누적 후 측정 가능")
    elif t5_status["status"] == "ARTIFACT_MISSING":
        print(f"  miss5_count={miss5_count}건 → T+5 산출물 없음(Temp/prediction_accuracy_harness_v2.json) — 먼저 생성 필요")
    else:
        print(f"  miss5_count={miss5_count}건 → T+5={t5_status['t5_match_rate_pct']}% (as_of={t5_status.get('as_of_date')}) → passive_signal_quality 개선 핵심")

    result = {
        "status": "CALIBRATION_PRIORITY_OK",
        "cases_analyzed": cases_analyzed,
        "miss5_count": miss5_count,
        "priority_count": len(priority_list),
        "priority_list": priority_list,
        "roadmap": {
            "step1": "표본 누적 — 매 거래일 T+5 결과 자동 수집",
            "step2": "30건 후: ALEG_V2_GATE1_BLOCK_PCT 3.0% → 실측 최적값 PROVISIONAL 승격",
            "step3": "50건 후: DSD_V1 가중치 logistic regression 최적화",
            "step4": "100건 후: K2_SPLIT_RATIO 30/70~60/40 backtest → CALIBRATED",
        },
        "priority_basis": "alpha_feedback_loop_v2" if adjustments else "registry_warning_fallback",
        "registry_health": registry_health,
        "target_improvement": {
            "t5_status": t5_status["status"],
            "current_t5_pct": t5_status["t5_match_rate_pct"],
            "t5_as_of_date": t5_status.get("as_of_date"),
            "target_t5_pct": 55.0,
            "key_lever": f"passive_signal_quality (miss5_count={miss5_count}건 개선)",
        },
    }

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

    print(f"\n  → 결과 저장: {OUTPUT}")
    print(f"  CALIBRATION_PRIORITY_OK\n")
    return 0


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