I am trying to extract data from two different channels of a websocket server simultaneously. In particular, I am extracting the orderbooks of two pairs of crypto BTC/USD and ZRX/USD from the FTX Exchange. The Websocket sends an initial snapshot of the orderbooks and then it sends updates. At every message, I apply the updates.
I want to store the orderbooks in a dictionary (maybe there is a better data structure?) and I am running the two websocket connections in two different processes with multprocessing. I am defining two shared-in-memory dictionaries, asks and bids which will store the asks and bids side as a dictionary of a dictionary.
So for example, asks = {'ZRX/USD':{best_askZRX: quantity_bestZRX, second_best_askZRX: quantity_second_bestZRX,..},'BTC/USD':{best_askBTC: quantity_bestBTC, second_best_askBTC: quantity_second_bestBTC,..} } and similarly for the bids.
In the main function, I define the dictionaries and the two processes
manager = Manager()
asks = manager.dict()
bids = manager.dict()
asks['ZRX/USD'] = manager.dict()
bids['ZRX/USD'] = manager.dict()
asks['BTC/USD'] = manager.dict()
bids['BTC/USD'] = manager.dict()
extract_data1 = Process(target=get_orderbook, args=('ZRX/USD', asks, bids))
extract_data2 = Process(target=get_orderbook, args=('BTC/USD', asks, bids))
Each process calls the function get_orderbook defined in the code. Then I run the processes. However, the processes do not run in parallel, but extract_data1 starts first and extract_data2 starts only when I manually kill extract_data1. See the code below. What is happening here? What am I doing wrong?
from multiprocessing import Process, Manager
import websocket, json
import time
def on_open(ws, market):
print('opened connection')
subscribe_message = {'op': 'subscribe', 'channel': 'orderbook', 'market': market}
print(subscribe_message)
ws.send(json.dumps(subscribe_message))
def init_orderbook(ws, asks, bids, market, asks_snapshot, bids_snapshot):
proxy_asks, proxy_bids = asks, bids
for level in asks_snapshot:
proxy_asks[level[0]] = level[1]
for level in bids_snapshot:
proxy_bids[level[0]] = level[1]
asks, bids = proxy_asks, proxy_bids
def apply_changes(ws, asks, bids, market, asks_snapshot, bids_snapshot):
proxy_asks, proxy_bids = asks, bids
for level in asks_snapshot:
if level[1] == 0:
del proxy_asks[level[0]]
else:
proxy_asks[level[0]] = level[1]
for level in bids_snapshot:
if level[1] == 0:
del proxy_bids[level[0]]
else:
proxy_bids[level[0]] = level[1]
asks, bids = proxy_asks, proxy_bids
def on_message(ws, message, asks, bids):
js = json.loads(message)
if js['type'] == 'subscribed':
print('Subscribed, ', js)
elif js['type'] == 'update':
market = js['market']
update_asks = js['data']['asks']
update_bids = js['data']['asks']
apply_changes(ws, asks[market], bids[market], market, update_asks, update_bids)
elif js['type'] == 'partial':
print('Storing snapshot, ', js)
market = js['market']
update_asks = js['data']['asks']
update_bids = js['data']['asks']
init_orderbook(ws, asks[market], bids[market], market, update_asks, update_bids)
def get_orderbook(market, asks, bids):
socket = "wss://ftx.com/ws/"
ws = websocket.WebSocketApp(socket)
ws.on_open = lambda *x: on_open(*x, market)
ws.on_message = lambda ws, msg: on_message(ws, msg, asks, bids)
ws.run_forever()
if __name__ == '__main__':
manager = Manager()
asks = manager.dict()
bids = manager.dict()
asks['ZRX/USD'] = manager.dict()
bids['ZRX/USD'] = manager.dict()
asks['BTC/USD'] = manager.dict()
bids['BTC/USD'] = manager.dict()
extract_data1 = Process(target=get_orderbook, args=('ZRX/USD', asks, bids))
extract_data2 = Process(target=get_orderbook, args=('BTC/USD', asks, bids))
extract_data1.run()
extract_data2.run()
extract_data1.join()
extract_data2.join()