QuantEngineByItz/tools/build_portfolio_alpha_confidence_per_ticker_v1.py

"""PORTFOLIO_ALPHA_CONFIDENCE_PER_TICKER_V1 — 종목별 알파 신뢰도 산출기.

data_feed의 시그널들을 결합하여 종목별 PAC(Portfolio Alpha Confidence) 점수를 산출.

점수 구성 (-100 ~ +100):
  entry_freshness  35점: MA20 이격도 기반 (낮을수록 신선한 진입)
  breakout_quality 25점: Breakout_Score 기반
  flow_acceleration 20점: Val_Surge_Pct (거래량 급등 비율)
  fundamental_signal 10점: fundamental_multifactor_v3 점수 기반
  rs_slope         10점: RS_Line_20D_Slope 기반

라벨:
  BULLISH  ≥ +30
  NEUTRAL  ≥ -10
  BEARISH  <  -10

출력: Temp/portfolio_alpha_confidence_per_ticker_v1.json
"""
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_FUND = ROOT / "Temp" / "fundamental_multifactor_v3.json"
DEFAULT_OUT = ROOT / "Temp" / "portfolio_alpha_confidence_per_ticker_v1.json"


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


def _rows(v: Any) -> list[dict[str, Any]]:
    if isinstance(v, list):
        return [r for r in v if isinstance(r, dict)]
    if isinstance(v, str):
        try:
            return _rows(json.loads(v))
        except Exception:
            return []
    return []


def _f(v: Any, default: float = 0.0) -> float:
    try:
        return float(v)
    except Exception:
        return default


def _entry_freshness_score(disparity_pct: float, rsi14: float) -> float:
    """진입 신선도: 이격도 낮고 RSI 적정 구간이면 좋음."""
    # disparity가 0~5% 이내면 신선함: +35
    # 10% 이상이면 오버슈팅: -35
    d_score = 0.0
    if disparity_pct <= 5.0:
        d_score = 35.0
    elif disparity_pct <= 10.0:
        d_score = 20.0
    elif disparity_pct <= 20.0:
        d_score = 0.0
    else:
        d_score = -20.0

    # RSI: 30~60이면 좋음
    r_score = 0.0
    if 30 <= rsi14 <= 60:
        r_score = 15.0
    elif 20 <= rsi14 < 30 or 60 < rsi14 <= 70:
        r_score = 5.0
    elif rsi14 > 70:
        r_score = -10.0

    return d_score + r_score


def _breakout_quality_score(breakout_score: float) -> float:
    """브레이크아웃 품질 (0~100 스케일의 breakout_score → -25~+25)."""
    if breakout_score >= 60:
        return 25.0
    elif breakout_score >= 40:
        return 10.0
    elif breakout_score >= 20:
        return 0.0
    else:
        return -15.0


def _flow_accel_score(val_surge_pct: float) -> float:
    """거래량 급등 비율 → 모멘텀 점수."""
    if val_surge_pct >= 3.0:
        return 20.0
    elif val_surge_pct >= 1.5:
        return 10.0
    elif val_surge_pct >= 0.5:
        return 5.0
    else:
        return 0.0


def _fund_signal_score(grade: str) -> float:
    """펀더멘털 등급 → 알파 기여도."""
    grade_map = {"A": 10.0, "B": 7.0, "C": 3.0, "D": -3.0, "F": -8.0, "ETF": 0.0}
    return grade_map.get(grade, 0.0)


def _rs_slope_score(rs_slope: float) -> float:
    """RS Line 20일 기울기 → 상대강도 기여도."""
    if rs_slope > 0.5:
        return 10.0
    elif rs_slope > 0:
        return 5.0
    elif rs_slope > -0.5:
        return 0.0
    else:
        return -8.0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--json", default=str(DEFAULT_JSON))
    ap.add_argument("--fund", default=str(DEFAULT_FUND))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()
    jp = Path(args.json)
    fp = Path(args.fund)
    op = Path(args.out)
    if not jp.is_absolute():
        jp = ROOT / jp
    if not fp.is_absolute():
        fp = ROOT / fp
    if not op.is_absolute():
        op = ROOT / op

    payload = _load(jp)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    df_list = _rows(data.get("data_feed"))

    # 펀더멘털 등급 조회
    fund_rows = _rows(_load(fp).get("rows"))
    fund_map = {str(r.get("ticker") or ""): r for r in fund_rows}

    rows = []
    vals: list[float] = []
    for r in df_list:
        t = str(r.get("Ticker") or r.get("ticker") or "")
        name = r.get("Name") or r.get("name") or ""

        disparity = _f(r.get("Disparity"))
        rsi14 = _f(r.get("RSI14"), 50.0)
        breakout = _f(r.get("Breakout_Score"))
        val_surge = _f(r.get("Val_Surge_Pct"))
        rs_slope = _f(r.get("RS_Line_20D_Slope"))

        fund_info = fund_map.get(t, {})
        grade = str(fund_info.get("grade") or "F")

        s_entry     = _entry_freshness_score(abs(disparity), rsi14)
        s_breakout  = _breakout_quality_score(breakout)
        s_flow      = _flow_accel_score(val_surge)
        s_fund      = _fund_signal_score(grade)
        s_rs        = _rs_slope_score(rs_slope)

        pac = round(s_entry + s_breakout + s_flow + s_fund + s_rs, 2)
        pac = max(-100.0, min(100.0, pac))
        vals.append(pac)

        label = "BULLISH" if pac >= 30 else ("NEUTRAL" if pac >= -10 else "BEARISH")
        rows.append({
            "ticker": t,
            "name": name,
            "pac_score": pac,
            "pac_label": label,
            "breakdown": {
                "entry_freshness": round(s_entry, 2),
                "breakout_quality": round(s_breakout, 2),
                "flow_accel": round(s_flow, 2),
                "fundamental": round(s_fund, 2),
                "rs_slope": round(s_rs, 2),
            },
            "fundamental_grade": grade,
            "formula_id": "PORTFOLIO_ALPHA_CONFIDENCE_PER_TICKER_V1",
        })

    mean = sum(vals) / len(vals) if vals else 0.0
    var = sum((v - mean) ** 2 for v in vals) / len(vals) if vals else 0.0
    std = var ** 0.5
    label_diversity = len({r["pac_label"] for r in rows})

    label_summary: dict[str, int] = {}
    for r in rows:
        lbl = r["pac_label"]
        label_summary[lbl] = label_summary.get(lbl, 0) + 1

    gate = "PASS" if (std >= 5.0 and label_diversity >= 2) else (
        "CAUTION" if rows else "FAIL"
    )

    out = {
        "formula_id": "PORTFOLIO_ALPHA_CONFIDENCE_PER_TICKER_V1",
        "gate": gate,
        "rows": rows,
        "row_count": len(rows),
        "stddev": round(std, 2),
        "label_diversity": label_diversity,
        "label_summary": label_summary,
    }
    op.parent.mkdir(parents=True, exist_ok=True)
    op.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({
        "formula_id": out["formula_id"],
        "gate": gate,
        "stddev": out["stddev"],
        "label_summary": label_summary,
    }, ensure_ascii=False))
    return 0


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