QuantEngineByItz/tools/build_sell_waterfall_engine_v2.py

"""SELL_WATERFALL_ENGINE_V2 — 매도 폭포수 엔진 V2.

V1 4단계 유지 + 신규 추가:
  (a) 호가단위 기반 슬리피지 시뮬레이션 (est_slippage_bps)
  (b) 유동성 라벨(LIQUIDITY_FLOW) 기반 exec_mode 결정 (MARKET/TWAP/LIMIT)
  (c) 부분체결 시 잔량 자동 stage 승격 (escalation_rule)

Stage 정의:
  1 = EMERGENCY_FULL      — stop_breach or emergency_full_sell
  2 = URGENT_TRIM         — urgent_liquidity_trim / cash preservation ≥ 50%
  3 = DISTRIBUTION_EXIT   — 분산 매도 (여러 날 분할)
  4 = TRIM_ONLY           — 소규모 이익실현 / 리밸런싱

exec_mode:
  MARKET  — AvgTradeValue ≥ 500억: 시장가
  TWAP    — 50억 ≤ AvgTradeValue < 500억: 시간분할(TWAP)
  LIMIT   — AvgTradeValue < 50억: 지정가

escalation_rule: 부분체결 < 80% 시 자동 stage+1 승격 (최대 stage 1)
"""
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" / "sell_waterfall_engine_v2.json"

# 유동성 임계값 (억 단위: AvgTradeValue_20D_M / 1e8)
_LIQUIDITY_HIGH_THRESHOLD = 500.0   # ≥ 500억 → MARKET
_LIQUIDITY_MED_THRESHOLD = 50.0     # 50~500억 → TWAP
# < 50억 → LIMIT

# 슬리피지: 호가단위(tick) × 배수
_SLIPPAGE_TICKS = 1  # 1 tick 슬리피지 가정

# 부분체결 에스컬레이션 임계
_PARTIAL_FILL_THRESHOLD = 0.80  # 80% 미만 체결 시 escalation

# 단계 skip 금지 — 1→2→3→4 순서 유지 (단, 현 단계에서 승격 허용)


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


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 = 0.0) -> float:
    try:
        return float(v) if v is not None else default
    except (TypeError, ValueError):
        return default


def _tick_size(price: float) -> float:
    """한국 주식 호가단위 (KRX 기준)."""
    if price < 1_000:
        return 1.0
    if price < 5_000:
        return 5.0
    if price < 10_000:
        return 10.0
    if price < 50_000:
        return 50.0
    if price < 100_000:
        return 100.0
    if price < 500_000:
        return 500.0
    return 1_000.0


def _slippage_bps(price: float, ticks: int = _SLIPPAGE_TICKS) -> float:
    """예상 슬리피지 (basis points = 0.01%)."""
    if price <= 0:
        return 0.0
    tick = _tick_size(price)
    return round((tick * ticks / price) * 10_000, 1)


def _exec_mode(avg_trade_value_m: float | None) -> str:
    """유동성 기반 체결 방식 결정.
    avg_trade_value_m: AvgTradeValue_20D_M (억 단위가 아닌 백만 단위임 주의)
    """
    if avg_trade_value_m is None:
        return "LIMIT"
    # AvgTradeValue_20D_M 은 백만원(M) 단위
    # 500억 = 50,000M, 50억 = 5,000M
    if avg_trade_value_m >= 50_000:   # ≥ 500억
        return "MARKET"
    if avg_trade_value_m >= 5_000:    # 50~500억
        return "TWAP"
    return "LIMIT"


def _stage_from_style(execution_style: str, emergency_full_sell: bool) -> int:
    """execution_style → stage 번호."""
    if emergency_full_sell:
        return 1
    s = (execution_style or "").upper()
    if "EMERGENCY" in s:
        return 1
    if "URGENT" in s:
        return 2
    if "DISTRIBUTION" in s or "OVERSOLD" in s:
        return 3
    if "TRIM" in s or "PROTECT" in s or "PROFIT" in s:
        return 4
    return 3  # default


def _split_count(exec_mode: str, stage: int, qty: int) -> int:
    """분할 횟수: TWAP/LIMIT는 분할, MARKET은 1회."""
    if exec_mode == "MARKET":
        return 1
    if exec_mode == "TWAP":
        # stage별 분할 수: 1→1, 2→2, 3→3, 4→2
        return {1: 1, 2: 2, 3: 3, 4: 2}.get(stage, 2)
    # LIMIT: 더 세분화
    return {1: 1, 2: 3, 3: 5, 4: 3}.get(stage, 3)


def _escalation_rule(stage: int, exec_mode: str) -> str:
    """부분체결 에스컬레이션 규칙 설명."""
    if stage <= 1:
        return f"partial<{int(_PARTIAL_FILL_THRESHOLD*100)}%→MARKET_OVERRIDE(단계 최상위이므로 mode만 변경)"
    return (
        f"partial<{int(_PARTIAL_FILL_THRESHOLD*100)}%→stage{stage-1}승격_"
        f"({_stage_name(stage)}→{_stage_name(stage-1)})"
        f"_단계건너뜀금지"
    )


