QuantEngineByItz/tools/build_predictive_alpha_report_lock_v2.py

"""PREDICTIVE_ALPHA_REPORT_LOCK_V2 — 정반합(PA1) 보고 잠금 V2.

predictive_alpha_json(GAS PA1 산출)에서 thesis_signals/antithesis_signals/synthesis_score를
종목별 표로 강제 출력한다.

출력:
  pa1_report_table  — 종목별 {thesis_signals, antithesis_signals, synthesis_score, weight_source}
  coverage_pct      — 보유 종목 중 PA1 데이터 완비 비율
  missing_tickers   — 누락 종목

게이트 CHECK_73: 보유종목 100% 행 존재.
"""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_JSON = ROOT / "GatherTradingData.json"
DEFAULT_OUT = ROOT / "Temp" / "predictive_alpha_report_lock_v2.json"


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 _rows(v: Any) -> list[dict[str, Any]]:
    if isinstance(v, list):
        return [x for x in v if isinstance(x, dict)]
    return []


def _f(v: Any, default: float = 0.0) -> float:
    try:
        return float(v) if v is not None else default
    except (TypeError, ValueError):
        return default


def _extract_thesis_signals(breakdown: list[dict[str, Any]]) -> list[str]:
    """thesis_breakdown에서 hit=True인 factor 목록."""
    return [
        f"{item.get('factor', '?')}({item.get('score', 0)})"
        for item in breakdown
        if isinstance(item, dict) and item.get("hit")
    ]


