إضافة عمق سجل الأوامر المباشر عبر SAHMK WebSocket ‏(JavaScript + Python)

wss://api.sahmk.sa/ws/v1/market/depth/?api_key=YOUR_API_KEY

{"action":"snapshot","symbol":"2222","levels":5}
{"action":"subscribe","symbols":["2222","1120"],"levels":5}
{"action":"unsubscribe","symbols":["1120"]}
{"action":"ping"}
{"action":"subscribe","symbols":["*"],"levels":20} // Enterprise only

const API_KEY = "YOUR_API_KEY";
const URL = `wss://api.sahmk.sa/ws/v1/market/depth/?api_key=${API_KEY}`;

// Keep these aligned with your account and connected.limits.
const REQUESTED_LEVELS = 5;
const SYMBOLS = ["2222", "1120", "2010"];

let ws = null;
let pingTimer = null;
let reconnectAttempt = 0;
let manualClose = false;
let lastErrorSignal = null;
let connectedLimits = null;
let desiredSymbols = [...SYMBOLS];
const lastSnapshotAtBySymbol = new Map();

const MIN_ACTION_INTERVAL_MS = 250;
let queueTimer = null;
const actionQueue = [];

function enqueueAction(payload) {
  actionQueue.push(payload);
  if (!queueTimer) flushQueue();
}

function flushQueue() {
  if (!ws || ws.readyState !== WebSocket.OPEN) {
    queueTimer = null;
    return;
  }
  const item = actionQueue.shift();
  if (!item) {
    queueTimer = null;
    return;
  }
  ws.send(JSON.stringify(item));
  queueTimer = setTimeout(flushQueue, MIN_ACTION_INTERVAL_MS);
}

function nextDelayMs(attempt) {
  const base = 1000;
  const cap = 30000;
  const exp = Math.min(cap, base * 2 ** attempt);
  const jitter = Math.floor(Math.random() * 500);
  return exp + jitter;
}

function parseRetryAfterSeconds(details) {
  if (!details || typeof details !== "object") return null;
  const value = Number(details.retry_after_seconds);
  return Number.isFinite(value) && value > 0 ? value : null;
}

function perCallLimit() {
  const runtime = Number(connectedLimits?.max_symbols_per_call);
  if (Number.isFinite(runtime) && runtime > 0) return runtime;
  return 20;
}

function chunk(arr, size) {
  const out = [];
  for (let i = 0; i < arr.length; i += size) out.push(arr.slice(i, i + size));
  return out;
}

function subscribeSymbols(symbols) {
  for (const group of chunk(symbols, perCallLimit())) {
    enqueueAction({ action: "subscribe", symbols: group, levels: REQUESTED_LEVELS });
  }
}

function requestSnapshotSafe(symbol) {
  const now = Date.now();
  const last = lastSnapshotAtBySymbol.get(symbol) || 0;
  if (now - last < 1500) return; // snapshot cooldown per symbol
  lastSnapshotAtBySymbol.set(symbol, now);
  enqueueAction({ action: "snapshot", symbol, levels: REQUESTED_LEVELS });
}

function startPing() {
  clearInterval(pingTimer);
  pingTimer = setInterval(() => {
    if (ws && ws.readyState === WebSocket.OPEN) enqueueAction({ action: "ping" });
  }, 30000);
}

function stopPing() {
  clearInterval(pingTimer);
  pingTimer = null;
}

function connect() {
  ws = new WebSocket(URL);

  ws.onopen = () => {
    reconnectAttempt = 0;
    lastErrorSignal = null;
    subscribeSymbols(desiredSymbols);
    startPing();
  };

  ws.onmessage = (event) => {
    let msg;
    try {
      msg = JSON.parse(event.data);
    } catch {
      console.error("Invalid JSON from server:", event.data);
      return;
    }

    if (msg.type === "connected") {
      connectedLimits = msg.limits || null;
      console.log("Connected depth limits:", connectedLimits);
      return;
    }

    if (msg.type === "subscribed") {
      console.log("Subscribed symbols:", msg.symbols);
      return;
    }

    if (msg.type === "depth_snapshot") {
      // Render depth snapshot in UI/store.
      // msg: { symbol, bids, asks, entitled_levels, total_bid_quantity, total_ask_quantity, level_imbalance, ... }
      console.log("Depth snapshot:", msg.symbol);
      return;
    }

    if (msg.type === "error") {
      lastErrorSignal = { code: msg.code || null, details: msg.details || null };
      console.error("Depth server error:", msg);
      return;
    }
  };

  ws.onclose = (event) => {
    stopPing();
    console.warn(`Depth closed: code=${event.code} reason=${event.reason}`);

    if (event.code === 4401) {
      console.error("4401 auth error. Fix API key before reconnecting.");
      return;
    }
    if (event.code === 4403) {
      console.error(
        "4403 access error. Open Realtime Access, confirm Market Depth is approved/activated, then reconnect."
      );
      return;
    }
    if (event.code === 4429) {
      const retryAfter = parseRetryAfterSeconds(lastErrorSignal?.details);
      const requestedDelay = ((retryAfter || 1) * 1000) + Math.floor(Math.random() * 500);
      const delay = Math.max(nextDelayMs(reconnectAttempt++), requestedDelay);
      console.warn(`4429 temporary throttle. Reconnecting in ${delay}ms`);
      if (!manualClose) setTimeout(connect, delay);
      return;
    }

    if (!manualClose) {
      const delay = nextDelayMs(reconnectAttempt++);
      setTimeout(connect, delay);
    }
  };
}

