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 from decimal import Decimal from loguru import logger def empty_call(msg): # print(msg) pass class KucoinUsdtSwapWs: def __init__(self, params: model.ClientParams, colo=0, is_print=0): if colo: print('不支持colo高速线路') self.BaseURL = "https://api-futures.kucoin.com" else: self.BaseURL = "https://api-futures.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 + "M" # 处理特殊情况 if self.symbol == 'BTCUSDTM': self.symbol = 'XBTUSDTM' 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.ticker_info = {"name":self.name,'bp':0.0,'ap':0.0} self.multiplier = None 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']['sequence'] > self.update_t: self.update_t = msg['data']['sequence'] 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_ticker(self, msg): if msg['data']['sequence'] > self.update_t: self.update_t = msg['data']['sequence'] self.ticker_info['bp'] = float(msg['data']['bestBidPrice']) self.ticker_info['ap'] = float(msg['data']['bestAskPrice']) self.callback['onTicker'](self.ticker_info) bp = float(msg['data']['bestBidPrice']) bv = float(msg['data']['bestBidSize']) ap = float(msg['data']['bestAskPrice']) av = float(msg['data']['bestAskSize']) 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'])*self.multiplier 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_position(self, msg): pos = model.Position() if "currentQty" in msg['data']: amt = float(msg["data"]["currentQty"]) * self.multiplier ep = float(msg["data"]["avgEntryPrice"]) if amt == 0: self.callback["onPosition"](pos) elif amt > 0: pos.longPos = amt pos.longAvg = ep self.callback["onPosition"](pos) elif amt < 0: pos.shortPos = -amt pos.shortAvg = ep self.callback["onPosition"](pos) def _update_account(self, msg): pass # if msg['data']['currency'] == 'USDT' and msg['subject'] == "availableBalance.change": # cash = float(msg['data']['availableBalance']) + float(msg['data']['holdBalance']) # self.callback['onEquity'] = {self.quote:cash} def _update_order(self, msg): self.logger.debug(f"ws订单推送 {msg}") if '/contractMarket/tradeOrders' in msg['topic']: if msg["data"]["symbol"] == self.symbol: if msg["data"]["status"] == 'open': # 新增订单 order_event = dict() order_event['status'] = "NEW" order_event['filled'] = 0 order_event['filled_price'] = 0 order_event['client_id'] = msg["data"]["clientOid"] if "clientOid" in msg["data"] else "" order_event['order_id'] = msg["data"]['orderId'] self.callback["onOrder"](order_event) elif msg["data"]["type"] in ['filled','canceled']: # 删除订单 order_event = dict() order_event['status'] = "REMOVE" order_event['client_id'] = msg["data"]["clientOid"] if "clientOid" in msg["data"] else "" order_event['order_id'] = msg["data"]['orderId'] order_event['filled'] = float(Decimal(msg["data"]["filledSize"])*Decimal(str(self.multiplier))) if 'price' in msg["data"]: if msg['data']['price'] != '': order_event['filled_price'] = float(msg["data"]["price"]) else: order_event['filled_price'] = 0 else: order_event['filled_price'] = 0 self.callback["onOrder"](order_event) 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']({'name': self.name,'data':market_data}) except: traceback.print_exc() await asyncio.sleep(interval) async def get_token(self, is_auth): # 获取 合约系数 session = aiohttp.ClientSession() response = await session.get( "https://api-futures.kucoin.com/api/v1/contracts/active", proxy=self.proxy ) res = await response.json() for i in res['data']: if i['symbol'] == self.symbol: self.multiplier = float(i["multiplier"]) print(f"合约乘数为 {self.multiplier}") self.logger.debug(f"合约乘数为 {self.multiplier}") await session.close() # 获取 token 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 usdt swap 获取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连接成功') # 订阅 channels=[ # f"/contractMarket/tickerV2:{self.symbol}", f"/contractMarket/level2Depth50:{self.symbol}", ] if sub_trade: channels += [f"/contractMarket/execution:{self.symbol}"] if is_auth: channels += [ f"/contractAccount/wallet", f"/contract/position:{self.symbol}", f"/contractMarket/tradeOrders:{self.symbol}", ] for i in channels: sub_str = ujson.dumps({"topic": i, "type": "subscribe"}) if "/contractMarket/level2Depth50" not in i \ and "/contractMarket/execution" not in i \ and "/contractMarket/tickerV2" not in i: # print(i) sub_str = ujson.dumps({"topic": i, "type": "subscribe", "privateChannel": True, "response": True}) await _ws.send_str(sub_str) while True: # 接受消息 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) # print(msg) # 处理消息 if 'data' in msg: # if 'level2' not in msg['subject']:print(msg) if 'level2' in msg['subject']:self._update_depth(msg) elif "tickerV2" in msg["subject"]:self._update_ticker(msg) elif 'match' in msg['subject']:self._update_trade(msg) elif 'orderMargin.change' in msg['subject']:self._update_account(msg) elif 'symbolOrderChange' in msg['subject']:self._update_order(msg) elif 'position.change' in msg['subject']:self._update_position(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: traceback.print_exc() print(f'{self.name} ws连接失败 开始重连...') self.logger.error(f'{self.name} ws连接失败 开始重连...') # await asyncio.sleep(1)