"""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())