QuantEngineByItz/tools/build_growth_rate_signal_v1.py

"""GROWTH_RATE_SIGNAL_V1 — 성장률 시그널 산출기.

EPS YoY / 매출 YoY / 영업이익 YoY를 결정론적으로 합산하여 성장 라벨을 부여한다.

주 소스: GatherTradingData.json → EPS_Growth_1Y_Pct, Revenue_Growth_Pct
보완 소스: fundamental_raw_v1.json → eps_krw (현재 EPS 확인)
EPS 프록시: EPS 존재 여부 + Forward_PE 구간 (주 소스 없을 때)

라벨:
  HYPER_GROWTH  ← EPS_Growth ≥ 30% AND Revenue_Growth ≥ 20%
  GROWTH        ← EPS_Growth ≥ 10% OR Revenue_Growth ≥ 10%
  FLAT          ← -10% ≤ growth < 10%
  DECLINE       ← growth < -10%
  DATA_MISSING  ← 모든 소스 결손

buy_modifier:
  HYPER_GROWTH → +15
  GROWTH       → +8
  FLAT         →  0
  DECLINE      → -12
  DATA_MISSING → -3

단기/중기/장기 horizon 적합도:
  HYPER_GROWTH → short=HIGH, mid=HIGH, long=MEDIUM
  GROWTH       → short=MEDIUM, mid=HIGH, long=HIGH
  FLAT         → short=LOW, mid=MEDIUM, long=MEDIUM
  DECLINE      → short=LOW, mid=LOW, long=LOW
  DATA_MISSING → short=UNKNOWN, mid=UNKNOWN, long=UNKNOWN
"""
from __future__ import annotations

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

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

_BUY_MODIFIER: dict[str, int] = {
    "HYPER_GROWTH": 15,
    "GROWTH": 8,
    "FLAT": 0,
    "DECLINE": -12,
    "DATA_MISSING": -3,
    "ETF_EXCLUDED": 0,
}

_HORIZON_FIT: dict[str, dict[str, str]] = {
    "HYPER_GROWTH":  {"short": "HIGH",    "mid": "HIGH",    "long": "MEDIUM"},
    "GROWTH":        {"short": "MEDIUM",  "mid": "HIGH",    "long": "HIGH"},
    "FLAT":          {"short": "LOW",     "mid": "MEDIUM",  "long": "MEDIUM"},
    "DECLINE":       {"short": "LOW",     "mid": "LOW",     "long": "LOW"},
    "DATA_MISSING":  {"short": "UNKNOWN", "mid": "UNKNOWN", "long": "UNKNOWN"},
    "ETF_EXCLUDED":  {"short": "N/A",     "mid": "N/A",     "long": "N/A"},
}


