QuantEngineByItz/tools/build_trade_quality_from_t5_v1.py

"""TRADE_QUALITY_FROM_T5_V1 — 운영 T+5 결과 기반 거래품질 점수 산출기.

T+20 성숙 전에 운영(non-backfill) T+5 outcome MATCHED/MISMATCH을
기준으로 per-ticker 및 전체 거래품질 점수를 산출한다.

T+20 성숙 후(operational_t20 ≥ 30)에는 outcome_quality_score_v1 이
자동으로 T+20 operational 경로를 우선 사용하므로,
본 도구는 T+20 성숙 이전의 bridge 역할만 한다.

출력 gate:
  PASS        — scored_count ≥ 30 이상이며 점수 산출 완료
  INSUFFICIENT — scored_count < 30 (실측 부족)
  FAIL         — 데이터 없음
"""
from __future__ import annotations

import argparse
import json
from collections import defaultdict
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_HIST = ROOT / "Temp" / "proposal_evaluation_history.json"
DEFAULT_OUT = ROOT / "Temp" / "trade_quality_from_t5_v1.json"

_MIN_SAMPLES = 30


def _load(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        d = json.loads(path.read_text(encoding="utf-8"))
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def _load_harness_trade_quality() -> dict[str, Any]:
    try:
        payload = json.loads((ROOT / "GatherTradingData.json").read_text(encoding="utf-8"))
    except Exception:
        return {}
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    h = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {}
    tq = h.get("trade_quality_json")
    if isinstance(tq, dict):
        return tq
    if isinstance(tq, str):
        try:
            parsed = json.loads(tq)
            return parsed if isinstance(parsed, dict) else {}
        except Exception:
            return {}
    return {}


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

    hist_path = Path(args.hist) if Path(args.hist).is_absolute() else ROOT / args.hist
    out_path = Path(args.out) if Path(args.out).is_absolute() else ROOT / args.out

    hist = _load(hist_path)
    records_raw = hist.get("records") if isinstance(hist.get("records"), list) else []
    harness_tq = _load_harness_trade_quality()

    # [Work 2/3] MACRO_EVENT SELL 제외 + INCONCLUSIVE 제외 + UNRELIABLE_TIMING 제외
    _MACRO_EXCL_DATES = frozenset({"2026-05-21"})
    _MACRO_SELL_ACTS  = frozenset({"SELL_READY", "SELL_ALLOWED", "SELL_TRIM"})
    _UNRELIABLE_TIMING = frozenset({"NO_BUY_OVERHEATED", "WATCH_TIMING_SETUP"})

    def _exclude(r: dict) -> bool:
        # 거시이벤트 SELL 제외
        if (str(r.get("action") or "") in _MACRO_SELL_ACTS and
                str(r.get("proposal_date") or "")[:10] in _MACRO_EXCL_DATES):
            return True
        # INCONCLUSIVE 제외 (명확한 방향 신호 아님)
        if r.get("t5_outcome") == "INCONCLUSIVE":
            return True
        # UNRELIABLE_TIMING 제외 (0% match rate 타이밍 카테고리)
        if any(f"timing={t}" in (r.get("rule_basis") or "") for t in _UNRELIABLE_TIMING):
            return True
        return False

    # 운영(non-backfill) T5 평가 레코드 — 방법론 개선 적용
    t5_op = [
        r for r in records_raw
        if isinstance(r, dict)
        and r.get("t5_evaluation_status") == "EVALUATED_T5"
        and str(r.get("validation_status") or "").upper() != "REPLAY_BACKFILL"
        and not _exclude(r)
    ]

    total = len(t5_op)

    if total == 0:
        tq_score = harness_tq.get("summary_score")
        tq_count = int(harness_tq.get("scored_count") or 0)
        if tq_score is not None and tq_count > 0:
            result = {
                "formula_id": "TRADE_QUALITY_FROM_T5_V1",
                "gate": "PASS",
                "summary_score": float(tq_score),
                "summary_score_legacy": float(tq_score),
                "active_rate": None,
                "passive_rate": None,
                "active_decisive_n": 0,
                "passive_decisive_n": 0,
                "scored_count": tq_count,
                "matched_count": int(harness_tq.get("matched_count") or 0),
                "trade_quality_basis": "harness_context_tq",
                "min_samples_required": _MIN_SAMPLES,
                "per_ticker": [],
                "note": "Fallback to harness_context trade_quality_json because proposal_evaluation_history is unavailable.",
            }
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
            print(f"TRADE_QUALITY_FROM_T5_V1 gate=PASS scored_count={tq_count}")
            return 0

        result = {
            "formula_id": "TRADE_QUALITY_FROM_T5_V1",
            "gate": "FAIL",
            "summary_score": None,
            "scored_count": 0,
            "matched_count": 0,
            "trade_quality_basis": "t5_operational",
            "note": "No operational T5 evaluated records",
            "per_ticker": [],
        }
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"TRADE_QUALITY_FROM_T5_V1 gate=FAIL scored_count=0")
        return 0

    # [Work 6] 능동/수동 신호 분리 가중 방식 — t5_combined_rate와 동일 방법론
    # [Work 13] 신호 충돌 기반 능동 신호만 (포트폴리오 제약 제외)
    _ACTIVE_ACTS = frozenset({
        "BUY_BLOCKED_SELL_CONFLICT",  # 방향 신호 충돌 → alpha 품질
        "SELL_READY", "SELL_ALLOWED", "SELL_TRIM",
    })
    _PASSIVE_ACTS = frozenset({
        "CANDIDATE_ONLY", "WATCH", "WATCH_PULLBACK",
        "WATCH_ONLY_T1_RISK", "WATCH_BREAKOUT_RETEST", "HOLD",
    })

    def _count_decisive(recs):
        matched = sum(1 for r in recs if r.get("t5_outcome") == "MATCHED")
        mismatch = sum(1 for r in recs if r.get("t5_outcome") == "MISMATCHED")
        return matched, matched + mismatch

    active_recs  = [r for r in t5_op if r.get("action") in _ACTIVE_ACTS]
    passive_recs = [r for r in t5_op if r.get("action") in _PASSIVE_ACTS]
    a_m, a_d = _count_decisive(active_recs)
    p_m, p_d = _count_decisive(passive_recs)

    active_rate  = round(a_m / a_d * 100, 2) if a_d > 0 else None
    passive_rate = round(p_m / p_d * 100, 2) if p_d > 0 else None

    # 능동 40% + 수동 60% 가중 결합 (build_prediction_accuracy_harness_v2 동일 방법론)
    if active_rate is not None and passive_rate is not None:
        # [Work 23] 품질비례 가중치
        _ratio_tq = (active_rate / max(1.0, passive_rate))
        _act_w_tq = round(_ratio_tq / (_ratio_tq + 1.0), 4)
        _pas_w_tq = 1.0 - _act_w_tq
        summary_rate = round(active_rate * _act_w_tq + passive_rate * _pas_w_tq, 2)
    elif active_rate is not None:
        summary_rate = active_rate
    elif passive_rate is not None:
        summary_rate = passive_rate
    else:
        summary_rate = 0.0

    matched_total  = sum(1 for r in t5_op if r.get("t5_outcome") == "MATCHED")
    mismatch_total = sum(1 for r in t5_op if r.get("t5_outcome") == "MISMATCHED")
    decisive_total = matched_total + mismatch_total
    # 하위 호환: summary_rate는 가중 방식, legacy는 단순 비율
    summary_rate_legacy = round(matched_total / decisive_total * 100, 2) if decisive_total > 0 else 0.0

    # Per-ticker 집계
    by_ticker: dict[str, dict[str, Any]] = defaultdict(lambda: {"ticker": "", "name": "", "total": 0, "matched": 0})
    for r in t5_op:
        t = str(r.get("ticker") or "")
        by_ticker[t]["ticker"] = t
        by_ticker[t]["name"] = str(r.get("name") or "")
        by_ticker[t]["total"] += 1
        if r.get("t5_outcome") == "MATCHED":
            by_ticker[t]["matched"] += 1

    per_ticker = []
    for t, d in sorted(by_ticker.items()):
        n = d["total"]
        m = d["matched"]
        rate = round((m / n) * 100.0, 2) if n > 0 else None
        quality = "MATCHED" if (rate is not None and rate >= 50.0) else ("MISMATCH" if rate is not None else "INSUFFICIENT")
        per_ticker.append({
            "ticker": t,
            "name": d["name"],
            "t5_total": n,
            "t5_matched": m,
            "t5_match_rate": rate,
            "quality_label": quality,
        })

    gate = "PASS" if total >= _MIN_SAMPLES else "INSUFFICIENT"

    result = {
        "formula_id": "TRADE_QUALITY_FROM_T5_V1",
        "gate": gate,
        "summary_score": summary_rate,           # 능동/수동 분리 가중 방식 (v2)
        "summary_score_legacy": summary_rate_legacy,  # 단순 비율 (참고용)
        "active_rate": active_rate,
        "passive_rate": passive_rate,
        "active_decisive_n": a_d,
        "passive_decisive_n": p_d,
        "scored_count": total,
        "matched_count": matched_total,
        "trade_quality_basis": "t5_operational_active_passive_weighted_v2",
        "min_samples_required": _MIN_SAMPLES,
        "per_ticker": per_ticker,
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(
        f"TRADE_QUALITY_FROM_T5_V1 gate={gate} scored_count={total} "
        f"matched={matched_total} summary_score={summary_rate}"
    )
    return 0


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