仮想通貨マイニングセンター@カブキ

マイニングや仮想通貨,競技プログラミングなど管理人の趣味に関する記事を書いていきます。

Bybitの板情報をPythonで扱いたいbotterの話

みなさんこんにちは.

みなさんはBitMEXクローンの仮想通貨取引所は数多あることをご存知だと思います.

今回はあのMAXも使っているBybitで板情報をwebsocketからとって扱おうとして苦労したことなどについて書いていきます

Bybit公式のWebsocketコネクタを使ってみる

BybitはBitMEXと同じように,公式でPythonコネクタが用意されています.

試しにOrderBook情報を扱おうとこんなコードを書いてみます.

 bybitws = BybitWebsocket(wsURL="wss://stream-testnet.bybit.com/realtime", api_key=None, api_secret=None, symbol='BTCUSD')
    bybitws.subscribe_orderBookL2()

    while True:
        sleep(5)
        print(bybitws.get_data('orderBookL2_25.BTCUSD'))

すると返ってくるのは...

'delete': [{'price': '6774.00', 'symbol': 'BTCUSD', 'id': 67740000, 'side': 'Buy'}], 'update': [], 'insert': [{'price': '6773.50', 'symbol': 'BTCUSD', 'id': 67735000, 'side': 'Buy', 'size': 1633}], 'transactTimeE6': 0}
{'delete': [], 'update': [{'price': '6772.00', 'symbol': 'BTCUSD', 'id': 67720000, 'side': 'Buy', 'size': 35784}], 'insert': [], 'transactTimeE6': 0}
{'delete': [{'price': '6776.50', 'symbol': 'BTCUSD', 'id': 67765000, 'side': 'Buy'}, {'price': '6776.00', 'symbol': 'BTCUSD', 'id': 67760000, 'side': 'Buy'}, {'price': '6777.00', 'symbol': 'BTCUSD', 'id': 67770000, 'side': 'Buy'}], 'update': [{'price': '6787.00', 'symbol': 'BTCUSD', 'id': 67870000, 'side': 'Sell', 'size': 17609}, {'price': '6786.50', 'symbol': 'BTCUSD', 'id': 67865000, 'side': 'Sell', 'size': 21198}, {'price': '6785.00', 'symbol': 'BTCUSD', 'id': 67850000, 'side': 'Sell', 'size': 23983}, {'price': '6775.00', 'symbol': 'BTCUSD', 'id': 67750000, 'side': 'Buy', 'size': 2266}], 'insert': [{'price': '6762.00', 'symbol': 'BTCUSD', 'id': 67620000, 'side': 'Buy', 'size': 8432}, {'price': '6775.50', 'symbol': 'BTCUSD', 'id': 67755000, 'side': 'Buy', 'size': 23871}, {'price': '6777.50', 'symbol': 'BTCUSD', 'id': 67775000, 'side': 'Buy', 'size': 45646}], 'transactTimeE6': 0}
{'delete': [{'price': '6798.50', 'symbol': 'BTCUSD', 'id': 67985000, 'side': 'Sell'}, {'price': '6765.50', 'symbol': 'BTCUSD', 'id': 67655000, 'side': 'Buy'}, {'price': '6765.00', 'symbol': 'BTCUSD', 'id': 67650000, 'side': 'Buy'}], 'update': [{'price': '6778.00', 'symbol': 'BTCUSD', 'id': 67780000, 'side': 'Buy', 'size': 22444}, {'price': '6778.50', 'symbol': 'BTCUSD', 'id': 67785000, 'side': 'Buy', 'size': 37785}, {'price': '6779.00', 'symbol': 'BTCUSD', 'id': 67790000, 'side': 'Buy', 'size': 45131}, {'price': '6786.50', 'symbol': 'BTCUSD', 'id': 67865000, 'side': 'Sell', 'size': 27567}], 'insert': [{'price': '6780.50', 'symbol': 'BTCUSD', 'id': 67805000, 'side': 'Buy', 'size': 40484}, {'price': '6780.00', 'symbol': 'BTCUSD', 'id': 67800000, 'side': 'Buy', 'size': 43470}, {'price': '6787.50', 'symbol': 'BTCUSD', 'id': 67875000, 'side': 'Sell', 'size': 17584}], 'transactTimeE6': 0}

...

...

...

...

...

....

...

そう. 受け取った情報をそのまま流してくるのである!