def _load(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    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 [x for x in v if isinstance(x, dict)]
    return []


def _f(v: Any, default: float | None = None) -> float | None:
    if v is None or v == "" or v == "N/A":
        return default
    try:
        return float(v)
    except (TypeError, ValueError):
        return default


def _classify_from_growth(eps_growth: float | None, rev_growth: float | None) -> tuple[str, str]:
    """성장률 수치에서 라벨 산출."""
    if eps_growth is None and rev_growth is None:
        return "DATA_MISSING", "no_growth_data"

    # 양쪽 모두 있으면 우선 복합 판단
    if eps_growth is not None and rev_growth is not None:
        if eps_growth >= 30.0 and rev_growth >= 20.0:
            return "HYPER_GROWTH", f"eps_g={eps_growth:.1f}%_rev_g={rev_growth:.1f}%"
        if eps_growth >= 10.0 or rev_growth >= 10.0:
            return "GROWTH", f"eps_g={eps_growth:.1f}%_rev_g={rev_growth:.1f}%"
        if eps_growth >= -10.0 and rev_growth >= -10.0:
            return "FLAT", f"eps_g={eps_growth:.1f}%_rev_g={rev_growth:.1f}%"
        return "DECLINE", f"eps_g={eps_growth:.1f}%_rev_g={rev_growth:.1f}%"

    # 한쪽만 있을 때
    g = eps_growth if eps_growth is not None else rev_growth
    label_str = "eps_g" if eps_growth is not None else "rev_g"
    assert g is not None
    if g >= 30.0:
        return "HYPER_GROWTH", f"{label_str}={g:.1f}%"
    if g >= 10.0:
        return "GROWTH", f"{label_str}={g:.1f}%"
    if g >= -10.0:
        return "FLAT", f"{label_str}={g:.1f}%"
    return "DECLINE", f"{label_str}={g:.1f}%"


def _classify_proxy_pe(eps: float | None, pe: float | None) -> tuple[str, str, str]:
    """EPS + Forward_PE 기반 성장 프록시 라벨."""
    if eps is None:
        return "DATA_MISSING", "no_eps", "NONE"
    if eps <= 0:
        return "DECLINE", f"eps_neg({eps:.0f})", "LOW"
    # EPS > 0 → PE 구간으로 시장 기대 성장률 추정
    if pe is None:
        return "DATA_MISSING", "eps_positive_no_pe", "NONE"
    pe_f = float(pe)
    if pe_f <= 0:
        return "DATA_MISSING", f"pe_invalid({pe_f:.1f})", "NONE"
    # 낮은 PE → 시장이 저성장 기대 or 저평가
    if pe_f < 10:
        return "FLAT", f"pe_low({pe_f:.1f})", "VERY_LOW"
    if pe_f < 20:
        return "FLAT", f"pe_moderate_low({pe_f:.1f})", "VERY_LOW"
    if pe_f < 35:
        return "GROWTH", f"pe_moderate({pe_f:.1f})", "VERY_LOW"
    if pe_f < 60:
        return "GROWTH", f"pe_high({pe_f:.1f})", "VERY_LOW"
    # PE > 60 → 매우 높은 성장 기대 OR 과열
    return "HYPER_GROWTH", f"pe_extreme({pe_f:.1f})", "VERY_LOW"


def _process_ticker(
    ticker: str,
    name: str,
    raw_row: dict[str, Any] | None,
    df_row: dict[str, Any] | None,
    is_etf: bool,
) -> dict[str, Any]:
    if is_etf:
        return {
            "ticker": ticker,
            "name": name,
            "label": "ETF_EXCLUDED",
            "buy_modifier": 0,
            "confidence": "N/A",
            "data_source": "etf_skip",
            "proxy_basis": None,
            "missing_fields": [],
            "horizon_fit": _HORIZON_FIT["ETF_EXCLUDED"],
            "is_etf": True,
        }

    missing_fields: list[str] = []
    label = "DATA_MISSING"
    confidence = "NONE"
    data_source = "none"
    proxy_basis: str | None = None

    # ── 1순위: data_feed EPS_Growth_1Y_Pct + Revenue_Growth_Pct ─────────────
    eps_g = _f(df_row.get("EPS_Growth_1Y_Pct") if df_row else None)
    rev_g = _f(df_row.get("Revenue_Growth_Pct") if df_row else None)

    if eps_g is not None or rev_g is not None:
        label, proxy_basis = _classify_from_growth(eps_g, rev_g)
        confidence = "HIGH" if (eps_g is not None and rev_g is not None) else "MEDIUM"
        data_source = "data_feed.EPS_Growth+Revenue_Growth"
    else:
        missing_fields += ["data_feed.EPS_Growth_1Y_Pct", "data_feed.Revenue_Growth_Pct"]

        # ── 2순위: EPS 절대값 + Forward_PE 프록시 ─────────────────────────────
        eps = _f(df_row.get("EPS") if df_row else None)
        pe = _f(df_row.get("Forward_PE") if df_row else None)
        if eps is None:
            missing_fields.append("data_feed.EPS")
        if pe is None:
            missing_fields.append("data_feed.Forward_PE")

        label, proxy_basis, confidence = _classify_proxy_pe(eps, pe)
        if confidence != "NONE":
            data_source = "proxy.eps_forward_pe"

    buy_modifier = _BUY_MODIFIER.get(label, -3)
    horizon_fit = _HORIZON_FIT.get(label, _HORIZON_FIT["DATA_MISSING"])

    return {
        "ticker": ticker,
        "name": name,
        "label": label,
        "buy_modifier": buy_modifier,
        "confidence": confidence,
        "data_source": data_source,
        "proxy_basis": proxy_basis,
        "missing_fields": missing_fields,
        "horizon_fit": horizon_fit,
        "is_etf": False,
    }


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

    raw_path = Path(args.raw) if Path(args.raw).is_absolute() else ROOT / args.raw
    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

    raw_data = _load(raw_path)
    raw_map: dict[str, dict[str, Any]] = {
        str(r.get("ticker") or ""): r
        for r in _rows(raw_data.get("rows"))
    }

    gtd = _load(json_path)
    df_list = _rows((gtd.get("data") or {}).get("data_feed"))
    df_map: dict[str, dict[str, Any]] = {str(r.get("Ticker") or ""): r for r in df_list}

    tickers_seen: set[str] = set()
    rows: list[dict[str, Any]] = []
    label_counts: dict[str, int] = {}

    for df_row in df_list:
        ticker = str(df_row.get("Ticker") or "")
        if not ticker or ticker in tickers_seen:
            continue
        tickers_seen.add(ticker)
        name = str(df_row.get("Name") or "")
        # ETF 판별: EPS/Forward_PE/PBR 모두 없으면 ETF
        is_etf = (
            df_row.get("EPS") is None
            and df_row.get("Forward_PE") is None
            and df_row.get("PBR") is None
        )
        raw_row = raw_map.get(ticker)
        if raw_row is not None:
            is_etf = bool(raw_row.get("is_etf", is_etf))

        result = _process_ticker(ticker, name, raw_row, df_row, is_etf)
        rows.append(result)
        lbl = result["label"]
        label_counts[lbl] = label_counts.get(lbl, 0) + 1

    non_etf = [r for r in rows if not r["is_etf"]]
    data_missing_pct = (
        sum(1 for r in non_etf if r["label"] == "DATA_MISSING") / len(non_etf) * 100
        if non_etf else 0.0
    )
    gate = "PASS" if non_etf else "FAIL"

    out = {
        "formula_id": "GROWTH_RATE_SIGNAL_V1",
        "gate": gate,
        "data_missing_pct": round(data_missing_pct, 1),
        "label_counts": label_counts,
        "row_count": len(rows),
        "non_etf_count": len(non_etf),
        "rows": rows,
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    status = "GROWTH_RATE_SIGNAL_V1_OK" if gate != "FAIL" else "GROWTH_RATE_SIGNAL_V1_FAIL"
    print(
        f"GROWTH_RATE_SIGNAL_V1 gate={gate} rows={len(rows)} "
        f"non_etf={len(non_etf)} data_missing_pct={data_missing_pct:.1f}% labels={label_counts}"
    )
    print(status)
    return 0


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