window.addEventListener("beforeunload", () => {
  manualClose = true;
  if (ws && ws.readyState === WebSocket.OPEN && desiredSymbols.length) {
    enqueueAction({ action: "unsubscribe", symbols: desiredSymbols });
  }
  stopPing();
  ws?.close(1000, "Page unload");
});

// Public method your UI can call
window.depthRequestSnapshot = requestSnapshotSafe;

connect();

import asyncio
import contextlib
import json
import random
import time
import websockets

API_KEY = "YOUR_API_KEY"
URL = f"wss://api.sahmk.sa/ws/v1/market/depth/?api_key={API_KEY}"
REQUESTED_LEVELS = 5
SYMBOLS = ["2222", "1120", "2010"]

MIN_ACTION_INTERVAL_SECONDS = 0.25
SNAPSHOT_MIN_INTERVAL_SECONDS = 1.5

def next_delay_seconds(attempt: int) -> float:
    base = 1.0
    cap = 30.0
    return min(cap, base * (2 ** attempt)) + random.uniform(0.0, 0.5)

def parse_retry_after_seconds(details):
    if not isinstance(details, dict):
        return None
    raw = details.get("retry_after_seconds")
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return None
    return value if value > 0 else None

def chunk(items, size):
    out = []
    for i in range(0, len(items), size):
        out.append(items[i : i + size])
    return out

async def action_sender(ws, action_queue):
    while True:
        payload = await action_queue.get()
        await ws.send(json.dumps(payload))
        await asyncio.sleep(MIN_ACTION_INTERVAL_SECONDS)

async def stream_forever():
    attempt = 0
    connected_limits = {}
    last_error_signal = {}
    snapshot_last_at = {}
    desired_symbols = SYMBOLS[:]

    while True:
        try:
            async with websockets.connect(URL, ping_interval=None) as ws:
                attempt = 0
                connected_limits = {}
                last_error_signal = {}
                action_queue = asyncio.Queue()
                sender_task = asyncio.create_task(action_sender(ws, action_queue))

                async def queue_subscribe_all():
                    per_call = int(connected_limits.get("max_symbols_per_call", 20) or 20)
                    for group in chunk(desired_symbols, per_call):
                        await action_queue.put(
                            {"action": "subscribe", "symbols": group, "levels": REQUESTED_LEVELS}
                        )

                try:
                    # Proactive ping loop
                    async def ping_loop():
                        while True:
                            await asyncio.sleep(30)
                            await action_queue.put({"action": "ping"})

                    pinger = asyncio.create_task(ping_loop())

                    # Subscribe once; if connected limits arrive later, next reconnect applies them.
                    await queue_subscribe_all()

                    async for raw in ws:
                        try:
                            msg = json.loads(raw)
                        except json.JSONDecodeError:
                            print("Invalid JSON payload:", raw)
                            continue

                        msg_type = msg.get("type")
                        if msg_type == "connected":
                            connected_limits = msg.get("limits") or {}
                            print("Connected limits:", connected_limits)
                        elif msg_type == "subscribed":
                            print("Subscribed:", msg.get("symbols"))
                        elif msg_type == "depth_snapshot":
                            print("Depth snapshot:", msg.get("symbol"))
                        elif msg_type == "error":
                            last_error_signal = {
                                "code": msg.get("code"),
                                "details": msg.get("details"),
                            }
                            print("Server error:", msg)

                        # Example snapshot throttle call path:
                        # symbol = __TOKEN_30__
                        # now = time.time()
                        # if now - snapshot_last_at.get(symbol, 0.0) >= SNAPSHOT_MIN_INTERVAL_SECONDS:
                        #     snapshot_last_at[symbol] = now
                        #     await action_queue.put({__TOKEN_31__:__TOKEN_32__,__TOKEN_33__:symbol,__TOKEN_34__:REQUESTED_LEVELS})
                finally:
                    pinger.cancel()
                    sender_task.cancel()
                    with contextlib.suppress(asyncio.CancelledError):
                        await pinger
                    with contextlib.suppress(asyncio.CancelledError):
                        await sender_task
        except websockets.exceptions.ConnectionClosed as exc:
            print(f"Connection closed: code={exc.code} reason={exc.reason}")
            if exc.code == 4401:
                print("4401 auth error. Fix API key before reconnect.")
                return
            if exc.code == 4403:
                print(
                    "4403 access error. Confirm Market Depth is approved/activated in Realtime Access, then reconnect."
                )
                return
            if exc.code == 4429:
                retry_after = parse_retry_after_seconds(last_error_signal.get("details"))
                delay = max(next_delay_seconds(attempt), (retry_after or 1.0) + random.uniform(0.0, 0.5))
                print(f"4429 temporary throttle. Reconnecting in {delay:.2f}s")
                await asyncio.sleep(delay)
                attempt += 1
                continue
        except Exception as exc:
            print(f"Unexpected error: {exc}")

        delay = next_delay_seconds(attempt)
        print(f"Reconnecting in {delay:.2f}s")
        await asyncio.sleep(delay)
        attempt += 1

if __name__ == "__main__":
    asyncio.run(stream_forever())

pip install websockets

curl -X GET "https://api.sahmk.sa/api/v1/market/depth/2222/?levels=5" \
  -H "X-API-Key: YOUR_API_KEY"