py-as-zhong/exchange/kucoin_spot_ws.py

import aiohttp
import time
import asyncio
import zlib
import json, ujson
import zlib
import hashlib
import hmac
import base64
import traceback
import random
import gzip, csv, sys
from uuid import uuid4
import logging, logging.handlers
import utils
import model

def empty_call(msg):
    pass


class KucoinSpotWs:

    def __init__(self, params: model.ClientParams, colo=0, is_print=0):
        if colo:
            print('不支持colo高速线路')
            self.BaseURL = "https://api.kucoin.com"
        else:
            self.BaseURL = "https://api.kucoin.com"
        self.params = params
        self.name = self.params.name
        self.base = self.params.pair.split('_')[0].upper()
        self.quote = self.params.pair.split('_')[1].upper()
        self.symbol = self.base + '-' + self.quote
        self.callback = {
            "onMarket":self.save_market,
            "onPosition":empty_call,
            "onEquity":empty_call,
            "onOrder":empty_call,
            "onTicker":empty_call,
            "onDepth":empty_call,
            "onExit":empty_call,
            }
        self.is_print = is_print
        self.proxy = None
        if 'win' in sys.platform:
            self.proxy = self.params.proxy
        self.logger = self.get_logger()
        self.local_orders = dict()
        self.ticker_info = {"name":self.name,'bp':0.0,'ap':0.0}
        self.stop_flag = 0
        self.max_buy = 0.0
        self.min_sell = 0.0
        self.buy_v = 0.0
        self.buy_q = 0.0
        self.sell_v = 0.0
        self.sell_q = 0.0
        self.update_t = 0.0
        self.depth = []
        #### 指定发包ip
        iplist = utils.get_local_ip_list()
        self.ip = iplist[int(self.params.ip)]
    
    def get_logger(self):
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.DEBUG)
        # log to txt
        formatter = logging.Formatter('[%(asctime)s] - %(levelname)s - %(message)s')
        handler = logging.handlers.RotatingFileHandler(f"log.log",maxBytes=1024*1024)
        handler.setLevel(logging.DEBUG)
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        return logger

    def save_market(self, msg):
        date = time.strftime('%Y-%m-%d',time.localtime())
        interval = float(self.params.interval)
        if msg:
            exchange = msg['name']
            if len(msg['data']) > 1:
                with open(f'./history/{exchange}_{self.symbol}_{interval}_{date}.csv',
                            'a',
                            newline='',
                            encoding='utf-8') as f:
                    writer = csv.writer(f, delimiter=',')
                    writer.writerow(msg['data'])
                if self.is_print:print(f'写入行情 {self.symbol}')

    async def get_sign(self):
        headers = {}
        headers['Content-Type'] = 'application/json'
        headers['X-MBX-APIKEY'] = self.params.access_key
        params = {
            'timestamp':int(time.time())*1000,
            'recvWindow':5000,
        }
        query_string = "&".join(["{}={}".format(k, params[k]) for k in sorted(params.keys())])
        signature = hmac.new(self.params.secret_key.encode(), msg=query_string.encode(), digestmod=hashlib.sha256).hexdigest()
        params['signature']=signature
        url = 'https://fapi.binance.com/fapi/v1/listenKey'
        session = aiohttp.ClientSession()
        response = await session.post(
            url, 
            params=params,
            headers=headers, 
            timeout=5, 
            proxy=self.proxy
            )
        login_str = await response.text()
        await session.close()
        return ujson.loads(login_str)['listenKey']

    def _update_depth(self, msg):
        if msg['data']['timestamp'] > self.update_t:
            self.update_t = msg['data']['timestamp']
            self.ticker_info["bp"] = float(msg['data']['bids'][0][0])
            self.ticker_info["ap"] = float(msg['data']['asks'][0][0])
            self.callback['onTicker'](self.ticker_info)
            ##### 标准化深度
            mp = (self.ticker_info["bp"] + self.ticker_info["ap"])*0.5
            step = mp * utils.EFF_RANGE / utils.LEVEL
            bp = []
            ap = []
            bv = [0 for _ in range(utils.LEVEL)]
            av = [0 for _ in range(utils.LEVEL)]
            for i in range(utils.LEVEL):
                bp.append(self.ticker_info["bp"]-step*i)
            for i in range(utils.LEVEL):
                ap.append(self.ticker_info["ap"]+step*i)
            # 
            price_thre = self.ticker_info["bp"] - step
            index = 0
            for bid in msg['data']['bids']:
                price = float(bid[0])
                amount = float(bid[1])
                if price > price_thre:
                    bv[index] += amount
                else:
                    price_thre -= step
                    index += 1
                    if index == utils.LEVEL:
                        break
                    bv[index] += amount
            price_thre = self.ticker_info["ap"] + step
            index = 0
            for ask in msg['data']['asks']:
                price = float(ask[0])
                amount = float(ask[1])
                if price < price_thre:
                    av[index] += amount
                else:
                    price_thre += step
                    index += 1
                    if index == utils.LEVEL:
                        break
                    av[index] += amount
            self.depth = bp + bv + ap + av
            self.callback['onDepth']({'name':self.name,'data':self.depth})
    
    def _update_trade(self, msg):
        price = float(msg["data"]['price'])
        side = msg["data"]['side']
        amount = float(msg["data"]['size'])
        if price > self.max_buy or self.max_buy == 0.0:
            self.max_buy = price
        if price < self.min_sell or self.min_sell == 0.0:
            self.min_sell = price
        if side == 'buy':
            self.buy_q += amount
            self.buy_v += amount*price
        elif side == 'sell':
            self.sell_q += amount
            self.sell_v += amount*price

    def _update_account(self, msg):
        if 'trade' in msg['data']["relationEvent"]:
            if msg['data']["currency"].upper() == self.base:
                coin = abs(float(msg['data']["total"]))
                self.callback['onEquity']({
                    self.base:coin
                })
            if msg['data']["currency"].upper() == self.quote:
                self.callback['onEquity']({
                    self.quote:float(msg['data']["total"])
                })
    
    def _update_order(self, msg):
        self.logger.debug(f"ws订单推送 {msg}")
        if msg['topic'] == '/spotMarket/tradeOrders-batch':
            for i in msg['data']:
                if i["symbol"] == self.symbol:
                    if i["status"] == 'open':  # 新增订单
                        order_event = dict()
                        order_event['filled'] = 0
                        order_event['filled_price'] = 0
                        order_event['client_id'] = i["clientOid"]
                        order_event['order_id'] = i['orderId']
                        order_event['status'] = "NEW"
                        self.callback["onOrder"](order_event)
                        self.local_orders[i["clientOid"]] = order_event
                    elif i["status"] == 'done':  # 删除订单
                        if "price" in i:
                            price = float(i["price"])
                        else:
                            if i["clientOid"] in self.local_orders:
                                price = self.local_orders[i["clientOid"]]["price"]
                            else:
                                # 应该是非本策略订单 忽略price
                                price = 0
                        order_event = dict()
                        order_event['amount'] = float(i["size"])
                        order_event['filled'] = float(i["filledSize"])
                        order_event['filled_price'] = price
                        order_event['client_id'] = i["clientOid"]
                        order_event['order_id'] = i['orderId']
                        order_event['status'] = "REMOVE"
                        order_event['fee'] = float(i["fee"]) if "fee" in i["data"] else 0.0
                        self.callback["onOrder"](order_event)
                        if i["clientOid"] in self.local_orders:
                            del(self.local_orders[i["clientOid"]])
        else:
            if msg["data"]["symbol"] == self.symbol:
                if msg["data"]["status"] == 'open':  # 新增订单
                    order_event = dict()
                    order_event['filled'] = 0
                    order_event['filled_price'] = 0
                    order_event['client_id'] = msg["data"]["clientOid"]
                    order_event['order_id'] = msg["data"]['orderId']
                    order_event['status'] = "NEW"
                    self.callback["onOrder"](order_event)
                    self.local_orders[msg["data"]["clientOid"]] = order_event
                elif msg["data"]["status"] == 'done':  # 删除订单
                    if "price" in msg['data']:
                        price = float(msg["data"]["price"])
                    else:
                        if msg["data"]["clientOid"] in self.local_orders:
                            price = self.local_orders[msg["data"]["clientOid"]]["price"]
                        else:
                            # 应该是非本策略订单 忽略price
                            price = 0
                    order_event = dict()
                    order_event['filled'] = float(msg["data"]["filledSize"])
                    order_event['filled_price'] = price
                    order_event['client_id'] = msg["data"]["clientOid"]
                    order_event['order_id'] = msg["data"]['orderId']
                    order_event['fee'] = float(msg["data"]["fee"]) if "fee" in msg["data"] else 0.0
                    order_event['status'] = "REMOVE"
                    self.callback["onOrder"](order_event)
                    if msg["data"]["clientOid"] in self.local_orders:
                        del(self.local_orders[msg["data"]["clientOid"]])

    def _get_data(self):
        market_data = self.depth + [self.max_buy, self.min_sell]
        self.max_buy = 0.0
        self.min_sell = 0.0
        self.buy_v = 0.0
        self.buy_q = 0.0
        self.sell_v = 0.0
        self.sell_q = 0.0
        return {'name': self.name,'data':market_data}

    async def go(self):
        interval = float(self.params.interval)
        if self.is_print:print(f'Ws循环器启动 interval {interval}')
        ### onTrade
        while 1:
            try:
                # 更新市场信息
                market_data = self._get_data()
                self.callback['onMarket'](market_data)
            except:
                traceback.print_exc()
            await asyncio.sleep(interval)

    async def get_token(self, is_auth):
        if is_auth:
            uri = "/api/v1/bullet-private"
        else:
            uri = "/api/v1/bullet-public"
        headers = {}
        if is_auth:
            now_time = int(time.time()) * 1000
            str_to_sign = str(now_time) + "POST" + uri
            sign = base64.b64encode(hmac.new(self.params.secret_key.encode('utf-8'), str_to_sign.encode('utf-8'), hashlib.sha256).digest())
            passphrase = base64.b64encode(hmac.new(self.params.secret_key.encode('utf-8'), self.params.pass_key.encode('utf-8'), hashlib.sha256).digest())
            headers = {
                "KC-API-SIGN": sign.decode(),
                "KC-API-TIMESTAMP": str(now_time),
                "KC-API-KEY": self.params.access_key,
                "KC-API-PASSPHRASE": passphrase.decode(),
                "Content-Type": "application/json",
                "KC-API-KEY-VERSION": "2"
            }
        headers["User-Agent"] = "kucoin-python-sdk/v1.0"
        session = aiohttp.ClientSession()
        response = await session.post(
            self.BaseURL+uri, 
            timeout=5, 
            headers=headers,
            proxy=self.proxy
            )
        res = await response.text()
        res = ujson.loads(res)
        await session.close()
        if res["code"] == "200000":
            token = res["data"]["token"]
            ws_connect_id = str(uuid4()).replace('-', '')
            endpoint = res["data"]['instanceServers'][0]['endpoint']
            ws_endpoint = f"{endpoint}?token={token}&connectId={ws_connect_id}"
            encrypt = res["data"]['instanceServers'][0]['encrypt']
            if is_auth:
                ws_endpoint += '&acceptUserMessage=true'
            return ws_endpoint, encrypt
        else:
            raise Exception("kucoin spot 获取token错误")

    async def run(self, is_auth=0, sub_trade=0, sub_fast=0):
        while True:
            try:
                ping_time = time.time()
                # 尝试连接
                print(f'{self.name} 尝试连接ws')
                # 获取token
                ws_endpoint, encrypt = await self.get_token(is_auth)
                # 登陆
                ws_url = ws_endpoint
                async with aiohttp.ClientSession(
                        connector = aiohttp.TCPConnector(
                                limit=50,
                                keepalive_timeout=120,
                                verify_ssl=False,
                                local_addr=(self.ip,0)
                            )
                        ).ws_connect(
                            ws_url,
                            proxy=self.proxy,
                            timeout=30,
                            receive_timeout=30,
                            ) as _ws:
                    print(f'{self.name} ws连接成功')
                    self.logger.info(f'{self.name} ws连接成功')
                    # 订阅 ticker来的很慢
                    channels=[
                        # f"/market/ticker:{self.symbol}",
                        f"/spotMarket/level2Depth50:{self.symbol}",
                        ]
                    if sub_trade:
                        channels.append(f"/market/match:{self.symbol}")
                    if is_auth:
                        channels.append(f"/spotMarket/tradeOrders")
                        channels.append(f"/account/balance")
                    for i in channels:
                        sub_str = ujson.dumps({"topic": i, "type":"subscribe"})
                        await _ws.send_str(sub_str)
                    while True:
                        # 停机信号
                        if self.stop_flag:
                            await _ws.close()
                            return
                        # 接受消息
                        try:
                            msg = await _ws.receive(timeout=30)
                        except:
                            print(f'{self.name} ws长时间没有收到消息 准备重连...')
                            self.logger.error(f'{self.name} ws长时间没有收到消息 准备重连...')
                            break
                        msg = ujson.loads(msg.data)
                        # self.logger.debug(msg)
                        # print(msg)
                        # 处理消息
                        if 'data' in msg:
                            if 'level2' in msg['subject']:self._update_depth(msg)
                            elif 'trade.l3match' in msg['subject']:self._update_trade(msg)
                            elif 'account.balance' in msg['subject']:self._update_account(msg)
                            elif 'orderChange' in msg['subject']:self._update_order(msg)
                        # heartbeat
                        if time.time() - ping_time > 30:
                            msg = {
                                'id': str(int(time.time() * 1000)),
                                'type': 'ping'
                            }
                            await _ws.send_str(ujson.dumps(msg))
                            ping_time = time.time()
            except Exception as e:
                traceback.print_exc()
                print(f'{self.name} ws连接失败 开始重连...')
                self.logger.error(f'{self.name} ws连接失败 开始重连...')
                self.logger.error(traceback.format_exc())
                # await asyncio.sleep(1)