Merge pull request #40 from slazarov/0.0.6.3

0.0.6.3

README.md

@@ -61,20 +61,7 @@ I have been largely motivated by the following projects and people:
 # Road map
 
 ### Notices
-Version 0.0.6 will change the names of the unsubscribe methods from:
-```
-unsubscribe_to_orderbook
-unsubscribe_to_orderbook_update
-unsubscribe_to_trades
-unsubscribe_to_ticker_update
-```
-to
-```
-unsubscribe_from_orderbook
-unsubscribe_from_orderbook_update
-unsubscribe_from_trades
-unsubscribe_from_ticker_update
-```
+None right now.
 
 ### Currently in development
 * Socket reconnection handling
@@ -426,6 +413,10 @@ if __name__ == "__main__":
     main()
 ```
 # Change log
+0.0.6.3 - 18/02/2018
+* Major changes to how the code handles order book syncing. Syncing is done significantly faster than previous versions, i.e full sync of all Bittrex tickers takes ca. 4 minutes.
+* Fixed `on_open` bug as per [Issue #21](https://github.com/slazarov/python-bittrex-websocket/issues/21)
+
 0.0.6.2.2
 * Update cfscrape>=1.9.2 and gevent>=1.3a1
 * Reorder imports in websocket_client to safeguard against SSL recursion errors.

bittrex_websocket/_auxiliary.py

@@ -40,7 +40,7 @@ def _create_structure(self):
         d = \
             {
                 SUB_TYPE_ORDERBOOK: dict(self._set_default_subscription(),
-                                         **{'SnapshotState': 0,
+                                         **{'SnapshotState': SNAPSHOT_OFF,
                                             'OrderBookDepth': 10,
                                             'NouncesRcvd': 0,
                                             'InternalQueue': None}),
@@ -297,7 +297,7 @@ def get_nounces(self, ticker):
 
     def reset_snapshot(self, ticker):
         self.list[ticker][SUB_TYPE_ORDERBOOK]['NouncesRcvd'] = 0
-        self.list[ticker][SUB_TYPE_ORDERBOOK]['SnapshotState'] = 0
+        self.list[ticker][SUB_TYPE_ORDERBOOK]['SnapshotState'] = SNAPSHOT_OFF
         self.list[ticker][SUB_TYPE_ORDERBOOK]['InternalQueue'] = None
         logger.info(
             '[Subscription][{}][{}]: Snapshot nounce, state and internal queue are reset.'.format(SUB_TYPE_ORDERBOOK,

bittrex_websocket/constants.py

@@ -7,7 +7,8 @@
 INVALID_SUB = 'Subscription type is invalid or not implemented. ' \
               'Available options: OrderBook, OrderBookUpdate, Trades'
 INVALID_SUB_CHANGE = 'Subscription change is invalid. Available options: True/False'
-SNAPSHOT_OFF = 0  # 'Not initiated'
+SNAPSHOT_OFF = -1  # 'Not initiated'
+SNAPSHOT_QUEUED = 0  # Sent to queue for processing
 SNAPSHOT_SENT = 1  # Invoked, not processed
 SNAPSHOT_RCVD = 2  # Received, not processed
 SNAPSHOT_ON = 3  # Received, processed
@@ -35,6 +36,8 @@
 MSG_INFO_CONN_ESTABLISHING = _CONN_PREFIX + 'Trying to establish connection to Bittrex through {}.'
 MSG_INFO_RECONNECT = _SUB_PREFIX + 'Initiating reconnection procedure.'
 MSG_INFO_CONN_INIT_RECONNECT = _CONN_PREFIX + 'Initiating reconnection procedure for all relevant subscriptions.'
+NSG_INFO_ORDER_BOOK_REQUESTED = _SUB_PREFIX + 'Order book snapshot requested.'
+NSG_INFO_ORDER_BOOK_RECEIVED = _SUB_PREFIX + 'Order book snapshot synced.'
 MSG_ERROR_CONN_SOCKET = _CONN_PREFIX + 'Timeout for url {}. Please check your internet connection is on.'
 MSG_ERROR_CONN_FAILURE = _CONN_PREFIX + 'Failed to establish connection through supplied URLS. Leaving to ' \
                                         'watchdog...'

bittrex_websocket/websocket_client.py

@@ -273,7 +273,11 @@ def _handle_connect(self, conn_event):
         self.threads[thread.getName()] = thread
         conn_event.conn_obj.assign_thread(thread.getName())
         self.connections.update({conn_event.conn_obj.id: conn_event.conn_obj})
-        thread.start()
+        try:
+            thread.start()
+        except WebSocketConnectionClosedException:
+            print(WebSocketBadStatusException)
+            print('Received in _handle_connect. Report to github.')
 
     def _init_connection(self, conn_obj):
         """
@@ -439,8 +443,8 @@ def _handle_subscribe(self, sub_event):
                     for cb in server_callback:
                         for ticker in tickers:
                             conn.corehub.server.invoke(cb, ticker)
-                            if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOK) is True:
-                                self._get_snapshot([ticker])
+                            # if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOK) is True:
+                            #     self._get_snapshot([ticker])
                             conn.increment_ticker()
                 if server_callback_no_payload is not None:
                     if tickers == ALL_TICKERS:
@@ -498,20 +502,13 @@ def _handle_get_snapshot(self, snapshot_event):
         """
         conn, ticker = snapshot_event.conn_object, snapshot_event.ticker
         method = 'queryExchangeState'
-        # Wait for the connection to start successfully and record N nounces of data
-        while conn.state is False or self.tickers.get_nounces(ticker) < 5:
-            sleep(0.1)
-        else:
-            try:
-                logger.info('[Subscription][{}][{}]: Order book snapshot '
-                            'requested.'.format(SUB_TYPE_ORDERBOOK, ticker))
-                conn.corehub.server.invoke(method, ticker)
-                self.tickers.set_snapshot_state(ticker, SNAPSHOT_SENT)
-            except Exception as e:
-                print(e)
-                print('Failed to invoke snapshot query')
-        while self.tickers.get_snapshot_state(ticker) is not SNAPSHOT_ON:
-            sleep(0.5)
+        try:
+            logger.info(NSG_INFO_ORDER_BOOK_REQUESTED.format(SUB_TYPE_ORDERBOOK, ticker))
+            conn.corehub.server.invoke(method, ticker)
+            self.tickers.set_snapshot_state(ticker, SNAPSHOT_SENT)
+        except Exception as e:
+            print(e)
+            print('Failed to invoke snapshot query')
 
     def _handle_reconnect(self, reconnect_event):
         ticker, sub_type, book_depth = reconnect_event.tickers, reconnect_event.sub_type, reconnect_event.book_depth
@@ -597,7 +594,18 @@ def _get_snapshot(self, tickers):
                     conn_id = self.tickers.get_sub_type_conn_id(ticker_name, SUB_TYPE_ORDERBOOK)
                 else:
                     break
-            self.control_queue.put(SnapshotEvent(ticker_name, conn))
+            self.tickers.set_snapshot_state(ticker_name, SNAPSHOT_QUEUED)
+            ### EXPERIMENTAL ###
+            method = 'queryExchangeState'
+            try:
+                logger.info(NSG_INFO_ORDER_BOOK_REQUESTED.format(SUB_TYPE_ORDERBOOK, ticker_name))
+                conn.corehub.server.invoke(method, ticker_name)
+                self.tickers.set_snapshot_state(ticker_name, SNAPSHOT_SENT)
+            except Exception as e:
+                print(e)
+                print('Failed to invoke snapshot query')
+            ###
+            # self.control_queue.put(SnapshotEvent(ticker_name, conn))
 
     def _is_order_queue(self):
         if self.order_queue is None:
@@ -617,7 +625,7 @@ def _start_order_queue(self):
                 if order_event is not None:
                     ticker = order_event['MarketName']
                     snapshot_state = self.tickers.get_snapshot_state(ticker)
-                    if snapshot_state in [SNAPSHOT_OFF, SNAPSHOT_SENT]:
+                    if snapshot_state in [SNAPSHOT_OFF, SNAPSHOT_QUEUED, SNAPSHOT_SENT]:
                         self._init_backorder_queue(ticker, order_event)
                     elif snapshot_state == SNAPSHOT_RCVD:
                         if self._transfer_backorder_queue(ticker):
@@ -930,16 +938,21 @@ def _is_orderbook_snapshot(self, msg):
         # Detect if the message contains order book snapshots and manipulate them.
         if 'R' in msg and type(msg['R']) is not bool:
             if 'MarketName' in msg['R'] and msg['R']['MarketName'] is None:
-                for ticker in self.tickers.list.values():
-                    if ticker[SUB_TYPE_ORDERBOOK]['SnapshotState'] == SNAPSHOT_SENT:
-                        msg['R']['MarketName'] = ticker['Name']
-                        del msg['R']['Fills']
-                        self.order_books[ticker['Name']] = msg['R']
-                        self.tickers.set_snapshot_state(ticker['Name'], SNAPSHOT_RCVD)
-                        break
-                logger.info(
-                    '[Subscription][{}][{}]: Order book snapshot received.'.format(SUB_TYPE_ORDERBOOK,
-                                                                                   msg['R']['MarketName']))
+                thread_name = current_thread().getName()
+                conn_id = self._return_conn_by_thread_name(thread_name).id
+                subs = self.tickers.sort_by_conn_id(conn_id)['OrderBook']
+                for ticker in subs.keys():
+                    # for ticker in self.tickers.list.values():
+                    if self.tickers.get_snapshot_state(ticker) is SNAPSHOT_SENT:
+                        # if ticker[SUB_TYPE_ORDERBOOK]['SnapshotState'] == SNAPSHOT_SENT:
+                        ### experimental - confirm
+                        if self._transfer_backorder_queue2(ticker, msg['R']):
+                            msg['R']['MarketName'] = ticker
+                            del msg['R']['Fills']
+                            self.order_books[ticker] = msg['R']
+                            self.tickers.set_snapshot_state(ticker, SNAPSHOT_RCVD)
+                            break
+                logger.info(NSG_INFO_ORDER_BOOK_RECEIVED.format(SUB_TYPE_ORDERBOOK, msg['R']['MarketName']))
 
     def _init_backorder_queue(self, ticker, msg):
         sub = self.tickers.list[ticker][SUB_TYPE_ORDERBOOK]
@@ -966,6 +979,27 @@ def _transfer_backorder_queue(self, ticker):
                     self.tickers.set_snapshot_state(ticker, SNAPSHOT_ON)
                 q.task_done()
 
+    def _transfer_backorder_queue2(self, ticker, snapshot):
+        confirmed = False
+        sub = self.tickers.list[ticker][SUB_TYPE_ORDERBOOK]
+        q = sub['InternalQueue']
+        q2 = queue.Queue()
+        while True:
+            try:
+                e = q.get(False)
+                q2.put(e)
+            except queue.Empty:
+                sub['InternalQueue'] = q2
+                return confirmed
+            except AttributeError:
+                raise NotImplementedError('Please report error to '
+                                          'https://github.com/slazarov/python-bittrex-websocket, '
+                                          'Error:_transfer_backorder_queue:AttributeError')
+            else:
+                if self._confirm_order_book(snapshot, e):
+                    confirmed = True
+                q.task_done()
+
     # ========================
     # Private Channels Methods
     # ========================
@@ -976,48 +1010,64 @@ def _on_debug(self, **kwargs):
         Don't edit unless you know what you are doing.
         Redirect full order book snapshots to on_message
         """
-        if self._is_close_me():
-            return
-        self._is_orderbook_snapshot(kwargs)
+        try:
+            if self._is_close_me():
+                return
+            self._is_orderbook_snapshot(kwargs)
+        except Exception as e:
+            print(e)
+            print('Got this exception from _on_debug. This is bug testing. Please report to '
+                  'https://github.com/slazarov/python-bittrex-websocket with this message')
 
     def _on_tick_update(self, msg):
-        if self._is_close_me():
-            return
-        ticker = msg['MarketName']
-        subs = self.tickers.get_ticker_subs(ticker)
-        if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOK) is SUB_STATE_ON:
-            self.order_queue.put(msg)
-        if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOKUPDATE) is SUB_STATE_ON:
-            d = dict(self._create_base_layout(msg),
-                     **{'bids': msg['Buys'],
-                        'asks': msg['Sells']})
-            self.orderbook_update.on_change(d)
-        if self.tickers.get_sub_state(ticker, SUB_TYPE_TRADES) is SUB_STATE_ON:
-            if msg['Fills']:
+        try:
+            if self._is_close_me():
+                return
+            ticker = msg['MarketName']
+            if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOK) is SUB_STATE_ON:
+                if self.tickers.get_snapshot_state(ticker) is SNAPSHOT_OFF:
+                    self._get_snapshot([ticker])
+                self.order_queue.put(msg)
+            if self.tickers.get_sub_state(ticker, SUB_TYPE_ORDERBOOKUPDATE) is SUB_STATE_ON:
                 d = dict(self._create_base_layout(msg),
-                         **{'trades': msg['Fills']})
-                self.trades.on_change(d)
+                         **{'bids': msg['Buys'],
+                            'asks': msg['Sells']})
+                self.orderbook_update.on_change(d)
+            if self.tickers.get_sub_state(ticker, SUB_TYPE_TRADES) is SUB_STATE_ON:
+                if msg['Fills']:
+                    d = dict(self._create_base_layout(msg),
+                             **{'trades': msg['Fills']})
+                    self.trades.on_change(d)
+        except Exception as e:
+            print(e)
+            print('Got this exception from _on_tick_update. This is bug testing. Please report to '
+                  'https://github.com/slazarov/python-bittrex-websocket with this message')
 
     def _on_ticker_update(self, msg):
         """
         Invoking summary state updates for specific filter
         doesn't work right now. So we will filter them manually.
         """
-        if self._is_close_me():
-            return
-        if 'Deltas' in msg:
-            for update in msg['Deltas']:
-                if self.tickers.get_sub_state(ALL_TICKERS, SUB_TYPE_TICKERUPDATE) is SUB_STATE_ON:
-                    self.updateSummaryState.on_change(msg['Deltas'])
-                else:
-                    try:
-                        ticker = update['MarketName']
-                        subs = self.tickers.get_ticker_subs(ticker)
-                    except KeyError:  # not in the subscription list
-                        continue
+        try:
+            if self._is_close_me():
+                return
+            if 'Deltas' in msg:
+                for update in msg['Deltas']:
+                    if self.tickers.get_sub_state(ALL_TICKERS, SUB_TYPE_TICKERUPDATE) is SUB_STATE_ON:
+                        self.updateSummaryState.on_change(msg['Deltas'])
                     else:
-                        if subs['TickerUpdate']['Active']:
-                            self.updateSummaryState.on_change(update)
+                        try:
+                            ticker = update['MarketName']
+                            subs = self.tickers.get_ticker_subs(ticker)
+                        except KeyError:  # not in the subscription list
+                            continue
+                        else:
+                            if subs['TickerUpdate']['Active']:
+                                self.updateSummaryState.on_change(update)
+        except Exception as e:
+            print(e)
+            print('Got this exception from _on_ticker_update. This is bug testing. Please report to '
+                  'https://github.com/slazarov/python-bittrex-websocket with this message')
 
     # -------------------------------------
     # Private Channels Supplemental Methods
@@ -1101,6 +1151,43 @@ def _sync_order_book(self, ticker, order_data):
             event = ReconnectEvent([ticker], SUB_TYPE_ORDERBOOK, book_depth)
             self.control_queue.put(event)
 
+    def _confirm_order_book(self, snapshot, nounce_data):
+        # Syncs the order book for the pair, given the most recent data from the socket
+        nounce_diff = nounce_data['Nounce'] - snapshot['Nounce']
+        if nounce_diff == 1:
+            # Start confirming
+            for side in [['Buys', True], ['Sells', False]]:
+                made_change = False
+                for item in nounce_data[side[0]]:
+                    # TYPE 1: Cancelled / filled order entries at matching price
+                    # -> DELETE from the order book
+                    if item['Type'] == 1:
+                        for i, existing_order in enumerate(
+                                self.order_books[snapshot][side[0]]):
+                            if existing_order['Rate'] == item['Rate']:
+                                del self.order_books[snapshot][side[0]][i]
+                                made_change = True
+                                break
+
+                    # TYPE 2: Changed order entries at matching price (partial fills, cancellations)
+                    # -> EDIT the order book
+                    elif item['Type'] == 2:
+                        for existing_order in self.order_books[snapshot][side[0]]:
+                            if existing_order['Rate'] == item['Rate']:
+                                existing_order['Quantity'] = item['Quantity']
+                                made_change = True
+                                break
+                if made_change:
+                    return True
+                else:
+                    return False
+        # The next nounce will trigger a sync.
+        elif nounce_diff == 0:
+            return True
+        # The order book snapshot nounce is ahead. Discard this nounce.
+        elif nounce_diff < 0:
+            return False
+
     def _is_close_me(self):
         thread_name = current_thread().getName()
         conn_object = self._return_conn_by_thread_name(thread_name)
@@ -1134,8 +1221,13 @@ def on_close(self):
 
     def on_error(self, error):
         # Error handler
-        print(error)
-        self.disconnect()
+        try:
+            print(error)
+            self.disconnect()
+        except Exception as e:
+            print(e)
+            print('Got this exception from on_error. This is bug testing. Please report to '
+                  'https://github.com/slazarov/python-bittrex-websocket with this message')
 
     def on_orderbook(self, msg):
         # The main channel of subscribe_to_orderbook().

requirements.txt

@@ -1,4 +1,4 @@
-cfscrape>=1.9.2
+cfscrape>=1.9.4
 signalr-client==0.0.7
 requests[security]==2.18.4
 Events==0.3

setup.py

@@ -4,7 +4,7 @@
 
 install_requires = \
     [
-        'cfscrape>=1.9.2',
+        'cfscrape>=1.9.4',
         'signalr-client==0.0.7',
         'requests[security]==2.18.4',
         'Events==0.3',