これを扱いたいBotterは,削除やアップデートの各クエリに対してのプログラムを書かなくてはいけない. BitMEXとは大違いである.

仕方ないので自分で書く

なので,自分の記事 を参考に,板情報をpandasのDataFrame型で返す関数を追加したBybitWebsocketクラスを書いた.

import sys
import websocket
import threading
import traceback
import pandas as pd
from time import sleep
import json
import logging
import urllib
import math
import time
import hmac
from logging import Formatter, getLogger, StreamHandler, DEBUG, INFO, WARNING, basicConfig

# This is a simple adapters for connecting to Bybit's websocket API.
# You could use methods whose names begin with “subscribe”, and get result by "get_data" method.
# All the provided websocket APIs are implemented, includes public and private topic.
# Private methods are only available after authorization, but public methods do not.
# If you have any questions during use, please contact us vim the e-mail "support@bybit.com".


class BybitWebsocket:

    #User can ues MAX_DATA_CAPACITY to control memory usage.
    MAX_DATA_CAPACITY = 200
    PRIVATE_TOPIC = ['position', 'execution', 'order']
    def __init__(self, wsURL, symbol, api_key, api_secret):
        
        self.df = pd.DataFrame()
        self.formatter = Formatter('%(asctime)-15s - %(levelname)-8s - %(message)s')
        self.logger = getLogger(__name__)
        self.handler = StreamHandler(sys.stdout)
        self.handler.setLevel(DEBUG)
        self.handler.setFormatter(self.formatter)
        self.logger.setLevel(DEBUG)
        self.logger.addHandler(self.handler)
        self.logger.debug("Initializing WebSocket.")
        self.symbol = symbol

        if api_key is not None and api_secret is None:
            raise ValueError('api_secret is required if api_key is provided')
        if api_key is None and api_secret is not None:
            raise ValueError('api_key is required if api_secret is provided')

        self.api_key = api_key
        self.api_secret = api_secret

        self.data = {}
        self.exited = False
        self.auth = False
        # We can subscribe right in the connection querystring, so let's build that.
        # Subscribe to all pertinent endpoints
        self.logger.info("Connecting to %s" % wsURL)
        self.__connect(wsURL)
    def exit(self):
        '''Call this to exit - will close websocket.'''
        self.exited = True
        self.ws.close()

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

        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,
                                         keep_running=True)

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

        # Wait for connect before continuing
        retry_times = 5
        while not self.ws.sock or not self.ws.sock.connected and retry_times:
            sleep(1)
            retry_times -= 1
        if retry_times == 0 and not self.ws.sock.connected: 
            self.logger.error("Couldn't connect to WebSocket! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Error!Couldn not connect to WebSocket!.')

        if self.api_key and self.api_secret:
            self.__do_auth()

    def generate_signature(self, expires):
        """Generate a request signature."""
        _val = 'GET/realtime' + expires
        return str(hmac.new(bytes(self.api_secret, "utf-8"), bytes(_val, "utf-8"), digestmod="sha256").hexdigest())

    def __do_auth(self):

        expires = str(int(round(time.time())+1))+"000"
        signature = self.generate_signature(expires)

        auth = {}
        auth["op"] = "auth"
        auth["args"] = [self.api_key, expires, signature]

        args = json.dumps(auth)

        self.ws.send(args)

    def __on_message(self, message):
        '''Handler for parsing WS messages.'''
        message = json.loads(message)

        if 'success' in message and message["success"]:   
            if 'request' in message and message["request"]["op"] == 'auth':
                self.auth = True
                self.logger.info("Authentication success.")
            if 'ret_msg' in message and message["ret_msg"] == 'pong':
                self.data["pong"].append("PING success")

        if 'topic' in message:
            self.data[message["topic"]].append(message["data"])

            if message['topic'] == 'orderBookL2_25.{}'.format(self.symbol):
                typ = message['type']
                data = message['data']
                if typ == 'snapshot':
                    self.df = pd.io.json.json_normalize(message['data'])
                elif typ == 'delta':
                    # self.logger.debug(data)
                    for key, value in data.items():                
                        if key == 'transactTimeE6':
                            continue
                        elif key == 'update':
                            df_data = pd.io.json.json_normalize(value)
                            indicies, Is = self.search_index(self.df, df_data)            

                            for i in range(len(indicies)):
                                index = indicies[i]
                                id = self.df.iloc[index]['id']
                                price = self.df.loc[index, 'price']

                                self.df.loc[index] = df_data.iloc[Is[i]]
                                self.df.loc[index, 'price'] = price

                        elif key == 'delete':
                            df_data = pd.io.json.json_normalize(value)
                            
                            indicies, Is = self.search_index(self.df, df_data)
                            self.df.drop(index=self.df.index[indicies], inplace=True)
                            self.df = self.df.sort_values('price', ascending=False)
                            self.df.reset_index(inplace=True, drop=True)
        
                        elif key == "insert":
                            df_data = pd.io.json.json_normalize(value)
                            self.df = pd.concat([self.df, df_data], sort=True)
                            self.df = self.df.sort_values('price', ascending=False)
                            self.df.reset_index(inplace=True, drop=True)

                        
                            
            if len(self.data[message["topic"]]) > BybitWebsocket.MAX_DATA_CAPACITY:
                self.data[message["topic"]] = self.data[message["topic"]][BybitWebsocket.MAX_DATA_CAPACITY//2:]

    def __on_error(self, error):
        '''Called on fatal websocket errors. We exit on these.'''
        if not self.exited:
            self.logger.error("Error : %s" % error)
            raise websocket.WebSocketException(error)

    def __on_open(self):
        '''Called when the WS opens.'''
        self.logger.debug("Websocket Opened.")

    def __on_close(self):
        '''Called on websocket close.'''
        self.logger.info('Websocket Closed')

    def ping(self):
        self.ws.send('{"op":"ping"}')
        if 'pong' not in self.data:
            self.data['pong'] = []

    def subscribe_kline(self, interval:str):
        param = {}
        param['op'] = 'subscribe'
        param['args'] = ['kline.' + self.symbol + '.' + interval]
        self.ws.send(json.dumps(param))
        if 'kline.' + self.symbol + '.' + interval not in self.data:
            self.data['kline.' + self.symbol + '.' + interval] = []

    def subscribe_trade(self):
        self.ws.send('{"op":"subscribe","args":["trade"]}')
        if "trade.BTCUSD" not in self.data:
            self.data["trade.BTCUSD"] = []
            self.data["trade.ETHUSD"] = []
            self.data["trade.EOSUSD"] = []
            self.data["trade.XRPUSD"] = []

    def subscribe_insurance(self):
        self.ws.send('{"op":"subscribe","args":["insurance"]}')
        if 'insurance.BTC' not in self.data:
            self.data['insurance.BTC'] = []
            self.data['insurance.XRP'] = []
            self.data['insurance.EOS'] = []
            self.data['insurance.ETH'] = []

    def subscribe_orderBookL2(self):
        param = {}
        param['op'] = 'subscribe'
        param['args'] = ['orderBookL2_25.' + self.symbol]
        self.ws.send(json.dumps(param))
        if 'orderBookL2_25.' + self.symbol not in self.data:
            self.data['orderBookL2_25.' + self.symbol] = []

    def subscribe_instrument_info(self):
        param = {}
        param['op'] = 'subscribe'
        param['args'] = ['instrument_info.100ms.' + self.symbol]
        self.ws.send(json.dumps(param))
        if 'instrument_info.100ms.' + self.symbol not in self.data:
            self.data['instrument_info.100ms.' + self.symbol] = []

    def subscribe_position(self):
        self.ws.send('{"op":"subscribe","args":["position"]}')
        if 'position' not in self.data:
            self.data['position'] = []

    def subscribe_execution(self):
        self.ws.send('{"op":"subscribe","args":["execution"]}')
        if 'execution' not in self.data:
            self.data['execution'] = []

    def subscribe_order(self):
        self.ws.send('{"op":"subscribe","args":["order"]}')
        if 'order' not in self.data:
            self.data['order'] = []

    def get_data(self, topic):
        if topic not in self.data:
            self.logger.info(" The topic %s is not subscribed." % topic)
            return []
        if topic.split('.')[0] in BybitWebsocket.PRIVATE_TOPIC and not self.auth:
            self.logger.info("Authentication failed. Please check your api_key and api_secret. Topic: %s" % topic)
            return []
        else:
            if len(self.data[topic]) == 0:
                # self.logger.info(" The topic %s is empty." % topic)
                return []
            # while len(self.data[topic]) == 0 :
            #     sleep(0.1)
            return self.data[topic].pop()

    def get_df(self):
        return self.df

    def search_index(self, a, b):
        # DataFrame a から DataFrame Bと一致する行を探す
        indicies = list()
        Is = list()

        for i in range(len(b)):
            id = b.iloc[i]['id']
            side = b.iloc[i]['side']

            index = a.query('id == {} and side == "{}"'.format(id, side)).index[0]
            indicies.append(index)
            Is.append(i)

        return indicies, Is

if __name__ == "__main__":
    bybitws = BybitWebsocket(wsURL="wss://stream-testnet.bybit.com/realtime", api_key=None, api_secret=None, symbol='BTCUSD')
    bybitws.subscribe_orderBookL2()

    while True:
        sleep(5)
        print(bybitws.get_data('orderBookL2_25.BTCUSD'))

使い方

もとのコネクタと同じように bybitws = BybitWebsocket(wsURL="wss://stream-testnet.bybit.com/realtime", api_key=None, api_secret=None, symbol='BTCUSD') として,get_df関数を呼ぶことで,OrderBook情報がpandasのDataFrame型として返ってくる. bybitws.get_df()

         id    price  side   size  symbol
0   67980000  6798.00  Sell   4168  BTCUSD
1   67975000  6797.50  Sell  16294  BTCUSD
2   67970000  6797.00  Sell    342  BTCUSD
3   67965000  6796.50  Sell  17166  BTCUSD
4   67960000  6796.00  Sell  10858  BTCUSD
5   67955000  6795.50  Sell   8516  BTCUSD
6   67950000  6795.00  Sell  27528  BTCUSD
7   67945000  6794.50  Sell  26413  BTCUSD
8   67940000  6794.00  Sell  23465  BTCUSD
9   67935000  6793.50  Sell  21457  BTCUSD
10  67930000  6793.00  Sell   6215  BTCUSD
11  67925000  6792.50  Sell  18226  BTCUSD
12  67920000  6792.00  Sell   4577  BTCUSD
13  67915000  6791.50  Sell  12060  BTCUSD
14  67910000  6791.00  Sell  16535  BTCUSD
15  67905000  6790.50  Sell  22345  BTCUSD
16  67900000  6790.00  Sell   7112  BTCUSD
17  67895000  6789.50  Sell  23136  BTCUSD
18  67890000  6789.00  Sell  18644  BTCUSD
19  67885000  6788.50  Sell   8303  BTCUSD
20  67880000  6788.00  Sell    322  BTCUSD
21  67865000  6786.50  Sell   8690  BTCUSD
22  67860000  6786.00  Sell  15285  BTCUSD
23  67855000  6785.50  Sell  14570  BTCUSD
24  67850000  6785.00  Sell  10832  BTCUSD
25  67805000  6780.50   Buy   9987  BTCUSD
26  67800000  6780.00   Buy  14040  BTCUSD
27  67795000  6779.50   Buy   3923  BTCUSD
28  67790000  6779.00   Buy   9540  BTCUSD
29  67780000  6778.00   Buy    513  BTCUSD
30  67775000  6777.50   Buy  30129  BTCUSD
31  67770000  6777.00   Buy  29851  BTCUSD
32  67765000  6776.50   Buy  11973  BTCUSD
33  67760000  6776.00   Buy  11096  BTCUSD
34  67755000  6775.50   Buy   8102  BTCUSD
35  67745000  6774.50   Buy  16637  BTCUSD
36  67740000  6774.00   Buy  31585  BTCUSD
37  67735000  6773.50   Buy  10602  BTCUSD
38  67730000  6773.00   Buy  14331  BTCUSD
39  67725000  6772.50   Buy  31113  BTCUSD
40  67720000  6772.00   Buy  26608  BTCUSD
41  67715000  6771.50   Buy   5231  BTCUSD
42  67710000  6771.00   Buy  44615  BTCUSD
43  67705000  6770.50   Buy  13054  BTCUSD
44  67700000  6770.00   Buy  28312  BTCUSD
45  67695000  6769.50   Buy   3325  BTCUSD
46  67690000  6769.00   Buy  16470  BTCUSD
47  67685000  6768.50   Buy  13675  BTCUSD
48  67680000  6768.00   Buy  22811  BTCUSD
49  67675000  6767.50   Buy  16363  BTCUSD

おわりに

過去のコードがほとんどそのまま使えたので助かった.

もしなにかこのコードにバグや不明な点などがあったら気軽にコメントお願いします.

では.