"""주문 실행 큐 intent — Signal(판단)과 Execution(place) 분리용.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Dict # 우선순위: 숫자 작을수록 먼저 (동순위는 seq FIFO) PRIO_URGENT_SELL = 0 PRIO_TICK_SELL = 1 PRIO_SCAN_SELL = 2 PRIO_BUY = 3 def resolve_order_priority(side: str, signal: Dict[str, Any], source: str) -> int: """side/source/reason 으로 큐 우선순위 결정.""" side_u = str(side or "").upper() src = str(source or "").strip().lower() if side_u == "BUY": return PRIO_BUY if src in ("halt", "eod", "pending", "risk", "force"): return PRIO_URGENT_SELL reason = str((signal or {}).get("reason") or "").lower() urgent_keys = ("stop", "손절", "eod", "halt", "risk", "긴급", "force", "pending") if any(k in reason for k in urgent_keys): return PRIO_URGENT_SELL if src == "tick": return PRIO_TICK_SELL return PRIO_SCAN_SELL @dataclass(order=True) class OrderIntent: """AccountOrderWorker 큐 1건 — strategy 참조는 strategy_id 로 Worker 가 조회.""" priority: int seq: int strategy_id: str side: str signal: Dict[str, Any] source: str = "scan"