def _stage_name(stage: int) -> str:
    return {1: "EMERGENCY_FULL", 2: "URGENT_TRIM", 3: "DISTRIBUTION_EXIT", 4: "TRIM_ONLY"}.get(stage, f"STAGE{stage}")


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

    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

    payload = _load(json_path)
    data_block = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    h = data_block.get("_harness_context") if isinstance(data_block.get("_harness_context"), dict) else {}
    df_list = _rows(data_block.get("data_feed"))

    # 가격 + AvgTradeValue lookup
    price_lut: dict[str, float] = {}
    atv_lut: dict[str, float] = {}
    for r in df_list:
        t = str(r.get("Ticker") or "")
        if t:
            price_lut[t] = _f(r.get("Close") or r.get("close"), 0.0)
            atv_lut[t] = _f(r.get("AvgTradeValue_20D_M"), 0.0)

    # V1 waterfall plan
    wp_raw = h.get("waterfall_plan_json", [])
    if isinstance(wp_raw, str):
        try:
            wp_raw = json.loads(wp_raw)
        except Exception:
            wp_raw = []
    wp = _rows(wp_raw)

    # K2 rebound trigger lookup: ticker → rebound_trigger_price
    k2_raw = h.get("k2_staged_rebound_sell_json", [])
    if isinstance(k2_raw, str):
        try:
            k2_raw = json.loads(k2_raw)
        except Exception:
            k2_raw = []
    k2_lut: dict[str, dict] = {}
    for k2r in _rows(k2_raw):
        t = str(k2r.get("ticker") or "")
        if t:
            k2_lut[t] = k2r

    # deadline_date lookup from sell_candidates
    sc_raw = h.get("sell_candidates_json", [])
    if isinstance(sc_raw, str):
        try:
            sc_raw = json.loads(sc_raw)
        except Exception:
            sc_raw = []
    deadline_lut: dict[str, str] = {}
    for scr in _rows(sc_raw):
        t = str(scr.get("ticker") or "")
        dd = scr.get("deadline_date") or scr.get("target_date") or scr.get("rebound_deadline")
        if t and dd:
            deadline_lut[t] = str(dd)

    if not wp:
        result = {
            "formula_id": "SELL_WATERFALL_ENGINE_V2",
            "gate": "FAIL",
            "note": "waterfall_plan_json is empty or missing",
            "rows": [],
        }
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        print("SELL_WATERFALL_ENGINE_V2 gate=FAIL rows=0")
        return 0

    rows_out = []
    skip_count = 0
    stage_counts: dict[int, int] = {}

    for item in wp:
        ticker = str(item.get("ticker") or "")
        name = str(item.get("name") or "")
        execution_style = str(item.get("execution_style") or "")
        emergency_full_sell = bool(item.get("emergency_full_sell"))
        immediate_qty = int(item.get("immediate_qty") or 0)
        max_daily_qty = int(item.get("max_daily_qty") or immediate_qty or 0)
        rank = int(item.get("rank") or 0)

        price = price_lut.get(ticker, 0.0)
        atv = atv_lut.get(ticker, None) if atv_lut.get(ticker, 0.0) > 0 else None

        stage = _stage_from_style(execution_style, emergency_full_sell)
        mode = _exec_mode(atv)
        slip_bps = _slippage_bps(price)
        n_splits = _split_count(mode, stage, immediate_qty)
        escal = _escalation_rule(stage, mode)

        # 최소 수량 검증 (qty 빈 경우)
        sell_qty = max(0, immediate_qty)
        qty_note = "OK" if sell_qty > 0 else "ZERO_QTY"

        stage_counts[stage] = stage_counts.get(stage, 0) + 1

        # rebound_trigger_price from K2 lookup; "N/A" for non-K2 stages
        k2_row = k2_lut.get(ticker, {})
        rebound_trigger = k2_row.get("rebound_trigger_price")
        if rebound_trigger is None:
            rebound_trigger = "N/A"

        # deadline_date from sell_candidates lookup; "N/A" when not set
        deadline = deadline_lut.get(ticker, "N/A")

        rows_out.append({
            "rank": rank,
            "ticker": ticker,
            "name": name,
            "stage": stage,
            "stage_label": _stage_name(stage),
            "execution_style": execution_style,
            "exec_mode": mode,
            "sell_qty": sell_qty,
            "max_daily_qty": max_daily_qty,
            "split_count": n_splits,
            "est_slippage_bps": slip_bps,
            "escalation_rule": escal,
            "qty_note": qty_note,
            "close_price": price,
            "avg_trade_value_m": atv,
            "rebound_trigger_price": rebound_trigger,
            "deadline_date": deadline,
        })

    # 단계 건너뜀 검증 (stage는 rank 순서대로 증가하거나 같아야 한다는 것은 아니지만
    # 한 ticker 안에서는 단계를 건너뛰지 않도록 — 이미 per-ticker 1행이므로 skip=0)
    has_all_stage_labels = all(r["stage"] in {1, 2, 3, 4} for r in rows_out)
    has_exec_mode = all(r["exec_mode"] in {"MARKET", "TWAP", "LIMIT"} for r in rows_out)
    has_split_count = all(r["split_count"] >= 1 for r in rows_out)

    gate = "PASS" if (rows_out and has_all_stage_labels and has_exec_mode and has_split_count and skip_count == 0) else "CAUTION"

    result = {
        "formula_id": "SELL_WATERFALL_ENGINE_V2",
        "gate": gate,
        "row_count": len(rows_out),
        "stage_counts": stage_counts,
        "escalation_skip_violations": skip_count,
        "partial_fill_threshold_pct": int(_PARTIAL_FILL_THRESHOLD * 100),
        "validation": {
            "all_stage_labels_valid": has_all_stage_labels,
            "all_exec_modes_valid": has_exec_mode,
            "all_split_counts_valid": has_split_count,
        },
        "rows": rows_out,
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(
        f"SELL_WATERFALL_ENGINE_V2 gate={gate} rows={len(rows_out)} "
        f"stages={stage_counts} skip_violations={skip_count}"
    )
    return 0


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