#!/usr/bin/env python3 import argparse import json import threading import time from typing import Any, Optional import websocket # pip install websocket-client def build_msg(endpoint: str, msg_id: int, query: str = "", body: Any = "") -> str: if body is None: body_str = "" elif isinstance(body, str): body_str = body else: body_str = json.dumps(body, separators=(",", ":")) return f"{endpoint}\n{msg_id}\n{query}\n{body_str}" class Stage2InteractiveWSClient: def __init__(self, url: str, token: str, hb_idle_ms: int = 2500): self.url = url self.token = token self.hb_idle_ms = hb_idle_ms self.ws_app: Optional[websocket.WebSocketApp] = None self._stop = threading.Event() self._connected = threading.Event() self._send_lock = threading.Lock() self._last_recv_ts = time.time() self._got_open_frame = False self._authorized = False self._syncrequest_sent = False self._next_msg_id = 2 # 0=authorize, 1=syncrequest def start(self): self.ws_app = websocket.WebSocketApp( self.url, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close, ) ws_thread = threading.Thread( target=lambda: self.ws_app.run_forever(ping_interval=20, ping_timeout=10), daemon=True, ) ws_thread.start() print("[INFO] Connecting...") self._connected.wait(timeout=15) threading.Thread(target=self._heartbeat_loop, daemon=True).start() self._stdin_loop() def stop(self): self._stop.set() try: if self.ws_app: self.ws_app.close() except Exception: pass def send_raw(self, payload: str): with self._send_lock: if not self.ws_app: print("[ERR] WebSocket not initialized") return self.ws_app.send(payload) def send_endpoint(self, endpoint: str, msg_id: int, body: Any = "", query: str = ""): payload = build_msg(endpoint, msg_id, query, body) self.send_raw(payload) print(f"[SENT] {endpoint} (i={msg_id})") def next_msg_id(self) -> int: msg_id = self._next_msg_id self._next_msg_id += 1 return msg_id # -------------------- WS callbacks -------------------- def on_open(self, ws): print("[OPEN] connected") self._connected.set() self._last_recv_ts = time.time() def on_message(self, ws, message: str): self._last_recv_ts = time.time() if message == "o": print("[RECV] o (open frame)") self._got_open_frame = True self._send_authorize() return if message == "h": try: ws.send("[]") except Exception: pass return if not self._got_open_frame and not self._authorized: self._send_authorize() if message.startswith("a"): try: arr = json.loads(message[1:]) except Exception as e: print("[RECV] a... parse error:", e, message[:200]) return for item in arr: self._handle_item(item) return try: obj = json.loads(message) self._handle_item(obj) except Exception: print("[RECV] raw:", message[:200]) def _send_authorize(self): if self._authorized: return self.send_endpoint("authorize", 0, self.token) def _send_syncrequest(self): if self._syncrequest_sent: return request_body = { "splitResponses": True, "shardingExpression": { "expressionType": "modUserId", "divisor": 1, "remainder": 0, }, "entityTypes": [ "account", "accountRiskStatus", "cashBalance", "commandReport", "command", "executionReport", "fill", "fillPair", "order", "orderStrategy", "position", "product", "user", ], } self.send_endpoint("user/syncrequest", 1, request_body) self._syncrequest_sent = True def _handle_item(self, item: Any): if isinstance(item, str): try: item = json.loads(item) except Exception: print("[RECV] item(str):", item[:200]) return if isinstance(item, dict): print("[RECV] item(json):", json.dumps(item, indent=2)) if item.get("i") == 0 and "s" in item: if item["s"] == 200: self._authorized = True print("[INFO] authorized OK") self._send_syncrequest() print("[INFO] ready for commands") else: print("[ERROR] authorize failed:", item) self.stop() else: print("[RECV] item(other):", item) # -------------------- heartbeat -------------------- def _heartbeat_loop(self): while not self._stop.is_set(): idle_ms = (time.time() - self._last_recv_ts) * 1000.0 if idle_ms > self.hb_idle_ms: try: self.send_raw("[]") self._last_recv_ts = time.time() except Exception: return time.sleep(0.25) # -------------------- FIXED stdin loop -------------------- def _stdin_loop(self): print("\nPaste WS message → END to send | HELP for examples | QUIT to exit\n") while not self._stop.is_set(): lines = [] while True: try: line = input() except EOFError: self.stop() return cmd = line.strip().upper() if cmd == "QUIT": self.stop() return if cmd == "HELP": print("\nExample order:\n") print("order/placeorder") print("2\n") print('{"accountId":284301,"accountSpec":"Tradesterdev1","symbol":"ESM6","action":"Buy","orderQty":1,"orderType":"Market","isAutomated":true}') print("END\n") lines = [] break if line.strip() == "END": break lines.append(line) if not lines: continue # 🔥 FIX: strip leading/trailing blank lines while lines and not lines[0].strip(): lines.pop(0) while lines and not lines[-1].strip(): lines.pop() if not lines: continue raw = "\n".join(lines) # 🔍 debug (safe to keep) print("[DEBUG] sending:", repr(raw)) self.send_raw(raw) first = raw.splitlines()[0] if raw.splitlines() else "" print(f"[SENT] {first} (first line / endpoint)") # -------------------- def on_error(self, ws, error): print("[ERROR]", error) def on_close(self, ws, code, msg): print(f"[CLOSE] code={code}, msg={msg}") self._stop.set() def main(): ap = argparse.ArgumentParser() ap.add_argument("--url", required=True) ap.add_argument("--token", required=True) ap.add_argument("--hb_idle_ms", type=int, default=2500) args = ap.parse_args() client = Stage2InteractiveWSClient(args.url, args.token, args.hb_idle_ms) try: client.start() except KeyboardInterrupt: client.stop() if __name__ == "__main__": main()