diff --git a/examples/websocket/public/order_book.py b/examples/websocket/public/order_book.py index ef6e31c..185ca56 100644 --- a/examples/websocket/public/order_book.py +++ b/examples/websocket/public/order_book.py @@ -2,7 +2,9 @@ from collections import OrderedDict -from typing import List +from typing import List, Dict + +import crcmod from bfxapi import Client, PUB_WSS_HOST @@ -14,10 +16,13 @@ class OrderBook: def __init__(self, symbols: List[str]): self.__order_book = { symbol: { - "bids": OrderedDict(), "asks": OrderedDict() + "bids": OrderedDict(), "asks": OrderedDict() } for symbol in symbols } + self.cooldown: Dict[str, bool] = \ + { symbol: False for symbol in symbols } + def update(self, symbol: str, data: TradingPairBook) -> None: price, count, amount = data.price, data.count, data.amount @@ -34,6 +39,31 @@ class OrderBook: if price in self.__order_book[symbol][kind]: del self.__order_book[symbol][kind][price] + def verify(self, symbol: str, checksum: int) -> bool: + values: List[int] = [ ] + + bids = sorted([ (data["price"], data["count"], data["amount"]) \ + for _, data in self.__order_book[symbol]["bids"].items() ], + key=lambda data: -data[0]) + + asks = sorted([ (data["price"], data["count"], data["amount"]) \ + for _, data in self.__order_book[symbol]["asks"].items() ], + key=lambda data: data[0]) + + if len(bids) < 25 or len(asks) < 25: + raise AssertionError("Not enough bids or asks (need at least 25).") + + for _i in range(25): + bid, ask = bids[_i], asks[_i] + values.extend([ bid[0], bid[2] ]) + values.extend([ ask[0], ask[2] ]) + + local = ":".join(str(value) for value in values).encode("UTF-8") + + crc32 = crcmod.mkCrcFun(0x104C11DB7, initCrc=0, xorOut=0xFFFFFFFF) + + return crc32(local) == checksum + SYMBOLS = [ "tBTCUSD", "tLTCUSD", "tLTCBTC", "tETHUSD", "tETHBTC" ] order_book = OrderBook(symbols=SYMBOLS) @@ -62,4 +92,18 @@ def on_t_book_snapshot(subscription: Book, snapshot: List[TradingPairBook]): def on_t_book_update(subscription: Book, data: TradingPairBook): order_book.update(subscription["symbol"], data) +@bfx.wss.on("checksum") +async def on_checksum(subscription: Book, value: int): + symbol = subscription["symbol"] + + if order_book.verify(symbol, value): + order_book.cooldown[symbol] = False + elif not order_book.cooldown[symbol]: + print("Mismatch between local and remote checksums: " + f"restarting book for symbol <{symbol}>...") + + await bfx.wss.resubscribe(sub_id=subscription["subId"]) + + order_book.cooldown[symbol] = True + bfx.wss.run()