QuantEngineByItz/tools/build_horizon_classification_v1.py

"""HORIZON_CLASSIFICATION_V1 — 종목별 투자 기간 분류기.

data_feed 및 fundamental_multifactor_v3 결과를 결합하여
각 보유 종목의 투자 기간(단/중/장기)을 결정론적으로 분류한다.

분류 결정 트리:
  LONG  ← 핵심 주도주(005930/000660) + 펀더멘털 B등급
  MID   ← 그 외의 펀더멘털 C/D등급 또는 중립 구간
  SHORT ← 과열/약세가 동시에 강한 종목(고RSI, 강한 음의 이격도, 고ATR)
  ETF   ← ETF 종목
  UNKNOWN ← 데이터 부족

출력: Temp/horizon_classification_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" / "horizon_classification_v1.json"
CORE_LONG_TICKERS = {"005930", "000660"}


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)]
    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 _classify_horizon(
    ticker: str,
    grade: str,
    disparity: float,
    atr_pct: float,
    rsi14: float,
    is_etf: bool,
) -> str:
    """결정론적 horizon 분류."""
    if is_etf:
        return "ETF"

    # 핵심 주도주는 변동성이 다소 높아도 장기 호라이즌으로 우선 분류한다.
    if ticker in CORE_LONG_TICKERS and grade in ("A", "B") and abs(disparity) <= 5 and atr_pct <= 8.0:
        return "LONG"

    # 과열 신호 → 단기
    if rsi14 > 70 or disparity > 15:
        return "SHORT"

    # 펀더멘털 F → 단기 또는 알 수 없음
    if grade == "F":
        return "SHORT"

    # 강한 약세/변동성 조합은 단기
    if grade == "B" and disparity <= -8 and rsi14 < 45 and atr_pct >= 7.0:
        return "SHORT"
    if grade == "C" and disparity <= -12 and rsi14 < 40 and atr_pct >= 9.0:
        return "SHORT"

    # 펀더멘털 B + 과열/약세가 아닌 눌림 구간은 장기 후보로 본다.
    if grade == "B" and disparity <= 0 and abs(disparity) <= 5 and atr_pct <= 8.0:
        return "LONG"

    # 펀더멘털 C/D → 중기
    if grade in ("C", "D"):
        return "MID"

    # 펀더멘털 A/B + 이격도 5~15% → 중기 (추가 상승 여력 모니터링)
    if grade in ("A", "B") and abs(disparity) <= 15:
        return "MID"

    return "UNKNOWN"


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 = []
    summary: dict[str, int] = {"SHORT": 0, "MID": 0, "LONG": 0, "ETF": 0, "UNKNOWN": 0}
    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"))
        atr_pct = _f(r.get("ATR20_Pct"))
        rsi14 = _f(r.get("RSI14"), 50.0)

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

        hz = _classify_horizon(t, grade, disparity, atr_pct, rsi14, is_etf)
        summary[hz] = summary.get(hz, 0) + 1

        rows.append({
            "ticker": t,
            "name": name,
            "horizon": hz,
            "fundamental_grade": grade,
            "disparity_pct": round(disparity, 2),
            "atr20_pct": round(atr_pct, 2),
            "rsi14": round(rsi14, 1),
            "formula_id": "HORIZON_CLASSIFICATION_V1",
        })

    # horizon allocation (비ETF 기준)
    non_etf = [r for r in rows if r["horizon"] != "ETF"]
    total_non_etf = len(non_etf) or 1
    allocation_pct = {
        "SHORT": round(summary.get("SHORT", 0) / total_non_etf * 100, 1),
        "MID": round(summary.get("MID", 0) / total_non_etf * 100, 1),
        "LONG": round(summary.get("LONG", 0) / total_non_etf * 100, 1),
    }
    classified_pct = allocation_pct["SHORT"] + allocation_pct["MID"] + allocation_pct["LONG"]
    gate = "PASS" if classified_pct >= 80 else ("CAUTION" if rows else "FAIL")

    out = {
        "formula_id": "HORIZON_CLASSIFICATION_V1",
        "gate": gate,
        "rows": rows,
        "row_count": len(rows),
        "summary": summary,
        "allocation_pct": allocation_pct,
        "classified_pct": classified_pct,
    }
    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,
        "summary": summary,
        "allocation_pct": allocation_pct,
    }, ensure_ascii=False))
    return 0


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