""" build_alpha_feedback_loop_v2.py 목적: proposal_evaluation_history T5 운영 데이터를 분석해 PA1 팩터 가중치 조정 권고를 생성한다. 기존 ALPHA_FEEDBACK_LOOP_V1은 T20 데이터만 사용해 DATA_INSUFFICIENT. V2는 T5 운영 데이터(≥10건)로 즉시 동작한다. AGENTS.md AFL 원칙: "권고만 출력, 자동 적용 금지" → 이 도구는 recommended_adjustments를 생성하지만 자동으로 settings를 수정하지 않는다. → 사용자 승인 후 settings 시트에서 수동 반영. 출력: Temp/alpha_feedback_loop_v2.json """ from __future__ import annotations import argparse import json import statistics from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[1] OUT_PATH = ROOT / "Temp" / "alpha_feedback_loop_v2.json" _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"}) _MIN_SAMPLES = 10 def _load(p: Path) -> dict: if not p.exists(): return {} try: return json.loads(p.read_text(encoding="utf-8")) except Exception: return {} def _exclude(r: dict) -> bool: if (str(r.get("action") or "") in _MACRO_SELL_ACTS and str(r.get("proposal_date") or "")[:10] in _MACRO_EXCL_DATES): return True if r.get("t5_outcome") == "INCONCLUSIVE": return True if any(f"timing={t}" in (r.get("rule_basis") or "") for t in _UNRELIABLE_TIMING): return True return False def _parse_rule(rb: str) -> dict: rb = rb or "" return {p.split("=")[0]: p.split("=")[1] for p in rb.split("|") if "=" in p} def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--hist", default=str(ROOT / "Temp" / "proposal_evaluation_history.json")) ap.add_argument("--out", default=str(OUT_PATH)) args = ap.parse_args() hist = _load(Path(args.hist)) recs_raw = hist.get("records") or [] op_t5 = [ r for r in recs_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) ] if len(op_t5) < _MIN_SAMPLES: result = { "formula_id": "ALPHA_FEEDBACK_LOOP_V2", "status": "DATA_INSUFFICIENT", "cases_analyzed": len(op_t5), "recommended_adjustments": [], } Path(args.out).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"ALPHA_FEEDBACK_LOOP_V2: DATA_INSUFFICIENT (n={len(op_t5)} < {_MIN_SAMPLES})") return 0 # ── 조건 컴포넌트별 T5 성과 분석 ───────────────────────────────────────── component_stats: dict[str, dict] = defaultdict(lambda: {"total": 0, "matched": 0}) for r in op_t5: rb = _parse_rule(r.get("rule_basis")) matched = r.get("t5_outcome") == "MATCHED" for key in ["quality", "timing", "t1", "sell_conflict"]: val = rb.get(key) if val: k = f"{key}={val}" component_stats[k]["total"] += 1 if matched: component_stats[k]["matched"] += 1 # ── 능동/수동 분리 성과 ────────────────────────────────────────────────── _ACTIVE = frozenset({"BUY_BLOCKED_SELL_CONFLICT", "BUY_BLOCKED_PORTFOLIO_GUARD", "BUY_BLOCKED_TRIM_REQUIRED", "SELL_READY"}) _PASSIVE = frozenset({"CANDIDATE_ONLY", "WATCH", "WATCH_PULLBACK", "HOLD"}) def _rate(recs): m = sum(1 for r in recs if r.get("t5_outcome") == "MATCHED") mm = sum(1 for r in recs if r.get("t5_outcome") == "MISMATCHED") n = m + mm return round(m / n * 100, 2) if n > 0 else None, n active_recs = [r for r in op_t5 if r.get("action") in _ACTIVE] passive_recs = [r for r in op_t5 if r.get("action") in _PASSIVE] active_rate, active_n = _rate(active_recs) passive_rate, passive_n = _rate(passive_recs) # ── PA1 팩터 효과 추정 ─────────────────────────────────────────────────── # 현재 PA1 가중치 읽기 json_path = ROOT / "GatherTradingData.json" jdata = _load(json_path) settings = jdata.get("data", {}).get("settings", {}) pa1_current = {k.replace("pa1_w_", ""): v for k, v in (settings.items() if isinstance(settings, dict) else {}.items()) if k.startswith("pa1_w_")} thesis_f = ["pullback_entry", "flow_strong", "rs_leader", "volume_confirm", "rsi_healthy", "brt_leader"] anti_f = ["chase_risk", "distribution", "foreign_sell", "rsi_overbought", "usd_krw_weak", "stale_position"] thesis_sum = sum(pa1_current.get(f, 0) for f in thesis_f) anti_sum = sum(pa1_current.get(f, 0) for f in anti_f) # ── 권고 생성 ──────────────────────────────────────────────────────────── recommendations = [] # 1. sell_pass 정확도 기반 antithesis 조정 sell_recs = [r for r in op_t5 if r.get("action") in ("SELL_READY", "SELL_ALLOWED")] sell_rate, sell_n = _rate(sell_recs) if sell_rate is not None and sell_rate < 50 and sell_n >= 5: # sell 정확도가 낮다 → antithesis가 지나치게 강하다 # → antithesis 일부 완화, thesis 강화 권고 recommendations.append({ "factor": "antithesis_balance", "current_ratio": round(anti_sum / max(1, thesis_sum), 2), "target_ratio": "2.0~3.0x", "action": "antithesis 일부 완화 + thesis 강화", "details": { "pa1_w_usd_krw_weak": {"current": pa1_current.get("usd_krw_weak", 40), "recommended": 15}, "pa1_w_stale_position": {"current": pa1_current.get("stale_position", 40), "recommended": 20}, "pa1_w_flow_strong": {"current": pa1_current.get("flow_strong", 5), "recommended": 15}, "pa1_w_pullback_entry": {"current": pa1_current.get("pullback_entry", 5), "recommended": 15}, }, "rationale": ( f"SELL 신호 정확도={sell_rate:.1f}%(n={sell_n}) < 50% - " f"antithesis {anti_sum}pt가 thesis {thesis_sum}pt의 {anti_sum/max(1,thesis_sum):.1f}x로 " f"지나치게 강해 모든 종목이 획일적 EXIT 신호를 받음. " f"usd_krw_weak/stale_position은 종목 차별화에 기여하지 않으므로 완화." ), }) else: recommendations.append({ "factor": "antithesis_balance", "current_ratio": round(anti_sum / max(1, thesis_sum), 2), "action": "현행 유지", "rationale": f"sell_rate={sell_rate}% 또는 표본 부족(n={sell_n})", }) # 2. 수동신호 개선 권고 (passive_rate 낮은 경우) if passive_rate is not None and passive_rate < 35: # 수동신호 정확도가 낮다 → WATCH/CANDIDATE 진입 조건 강화 miss5_passive = [r for r in passive_recs if r.get("t5_outcome") == "MISMATCHED" and (r.get("t5_return_pct") or 0) >= 5] timing_none_n = sum(1 for r in miss5_passive if _parse_rule(r.get("rule_basis")).get("timing", "None") == "None") recommendations.append({ "factor": "passive_signal_quality", "passive_rate_pct": passive_rate, "passive_n": passive_n, "miss5_count": len(miss5_passive), "action": "timing=None CANDIDATE에 PULLBACK_ENTRY_TRIGGER_V1 조건 필수화", "spec_ref": "AGENTS.md Direction B1", "rationale": ( f"수동신호 정확도={passive_rate:.1f}%(n={passive_n}), " f"5%+ 급등 미포착={len(miss5_passive)}건 중 timing=None이 {timing_none_n}건. " f"timing 조건 없이 alpha_lead만으로 CANDIDATE 상태에 오른 종목들이 " f"갑작스러운 급등 시 대응 불가. PULLBACK_ENTRY_TRIGGER 조건 필수화 필요." ), }) # 3. 능동신호 강화 권고 (active_rate가 높을 때 → 이 신호에 더 의존) if active_rate is not None and active_rate >= 65: recommendations.append({ "factor": "active_signal_confidence", "active_rate_pct": active_rate, "active_n": active_n, "action": f"BUY_BLOCKED 신호 신뢰도 {active_rate:.1f}%로 높음 - 포지션 규모 보수 유지 가능", "rationale": "능동 차단 신호가 정확하므로 현 리스크 관리 체계 유지 권고.", }) # ── 컴포넌트 분석 요약 ─────────────────────────────────────────────────── component_analysis = [] for cond, stat in sorted(component_stats.items(), key=lambda x: -x[1]["total"]): n = stat["total"]; m = stat["matched"] if n >= 5: component_analysis.append({ "condition": cond, "total": n, "matched": m, "match_rate": round(m / n * 100, 1), }) # ── 점수 추정 ──────────────────────────────────────────────────────────── combined_rate = (active_rate or 0) * 0.40 + (passive_rate or 0) * 0.60 if (active_rate and passive_rate) else None result = { "formula_id": "ALPHA_FEEDBACK_LOOP_V2", "status": "ANALYZED", "cases_analyzed": len(op_t5), "active_signal_rate_pct": active_rate, "active_signal_n": active_n, "passive_signal_rate_pct": passive_rate, "passive_signal_n": passive_n, "combined_rate_pct": round(combined_rate, 2) if combined_rate else None, "sell_signal_rate_pct": sell_rate, "sell_signal_n": sell_n, "pa1_current_ratio": round(anti_sum / max(1, thesis_sum), 2), "pa1_thesis_sum": thesis_sum, "pa1_antithesis_sum": anti_sum, "recommended_adjustments": recommendations, "component_analysis": component_analysis[:20], "note": "AFL 권고는 사용자 승인 후 GAS settings 시트에서 수동 반영 (자동 적용 금지)", } Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"ALPHA_FEEDBACK_LOOP_V2: status=ANALYZED cases={len(op_t5)} " f"active={active_rate:.1f}%(n={active_n}) passive={passive_rate:.1f}%(n={passive_n}) " f"pa1_ratio={anti_sum}/{thesis_sum}={anti_sum/max(1,thesis_sum):.1f}x") print(f" 권고 수: {len(recommendations)}건") for rec in recommendations: print(f" [{rec['factor']}] {rec['action']}") return 0 if __name__ == "__main__": raise SystemExit(main())