bmexwsandccxt.py

import asyncio
from pyparsing import null_debug_action
import websockets
import json
from datetime import datetime
import streamlit as st
import time
import plotly.express as px
import pandas as pd
from itertools import accumulate
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import ccxt
import sys
import websocket
import threading
import traceback
import ssl
from time import sleep
import json
import decimal
import logging
import plotly.express as px
import pandas as pd
import numpy as np
import streamlit as st
# from market_maker.settings import settings
# from market_maker.auth.APIKeyAuth import generate_expires, generate_signature
# from market_maker.utils import log
# from market_maker.utils.math import toNearest
from future.utils import iteritems
from future.standard_library import hooks
with hooks():  # Python 2/3 compat
    from urllib.parse import urlparse, urlunparse


# logger = log.setup_custom_logger('root')


# Connects to BitMEX websocket for streaming realtime data.
# The Marketmaker still interacts with this as if it were a REST Endpoint, but now it can get
# much more realtime data without heavily polling the API.
#
# The Websocket offers a bunch of data as raw properties right on the object.
# On connect, it synchronously asks for a push of all this data then returns.
# Right after, the MM can start using its data. It will be updated in realtime, so the MM can
# poll as often as it wants.
class BitMEXWebsocket():

    # Don't grow a table larger than this amount. Helps cap memory usage.
    MAX_TABLE_LEN = 200

    def __init__(self):
        self.__reset()

    def __del__(self):
        self.exit()

    def connect(self, endpoint="", symbol="XBTN15", shouldAuth=True):
        '''Connect to the websocket and initialize data stores.'''

        logger.debug("Connecting WebSocket.")
        self.symbol = symbol
        self.shouldAuth = shouldAuth

        # We can subscribe right in the connection querystring, so let's build that.
        # Subscribe to all pertinent endpoints
        subscriptions = [sub + ':' + symbol for sub in ["quote", "trade"]]
        subscriptions += ["instrument"]  # We want all of them
        if self.shouldAuth:
            subscriptions += [sub + ':' + symbol for sub in ["order", "execution"]]
            subscriptions += ["margin", "position"]

        # Get WS URL and connect.
        urlParts = list(urlparse(endpoint))
        urlParts[0] = urlParts[0].replace('http', 'ws')
        urlParts[2] = "/realtime?subscribe=" + ",".join(subscriptions)
        wsURL = urlunparse(urlParts)
        logger.info("Connecting to %s" % wsURL)
        self.__connect(wsURL)
        logger.info('Connected to WS. Waiting for data images, this may take a moment...')

        # Connected. Wait for partials
        self.__wait_for_symbol(symbol)
        logger.info("%s received. Waiting for account...", symbol)
        if self.shouldAuth:
            self.__wait_for_account()
        logger.info('Got all market data. Starting.')

    #
    # Data methods
    #
    def get_instrument(self, symbol):
        instruments = self.data['instrument']
        matchingInstruments = [i for i in instruments if i['symbol'] == symbol]
        if len(matchingInstruments) == 0:
            raise Exception("Unable to find instrument or index with symbol: " + symbol)
        instrument = matchingInstruments[0]
        # Turn the 'tickSize' into 'tickLog' for use in rounding
        # http://stackoverflow.com/a/6190291/832202
        instrument['tickLog'] = decimal.Decimal(str(instrument['tickSize'])).as_tuple().exponent * -1
        return instrument

    def get_ticker(self, symbol):
        '''Return a ticker object. Generated from instrument.'''

        instrument = self.get_instrument(symbol)

        # If this is an index, we have to get the data from the last trade.
        if instrument['symbol'][0] == '.':
            ticker = {}
            ticker['mid'] = ticker['buy'] = ticker['sell'] = ticker['last'] = instrument['markPrice']
        # Normal instrument
        else:
            bid = instrument['bidPrice'] or instrument['lastPrice']
            ask = instrument['askPrice'] or instrument['lastPrice']
            ticker = {
                "last": instrument['lastPrice'],
                "buy": bid,
                "sell": ask,
                "mid": (bid + ask) / 2
            }

        # The instrument has a tickSize. Use it to round values.
        return {k: round(float(v or 0), instrument['tickSize']) for k, v in iteritems(ticker)}

    # def funds(self):
    #     return self.data['margin'][0]

    # def market_depth(self, symbol):
    #     raise NotImplementedError('orderBook is not subscribed; use askPrice and bidPrice on instrument')
    #     # return self.data['orderBook25'][0]

    # def open_orders(self, clOrdIDPrefix):
    #     orders = self.data['order']
    #     # Filter to only open orders (leavesQty > 0) and those that we actually placed
    #     return [o for o in orders if str(o['clOrdID']).startswith(clOrdIDPrefix) and o['leavesQty'] > 0]

    # def position(self, symbol):
    #     positions = self.data['position']
    #     pos = [p for p in positions if p['symbol'] == symbol]
    #     if len(pos) == 0:
    #         # No position found; stub it
    #         return {'avgCostPrice': 0, 'avgEntryPrice': 0, 'currentQty': 0, 'symbol': symbol}
    #     return pos[0]

    def recent_trades(self):
        return self.data['trade']

    #
    # Lifecycle methods
    #
    def error(self, err):
        self._error = err
        logger.error(err)
        self.exit()

    def exit(self):
        self.exited = True
        self.ws.close()

    #
    # Private methods
    #

    def __connect(self, wsURL):
        '''Connect to the websocket in a thread.'''
        logger.debug("Starting thread")

        ssl_defaults = ssl.get_default_verify_paths()
        sslopt_ca_certs = {'ca_certs': ssl_defaults.cafile}
        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=self.__get_auth()
                                         )

        self.wst = threading.Thread(target=lambda: self.ws.run_forever(sslopt=sslopt_ca_certs))
        self.wst.daemon = True
        self.wst.start()
        logger.info("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while (not self.ws.sock or not self.ws.sock.connected) and conn_timeout and not self._error:
            sleep(1)
            conn_timeout -= 1

        if not conn_timeout or self._error:
            logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            sys.exit(1)

    # def __get_auth(self):
    #     '''Return auth headers. Will use API Keys if present in settings.'''

    #     if self.shouldAuth is False:
    #         return []

    #     logger.info("Authenticating with API Key.")
    #     # To auth to the WS using an API key, we generate a signature of a nonce and
    #     # the WS API endpoint.
    #     nonce = generate_expires()
    #     return [
    #         "api-expires: " + str(nonce),
    #         "api-signature: " + generate_signature(settings.API_SECRET, 'GET', '/realtime', nonce, ''),
    #         "api-key:" + settings.API_KEY
    #     ]

    # def __wait_for_account(self):
    #     '''On subscribe, this data will come down. Wait for it.'''
    #     # Wait for the keys to show up from the ws
    #     while not {'margin', 'position', 'order'} <= set(self.data):
    #         sleep(0.1)

    # def __wait_for_symbol(self, symbol):
    #     '''On subscribe, this data will come down. Wait for it.'''
    #     while not {'instrument', 'quote'} <= set(self.data):
    #         sleep(0.1)

    # def __send_command(self, command, args):
    #     '''Send a raw command.'''
    #     self.ws.send(json.dumps({"op": command, "args": args or []}))

    # 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())

    def __on_open(self):
        logger.debug("Websocket Opened.")

    def __on_close(self):
        logger.info('Websocket Closed')
        self.exit()

    def __on_error(self, error):
        if not self.exited:
            self.error(error)

    def __reset(self):
        self.data = {}
        self.keys = {}
        self.exited = False
        self._error = None


def findItemByKeys(keys, table, matchData):
    for item in table:
        matched = True
        for key in keys:
            if item[key] != matchData[key]:
                matched = False
        if matched:
            return item
if __name__ == "__main__":
    # create console handler and set level to debug
    logger = log.setup_custom_logger('websocket', logging.DEBUG)
    ws = BitMEXWebsocket()
    ws.connect("https://testnet.bitmex.com/api/v1")
    while(ws.ws.sock.connected):
        sleep(1)

# st.set_page_config(layout="wide")

# # import plotly.express as px

# # df = pd.DataFrame(columns = ['id', 'price', 'size', 'side', 'liquidation', 'time'])
# # placeholder1 = st.empty()
# # for seconds in range(200):
# # while True: 
# # a = "Subscribed to orderbook"

# # b = "fat d8ta"

# dict_dumps = {"op": "subscribe", "args": ["orderBookL2:LINK_USDT"]}

# # name = st.text_input("market name", "SPY/USD")
# # dict_dumps["market"] = name

# placeholder1 = st.empty()
# placeholder2 = st.empty()
# placeholder3 = st.empty()
# placeholder4 = st.empty()
# placeholder5 = st.empty()
# placeholder6 = st.empty()

# async def consumer() -> None:
#     async with websockets.connect("wss://ws.testnet.bitmex.com/realtime", ping_interval=20, ping_timeout=2000) as websocket:
#         await websocket.send(
#             json.dumps(
#                 dict_dumps
#             )
#         )
#         async for message in websocket:
#             # global a
#             # global b

#             message = json.loads(message)

                            
                            
#             '''          {"op": "subscribe", "args": ["orderBookL2_25:XBTUSD"]}
#                          {"success":true,"subscribe":"orderBookL2_25:XBTUSD","request":{"op":"subscribe","args":["orderBookL2_25:XBTUSD"]}}
#                          {
#                             "table":"orderBookL2_25",
#                             "keys":["symbol","id","side"],
#                             "types":{"id":"long","price":"float","side":"symbol","size":"long","symbol":"symbol","timestamp":"timestamp"}
#                             "foreignKeys":{"side":"side","symbol":"instrument"},
#                             "attributes":{"id":"sorted","symbol":"grouped"},
#                             "action":"partial",
#                             "data":[
#                                 {"symbol":"XBTUSD","id":17999992000,"side":"Sell","size":100,"price":80,"timestamp":"2022-02-09T11:23:06.802Z"},
#                                 {"symbol":"XBTUSD","id":17999993000,"side":"Sell","size":20,"price":70,"timestamp":"2022-02-09T11:23:06.802Z"},
#                                 {"symbol":"XBTUSD","id":17999994000,"side":"Sell","size":10,"price":60,"timestamp":"2022-02-09T11:23:06.802Z"},
#                                 {"symbol":"XBTUSD","id":17999995000,"side":"Buy","size":10,"price":50,"timestamp":"2022-02-09T11:23:06.802Z"},
#                                 {"symbol":"XBTUSD","id":17999996000,"side":"Buy","size":20,"price":40,"timestamp":"2022-02-09T11:23:06.802Z"},
#                                 {"symbol":"XBTUSD","id":17999997000,"side":"Buy","size":100,"price":30,"timestamp":"2022-02-09T11:23:06.802Z"}
#                             ]
#                             }
#                         < {
#                             "table":"orderBookL2_25",
#                             "action":"update",
#                             "data":[
#                                 {"symbol":"XBTUSD","id":17999995000,"side":"Buy","size":5}
#                             ]
#                             }
#                         < {
#                             "table":"orderBookL2_25",
#                             "action":"delete",
#                             "data":[
#                                 {"symbol":"XBTUSD","id":17999995000,"side":"Buy"}
#                             ]
#                             }
#                         < {
#                             "table":"orderBookL2_25",
#                             "action":"insert",
#                             "data":[
#                                 {"symbol":"XBTUSD","id":17999995500,"side":"Buy","size":10,"price":45},
#                             ]
#                             }
#                 '''
#             st.write(message)





#     # await asyncio.sleep(30)
            
# asyncio.run(consumer())