py-as-zhong/exchange/kucoin_usdt_swap_rest.py

import aiohttp
import time
import asyncio
import json
import hmac
import base64
import hashlib
import traceback
from urllib.parse import urljoin
import sys
from urllib.parse import urlparse
import logging.handlers
import utils
import logging, logging.handlers
import model

from loguru import logger
logger.add("kucoin_limits.log", format="{time} {level} {message}", level="INFO")

def empty_call(msg):
    print(f'空的回调函数 {msg}')

def decimal_amount(amount, d):
    if int(d) == 0:
        return str(int(amount))
    elif int(d) > 0:
        return str(round(int(amount*10**int(d))*0.1**int(d), int(d)))

def decimal_price(price, d):
    if int(d) == 0:
        return str(int(price))
    elif int(d) > 0:
        return str(round(float(price), int(d)))


class KucoinUsdtSwapRest:

    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('不支持colo高速线路')
            self.HOST = "https://api-futures.kucoin.com"
        else:
            self.HOST = "https://api-futures.kucoin.com"
        self.params = params
        self.name = self.params.name
        self.base = self.params.pair.split('_')[0].upper()
        # 处理特殊情况
        if self.base == "BTC":self.base = "XBT"
        self.quote = self.params.pair.split('_')[1].upper()
        self.symbol = self.base + self.quote + "M"
        self._SESSIONS = dict()
        self.logger = self.get_logger()
        self.callback = {
            "onMarket":empty_call,
            "onPosition":empty_call,
            "onOrder":empty_call,
            "onEquity":empty_call,
            "onTicker":empty_call,
            "onDepth":empty_call,
            "onExit":empty_call,
            }
        self.exchange_info = dict()
        self.decimal_amount = 10
        self.decimal_price = 10
        self.delays = []
        self.max_delay = 0
        self.avg_delay = 0
        self.multiplier = None
        self.proxy = None
        if 'win' in sys.platform:
            self.proxy = self.params.proxy
        self.logger = self.get_logger()
        self.coin_value = 0.0
        self.cash_value = 0.0
        self.last_cash = 0.0
        #### 指定发包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 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("log.log",maxBytes=1024*1024,encoding='utf-8')
        handler.setLevel(logging.DEBUG)
        handler.setFormatter(formatter)
        # log to console
        console = logging.StreamHandler()
        console.setLevel(logging.WARNING)
        logger.addHandler(handler)
        logger.addHandler(console)
        return logger

    def _get_session(self, url):
        parsed_url = urlparse(url)
        key = parsed_url.netloc or parsed_url.hostname
        if key not in self._SESSIONS:
            tcp = aiohttp.TCPConnector(limit=50,keepalive_timeout=120,verify_ssl=False,local_addr=(self.ip,0))
            session = aiohttp.ClientSession(connector=tcp)
            self._SESSIONS[key] = session
        return self._SESSIONS[key]

    async def _request(self, method, uri, body=None, params=None, auth=False):
        url = urljoin(self.HOST, uri)
        headers = {}
        if auth:
            now_time = int(time.time()) * 1000
            str_to_sign = str(now_time) + method + uri
            if method in ['GET', 'DELETE']:
                data_json = ''
                if params:
                    strl = []
                    for key in params:
                        strl.append("{}={}".format(key, params[key]))
                    data_json += '&'.join(strl)
                    str_to_sign += '?' + data_json
            else:
                if body:str_to_sign += body
            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 = self._get_session(url)
        timeout = aiohttp.ClientTimeout(10)
        msg = "rest请求记录" + str(method) + str(url) + str(params) + str(body)
        self.logger.debug(msg)
        try:
            start_time = time.time()
            if method == "GET":
                response = await session.get(url, params=params, headers=headers, timeout=timeout, proxy=self.proxy)
            elif method == "POST":
                response = await session.post(url, params=None, data=body, headers=headers, timeout=timeout, proxy=self.proxy)
            elif method == "DELETE":
                response = await session.delete(url, params=params, data=body, headers=headers, timeout=timeout, proxy=self.proxy)
            code = response.status
            res = await response.json() 
            delay = int(1000*(time.time() - start_time))
            res_msg = msg + f' 回报 {res}'
            self.logger.debug(res_msg)
            self.delays.append(delay)
            if code not in (200, 201, 202, 203, 204, 205, 206) or \
                int(res['code']) not in (200, 201, 202, 203, 204, 205, 206, 200000):
                print(f'URL:{url} METHOD:{method} PARAMS:{params} body:{body} ERROR:{res}')
                self.logger.error('请求错误'+str(res))
                self.logger.error(res)
                return None, res
            return res, None
        except Exception as e:
            print(f'{self.name} rest 请求出错', str(e))
            self.logger.error('请求错误'+str(e))
            self.logger.error(traceback.format_exc())
            return None, e
    
    async def buy_token(self):
        '''买入平台币'''
        pass

    async def check_position(self, hold_coin=0.0):
        '''
            两次执行check_position之间必须停留足够时间 避免position没及时更新 导致重复下单 已支持全品种
        '''
        try:
            ##############################
            self.logger.info("清空挂单")
            params = {
                'status':"active",
                'tradeType':'TRADE',
            }
            response, error = await self._request('GET', '/api/v1/orders', params=params, auth=1)
            if response is not None and response['data']['items'] is not None:
                for i in response['data']['items']:
                    res = await self.cancel_order(order_id=i["id"])
                    self.logger.info(res)
            self.logger.info(f"{self.name} 全平仓位")
            ##############################
            # 更新仓位
            print("获取仓位")
            res, err = await self._request('GET','/api/v1/positions', params={}, auth=1)
            if err:self.logger.info(err)
            if res:
                for i in res['data']:
                    symbol = i['symbol']
                    amt = float(i['currentQty'])
                    # 转换为单位：张
                    if self.exchange_info == dict():
                        await self.before_trade()
                    if amt > 0:
                        self.logger.info( await self.take_order(
                                    symbol,
                                    amt*self.exchange_info[symbol].multiplier,
                                    "pd",
                                    1,
                                    utils.get_cid(),
                                    "market"
                                ))
                    elif amt < 0:
                        self.logger.info( await self.take_order(
                                    symbol,
                                    -amt*self.exchange_info[symbol].multiplier,
                                    "pk",
                                    1,
                                    utils.get_cid(),
                                    "market"
                                ))
                ##############################        
        except:
            self.logger.error("清仓程序执行出错")
            self.logger.error(traceback.format_exc())
        return

    async def take_order(self, symbol, amount, origin_side, price, cid="", order_type='limit'):
        '''
            kumex会合并多空方向的仓位 不支持双向交易 所以reduce_only要小心使用
        '''
        reduceOnly = False
        if origin_side =='kd':
            side = 'buy'
        elif origin_side =='pd':
            side = 'sell'
        elif origin_side =='kk':
            side = 'sell'
        elif origin_side =='pk':
            side = 'buy'
        else:
            return None
        # 转换为单位：张
        if symbol not in self.exchange_info:
            await self.before_trade()
        amount_paper = int(amount/self.exchange_info[symbol].multiplier)
        if amount_paper == 0:
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['client_id'] = cid
            order_event['filled'] = 0
            order_event['filled_price'] = 0
            self.callback["onOrder"](order_event)
            return
        params = {
            'clientOid':cid,
            'symbol': symbol, 
            'size': amount_paper,
            'side': side, 
            'leverage': 10, #10, 没kyc老报错,##############################################
            'reduceOnly':reduceOnly,
            'price':utils.num_to_str(price, self.exchange_info[symbol].tickSize), 
            'type':order_type,
        }
        # logger.info(f'下单指令 {params}')
        if self.params.debug == 'True':
            return await asyncio.sleep(0.1)
        else:
            # 发单
            response, error = await self._request('POST', '/api/v1/orders', body=json.dumps(params), auth=1)
            # 更新
            if response:
                # 增加新的
                if 'data' in response:
                    order_event = dict()
                    order_event['status'] = "NEW"
                    order_event['client_id'] = cid
                    order_event['order_id'] = response['data']["orderId"]
                    self.callback["onOrder"](order_event)
            if error:
                order_event = dict()
                order_event['status'] = "REMOVE"
                order_event['client_id'] = cid
                order_event['filled_price'] = 0
                order_event['filled'] = 0
                self.callback["onOrder"](order_event)
        return response
    
    async def cancel_order(self, order_id=None, client_id=None):
        if order_id:
            response, error = await self._request('DELETE', f'/api/v1/orders/{order_id}', auth=1)
        elif client_id:
            response, error = await self._request('DELETE', f'/api/v1/orders/byClientOid?clientOid={client_id}', auth=1)
        else:
            raise Exception("撤单出错 没指定订单号")
        if response:
            self.logger.debug(f'撤单回报 {response}')
            # 撤单成功不会返回成交信息 所以不触发回调
        if error:
            print("撤单失败",error)
            return error
        return response
    
    async def check_order(self, order_id=None, client_id=None):
        if order_id:
            response, error = await self._request('GET', f'/api/v1/orders/{order_id}', auth=1)
        elif client_id:
            response, error = await self._request('GET', f'/api/v1/orders/byClientOid?clientOid={client_id}', auth=1)
        else:
            return
        if response:
            self.logger.debug(f'查单回报 {response}')
            order_event = dict()
            if response["data"]['isActive'] == True:
                order_event['status'] = "NEW"
            elif response["data"]['isActive'] == False:
                order_event['status'] = "REMOVE"
            else:
                self.logger.error("错误的订单状态")
            if self.multiplier == None:
                await self.before_trade()
            order_event['filled'] = float(response["data"]["filledSize"])*self.multiplier
            order_event['filled_price'] = float(response["data"]["filledValue"])/float(response["data"]["filledSize"])/self.multiplier if float(response["data"]["filledSize"]) > 0 else 0
            order_event['client_id'] = response["data"]["clientOid"]
            order_event['order_id'] = response["data"]['id']
            self.callback["onOrder"](order_event)
        if error:
            print("查单失败",error)
            self.logger.error(error)
        return response

    async def get_order_list(self):
        params = {
            'symbol':self.symbol,
            'status':"active",
            'tradeType':'TRADE',
        }
        response, error = await self._request('GET', '/api/v1/orders', params=params, auth=1)
        orders = [] # 重置本地订单列表
        if response is not None:
            for i in response['data']['items']:
                order_event = dict()
                if i['isActive'] == True:
                    order_event['status'] = "NEW"
                elif i['isActive'] == False:
                    order_event['status'] = "REMOVE"
                else:
                    self.logger.error("错误的订单状态")
                order_event['filled'] = float(i["dealSize"])
                order_event['filled_price'] = float(i["dealFunds"])/float(i["dealSize"]) if float(i["dealSize"]) > 0 else 0
                order_event['client_id'] = i["clientOid"]
                order_event['order_id'] = i['id']
                self.callback["onOrder"](order_event)
        if error:
            print(error)
        return response
    
    async def get_server_time(self):
        params = {}
        response = await self._request('GET', '/api/v1/timestamp', params=params)
        return response

    async def get_account(self):
        return await self._request('GET','/api/v1/account-overview', params={"currency":self.quote}, auth=1)

    async def get_position(self):
        return await self._request('GET','/api/v1/position', params={"symbol":self.symbol}, auth=1)

    async def get_market_details(self):
        return await self._request('GET',f'api/v1/contracts/active', params={}, auth=1)

    async def get_ticker(self):
        res, err = await self._request('GET',f'api/v1/ticker', params={"symbol":self.symbol}, auth=0)
        if res:
            ap = float(res['data']['bestAskPrice']) 
            bp = float(res['data']['bestBidPrice'])
            mp = (ap+bp)/2
            ticker = {"name":self.name,'mp': mp, 'bp':bp, 'ap':ap}
            return ticker
        if err:
            self.logger.error(err)
        return

    async def before_trade(self):
        # 获取市场
        res, error = await self.get_market_details()
        if error:
            pass
        if res:
            for i in res['data']:
                if i['symbol'] == self.symbol:
                    # 1张多少个币
                    self.multiplier = float(i["multiplier"])
                    # 1张 币的数量精度   张 转换成 币 需要乘以乘数
                    self.stepSize =  i["lotSize"]*self.multiplier
                    self.tickSize = i['tickSize']
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i['symbol']
                exchange_info.multiplier = float(i["multiplier"])
                exchange_info.tickSize = i['tickSize']
                exchange_info.stepSize = i["lotSize"]*float(i["multiplier"])
                self.exchange_info[exchange_info.symbol] = exchange_info
        # 设置杠杆
        await self._request('GET',f'api/v1/contracts/active', params={}, auth=1)
        # 更新账户
        res, err = await self.get_position()
        if err:print(err)
        if res:
            amt = float(res['data']['currentQty']) * self.multiplier
            ep = float(res['data']['avgEntryPrice'])
            pos = model.Position()
            if amt == 0.0:
                pos.longPos = 0
                pos.longAvg = 0
                pos.shortPos = 0
                pos.shortAvg = 0
            elif amt > 0.0:
                pos.longPos = amt
                pos.longAvg = ep
                pos.shortPos = 0
                pos.shortAvg = 0
            elif amt < 0.0:
                pos.longPos = 0
                pos.longAvg = 0
                pos.shortPos = -amt
                pos.shortAvg = ep
            self.callback['onPosition'](pos)

    async def get_equity(self):
        # 更新账户
        res, err = await self.get_account()
        if err:print(err)
        if res:
            cash = float(res['data']['accountEquity'])
            if self.last_cash == 0:
                self.last_cash = cash
            self.callback['onEquity']({self.quote:cash})
            self.cash_value = cash

    async def go(self):
        interval = 60
        await self.before_trade()
        await asyncio.sleep(1)
        while 1:
            try:
                # 更新账户
                res, err = await self.get_account()
                if err:print(err)
                if res:
                    if res['data']['currency'] == "USDT":
                        cash = float(res['data']['accountEquity'])
                        # rest有可能获取到中间态 为了避免得到错误的账户信息 需进行判断
                        if self.last_cash == 0:
                            self.last_cash = cash
                            self.callback['onEquity']({self.quote:cash})
                        else:
                            # 判断净值是否出现大幅度偏离 两次更新之间的差值不应超过5%
                            if abs(self.last_cash - cash)/cash < 0.05:
                                self.last_cash = cash
                                self.callback['onEquity']({self.quote:cash})
                            else:
                                # 否则舍弃本次更新
                                pass
                # 更新账户
                res, err = await self.get_position()
                if err:print(err)
                if res:
                    amt = float(res['data']['currentQty']) * self.multiplier
                    ep = float(res['data']['avgEntryPrice'])
                    pos = model.Position()
                    if amt == 0.0:
                        pos.longPos = 0
                        pos.longAvg = 0
                        pos.shortPos = 0
                        pos.shortAvg = 0
                    elif amt > 0.0:
                        pos.longPos = amt
                        pos.longAvg = ep
                        pos.shortPos = 0
                        pos.shortAvg = 0
                    elif amt < 0.0:
                        pos.longPos = 0
                        pos.longAvg = 0
                        pos.shortPos = -amt
                        pos.shortAvg = ep
                    self.callback['onPosition'](pos)
                await asyncio.sleep(interval)
                # 打印延迟
                self.get_delay_info()
                self.logger.debug(f'报单延迟 平均{self.avg_delay}ms 最大{self.max_delay}ms')
            except:
                traceback.print_exc()
                await asyncio.sleep(10)

    def get_delay_info(self):
        if len(self.delays) > 100:
            self.delays = self.delays[-100:]
        if max(self.delays) > self.max_delay:self.max_delay = max(self.delays)
        self.avg_delay = round(sum(self.delays)/len(self.delays),1)
    
    async def handle_signals(self, orders):
        '''执行策略指令'''
        try:
            for order_name in orders:
                if 'Cancel' in order_name:
                    # cid = orders[order_name][0]
                    oid = orders[order_name][1]
                    # kumex 只能按oid撤单
                    if oid:
                        asyncio.get_event_loop().create_task(self.cancel_order(order_id=oid))
            for order_name in orders:
                if 'Limits' in order_name:
                    for i in orders[order_name]:
                        logger.info(i)
                        asyncio.get_event_loop().create_task(self.take_order(
                            self.symbol,
                            i[0],
                            i[1],
                            i[2],
                            i[3]
                        ))
                    if len(orders[order_name]) > 0:
                        logger.info("")
            for order_name in orders:
                if 'Check' in order_name:
                    # cid = orders[order_name][0]
                    oid = orders[order_name][1]
                    if oid:
                        asyncio.get_event_loop().create_task(self.check_order(order_id=oid))
        except:
            # traceback.print_exc()
            await asyncio.sleep(0.1)