Websocket demo

[tomstandly]

What is python demo function for using websocket?

[binbincai410]

Hi, you can refer to the following code

#coding: utf-8

import zlib
import gzip
import io

from websocket import create_connection

def gzip_str(string_: str) -> bytes:
    return gzip.compress(string_.encode())


def gunzip_bytes_obj(bytes_obj: bytes) -> str:
    return gzip.decompress(bytes_obj).decode()

def main():
  print ("connect")
  ws = create_connection("wss://open-ws-swap.bingbon.pro/ws")
  print ("before sub")
  ws.send('{ "id": "id1", "reqType": "sub", "dataType": "market.trade.detail.BTC-USDT"}')
  print ("after sub")
  while True:
      print ("Receiving...")
      result =  ws.recv()
      print ("Received '%s'", gunzip_bytes_obj(result))
  ws.close()

if __name__ == "__main__":
  main()

[tomstandly]

Thanks for your reply. function do not show anything and shows this errors:
connect
Traceback (most recent call last):
File "", line 29, in
File "", line 18, in main
File "\Python\Python38-32\lib\site-packages\websocket_core.py", line 601, in create_connection
websock.connect(url, **options)
File "\Python\Python38-32\lib\site-packages\websocket_core.py", line 248, in connect
self.handshake_response = handshake(self.sock, url, *addrs, **options)
File "\Python\Python38-32\lib\site-packages\websocket_handshake.py", line 57, in handshake
status, resp = _get_resp_headers(sock)
File "\Python\Python38-32\lib\site-packages\websocket_handshake.py", line 148, in _get_resp_headers
raise WebSocketBadStatusException("Handshake status %d %s", status, status_message, resp_headers)
websocket._exceptions.WebSocketBadStatusException: Handshake status 403 Forbidden

[binbincai410]

Try this

import time
import json
import zlib
import uuid
import websocket            # pip install websocket-client==1.3.1
from multiprocessing.dummy import Process


def ws_method(func):
    def wrapper(self, *args, **kwargs):
        for i in range(20):
            if self.is_alive is True:
                break
            time.sleep(1)
        else:
            raise Exception('waiting ws connection timeout!')

        return func(self, *args, **kwargs)
    return wrapper


class BingxWS:
    def __init__(self, handler, open_handler=None, close_handler=None):
        self.name = 'BingX'
        self.handler = handler
        self.open_handler = open_handler
        self.close_handler = close_handler

        self.base_url = 'wss://open-ws-swap.bingbon.pro/ws'
        self.headers = {
            'Host': 'open-ws-swap.bingbon.pro',
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X -1_0_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36',
        }

        self.ws = None
        self.is_alive = False
        self.want_to_open = False                        
        self.on_sub = []

    def _start_connection(self):
        while True:
            try:
                if self.want_to_open is True:
                    self.ws = websocket.WebSocketApp(self.base_url, header=self.headers, on_open=self._on_open, on_message=self._on_message, on_error=self._on_error, on_close=self._on_close)
                    # self.ws.run_forever(http_proxy_host='127.0.0.1', http_proxy_port=7890, proxy_type='http')              
                    self.ws.run_forever()

                    self._close_connection()
                    time.sleep(0.1)
                    continue
            except Exception as e:
                print('%s ws _start_connection error: %s' % (self.name, e))
            time.sleep(1)

    def _close_connection(self):
        try:
            self.ws.close()
            self.ws = None
        except Exception as e:
            pass

    def _on_open(self, ws):
        print('%s websocket connect!' % self.name)
        self.is_alive = True
        Process(target=self._re_subs).start()                 
        if self.open_handler is not None:
            self.open_handler()

    def _on_close(self, ws, k1, k2):
        print('%s websocket disconnect!' % self.name)
        self.is_alive = False
        if self.close_handler is not None:
            self.close_handler()

    def _on_error(self, ws, status_code):
        print(' %s websocket error: %s' % (self.name, status_code))

    def _on_ping(self):
        self.ws.send('Pong')

    def _on_message(self, ws, content):
        data = str(zlib.decompress(content, 16 + zlib.MAX_WBITS), encoding='utf-8')

        if data == 'Ping':
            self._on_ping()
        else:
            js = json.loads(data)
            self.handler(js)

    def _re_subs(self):
        for one in self.on_sub:
            if one[0] == 'depth':
                self.sub_depth(symbol=one[1], step=one[2], level=one[3])
            elif one[0] == 'trade':
                self.sub_trade(symbol=one[1])
            elif one[0] == 'kline':
                self.sub_kline(symbol=one[1], kline_type=one[2])
            time.sleep(0.2)

    def start(self):
        if self.want_to_open is False or self.is_alive is False:
            self.want_to_open = True
            Process(target=self._start_connection).start()

    def stop(self):
        self.want_to_open = False
        self._close_connection()
        self.on_sub = []

    @ws_method
    def sub_depth(self, symbol, step='step0', level='level10'):
        """
        :param symbol:
        :param step: step0\step1\step2\step3\step4\step5
        :param level: level5\level10\level20\level50\level100
        :return:
        """
        data = {
            'id': str(uuid.uuid4()),
            'reqType': 'sub',
            'dataType': 'market.depth.%s.%s.%s' % (symbol, step, level)
        }
        self.ws.send(json.dumps(data))

        elem = ['depth', symbol, step, level]
        if elem not in self.on_sub:
            self.on_sub.append(elem)

    @ws_method
    def sub_trade(self, symbol):
        """
        :param symbol:
        :return:
        """
        data = {
            'id': str(uuid.uuid4()),
            'reqType': 'sub',
            'dataType': 'market.trade.detail.%s' % symbol
        }
        self.ws.send(json.dumps(data))

        elem = ['trade', symbol]
        if elem not in self.on_sub:
            self.on_sub.append(elem)

    @ws_method
    def sub_kline(self, symbol, kline_type='1'):
        """
        :param symbol:
        :param kline_type: 1min\3min\5min\15min\30min\1hour\4hour\12hour\1day\1week\1month
        :return:
        """
        data = {
            'id': str(uuid.uuid4()),
            'reqType': 'sub',
            'dataType': 'market.kline.%s.%s' % (symbol, kline_type)
        }
        self.ws.send(json.dumps(data))

        elem = ['kline', symbol, kline_type]
        if elem not in self.on_sub:
            self.on_sub.append(elem)


def on_message(content):
    print(content)


if __name__ == "__main__":
    bingx = BingxWS(handler=on_message)
    bingx.start()

    bingx.sub_depth(symbol='BTC-USDT', step='step0', level='level5')
    bingx.sub_kline(symbol='BTC-USDT', kline_type='1min')
    bingx.sub_trade(symbol='BTC-USDT')

[ThisisAqib]

@binbincai410 Hi, do you know how to use the above code with API to get the account order updates?