QuantEngineByItz/tools/validate_order_grammar_v1.py

"""validate_order_grammar_v1.py — P7-T03 주문 문법 및 매도 우선순위 waterfall 검증기

1. 매도 주문에 다중 조건 접속사(AND, OR, &, +, , 등) 기반 문장이 없는지 검증 (단일 reason_code만 허용).
2. 매도 후보가 2개 이상인 경우, waterfall 순서가 맞는지 검증:
   STOP > CASH_FLOOR > DISTRIBUTION > VALUE_PRESERVE_TRIM > TAKE_PROFIT > HOLD
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

# Windows 로컬 인코딩 문제 해결을 위해 utf-8 강제
if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
    sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)

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

# 우선순위 정의 (STOP > CASH_FLOOR > DISTRIBUTION > VALUE_PRESERVE_TRIM > TAKE_PROFIT > HOLD)
PRIORITY_ORDER = [
    "STOP",
    "CASH_FLOOR",
    "DISTRIBUTION",
    "VALUE_PRESERVE_TRIM",
    "TAKE_PROFIT",
    "HOLD"
]

def load_harness(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    if isinstance(payload, dict) and isinstance(payload.get("data"), dict):
        maybe = payload["data"].get("_harness_context")
        if isinstance(maybe, dict):
            return maybe
    return payload if isinstance(payload, dict) else {}

def main() -> int:
    hctx = load_harness(DEFAULT_JSON)
    orders = hctx.get("order_blueprint_json")
    if not isinstance(orders, list):
        # order_blueprint_json이 문자열 형태일 수 있으므로 파싱 시도
        if isinstance(orders, str) and orders.strip():
            try:
                orders = json.loads(orders)
            except Exception:
                orders = []
        else:
            orders = []

    multi_condition_count = 0
    sell_priority_missing = 0
    errors: list[str] = []

    # 매도 후보 필터링
    sell_candidates: list[dict[str, Any]] = []
    sell_actions = {"SELL", "TRIM", "EXIT", "REDUCE"}
    
    for idx, order in enumerate(orders):
        if not isinstance(order, dict):
            continue
        order_type = str(order.get("order_type") or "").upper()
        action = str(order.get("action") or "").upper()
        is_sell = order_type in sell_actions or action in sell_actions
        
        if is_sell:
            sell_candidates.append(order)
            # 1. 다중 조건 접속사 검사
            # reason_code 또는 reason 필드를 확인
            reason_code = str(order.get("reason_code") or "")
            
            # 다중 조건 접속사 감지 (AND, OR, &, +, , 등)
            for sep in ["AND", "OR", "&", "+", ","]:
                rc_upper = reason_code.upper()
                if sep in ["&", "+", ","]:
                    if sep in reason_code:
                        multi_condition_count += 1
                        errors.append(f"order[{idx}] ({order.get('ticker')}): reason_code contains multiple conditions separated by '{sep}'")
                        break
                else:  # AND, OR
                    # 단어 경계 체크 (예: " AND ", " OR ")
                    if f" {sep} " in f" {rc_upper} ":
                        multi_condition_count += 1
                        errors.append(f"order[{idx}] ({order.get('ticker')}): reason_code contains multiple conditions separated by '{sep}'")
                        break

    # 2. Sell Priority Waterfall 검증
    if len(sell_candidates) >= 2:
        prev_priority_idx = -1
        for idx, order in enumerate(sell_candidates):
            rc = str(order.get("reason_code") or "").upper()
            
            # 매도 사유에 매핑되는 우선순위 찾기
            matched_priority_idx = -1
            for p_idx, p_name in enumerate(PRIORITY_ORDER):
                if p_name in rc:
                    matched_priority_idx = p_idx
                    break
            
            if matched_priority_idx == -1:
                sell_priority_missing += 1
                errors.append(f"order ({order.get('ticker')}): reason_code '{rc}' does not map to any priority in {PRIORITY_ORDER}")
            else:
                if matched_priority_idx < prev_priority_idx:
                    sell_priority_missing += 1
                    errors.append(
                        f"Waterfall precedence violation: '{PRIORITY_ORDER[matched_priority_idx]}' order "
                        f"appears after '{PRIORITY_ORDER[prev_priority_idx]}'"
                    )
                prev_priority_idx = matched_priority_idx

    status = "PASS" if not errors else "FAIL"
    
    result = {
        "formula_id": "ORDER_GRAMMAR_V1",
        "status": status,
        "errors": errors,
        "multi_condition_order_sentence_count": multi_condition_count,
        "sell_priority_missing_when_candidates_ge_2": sell_priority_missing,
        "sell_candidates_count": len(sell_candidates)
    }

    DEFAULT_OUT.parent.mkdir(parents=True, exist_ok=True)
    DEFAULT_OUT.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if status == "PASS":
        print("ORDER_GRAMMAR_V1_OK")
    else:
        print("ORDER_GRAMMAR_V1_FAIL")
        for e in errors:
            print(f"  ERROR: {e}")
            
    return 0 if status == "PASS" else 1

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