QuantEngineByItz/tools/build_liquidity_flow_signal_v1.py

"""LIQUIDITY_FLOW_SIGNAL_V1 — 종목별 유동성 등급 및 실행 모드 산출기.

data_feed의 AvgTradeValue_20D_M 기반으로 종목별 유동성을 분류하고
매도 실행 모드를 결정한다.

유동성 라벨:
  DEEP   → 20일 평균 거래대금 ≥ 200,000 M KRW/일  (MARKET_OK)
  NORMAL → ≥ 50,000 M KRW/일                      (LIMIT_NEAR_BID)
  THIN   → ≥ 5,000 M KRW/일                       (TWAP_SPLIT)
  FROZEN → < 5,000 M KRW/일                        (HOLD)

출력: Temp/liquidity_flow_signal_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_OUT = ROOT / "Temp" / "liquidity_flow_signal_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 [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) -> float:
    try:
        return float(v)
    except Exception:
        return 0.0


def _liquidity_label(trade_v_m: float) -> tuple[str, str]:
    """20일 평균 거래대금(M KRW) → (label, execution_mode)."""
    if trade_v_m >= 200_000:
        return "DEEP", "MARKET_OK"
    elif trade_v_m >= 50_000:
        return "NORMAL", "LIMIT_NEAR_BID"
    elif trade_v_m > 5_000:
        return "THIN", "TWAP_SPLIT"
    else:
        return "FROZEN", "HOLD"


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

    payload = _load(jp)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    # data_feed에서 직접 읽기 (prices_json이 비어있을 경우 fallback)
    df_list = _rows(data.get("data_feed"))

    rows = []
    label_set: set[str] = set()
    for r in df_list:
        t = str(r.get("Ticker") or r.get("ticker") or "")
        name = r.get("Name") or r.get("name") or ""
        # AvgTradeValue_20D_M 또는 AvgTradeValue_5D_M 사용
        trade_v = _f(r.get("AvgTradeValue_20D_M") or r.get("AvgTradeValue_5D_M") or
                     r.get("avg_trade_value_20d_m") or r.get("avgTradeVal20d") or 0)
        label, mode = _liquidity_label(trade_v)
        label_set.add(label)
        rows.append({
            "ticker": t,
            "name": name,
            "liquidity_label": label,
            "execution_mode": mode,
            "avg_trade_value_20d_m": round(trade_v, 2),
            "formula_id": "LIQUIDITY_FLOW_SIGNAL_V1",
        })

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

    gate = "PASS" if len(label_set) >= 2 else ("CAUTION" if rows else "FAIL")

    out = {
        "formula_id": "LIQUIDITY_FLOW_SIGNAL_V1",
        "gate": gate,
        "rows": rows,
        "row_count": len(rows),
        "label_summary": label_summary,
        "label_diversity": len(label_set),
    }
    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,
        "row_count": len(rows),
        "label_summary": label_summary,
    }, ensure_ascii=False))
    return 0


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