diff --git a/prod/clean_arch/dita_v2/__init__.py b/prod/clean_arch/dita_v2/__init__.py index 93ae0370..0a10dcbc 100644 --- a/prod/clean_arch/dita_v2/__init__.py +++ b/prod/clean_arch/dita_v2/__init__.py @@ -32,6 +32,7 @@ from .contracts import ( VenueEventStatus, VenueOrder, VenueOrderStatus, + VenueTelemetrySnapshot, ) from .journal import ClickHouseKernelJournal, KernelJournal, MemoryKernelJournal from .rust_backend import ExecutionKernel @@ -89,6 +90,7 @@ __all__ = [ "VenueEventStatus", "VenueOrder", "VenueOrderStatus", + "VenueTelemetrySnapshot", "ZincPlane", "ZincControlPlane", "build_position_state_row", diff --git a/prod/clean_arch/dita_v2/bingx_venue.py b/prod/clean_arch/dita_v2/bingx_venue.py index b0bc5708..2e40855a 100644 --- a/prod/clean_arch/dita_v2/bingx_venue.py +++ b/prod/clean_arch/dita_v2/bingx_venue.py @@ -27,6 +27,7 @@ from .contracts import ( KernelEventKind, KernelIntent, TradeSide, + VenueTelemetrySnapshot, VenueEvent, VenueEventStatus, VenueOrder, @@ -226,7 +227,7 @@ class BingxVenueAdapter(VenueAdapter): ) return cls._EXECUTOR - def __init__(self, backend: Any | None = None, *, config: Any | None = None) -> None: + def __init__(self, backend: Any | None = None, *, config: Any | None = None, zinc_plane: Any | None = None) -> None: if backend is None: if config is None: raise ValueError("BingxVenueAdapter requires a backend or config") @@ -234,6 +235,7 @@ class BingxVenueAdapter(VenueAdapter): backend = BingxDirectExecutionAdapter(config) self.backend = backend + self._telemetry_plane = zinc_plane self._event_seq = itertools.count(1) # Thread-safe snapshot cache — reads from a snapshot may arrive from # the kernel thread while _backend_snapshot writes from the pool thread. @@ -279,6 +281,73 @@ class BingxVenueAdapter(VenueAdapter): except Exception: pass + def set_telemetry_plane(self, zinc_plane: Any | None) -> None: + self._telemetry_plane = zinc_plane + + def _publish_telemetry( + self, + *, + phase: str, + status: str, + intent: KernelIntent | None = None, + order: VenueOrder | None = None, + endpoint: str = "", + method: str = "", + message: str = "", + retry_after_ms: int = 0, + venue_order_status: str = "", + venue_event_kind: str = "", + details: dict[str, Any] | None = None, + ) -> None: + plane = self._telemetry_plane + publish = getattr(plane, "publish_venue", None) if plane is not None else None + if publish is None: + return + slot_id = 0 + trade_id = "" + asset = "" + side = TradeSide.FLAT + action = "" + intent_id = "" + if intent is not None: + slot_id = int(getattr(intent, "slot_id", 0) or 0) + trade_id = str(getattr(intent, "trade_id", "") or "") + asset = str(getattr(intent, "asset", "") or "") + side = getattr(intent, "side", TradeSide.FLAT) + action = str(getattr(intent, "action", "") or "") + intent_id = str(getattr(intent, "intent_id", "") or "") + if order is not None: + slot_id = int(order.metadata.get("slot_id", slot_id) or slot_id) + trade_id = str(order.internal_trade_id or trade_id) + asset = str(order.metadata.get("asset") or asset) + side = order.side or side + try: + publish( + VenueTelemetrySnapshot( + phase=phase, + status=status, + venue="bingx", + endpoint=endpoint, + method=method, + intent_id=intent_id, + trade_id=trade_id, + slot_id=slot_id, + asset=asset, + side=side, + action=action, + order_id=str(getattr(order, "venue_order_id", "") or ""), + client_order_id=str(getattr(order, "venue_client_id", "") or ""), + venue_order_status=venue_order_status, + venue_event_kind=venue_event_kind, + message=message, + retry_after_ms=int(retry_after_ms or 0), + timestamp=datetime.now(timezone.utc), + details=dict(details or {}), + ) + ) + except Exception: + pass + def _call_backend(self, method_name: str, *args: Any, **kwargs: Any) -> Any: method = getattr(self.backend, method_name, None) if method is None: @@ -368,18 +437,37 @@ class BingxVenueAdapter(VenueAdapter): was fixed for submit via submit_async. This version awaits backend.cancel() directly in the caller's (main) event loop. """ + self._publish_telemetry( + phase="cancel:start", + status="REQUESTED", + order=order, + endpoint="/openApi/swap/v2/trade/order", + method="DELETE", + message=reason, + details={"asset": str(order.metadata.get("asset") or "")}, + ) cancel_fn = getattr(self.backend, "cancel", None) if cancel_fn is not None: response = await cancel_fn(order, reason=reason) else: response = None - return self._events_from_cancel(order, response, None, None, reason=reason) + events = self._events_from_cancel(order, response, None, None, reason=reason) + return events def cancel(self, order: VenueOrder, *, reason: str = "") -> List[VenueEvent]: # _events_from_cancel never reads before/after — snapshots are dead weight. # NOTE: if backend.cancel is async (BingxDirectExecutionAdapter), this sync # path goes through the thread-pool and will deadlock in a running event loop. # Use cancel_async() from async contexts (process_intent_async already does). + self._publish_telemetry( + phase="cancel:start", + status="REQUESTED", + order=order, + endpoint="/openApi/swap/v2/trade/order", + method="DELETE", + message=reason, + details={"asset": str(order.metadata.get("asset") or "")}, + ) response = None if hasattr(self.backend, "cancel"): response = self._call_backend("cancel", order, reason=reason) @@ -410,7 +498,8 @@ class BingxVenueAdapter(VenueAdapter): except BingxHttpError as exc: # W10: map HTTP error class to status — 429/5xx are transient, 4xx are real rejections response = {"status": _http_error_status(str(exc)), "msg": str(exc), "orderId": order.venue_order_id, "clientOrderId": order.venue_client_id} - return self._events_from_cancel(order, response, None, None, reason=reason) + events = self._events_from_cancel(order, response, None, None, reason=reason) + return events def open_orders(self) -> List[VenueOrder]: # Use backend._state (populated by await backend.connect()) rather than @@ -458,6 +547,13 @@ class BingxVenueAdapter(VenueAdapter): # entirely: the FSM stayed fill-blind (slot size 0 in ENTRY_WORKING), # the DecisionEngine saw "no position", and re-entered → the live # double-entries at 15:20 and 17:24 UTC. + self._publish_telemetry( + phase="reconcile:start", + status="REQUESTED", + endpoint="/openApi/swap/v2/trade/openOrders", + method="GET", + details={"include_history": True}, + ) recon_symbol = None kernel = getattr(self, "_kernel_ref", None) if kernel is not None: @@ -474,13 +570,60 @@ class BingxVenueAdapter(VenueAdapter): except Exception as exc: import logging as _log _log.getLogger(__name__).warning("reconcile: refresh_state failed: %s", exc) + self._publish_telemetry( + phase="reconcile:error", + status="ERROR", + endpoint="/openApi/swap/v2/trade/openOrders", + method="GET", + message=str(exc), + details={"symbol": recon_symbol or ""}, + ) return [] + self._publish_telemetry( + phase="reconcile:done", + status="OK", + endpoint="/openApi/swap/v2/trade/openOrders", + method="GET", + details={ + "symbol": recon_symbol or "", + "open_orders": len(getattr(snapshot, "open_orders", []) or []), + "positions": len(getattr(snapshot, "open_positions", {}) or {}), + "fills": len(getattr(snapshot, "all_fills", []) or []), + }, + ) return self._events_from_snapshot(snapshot) def submit(self, intent: KernelIntent) -> List[VenueEvent]: # Snapshots dropped: receipt executedQty fields take precedence (same as submit_async) + self._publish_telemetry( + phase="submit:start", + status="REQUESTED", + intent=intent, + endpoint="/openApi/swap/v2/trade/order", + method="POST", + details={"action": intent.action.value, "order_type": str(getattr(intent, "order_type", "MARKET") or "MARKET")}, + ) receipt = self._call_backend("submit_intent", self._legacy_intent(intent)) - return self._events_from_submit(intent, receipt, None, None) + events = self._events_from_submit(intent, receipt, None, None) + ack_row = dict(getattr(receipt, "raw_ack", {}) or {}) + self._publish_telemetry( + phase="submit:done", + status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="NEW")), + intent=intent, + endpoint="/openApi/swap/v2/trade/order", + method="POST", + message=_row_text(ack_row, "msg", "message", default=""), + order_id=_row_text(ack_row, "orderId", "orderID", default=str(getattr(receipt, "order_id", "") or "")), + client_order_id=_row_text(ack_row, "clientOrderID", "clientOrderId", default=str(getattr(receipt, "client_order_id", "") or intent.intent_id)), + venue_order_status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="")), + venue_event_kind=events[1].kind.value if len(events) > 1 else events[0].kind.value, + details={ + "filled_size": float((events[1].filled_size if len(events) > 1 else events[0].filled_size) or 0.0), + "event_count": len(events), + "asset": intent.asset, + }, + ) + return events async def submit_async(self, intent: KernelIntent) -> List[VenueEvent]: """Async submit — runs in the caller's event loop, no thread-pool deadlock. @@ -495,8 +638,35 @@ class BingxVenueAdapter(VenueAdapter): Passing None for snapshots makes _filled_size_from_snapshots return 0.0 (a safe fallback; the receipt fields take precedence). """ + self._publish_telemetry( + phase="submit:start", + status="REQUESTED", + intent=intent, + endpoint="/openApi/swap/v2/trade/order", + method="POST", + details={"action": intent.action.value, "order_type": str(getattr(intent, "order_type", "MARKET") or "MARKET")}, + ) receipt = await self.backend.submit_intent(self._legacy_intent(intent)) - return self._events_from_submit(intent, receipt, None, None) + events = self._events_from_submit(intent, receipt, None, None) + ack_row = dict(getattr(receipt, "raw_ack", {}) or {}) + self._publish_telemetry( + phase="submit:done", + status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="NEW")), + intent=intent, + endpoint="/openApi/swap/v2/trade/order", + method="POST", + message=_row_text(ack_row, "msg", "message", default=""), + order_id=_row_text(ack_row, "orderId", "orderID", default=str(getattr(receipt, "order_id", "") or "")), + client_order_id=_row_text(ack_row, "clientOrderID", "clientOrderId", default=str(getattr(receipt, "client_order_id", "") or intent.intent_id)), + venue_order_status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="")), + venue_event_kind=events[1].kind.value if len(events) > 1 else events[0].kind.value, + details={ + "filled_size": float((events[1].filled_size if len(events) > 1 else events[0].filled_size) or 0.0), + "event_count": len(events), + "asset": intent.asset, + }, + ) + return events def _events_from_submit(self, intent: KernelIntent, receipt: Any, before, after) -> List[VenueEvent]: # noqa: ANN001 ack_row = dict(getattr(receipt, "raw_ack", {}) or {}) @@ -612,6 +782,18 @@ class BingxVenueAdapter(VenueAdapter): raw = response if isinstance(response, dict) else {} status = _normalize_status(_row_text(raw, "status", default="CANCELED")) if status in {"RATE_LIMITED", "THROTTLED"}: + self._publish_telemetry( + phase="cancel:done", + status=status, + order=order, + endpoint="/openApi/swap/v2/trade/order", + method="DELETE", + message=reason or _row_text(raw, "msg", "message", default="BINGX_RATE_LIMITED"), + venue_order_status=VenueEventStatus.RATE_LIMITED.value, + venue_event_kind=KernelEventKind.RATE_LIMITED.value, + retry_after_ms=_rate_limit_retry_after_ms(raw), + details={"order_status": status, "asset": str(order.metadata.get("asset") or "")}, + ) return [ VenueEvent( timestamp=datetime.now(timezone.utc), @@ -637,6 +819,18 @@ class BingxVenueAdapter(VenueAdapter): kind = KernelEventKind.CANCEL_ACK if event_status == VenueEventStatus.CANCELED else KernelEventKind.CANCEL_REJECT if event_status == VenueEventStatus.CANCELED_REJECTED: kind = KernelEventKind.CANCEL_REJECT + self._publish_telemetry( + phase="cancel:done", + status=status or event_status.value, + order=order, + endpoint="/openApi/swap/v2/trade/order", + method="DELETE", + message=reason or _row_text(raw, "msg", "message", default=""), + venue_order_status=event_status.value, + venue_event_kind=kind.value, + retry_after_ms=_rate_limit_retry_after_ms(raw), + details={"order_status": status, "asset": str(order.metadata.get("asset") or "")}, + ) return [ VenueEvent( timestamp=datetime.now(timezone.utc), diff --git a/prod/clean_arch/dita_v2/contracts.py b/prod/clean_arch/dita_v2/contracts.py index f7a687cb..a1648d5e 100644 --- a/prod/clean_arch/dita_v2/contracts.py +++ b/prod/clean_arch/dita_v2/contracts.py @@ -144,6 +144,54 @@ class VenueOrder: return max(0.0, float(self.intended_size) - float(self.filled_size)) +@dataclass(frozen=True) +class VenueTelemetrySnapshot: + """Shared-memory surface for venue-side call boundaries and state.""" + + phase: str = "idle" + status: str = "IDLE" + venue: str = "bingx" + endpoint: str = "" + method: str = "" + intent_id: str = "" + trade_id: str = "" + slot_id: int = 0 + asset: str = "" + side: TradeSide = TradeSide.FLAT + action: str = "" + order_id: str = "" + client_order_id: str = "" + venue_order_status: str = "" + venue_event_kind: str = "" + message: str = "" + retry_after_ms: int = 0 + timestamp: Optional[datetime] = None + details: Dict[str, Any] = field(default_factory=dict) + + def as_dict(self) -> Dict[str, Any]: + return { + "phase": self.phase, + "status": self.status, + "venue": self.venue, + "endpoint": self.endpoint, + "method": self.method, + "intent_id": self.intent_id, + "trade_id": self.trade_id, + "slot_id": int(self.slot_id or 0), + "asset": self.asset, + "side": self.side.value if hasattr(self.side, "value") else str(self.side), + "action": self.action, + "order_id": self.order_id, + "client_order_id": self.client_order_id, + "venue_order_status": self.venue_order_status, + "venue_event_kind": self.venue_event_kind, + "message": self.message, + "retry_after_ms": int(self.retry_after_ms or 0), + "timestamp": self.timestamp.isoformat() if hasattr(self.timestamp, "isoformat") else None, + "details": dict(self.details), + } + + @dataclass class TradeSlot: """A single execution slot managed by the v2 kernel.""" diff --git a/prod/clean_arch/dita_v2/launcher.py b/prod/clean_arch/dita_v2/launcher.py index 2fb185e1..78d773c2 100644 --- a/prod/clean_arch/dita_v2/launcher.py +++ b/prod/clean_arch/dita_v2/launcher.py @@ -245,9 +245,15 @@ def _build_venue( mock_scenario: Optional[MockVenueScenario] = None, bingx_config: Optional[BingxExecClientConfig] = None, bingx_backend: Optional[Any] = None, + zinc_plane: Optional[ZincPlane] = None, venue: Optional[VenueAdapter] = None, ) -> VenueAdapter: if venue is not None: + if zinc_plane is not None and hasattr(venue, "set_telemetry_plane"): + try: + venue.set_telemetry_plane(zinc_plane) + except Exception: + pass return venue resolved_mode = venue_mode or _resolve_venue_mode() if resolved_mode is LauncherVenueMode.BINGX: @@ -256,7 +262,7 @@ def _build_venue( from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter backend = BingxDirectExecutionAdapter(bingx_config or build_bingx_exec_client_config()) - return BingxVenueAdapter(backend=backend) + return BingxVenueAdapter(backend=backend, zinc_plane=zinc_plane) return MockVenueAdapter(mock_scenario) @@ -342,6 +348,7 @@ def build_launcher_bundle( mock_scenario=mock_scenario, bingx_config=bingx_config, bingx_backend=bingx_backend, + zinc_plane=active_zinc_plane, venue=venue, ) kernel = ExecutionKernel( diff --git a/prod/clean_arch/dita_v2/real_zinc_plane.py b/prod/clean_arch/dita_v2/real_zinc_plane.py index 3a13375a..75bd5dba 100644 --- a/prod/clean_arch/dita_v2/real_zinc_plane.py +++ b/prod/clean_arch/dita_v2/real_zinc_plane.py @@ -16,7 +16,7 @@ import struct import sys import threading -from .contracts import KernelIntent, TradeSide, TradeSlot, TradeStage, VenueOrder, VenueOrderStatus +from .contracts import KernelIntent, TradeSide, TradeSlot, TradeStage, VenueOrder, VenueOrderStatus, VenueTelemetrySnapshot from .control import KernelControlSnapshot _ZINC_ADAPTER_PATH = Path(__file__).resolve().parents[3] / "zinc" / "adapters" / "python" @@ -130,6 +130,42 @@ def _decode_packet(buf: memoryview) -> Dict[str, Any]: return out +def _venue_from_payload(payload: Dict[str, Any]) -> VenueTelemetrySnapshot: + timestamp = payload.get("timestamp") + ts_value = None + if isinstance(timestamp, str) and timestamp: + try: + ts_value = datetime.fromisoformat(timestamp) + except Exception: + ts_value = None + try: + side = TradeSide(str(payload.get("side", TradeSide.FLAT.value))) + except Exception: + side = TradeSide.FLAT + details = payload.get("details", {}) + return VenueTelemetrySnapshot( + phase=str(payload.get("phase", "idle")), + status=str(payload.get("status", "IDLE")), + venue=str(payload.get("venue", "bingx")), + endpoint=str(payload.get("endpoint", "")), + method=str(payload.get("method", "")), + intent_id=str(payload.get("intent_id", "")), + trade_id=str(payload.get("trade_id", "")), + slot_id=int(payload.get("slot_id", 0) or 0), + asset=str(payload.get("asset", "")), + side=side, + action=str(payload.get("action", "")), + order_id=str(payload.get("order_id", "")), + client_order_id=str(payload.get("client_order_id", "")), + venue_order_status=str(payload.get("venue_order_status", "")), + venue_event_kind=str(payload.get("venue_event_kind", "")), + message=str(payload.get("message", "")), + retry_after_ms=int(payload.get("retry_after_ms", 0) or 0), + timestamp=ts_value, + details=dict(details) if isinstance(details, dict) else {}, + ) + + class RealZincPlane: """Shared-memory Zinc plane used by the Python prototype.""" @@ -148,18 +184,22 @@ class RealZincPlane: self.intent_name = f"{base}_intent" self.state_name = f"{base}_state" self.control_name = f"{base}_control" + self.venue_name = f"{base}_venue" self._intent_seq = 0 self._state_seq = 0 self._control_seq = 0 + self._venue_seq = 0 self._lock = threading.Lock() self._slot_cache: Dict[int, TradeSlot] = {i: TradeSlot(slot_id=i) for i in range(int(slot_count))} self._slot_count = int(slot_count) self._intent_cache: List[Dict[str, Any]] = [] self._control_cache = KernelControlSnapshot() + self._venue_cache = VenueTelemetrySnapshot() if create: self.intent_region = SharedRegion.create(self.intent_name, intent_capacity) self.state_region = SharedRegion.create(self.state_name, state_capacity) self.control_region = SharedRegion.create(self.control_name, control_capacity) + self.venue_region = SharedRegion.create(self.venue_name, control_capacity) self._write_region(self.control_region, self._control_seq, {"control": self._control_cache.as_dict()}) self._write_region( self.state_region, @@ -167,13 +207,16 @@ class RealZincPlane: {"slots": [self._slot_cache[key].to_dict() for key in range(self._slot_count)]}, ) self._write_region(self.intent_region, self._intent_seq, {"items": []}) + self._write_region(self.venue_region, self._venue_seq, {"venue": self._venue_cache.as_dict()}) else: self.intent_region = SharedRegion.open(self.intent_name) self.state_region = SharedRegion.open(self.state_name) self.control_region = SharedRegion.open(self.control_name) + self.venue_region = SharedRegion.open(self.venue_name) control_payload = _decode_packet(self.control_region.as_buffer()) state_payload = _decode_packet(self.state_region.as_buffer()) intent_payload = _decode_packet(self.intent_region.as_buffer()) + venue_payload = _decode_packet(self.venue_region.as_buffer()) if isinstance(control_payload.get("control"), dict): self._control_cache = KernelControlSnapshot(**control_payload["control"]) if isinstance(state_payload.get("slots"), list): @@ -183,11 +226,14 @@ class RealZincPlane: self._slot_cache[int(slot.slot_id)] = slot if isinstance(intent_payload.get("items"), list): self._intent_cache = list(intent_payload["items"]) + if isinstance(venue_payload.get("venue"), dict): + self._venue_cache = _venue_from_payload(venue_payload["venue"]) def close(self) -> None: self.intent_region.close() self.state_region.close() self.control_region.close() + self.venue_region.close() def publish_intent(self, intent: KernelIntent) -> None: with self._lock: @@ -246,6 +292,26 @@ class RealZincPlane: def notify_control(self) -> None: self.control_region.notify() + def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None: + with self._lock: + self._venue_seq += 1 + self._venue_cache = telemetry + self._write_region(self.venue_region, self._venue_seq, {"venue": telemetry.as_dict()}) + + def read_venue(self) -> VenueTelemetrySnapshot: + payload = _decode_packet(self.venue_region.as_buffer()) + venue = payload.get("venue") if isinstance(payload, dict) else None + if not isinstance(venue, dict): + return self._venue_cache + self._venue_cache = _venue_from_payload(venue) + return self._venue_cache + + def wait_on_venue(self, timeout_ms: int = 1000) -> bool: + return bool(self.venue_region.wait(timeout_ms)) + + def notify_venue(self) -> None: + self.venue_region.notify() + def wait_on_intent(self, timeout_ms: int = 1000) -> bool: return bool(self.intent_region.wait(timeout_ms)) diff --git a/prod/clean_arch/dita_v2/rust_backend.py b/prod/clean_arch/dita_v2/rust_backend.py index aaf41e68..c13f0900 100644 --- a/prod/clean_arch/dita_v2/rust_backend.py +++ b/prod/clean_arch/dita_v2/rust_backend.py @@ -107,6 +107,9 @@ def _crate_dir() -> Path: return Path(__file__).resolve().with_name("_rust_kernel") +_LOCAL_TARGET_DIR = Path("/root/.cargo/dita_v2_target") + + def _library_path() -> Path: if sys.platform == "darwin": name = "libdita_v2_kernel.dylib" @@ -114,6 +117,9 @@ def _library_path() -> Path: name = "dita_v2_kernel.dll" else: name = "libdita_v2_kernel.so" + local = _LOCAL_TARGET_DIR / "release" / name + if local.exists(): + return local return _crate_dir() / "target" / "release" / name @@ -121,10 +127,13 @@ def _build_library() -> None: crate_dir = _crate_dir() if not crate_dir.exists(): raise FileNotFoundError(f"Missing Rust kernel crate: {crate_dir}") + _LOCAL_TARGET_DIR.mkdir(parents=True, exist_ok=True) + env = {**os.environ, "CARGO_TARGET_DIR": str(_LOCAL_TARGET_DIR)} subprocess.run( ["cargo", "build", "--release", "--manifest-path", str(crate_dir / "Cargo.toml")], cwd=_repo_root(), check=True, + env=env, ) diff --git a/prod/clean_arch/dita_v2/zinc_plane.py b/prod/clean_arch/dita_v2/zinc_plane.py index 0d6f11a1..aea04e6f 100644 --- a/prod/clean_arch/dita_v2/zinc_plane.py +++ b/prod/clean_arch/dita_v2/zinc_plane.py @@ -12,7 +12,7 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Protocol import threading import time -from .contracts import KernelIntent, TradeSlot +from .contracts import KernelIntent, TradeSlot, VenueTelemetrySnapshot from .control import KernelControlSnapshot @@ -52,6 +52,18 @@ class ZincPlane(Protocol): def notify_control(self) -> None: ... + def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None: + ... + + def read_venue(self) -> VenueTelemetrySnapshot: + ... + + def wait_on_venue(self, timeout_ms: int = 1000) -> bool: + ... + + def notify_venue(self) -> None: + ... + @dataclass class InMemoryZincPlane: @@ -60,12 +72,15 @@ class InMemoryZincPlane: intent_region: List[KernelIntent] = field(default_factory=list) state_region: Dict[int, TradeSlot] = field(default_factory=dict) control_region: Optional[KernelControlSnapshot] = None + venue_region: VenueTelemetrySnapshot = field(default_factory=VenueTelemetrySnapshot) _intent_seq: int = field(default=0, init=False, repr=False) _state_seq: int = field(default=0, init=False, repr=False) _control_seq: int = field(default=0, init=False, repr=False) + _venue_seq: int = field(default=0, init=False, repr=False) _intent_observed_seq: int = field(default=0, init=False, repr=False) _state_observed_seq: int = field(default=0, init=False, repr=False) _control_observed_seq: int = field(default=0, init=False, repr=False) + _venue_observed_seq: int = field(default=0, init=False, repr=False) _signal: threading.Condition = field(default_factory=threading.Condition, init=False, repr=False) def publish_intent(self, intent: KernelIntent) -> None: @@ -118,6 +133,23 @@ class InMemoryZincPlane: self._control_seq += 1 self._signal.notify_all() + def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None: + with self._signal: + self.venue_region = telemetry + self._venue_seq += 1 + self._signal.notify_all() + + def read_venue(self) -> VenueTelemetrySnapshot: + return self.venue_region + + def wait_on_venue(self, timeout_ms: int = 1000) -> bool: + return self._wait_for_change("_venue_seq", "_venue_observed_seq", timeout_ms) + + def notify_venue(self) -> None: + with self._signal: + self._venue_seq += 1 + self._signal.notify_all() + def _wait_for_change(self, seq_attr: str, observed_attr: str, timeout_ms: int) -> bool: timeout_s = None if timeout_ms is None or timeout_ms < 0 else max(0.0, timeout_ms / 1000.0) deadline = None if timeout_s is None else time.monotonic() + timeout_s diff --git a/prod/clean_arch/violet/v4_execution_runner.py b/prod/clean_arch/violet/v4_execution_runner.py new file mode 100644 index 00000000..e4873252 --- /dev/null +++ b/prod/clean_arch/violet/v4_execution_runner.py @@ -0,0 +1,725 @@ +"""VIOLET V4 execution runner. + +This is the live-capable VIOLET runner boundary: + + NG7 scan -> BLUE-faithful VIOLET decision -> ExecIntent + -> DITAv2 KernelIntent -> DITAv2 BingX VST venue execution + +Safety is explicit rather than castrated: the module can submit real orders via +``ExecutionKernel.process_intent_async`` when the operator runs the V4 launcher +with VST keys and arming gates green. Unit tests inject fake kernels and never +touch a venue. +""" + +from __future__ import annotations + +import argparse +import asyncio +import gc +import json +import logging +import math +import os +import sys +import threading +import time +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Mapping, Optional + +from prod.clean_arch.dita_v2.contracts import ( + KernelCommandType, + KernelIntent, + KernelOutcome, + TradeSide, +) + +from .contracts_v3 import ExecIntent +from .decision_engine import ShadowDecision +from .exec_intent import to_exec_intent + +RUNNER_CONTRACT_VERSION = "violet-v4-exec-runner-contracts-20260626" +LOGGER = logging.getLogger("violet.v4_execution_runner") + +_ASEX_SRC = Path("/mnt/dolphinng5_predict/ASEx/src") +if _ASEX_SRC.exists() and str(_ASEX_SRC) not in sys.path: + sys.path.insert(0, str(_ASEX_SRC)) + +try: # pragma: no cover - exercised when ASEx package is installed/available. + from asex.guarded import ASExGuardedState + from asex.worker import ASExWorker +except Exception: # pragma: no cover - fallback is covered by unit tests. + ASExGuardedState = None + ASExWorker = None + + +def _finite_positive(value: float, name: str) -> float: + out = float(value) + if not math.isfinite(out) or out <= 0.0: + raise ValueError(f"{name} must be finite and > 0") + return out + + +def _non_empty(value: str, name: str) -> str: + out = str(value or "").strip() + if not out: + raise ValueError(f"{name} must be non-empty") + return out + + +def _trade_side(side: str) -> TradeSide: + normalized = str(side or "").strip().upper() + if normalized == "LONG": + return TradeSide.LONG + if normalized == "SHORT": + return TradeSide.SHORT + raise ValueError(f"unsupported VIOLET side: {side!r}") + + +def _action_from_intent(intent: ExecIntent, explicit: Optional[KernelCommandType]) -> KernelCommandType: + if explicit is not None: + return explicit + if intent.reason == "ENTRY": + return KernelCommandType.ENTER + if intent.reason == "EXIT": + return KernelCommandType.EXIT + raise ValueError(f"unsupported VIOLET intent reason: {intent.reason!r}") + + +def exec_intent_to_kernel_intent( + intent: ExecIntent, + *, + reference_price: float, + trade_id: str, + slot_id: int = 0, + intent_id: Optional[str] = None, + action: Optional[KernelCommandType] = None, + order_type: str = "MARKET", + limit_price: float = 0.0, + metadata: Optional[Mapping[str, Any]] = None, + timestamp: Optional[datetime] = None, +) -> KernelIntent: + """Convert a VIOLET ``ExecIntent`` into a DITAv2 ``KernelIntent``. + + The function is intentionally pure and strict. It does not fetch prices, + allocate state outside the returned object, or contact a venue. Live runner + code must pass a fresh reference price with provenance in ``metadata``. + """ + + price = _finite_positive(reference_price, "reference_price") + qty = _finite_positive(float(intent.qty), "intent.qty") + leverage = _finite_positive(float(intent.exchange_leverage), "intent.exchange_leverage") + tid = _non_empty(trade_id, "trade_id") + oid = _non_empty(intent_id or f"violet-v4-{uuid.uuid4().hex}", "intent_id") + order = _non_empty(order_type, "order_type").upper() + if order == "LIMIT": + _finite_positive(limit_price, "limit_price") + elif float(limit_price or 0.0) < 0.0: + raise ValueError("limit_price must be >= 0") + + merged_metadata: dict[str, Any] = dict(metadata or {}) + merged_metadata.update( + { + "violet_contract_version": RUNNER_CONTRACT_VERSION, + "violet_ts_ns": int(intent.ts_ns), + "violet_maker_policy": intent.maker_policy, + "violet_target_notional": float(intent.target_notional), + "violet_reason": intent.reason, + } + ) + + return KernelIntent( + timestamp=timestamp or datetime.now(timezone.utc), + intent_id=oid, + trade_id=tid, + slot_id=int(slot_id), + asset=_non_empty(intent.asset, "intent.asset"), + side=_trade_side(intent.side), + action=_action_from_intent(intent, action), + reference_price=price, + target_size=qty, + leverage=leverage, + reason=f"violet_v4:{intent.reason.lower()}", + metadata=merged_metadata, + order_type=order, + limit_price=float(limit_price or 0.0), + ) + + +@dataclass(frozen=True) +class VioletV4RunnerConfig: + """Operator/runtime knobs for the V4 execution runner.""" + + session_id: str + capital: float = 69_000.0 + slot_id: int = 0 + maker_policy: str = "maker_both" + order_type: str = "MARKET" + limit_price: float = 0.0 + poll_interval_s: float = 0.25 + queue_maxsize: int = 4096 + max_submissions: int = 0 + max_notional_usdt: float = 0.0 + assume_vol_ok: bool = False + require_arming: bool = True + report_dir: str = "/mnt/vp-VIOLET_main/prod/VIOLET_dev/reports" + hz_cluster: str = "dolphin" + hz_host: str = "localhost:5701" + hz_map: str = "DOLPHIN_FEATURES" + hz_key: str = "latest_eigen_scan" + snapshot_symbol: str = "BTCUSDT" + + @classmethod + def from_env(cls) -> "VioletV4RunnerConfig": + return cls( + session_id=os.environ.get("DOLPHIN_VIOLET_SESSION_ID", uuid.uuid4().hex), + capital=float(os.environ.get("DOLPHIN_VIOLET_CAPITAL", "69000")), + slot_id=int(os.environ.get("DOLPHIN_VIOLET_SLOT_ID", "0")), + maker_policy=os.environ.get("DOLPHIN_VIOLET_MAKER_POLICY", "maker_both"), + order_type=os.environ.get("DOLPHIN_VIOLET_ORDER_TYPE", "MARKET"), + limit_price=float(os.environ.get("DOLPHIN_VIOLET_LIMIT_PRICE", "0")), + poll_interval_s=float(os.environ.get("DOLPHIN_VIOLET_POLL_INTERVAL_SEC", "0.25")), + queue_maxsize=int(os.environ.get("DOLPHIN_VIOLET_SCAN_QUEUE_MAX", "4096")), + max_submissions=int(os.environ.get("DOLPHIN_VIOLET_MAX_SUBMISSIONS", "0")), + max_notional_usdt=float(os.environ.get("DOLPHIN_VIOLET_MAX_NOTIONAL_USDT", "0")), + assume_vol_ok=str(os.environ.get("DOLPHIN_VIOLET_ASSUME_VOL_OK", "0")).lower() + in {"1", "true", "yes", "on"}, + require_arming=str(os.environ.get("DOLPHIN_VIOLET_REQUIRE_ARMING", "1")).lower() + not in {"0", "false", "no", "off"}, + report_dir=os.environ.get( + "DOLPHIN_VIOLET_REPORT_DIR", + "/mnt/vp-VIOLET_main/prod/VIOLET_dev/reports", + ), + hz_cluster=os.environ.get("HZ_CLUSTER", "dolphin"), + hz_host=os.environ.get("HZ_HOST", "localhost:5701"), + hz_map=os.environ.get("DOLPHIN_VIOLET_SCAN_MAP", "DOLPHIN_FEATURES"), + hz_key=os.environ.get("DOLPHIN_VIOLET_SCAN_KEY", "latest_eigen_scan"), + snapshot_symbol=os.environ.get("DOLPHIN_VIOLET_SNAPSHOT_SYMBOL", "BTCUSDT"), + ) + + +class _LocalGuardedState: + """Tiny ASEx-compatible fallback for tests or missing ASEx install.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self.last_scan_number = -1 + self.scans_seen = 0 + self.decisions = 0 + self.submissions = 0 + self.outcomes = 0 + self.errors = 0 + self.last_trade_id = "" + self.last_intent_id = "" + + def mutate(self, mutation: Mapping[str, Any]) -> dict[str, Any]: + with self._lock: + return self._apply(dict(mutation)) + + def _apply(self, mutation: dict[str, Any]) -> dict[str, Any]: + op = str(mutation.get("op") or "") + if op == "scan": + sn = int(mutation.get("scan_number") or 0) + if sn <= self.last_scan_number: + return {"accepted": False, "reason": "duplicate_scan", "last_scan_number": self.last_scan_number} + self.last_scan_number = sn + self.scans_seen += 1 + return {"accepted": True, "last_scan_number": self.last_scan_number} + if op == "decision": + self.decisions += 1 + return {"accepted": True, "decisions": self.decisions} + if op == "submission": + self.submissions += 1 + self.last_trade_id = str(mutation.get("trade_id") or "") + self.last_intent_id = str(mutation.get("intent_id") or "") + return {"accepted": True, "submissions": self.submissions} + if op == "outcome": + self.outcomes += 1 + return {"accepted": True, "outcomes": self.outcomes} + if op == "error": + self.errors += 1 + return {"accepted": True, "errors": self.errors} + return {"accepted": False, "reason": f"unknown_op:{op}"} + + +if ASExGuardedState is not None: + + class VioletRunnerGuardedState(ASExGuardedState[dict[str, Any], dict[str, Any]]): + """ASEx single-writer state for scan/order lifecycle counters.""" + + def __init__(self) -> None: + super().__init__() + self._state = _LocalGuardedState() + + def _validate(self, mutation: dict[str, Any]) -> bool: + return str(mutation.get("op") or "") in { + "scan", + "decision", + "submission", + "outcome", + "error", + } + + def _apply(self, mutation: dict[str, Any]) -> dict[str, Any]: + return self._state._apply(dict(mutation)) + +else: + VioletRunnerGuardedState = _LocalGuardedState # type: ignore[assignment] + + +class SerialRunnerState: + """Small wrapper around ASExWorker with a deterministic local fallback.""" + + def __init__(self) -> None: + self.backend = VioletRunnerGuardedState() + self.worker = ASExWorker(self.backend) if ASExWorker is not None else None + + def mutate(self, mutation: Mapping[str, Any], *, timeout: float = 5.0) -> dict[str, Any]: + payload = dict(mutation) + if self.worker is not None: + return self.worker.mutate(payload).result(timeout=timeout) + return self.backend.mutate(payload) # type: ignore[attr-defined] + + def close(self) -> None: + if self.worker is not None: + self.worker.close(timeout=2.0) + + +def _parse_scan_payload(raw: Any) -> dict[str, Any]: + if raw is None: + return {} + if isinstance(raw, str): + data = json.loads(raw) + elif isinstance(raw, Mapping): + data = dict(raw) + else: + return {} + if isinstance(data, dict) and data.get("version") == "NG7": + try: + from prod.clean_arch.adapters.eigen_scan_normalizer import normalize_ng7_scan + + data = normalize_ng7_scan(data) + except Exception: + LOGGER.debug("NG7 normalize failed; using raw scan", exc_info=True) + return data if isinstance(data, dict) else {} + + +def reference_price_from_scan(payload: Mapping[str, Any], asset: str) -> float: + """Extract the decision asset price from an NG7/universe payload.""" + + symbol = str(asset or "").upper() + assets = payload.get("assets") or [] + prices = payload.get("asset_prices") or [] + if isinstance(assets, list) and isinstance(prices, list): + for raw_asset, raw_price in zip(assets, prices): + if str(raw_asset).upper() != symbol: + continue + price = _finite_positive(float(raw_price), "reference_price") + return price + + result = payload.get("result") if isinstance(payload, Mapping) else None + if isinstance(result, Mapping): + result_asset = str(result.get("asset") or payload.get("target_asset") or "").upper() + if result_asset == symbol: + return _finite_positive(float(result.get("price") or payload.get("price")), "reference_price") + + payload_asset = str(payload.get("asset") or payload.get("target_asset") or "").upper() + if payload_asset == symbol and payload.get("price") is not None: + return _finite_positive(float(payload.get("price")), "reference_price") + + raise ValueError(f"scan payload has no fresh price for {symbol}") + + +def _decision_from_shadow( + shadow: dict[str, Any], + payload: dict[str, Any], + *, + scan_number: int, + now_ns: int, + vel_div: float, + vol_ok: bool, +) -> Optional[ShadowDecision]: + """Run the existing BLUE-faithful VIOLET shadow path and return the decision.""" + + from .shadow_live_factors import _ensure_ob_engine + + shadow["engine"].observe(payload, scan_number) + live_source = shadow.get("live_source") + if live_source is None: + return None + + ob_engine = _ensure_ob_engine(shadow, payload) + live_result = live_source( + shadow["client"], + scan_history=shadow["scan_history"], + selector=shadow["selector"], + ob_engine=ob_engine, + bar_idx=shadow.get("ob_bar_idx", 0), + prior_boost_beta=shadow.get("prior_boost_beta"), + ) + shadow["last_live_source"] = live_result + shadow["ob_bar_idx"] = shadow.get("ob_bar_idx", 0) + 1 + ab = getattr(live_result, "acb_boost", None) + bb = getattr(live_result, "acb_beta", None) + if ab is not None and bb is not None: + shadow["prior_boost_beta"] = (ab, bb) + + decision = shadow["engine"].decide( + now_ns=now_ns, + scan_number=scan_number, + capital=shadow["capital"], + vel_div=vel_div, + vol_ok=vol_ok, + factors=live_result.factors, + ) + if decision is None: + return None + journal = shadow.get("journal") + if journal is not None: + journal.journal(decision, mono_ns=now_ns) + return decision + + +class HazelcastNG7ScanSource: + """Event-listener first, polling fallback source for NG7 scans.""" + + def __init__(self, config: VioletV4RunnerConfig) -> None: + self.config = config + self.client = None + self.map = None + self.blocking_map = None + self.queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=config.queue_maxsize) + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._last_polled_scan = -1 + + async def connect(self) -> None: + import hazelcast + + self._loop = asyncio.get_running_loop() + self.client = hazelcast.HazelcastClient( + cluster_name=self.config.hz_cluster, + cluster_members=[self.config.hz_host], + ) + self.map = self.client.get_map(self.config.hz_map) + self.blocking_map = self.map.blocking() + + def on_entry(event: Any) -> None: + try: + payload = _parse_scan_payload(getattr(event, "value", None)) + if not payload: + return + loop = self._loop + if loop is None: + return + loop.call_soon_threadsafe(self._offer_payload, payload) + except Exception: + LOGGER.debug("scan listener callback failed", exc_info=True) + + self.map.add_entry_listener( + include_value=True, + key=self.config.hz_key, + updated_func=on_entry, + added_func=on_entry, + ) + + def _offer_payload(self, payload: dict[str, Any]) -> None: + try: + self.queue.put_nowait(payload) + except asyncio.QueueFull: + try: + self.queue.get_nowait() + except asyncio.QueueEmpty: + pass + self.queue.put_nowait(payload) + + async def next_payload(self) -> dict[str, Any]: + try: + return await asyncio.wait_for(self.queue.get(), timeout=self.config.poll_interval_s) + except asyncio.TimeoutError: + raw = self.blocking_map.get(self.config.hz_key) + payload = _parse_scan_payload(raw) + sn = int(payload.get("scan_number") or 0) + if sn <= self._last_polled_scan: + return {} + self._last_polled_scan = sn + return payload + + async def close(self) -> None: + if self.client is not None: + self.client.shutdown() + + +class VioletV4ExecutionRunner: + """Live-capable runner: decision path to actual DITAv2 execution.""" + + def __init__( + self, + *, + bundle: Any, + shadow: dict[str, Any], + config: VioletV4RunnerConfig, + state: Optional[SerialRunnerState] = None, + decision_step: Optional[ + Callable[[dict[str, Any], dict[str, Any], int, int, float, bool], Optional[ShadowDecision]] + ] = None, + ) -> None: + self.bundle = bundle + self.shadow = shadow + self.config = config + self.state = state or SerialRunnerState() + self.decision_step = decision_step or ( + lambda sh, payload, sn, now_ns, vd, vol_ok: _decision_from_shadow( + sh, + payload, + scan_number=sn, + now_ns=now_ns, + vel_div=vd, + vol_ok=vol_ok, + ) + ) + self._closed = False + + async def connect(self) -> None: + connect = getattr(self.bundle.venue, "connect", None) + if connect is not None: + result = connect() + if asyncio.iscoroutine(result): + await result + + def close(self) -> None: + if self._closed: + return + self._closed = True + try: + close = getattr(self.bundle, "close", None) + if close is not None: + close() + finally: + self.state.close() + + def _capital(self) -> float: + try: + cap = float(self.bundle.kernel.account.snapshot.capital) + if math.isfinite(cap) and cap > 0.0: + return cap + except Exception: + pass + return float(self.config.capital) + + def _maybe_cap_exec_intent(self, intent: ExecIntent, reference_price: float) -> ExecIntent: + cap = float(self.config.max_notional_usdt or 0.0) + if cap <= 0.0 or float(intent.target_notional) <= cap: + return intent + return ExecIntent( + asset=intent.asset, + side=intent.side, + qty=cap / reference_price, + exchange_leverage=intent.exchange_leverage, + maker_policy=intent.maker_policy, + target_notional=cap, + ts_ns=intent.ts_ns, + reason=intent.reason, + ) + + async def process_scan_payload(self, raw_payload: Any) -> Optional[KernelOutcome]: + payload = _parse_scan_payload(raw_payload) + if not payload: + return None + scan_number = int(payload.get("scan_number") or 0) + if scan_number <= 0: + return None + scan_state = self.state.mutate({"op": "scan", "scan_number": scan_number}) + if not scan_state.get("accepted"): + return None + + vel_div = float(payload.get("vel_div") or 0.0) + if not math.isfinite(vel_div): + vel_div = 0.0 + vol_ok = bool(payload.get("vol_ok", True)) + if self.config.assume_vol_ok: + vol_ok = True + now_ns = int(time.monotonic_ns()) + decision = self.decision_step(self.shadow, payload, scan_number, now_ns, vel_div, vol_ok) + if decision is None: + return None + self.state.mutate({"op": "decision", "scan_number": scan_number, "asset": decision.asset}) + + reference_price = reference_price_from_scan(payload, decision.asset) + exec_intent = to_exec_intent( + decision, + capital=self._capital(), + reference_price=reference_price, + maker_policy=self.config.maker_policy, + ) + exec_intent = self._maybe_cap_exec_intent(exec_intent, reference_price) + + trade_id = f"VIOLET-{self.config.session_id[:8]}-{scan_number}" + intent_id = f"violet-v4-{self.config.session_id[:8]}-{scan_number}" + kernel_intent = exec_intent_to_kernel_intent( + exec_intent, + reference_price=reference_price, + trade_id=trade_id, + slot_id=self.config.slot_id, + intent_id=intent_id, + order_type=self.config.order_type, + limit_price=self.config.limit_price, + metadata={ + "violet_session_id": self.config.session_id, + "scan_number": scan_number, + "price_source": "ng7_scan", + "raw_notional_before_runner_cap": float(decision.target_exposure), + }, + ) + submission = self.state.mutate( + {"op": "submission", "scan_number": scan_number, "trade_id": trade_id, "intent_id": intent_id} + ) + if self.config.max_submissions and int(submission.get("submissions") or 0) > self.config.max_submissions: + LOGGER.critical("max submissions reached; refusing scan=%s trade_id=%s", scan_number, trade_id) + return None + + LOGGER.warning( + "VIOLET V4 LIVE SUBMIT scan=%s asset=%s side=%s qty=%.8f lev=%.2f ref=%.8f trade=%s", + scan_number, + kernel_intent.asset, + kernel_intent.side.value, + kernel_intent.target_size, + kernel_intent.leverage, + kernel_intent.reference_price, + trade_id, + ) + try: + outcome = await self.bundle.kernel.process_intent_async(kernel_intent) + self.state.mutate( + { + "op": "outcome", + "scan_number": scan_number, + "trade_id": trade_id, + "accepted": bool(getattr(outcome, "accepted", False)), + "diagnostic": str(getattr(getattr(outcome, "diagnostic_code", ""), "value", "")), + } + ) + return outcome + except Exception as exc: + self.state.mutate({"op": "error", "scan_number": scan_number, "error": str(exc)}) + raise + + +def _apply_runtime_optimizations() -> None: + if gc.isenabled(): + gc.disable() + + +def _build_shadow(config: VioletV4RunnerConfig) -> dict[str, Any]: + from prod.ch_writer import ch_put_violet + + from .decision_engine import VioletDecisionEngine + from .shadow_journal import VioletDecisionJournal + from .shadow_live_factors import build_shadow_live_source + + thr = float(os.environ.get("DOLPHIN_VIOLET_ENTRY_VEL_DIV_THRESHOLD", "-0.02")) + return { + "engine": VioletDecisionEngine(entry_vel_div_threshold=thr), + "journal": VioletDecisionJournal(sink=ch_put_violet, session_id=config.session_id), + "capital": float(config.capital), + **build_shadow_live_source(), + } + + +def _build_real_bundle(config: VioletV4RunnerConfig) -> Any: + from prod.clean_arch.dita_v2.launcher import build_launcher_bundle + from prod.launch_dolphin_violet import ( + DDL_APPLY_CMD, + _apply_violet_env, + _preflight_clickhouse, + _violet_table_present, + build_bingx_exec_client_config, + ) + from .v4_arming import assess_v4_arming, write_v4_arming_report + + _apply_violet_env() + missing = _preflight_clickhouse() + if missing: + raise RuntimeError(f"dolphin_violet tables missing: {missing}; run {DDL_APPLY_CMD}") + if not _violet_table_present("violet_decisions"): + raise RuntimeError(f"dolphin_violet table missing: violet_decisions; run {DDL_APPLY_CMD}") + report = assess_v4_arming(report_dir=config.report_dir, launcher_mode="execution") + write_v4_arming_report(report, report_dir=config.report_dir) + if config.require_arming and not report.can_arm: + raise RuntimeError(f"VIOLET V4 arming refused: {report.reasons}") + + return build_launcher_bundle( + venue_mode="BINGX", + max_slots=1, + prefix="violet", + bingx_config=build_bingx_exec_client_config(), + ) + + +async def run_live( + config: Optional[VioletV4RunnerConfig] = None, + *, + stop_after_first_submission: bool = False, +) -> None: + """Run the actual VIOLET V4 BingX VST execution loop.""" + + cfg = config or VioletV4RunnerConfig.from_env() + _apply_runtime_optimizations() + bundle = _build_real_bundle(cfg) + shadow = _build_shadow(cfg) + source = HazelcastNG7ScanSource(cfg) + runner = VioletV4ExecutionRunner(bundle=bundle, shadow=shadow, config=cfg) + await runner.connect() + await source.connect() + LOGGER.critical( + "VIOLET V4 EXECUTION RUNNER ARMED: session=%s venue=BINGX_VST hz=%s[%s]", + cfg.session_id, + cfg.hz_map, + cfg.hz_key, + ) + try: + while True: + payload = await source.next_payload() + if payload: + outcome = await runner.process_scan_payload(payload) + if stop_after_first_submission and outcome is not None: + return + await asyncio.sleep(0) + finally: + await source.close() + runner.close() + + +def main(argv: Optional[list[str]] = None) -> int: + parser = argparse.ArgumentParser(description="VIOLET V4 live BingX VST execution runner") + parser.add_argument("--once", action="store_true", help="process one queued/polled scan then exit") + args = parser.parse_args(argv) + config = VioletV4RunnerConfig.from_env() + if args.once: + config = dataclass_replace(config, max_submissions=1) + asyncio.run(run_live(config, stop_after_first_submission=args.once)) + return 0 + + +def dataclass_replace(config: VioletV4RunnerConfig, **updates: Any) -> VioletV4RunnerConfig: + data = config.__dict__.copy() + data.update(updates) + return VioletV4RunnerConfig(**data) + + +__all__ = [ + "HazelcastNG7ScanSource", + "RUNNER_CONTRACT_VERSION", + "SerialRunnerState", + "VioletV4ExecutionRunner", + "VioletV4RunnerConfig", + "exec_intent_to_kernel_intent", + "main", + "reference_price_from_scan", + "run_live", +] + + +if __name__ == "__main__": + raise SystemExit(main())