QuantEngineByItz/tools/build_alpha_feedback_loop_v2.py

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