market_maker/ws/ws_thread.py
def __on_message(self, message):
        '''Handler for parsing WS messages.'''
        message = json.loads(message)
        logger.debug(json.dumps(message))

        table = message['table'] if 'table' in message else None
        action = message['action'] if 'action' in message else None
        try:
            if 'subscribe' in message:
                if message['success']:
                    logger.debug("Subscribed to %s." % message['subscribe'])
                else:
                    self.error("Unable to subscribe to %s. Error: \"%s\" Please check and restart." %
                               (message['request']['args'][0], message['error']))
            elif 'status' in message:
                if message['status'] == 400:
                    self.error(message['error'])
                if message['status'] == 401:
                    self.error("API Key incorrect, please check and restart.")
            elif action:

                if table not in self.data:
                    self.data[table] = []

                if table not in self.keys:
                    self.keys[table] = []

                # There are four possible actions from the WS:
                # 'partial' - full table image
                # 'insert'  - new row
                # 'update'  - update row
                # 'delete'  - delete row
                if action == 'partial':
                    logger.debug("%s: partial" % table)
                    self.data[table] += message['data']
                    # Keys are communicated on partials to let you know how to uniquely identify
                    # an item. We use it for updates.
                    self.keys[table] = message['keys']
                elif action == 'insert':
                    logger.debug('%s: inserting %s' % (table, message['data']))
                    self.data[table] += message['data']

                    # Limit the max length of the table to avoid excessive memory usage.
                    # Don't trim orders because we'll lose valuable state if we do.
                    if table not in ['order', 'orderBookL2'] and len(self.data[table]) > BitMEXWebsocket.MAX_TABLE_LEN:
                        self.data[table] = self.data[table][(BitMEXWebsocket.MAX_TABLE_LEN // 2):]

                elif action == 'update':
                    logger.debug('%s: updating %s' % (table, message['data']))
                    # Locate the item in the collection and update it.
                    for updateData in message['data']:
                        item = findItemByKeys(self.keys[table], self.data[table], updateData)
                        if not item:
                            continue  # No item found to update. Could happen before push

                        # Log executions
                        if table == 'order':
                            is_canceled = 'ordStatus' in updateData and updateData['ordStatus'] == 'Canceled'
                            if 'cumQty' in updateData and not is_canceled:
                                contExecuted = updateData['cumQty'] - item['cumQty']
                                if contExecuted > 0:
                                    instrument = self.get_instrument(item['symbol'])
                                    logger.info("Execution: %s %d Contracts of %s at %.*f" %
                                             (item['side'], contExecuted, item['symbol'],
                                              instrument['tickLog'], item['price'] or updateData['price']))

                        # Update this item.
                        item.update(updateData)

                        # Remove canceled / filled orders
                        if table == 'order' and item['leavesQty'] <= 0:
                            self.data[table].remove(item)

                elif action == 'delete':
                    logger.debug('%s: deleting %s' % (table, message['data']))
                    # Locate the item in the collection and remove it.
                    for deleteData in message['data']:
                        item = findItemByKeys(self.keys[table], self.data[table], deleteData)
                        self.data[table].remove(item)
                else:
                    raise Exception("Unknown action: %s" % action)
        except:
            logger.error(traceback.format_exc())
Handler for parsing WS messages.
Handler for parsing WS messages.
Handler for parsing WS messages .
def __on_message(self, message):
        message = json.loads(message)
        logger.debug(json.dumps(message))
        table = message['table'] if 'table' in message else None
        action = message['action'] if 'action' in message else None
        try:
            if 'subscribe' in message:
                if message['success']:
                    logger.debug("Subscribed to %s." % message['subscribe'])
                else:
                    self.error("Unable to subscribe to %s. Error: \"%s\" Please check and restart." %
                               (message['request']['args'][0], message['error']))
            elif 'status' in message:
                if message['status'] == 400:
                    self.error(message['error'])
                if message['status'] == 401:
                    self.error("API Key incorrect, please check and restart.")
            elif action:
                if table not in self.data:
                    self.data[table] = []
                if table not in self.keys:
                    self.keys[table] = []
                if action == 'partial':
                    logger.debug("%s: partial" % table)
                    self.data[table] += message['data']
                    self.keys[table] = message['keys']
                elif action == 'insert':
                    logger.debug('%s: inserting %s' % (table, message['data']))
                    self.data[table] += message['data']
                    if table not in ['order', 'orderBookL2'] and len(self.data[table]) > BitMEXWebsocket.MAX_TABLE_LEN:
                        self.data[table] = self.data[table][(BitMEXWebsocket.MAX_TABLE_LEN // 2):]
                elif action == 'update':
                    logger.debug('%s: updating %s' % (table, message['data']))
                    for updateData in message['data']:
                        item = findItemByKeys(self.keys[table], self.data[table], updateData)
                        if not item:
                            continue  
                        if table == 'order':
                            is_canceled = 'ordStatus' in updateData and updateData['ordStatus'] == 'Canceled'
                            if 'cumQty' in updateData and not is_canceled:
                                contExecuted = updateData['cumQty'] - item['cumQty']
                                if contExecuted > 0:
                                    instrument = self.get_instrument(item['symbol'])
                                    logger.info("Execution: %s %d Contracts of %s at %.*f" %
                                             (item['side'], contExecuted, item['symbol'],
                                              instrument['tickLog'], item['price'] or updateData['price']))
                        item.update(updateData)
                        if table == 'order' and item['leavesQty'] <= 0:
                            self.data[table].remove(item)
                elif action == 'delete':
                    logger.debug('%s: deleting %s' % (table, message['data']))
                    for deleteData in message['data']:
                        item = findItemByKeys(self.keys[table], self.data[table], deleteData)
                        self.data[table].remove(item)
                else:
                    raise Exception("Unknown action: %s" % action)
        except:
            logger.error(traceback.format_exc())
def __on_message ( self , message ) : message = json . loads ( message ) logger . debug ( json . dumps ( message ) ) table = message [ 'table' ] if 'table' in message else None action = message [ 'action' ] if 'action' in message else None try : if 'subscribe' in message : if message [ 'success' ] : logger . debug ( "Subscribed to %s." % message [ 'subscribe' ] ) else : self . error ( "Unable to subscribe to %s. Error: \"%s\" Please check and restart." % ( message [ 'request' ] [ 'args' ] [ 0 ] , message [ 'error' ] ) ) elif 'status' in message : if message [ 'status' ] == 400 : self . error ( message [ 'error' ] ) if message [ 'status' ] == 401 : self . error ( "API Key incorrect, please check and restart." ) elif action : if table not in self . data : self . data [ table ] = [ ] if table not in self . keys : self . keys [ table ] = [ ] if action == 'partial' : logger . debug ( "%s: partial" % table ) self . data [ table ] += message [ 'data' ] self . keys [ table ] = message [ 'keys' ] elif action == 'insert' : logger . debug ( '%s: inserting %s' % ( table , message [ 'data' ] ) ) self . data [ table ] += message [ 'data' ] if table not in [ 'order' , 'orderBookL2' ] and len ( self . data [ table ] ) > BitMEXWebsocket . MAX_TABLE_LEN : self . data [ table ] = self . data [ table ] [ ( BitMEXWebsocket . MAX_TABLE_LEN // 2 ) : ] elif action == 'update' : logger . debug ( '%s: updating %s' % ( table , message [ 'data' ] ) ) for updateData in message [ 'data' ] : item = findItemByKeys ( self . keys [ table ] , self . data [ table ] , updateData ) if not item : continue if table == 'order' : is_canceled = 'ordStatus' in updateData and updateData [ 'ordStatus' ] == 'Canceled' if 'cumQty' in updateData and not is_canceled : contExecuted = updateData [ 'cumQty' ] - item [ 'cumQty' ] if contExecuted > 0 : instrument = self . get_instrument ( item [ 'symbol' ] ) logger . info ( "Execution: %s %d Contracts of %s at %.*f" % ( item [ 'side' ] , contExecuted , item [ 'symbol' ] , instrument [ 'tickLog' ] , item [ 'price' ] or updateData [ 'price' ] ) ) item . update ( updateData ) if table == 'order' and item [ 'leavesQty' ] <= 0 : self . data [ table ] . remove ( item ) elif action == 'delete' : logger . debug ( '%s: deleting %s' % ( table , message [ 'data' ] ) ) for deleteData in message [ 'data' ] : item = findItemByKeys ( self . keys [ table ] , self . data [ table ] , deleteData ) self . data [ table ] . remove ( item ) else : raise Exception ( "Unknown action: %s" % action ) except : logger . error ( traceback . format_exc ( ) )
Handler for parsing WS messages.
Handler for parsing WS messages .
'''Handler for parsing WS messages.''' # There are four possible actions from the WS: # 'partial' - full table image # 'insert'  - new row # 'update'  - update row # 'delete'  - delete row # Keys are communicated on partials to let you know how to uniquely identify # an item. We use it for updates. # Limit the max length of the table to avoid excessive memory usage. # Don't trim orders because we'll lose valuable state if we do. # Locate the item in the collection and update it. # No item found to update. Could happen before push # Log executions # Update this item. # Remove canceled / filled orders # Locate the item in the collection and remove it.
self
message