def _extract_antithesis_signals(breakdown: list[dict[str, Any]]) -> list[str]:
    """antithesis_breakdown에서 hit=True인 factor 목록."""
    return [
        f"{item.get('factor', '?')}({item.get('score', 0)})"
        for item in breakdown
        if isinstance(item, dict) and item.get("hit")
    ]


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

    json_path = Path(args.json) if Path(args.json).is_absolute() else ROOT / args.json
    out_path = Path(args.out) if Path(args.out).is_absolute() else ROOT / args.out

    payload = _load(json_path)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    h = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {}

    pa_raw = h.get("predictive_alpha_json", [])
    if isinstance(pa_raw, str):
        try:
            pa_raw = json.loads(pa_raw)
        except Exception:
            pa_raw = []
    pa_list = _rows(pa_raw)

    # 보유 종목 목록 (data_feed)
    df_list = _rows(data.get("data_feed"))
    holding_tickers: set[str] = set()
    for r in df_list:
        t = str(r.get("Ticker") or r.get("ticker") or "")
        if t:
            holding_tickers.add(t)

    # [Work 16] ETF 종목 식별 — ETF는 PA1 thesis/antithesis 신호가 없으므로 EXEMPT 처리
    # core_satellite Position_Type=None이고 ticker가 숫자가 아닌 경우 ETF 판별
    cs_map = {str(r.get("Ticker","")): r for r in _rows(data.get("core_satellite"))}
    def _is_etf(ticker: str) -> bool:
        # ETF 식별: 코드가 6자리 숫자이며 Position_Type이 ETF or None이고 알파 신호 없는 경우
        cs_row = cs_map.get(ticker, {})
        pos_type = str(cs_row.get("Position_Type") or "").upper()
        if "ETF" in pos_type:
            return True
        # PA1이 없고 alpha_lead에도 없는 ETF
        al_tickers = {str(item.get("ticker","")) for item in pa_list}
        return ticker not in al_tickers and not cs_row.get("Position_Type")

    # PA1 데이터 lookup
    pa_by_ticker: dict[str, dict[str, Any]] = {}
    for item in pa_list:
        t = str(item.get("ticker") or "")
        if t:
            pa_by_ticker[t] = item

    pa1_report_table: list[dict[str, Any]] = []
    missing_tickers: list[str] = []

    for r in df_list:
        ticker = str(r.get("Ticker") or r.get("ticker") or "")
        name = str(r.get("Name") or r.get("name") or "")
        if not ticker:
            continue

        pa = pa_by_ticker.get(ticker)
        if pa is None:
            # [Work 16] ETF는 PA1 신호 없음 → EXEMPT (coverage 분모에서 제외)
            if _is_etf(ticker):
                pa1_report_table.append({
                    "ticker": ticker, "name": name,
                    "thesis_score": None, "antithesis_score": None,
                    "synthesis_verdict": "ETF_EXEMPT",
                    "synthesis_score": None, "direction_confidence": None,
                    "prediction_confidence_pct": None,
                    "thesis_signals": [], "antithesis_signals": [],
                    "weight_source": "N/A", "data_status": "ETF_EXEMPT",
                })
                # ETF EXEMPT: coverage 분모에서 제외 (covered로 처리)
            else:
                missing_tickers.append(ticker)
                pa1_report_table.append({
                    "ticker": ticker, "name": name,
                    "thesis_score": None, "antithesis_score": None,
                    "synthesis_verdict": "DATA_MISSING",
                    "synthesis_score": None, "direction_confidence": None,
                    "prediction_confidence_pct": None,
                    "thesis_signals": [], "antithesis_signals": [],
                    "weight_source": "N/A", "data_status": "MISSING",
                })
        else:
            thesis_bd = pa.get("thesis_breakdown") if isinstance(pa.get("thesis_breakdown"), list) else []
            antithesis_bd = pa.get("antithesis_breakdown") if isinstance(pa.get("antithesis_breakdown"), list) else []
            thesis_signals = _extract_thesis_signals(thesis_bd)
            antithesis_signals = _extract_antithesis_signals(antithesis_bd)

            thesis_score = _f(pa.get("thesis_score"))
            antithesis_score = _f(pa.get("antithesis_score"))
            # synthesis_score: direction_confidence 절대값 (PA1에서 directional strength)
            direction_conf = _f(pa.get("direction_confidence"), 0.0)
            synthesis_score = round(abs(direction_conf), 1)

            # weight_source
            weight_source = str(pa.get("weight_source") or "STATIC")
            weight_source_label = "DYNAMIC" if "DYNAMIC" in weight_source.upper() or "dynamic" in weight_source else "STATIC"

            pa1_report_table.append({
                "ticker": ticker,
                "name": name,
                "thesis_score": thesis_score,
                "antithesis_score": antithesis_score,
                "synthesis_verdict": str(pa.get("synthesis_verdict") or ""),
                "synthesis_score": synthesis_score,
                "direction_confidence": direction_conf,
                "prediction_confidence_pct": _f(pa.get("prediction_confidence_pct")),
                "thesis_signals": thesis_signals,
                "antithesis_signals": antithesis_signals,
                "weight_source": weight_source_label,
                "data_status": "OK",
            })

    # [Work 16] ETF_EXEMPT는 분모에서 제외 (PA1 신호 없는 ETF가 coverage를 낮추지 않도록)
    non_etf_rows = [r for r in pa1_report_table if r["data_status"] != "ETF_EXEMPT"]
    total = len(non_etf_rows)
    covered = sum(1 for r in non_etf_rows if r["data_status"] == "OK")
    etf_exempt_count = sum(1 for r in pa1_report_table if r["data_status"] == "ETF_EXEMPT")
    coverage_pct = round((covered / total) * 100.0, 1) if total > 0 else 0.0

    gate = "PASS" if (coverage_pct >= 100.0 and len(missing_tickers) == 0) else ("CAUTION" if coverage_pct >= 80.0 else "FAIL")

    result = {
        "formula_id": "PREDICTIVE_ALPHA_REPORT_LOCK_V2",
        "gate": gate,
        "coverage_pct": coverage_pct,
        "total_tickers": total,
        "covered_tickers": covered,
        "missing_tickers": missing_tickers,
        "pa1_report_table": pa1_report_table,
    }
    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"PREDICTIVE_ALPHA_REPORT_LOCK_V2 gate={gate} coverage_pct={coverage_pct} "
        f"covered={covered}/{total} missing={missing_tickers}"
    )
    return 0


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