feat(okx): add listenOrderBook

src/exchanges/okx/okx.exchange.ts

@@ -9,6 +9,7 @@ import type {
   Candle,
   ExchangeOptions,
   OHLCVOptions,
+  OrderBook,
   Position,
   Ticker,
 } from '../../types';
@@ -312,4 +313,11 @@ export class OKXExchange extends BaseExchange {
   listenOHLCV = (opts: OHLCVOptions, callback: (candle: Candle) => void) => {
     return this.publicWebsocket.listenOHLCV(opts, callback);
   };
+
+  listenOrderBook = (
+    symbol: string,
+    callback: (orderBook: OrderBook) => void
+  ) => {
+    return this.publicWebsocket.listenOrderBook(symbol, callback);
+  };
 }

src/exchanges/okx/okx.types.ts

@@ -7,11 +7,11 @@ export const BASE_URL = 'https://okx.tuleep.trade';
 export const BASE_WS_URL = {
   public: {
     livenet: 'wss://wsaws.okx.com:8443/ws/v5/public',
-    testnet: 'wss://okx-testnet-wss.tuleep.trade/ws/v5/public',
+    testnet: 'wss://okx-testnet-wss.tuleep.trade/ws/v5/public?brokerId=9999',
   },
   private: {
     livenet: 'wss://wsaws.okx.com:8443/ws/v5/private',
-    testnet: 'wss://okx-testnet-wss.tuleep.trade/ws/v5/private',
+    testnet: 'wss://okx-testnet-wss.tuleep.trade/ws/v5/private?brokerId=9999',
   },
 };
 

src/exchanges/okx/okx.ws-public.ts

@@ -1,7 +1,14 @@
 import flatten from 'lodash/flatten';
 
-import type { Candle, OHLCVOptions } from '../../types';
+import type {
+  Candle,
+  OHLCVOptions,
+  OrderBook,
+  OrderBookOrders,
+} from '../../types';
+import { calcOrderBookTotal, sortOrderBook } from '../../utils/orderbook';
 import { roundUSD } from '../../utils/round-usd';
+import { multiply } from '../../utils/safe-math';
 import { BaseWebSocket } from '../base.ws';
 
 import type { OKXExchange } from './okx.exchange';
@@ -85,11 +92,11 @@ export class OKXPublicWebsocket extends BaseWebSocket<OKXExchange> {
   };
 
   listenOHLCV = (opts: OHLCVOptions, callback: (candle: Candle) => void) => {
-    let timeoutId: NodeJS.Timeout | null = null;
-
     const market = this.store.markets.find((m) => m.symbol === opts.symbol);
     if (!market) return () => {};
 
+    let timeoutId: NodeJS.Timeout | null = null;
+
     const topic = {
       channel: `candle${opts.interval}`,
       instId: market.id,
@@ -128,7 +135,118 @@ export class OKXPublicWebsocket extends BaseWebSocket<OKXExchange> {
       }
 
       if (this.isConnected) {
-        const payload = { op: 'unsubscribe', args: topic };
+        const payload = { op: 'unsubscribe', args: [topic] };
+        this.ws?.send?.(JSON.stringify(payload));
+      }
+    };
+  };
+
+  listenOrderBook = (
+    symbol: string,
+    callback: (orderBook: OrderBook) => void
+  ) => {
+    const market = this.store.markets.find((m) => m.symbol === symbol);
+    if (!market) return () => {};
+
+    let timeoutId: NodeJS.Timeout | null = null;
+    const sides = ['bids', 'asks'] as const;
+    const orderBook: OrderBook = { bids: [], asks: [] };
+    const topic = { channel: 'books', instId: market.id };
+
+    const waitForConnectedAndSubscribe = () => {
+      if (this.isConnected) {
+        if (!this.isDisposed) {
+          this.messageHandlers.books = (data: Data) => {
+            if (data.action === 'snapshot') {
+              const {
+                data: [snapshot],
+              } = data;
+
+              sides.forEach((side) => {
+                orderBook[side] = snapshot[side].reduce(
+                  (acc: OrderBookOrders[], [price, amount]: string[]) => {
+                    if (parseFloat(amount) === 0) return acc;
+                    return [
+                      ...acc,
+                      {
+                        price: parseFloat(price),
+                        amount: multiply(
+                          parseFloat(amount),
+                          market.precision.amount
+                        ),
+                      },
+                    ];
+                  },
+                  []
+                );
+              });
+            }
+
+            if (data.action === 'update') {
+              const {
+                data: [update],
+              } = data;
+
+              sides.forEach((side) => {
+                for (const [rPrice, rAmount] of update[side]) {
+                  const price = parseFloat(rPrice);
+                  const amount = parseFloat(rAmount);
+
+                  const index = orderBook[side].findIndex(
+                    (b) => b.price === price
+                  );
+
+                  if (amount === 0 && index !== -1) {
+                    orderBook[side].splice(index, 1);
+                    return;
+                  }
+
+                  if (amount !== 0) {
+                    if (index === -1) {
+                      orderBook[side].push({
+                        price,
+                        amount: multiply(amount, market.precision.amount),
+                        total: 0,
+                      });
+                      return;
+                    }
+
+                    orderBook[side][index].amount = multiply(
+                      amount,
+                      market.precision.amount
+                    );
+                  }
+                }
+              });
+            }
+
+            sortOrderBook(orderBook);
+            calcOrderBookTotal(orderBook);
+
+            callback(orderBook);
+          };
+
+          const payload = JSON.stringify({ op: 'subscribe', args: [topic] });
+          this.ws?.send?.(payload);
+        }
+      } else {
+        timeoutId = setTimeout(() => waitForConnectedAndSubscribe(), 100);
+      }
+    };
+
+    waitForConnectedAndSubscribe();
+
+    return () => {
+      orderBook.bids = [];
+      orderBook.asks = [];
+
+      if (timeoutId) {
+        clearTimeout(timeoutId);
+        timeoutId = null;
+      }
+
+      if (this.isConnected) {
+        const payload = { op: 'unsubscribe', args: [topic] };
         this.ws?.send?.(JSON.stringify(payload));
       }
     };