py-as-zhong/exchange/binance_coin_swap_rest.py

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

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


class BinanceCoinSwapRest:

    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('不支持colo高速线路')
            self.HOST = 'https://dapi.binance.com'
        else:
            self.HOST = 'https://dapi.binance.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
        if len(self.params.pair.split('_')) > 2:
            self.delivery = self.params.pair.split('_')[2] # 210924
            self.symbol += f"_{self.delivery}"
        else:
            self.symbol += '_PERP'
        self.data = {}
        self._SESSIONS = dict()
        self.callback = {
            "onMarket":empty_call,
            "onPosition":empty_call,
            "onOrder":empty_call,
            "onEquity":empty_call,
            "onExit":empty_call,
            }
        self.exchange_info = dict()
        self.stepSize = None
        self.tickSize = None
        self.delays = []
        self.avg_delay = 0
        self.max_delay = 0
        self.proxy = None
        self.broker_id = self.params.broker_id
        if 'win' in sys.platform:
            self.proxy = self.params.proxy
        self.logger = self.get_logger()
        self.multiplier = None
        self.mp = 0.0 # 初始mp 在check_position中更新
        #### 指定发包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_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, HOST=None):

        headers = {}
        headers["Content-Type"] = "application/json"
        headers['X-MBX-APIKEY'] = self.params.access_key

        params['timestamp']=int(time.time())*1000
        query_string = "&".join(["{}={}".format(k, params[k]) for k in params.keys()])
        signature = hmac.new(self.params.secret_key.encode(), msg=query_string.encode(), digestmod=hashlib.sha256).hexdigest()
        params['signature']=signature
        if HOST == None:
            url = self.HOST + uri
        else:
            url = HOST + uri
        # 发起请求
        timeout = aiohttp.ClientTimeout(10)
        session = self._get_session(url)
        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=params, 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()
            if code not in (200, 201, 202, 203, 204, 205, 206):
                print(f'URL:{url} PARAMS:{params} body:{body} ERROR:{res}')
                self.logger.error(res)
                return None, str(res)
            delay = int(1000*(time.time() - start_time))
            self.delays.append(delay)
            return res, None
        except Exception as e:
            print('网络请求错误')
            print(f'URL:{url} PARAMS:{params} ERROR:{e}')
            self.logger.error(e)
            self.logger.error(traceback.format_exc())
            return None, str(e)

    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 take_order(self, symbol, amount, origin_side, price, cid="", order_type='LIMIT'):
        '''
            下单接口
            input amount 单位为 币 需要转换为 张
        '''
        if symbol not in self.exchange_info:
            await self.before_trade()
        # price = float(Decimal(str(price//self.tickSize))*Decimal(str(self.tickSize)))
        # amount = amount * price / self.multiplier
        # amount = float(Decimal(str(amount//self.stepSize))*Decimal(str(self.stepSize)))
        # 转换为张
        if origin_side =='kd':
            side = 'BUY'
            positionSide = 'LONG'
        elif origin_side =='pd':
            side = 'SELL'
            positionSide = 'LONG'
        elif origin_side =='kk':
            side = 'SELL'
            positionSide = 'SHORT'
        elif origin_side =='pk':
            side = 'BUY'
            positionSide = 'SHORT'
        else:
            raise Exception(f'下单参数错误 side:{origin_side}')
        if amount <= 0: 
            self.logger.error(f'下单参数错误 amount:{amount}')
            return None
        if price <= 0:
            self.logger.error(f'下单参数错误 price:{price}')
            return None
        params = {
            'symbol': symbol, 
            'quantity': amount,
            'side': side, 
            'positionSide': positionSide, 
            'type':order_type,
            'newClientOrderId':cid,
        }
        if order_type in ['LIMIT','STOP','TAKE_PROFIT']:
            params['price'] = price
            params['timeInForce'] = 'GTC'
        if self.params.debug == 'True':
            await asyncio.sleep(0.1)
            return None, None
        else:
            # 再报单
            response, error = await self._request('POST', '/dapi/v1/order', params=params)
            # 再更新
            if response is not None:
                if 'orderId' in response:
                    order_event = dict()
                    order_event['status'] = "NEW"
                    order_event['client_id'] = params["newClientOrderId"]
                    order_event['order_id'] = response['orderId']
                    self.callback["onOrder"](order_event)
            if error:
                order_event = dict()
                order_event['status'] = "REMOVE"
                order_event['filled_price'] = 0
                order_event['filled'] = 0
                order_event['client_id'] = params["newClientOrderId"]
                self.callback["onOrder"](order_event)
        return response, error
    
    async def cancel_order(self, order_id=None, client_id=None, symbol=None):
        params = {
            "symbol": self.symbol if symbol==None else symbol,
        }
        if order_id:
            params["orderId"] = order_id
        if client_id:
            params["origClientOrderId"] = client_id
        if self.params.debug == 'True':
            await asyncio.sleep(0.1)
            return None
        else:
            response, error = await self._request('DELETE', f'/dapi/v1/order', params=params)
            if error:
                print("撤单失败",error)
                # if 'Unknown order sent.' in error: 
                # 撤单失败 可能已经撤单 是否发生成交需要rest查
                if client_id:await self.check_order(client_id=client_id)
                if order_id:await self.check_order(order_id=order_id)
                return error
            if response:
                if 'status' in response:
                    if response['status'] in ['CANCELED','EXPIRED']:  # 已撤销 删除本地订单表
                        order_event = dict()
                        order_event['status'] = "REMOVE"
                        order_event['client_id'] = response["clientOrderId"]
                        order_event['order_id'] = response["orderId"]
                        order_event['filled'] = float(response["executedQty"]) * self.multiplier / float(response["price"])
                        order_event['filled_price'] = float(response["price"])
                        self.callback['onOrder'](order_event)
                return response

    async def check_order(self, order_id=None, client_id=None, symbol=None):
        params = {
            "symbol": self.symbol if symbol==None else symbol,
        }
        if order_id:
            params["orderId"] = order_id
        if client_id:
            params["origClientOrderId"] = client_id
        if self.params.debug == 'True':
            await asyncio.sleep(0.1)
            return None
        else:
            response, error = await self._request('GET', f'/dapi/v1/order', params=params)
            if error:
                print("查单失败", error)
                if 'Order does not exist' in error: 
                    # 这种情况也可能还会有成交
                    # 在订单从引擎到数据库的间隙查单会提示不存在 但实际有成交
                    pass
                return error
            if response:
                if 'status' in response:
                    # 需要删除本地订单表的情况
                    if response['status'] in ['CANCELED','EXPIRED','FILLED']:  
                        order_event = dict()
                        order_event['status'] = "REMOVE"
                        order_event['client_id'] = response["clientOrderId"]
                        order_event['order_id'] = response["orderId"]
                        order_event['filled'] = float(response["executedQty"]) * self.multiplier / float(response["price"])
                        order_event['filled_price'] = float(response["price"])
                        self.callback['onOrder'](order_event)
                    elif response['status'] in ["NEW"]: # 需要更新本地表的情况
                        order_event = dict()
                        order_event['status'] = "NEW"
                        order_event['client_id'] = response["clientOrderId"]
                        order_event['order_id'] = response['orderId']
                        self.callback['onOrder'](order_event)
                return response
    
    async def get_order_list(self):
        '''
            获取挂单表
        '''
        response, error = await self._request('GET', '/dapi/v1/openOrders', params={'symbol':self.symbol})
        orders = [] # 查询当前挂单 只可能出现 new 和 partfill 默认成交为0 只有 done状态的订单才考虑是否有成交
        if response:
            for i in response:
                order_event = dict()
                order_event['status'] = "NEW"
                order_event['filled'] = 0
                order_event['filled_price'] = 0
                order_event['client_id'] = i["clientOrderId"]
                order_event['order_id'] = i['orderId']
                self.callback["onOrder"](order_event)
                orders.append(order_event)
        if error:
            print('查询列表出错',error)
        return orders
    
    async def get_server_time(self):
        params = {}
        response = await self._request('GET', '/dapi/v1/time', params=params)
        return response

    async def change_pos_side(self, dual='true'):
        '''
            获取仓位模式
        '''
        params = {'dualSidePosition':dual}
        response = await self._request('POST', '/dapi/v1/positionSide/dual', params=params)
        return response

    async def get_ticker(self):
        params = {'symbol': self.symbol}
        response = await self._request('GET', '/dapi/v1/ticker/bookTicker', params=params)
        ap = float(response[0][0]["askPrice"])
        bp = float(response[0][0]["bidPrice"])
        mp = (ap+bp)*0.5
        d = {"name":self.name,'mp': mp, 'bp':bp, 'ap':ap}
        return d

    async def before_trade(self):
        params = {}
        response, error = await self._request('GET', '/dapi/v1/exchangeInfo', params=params)
        if response:
            for i in response['symbols']:
                if self.symbol in i['symbol'].upper():
                    self.tickSize = float(i['filters'][0]['tickSize'])
                    self.stepSize = float(i['filters'][1]['stepSize'])
                    self.multiplier = float(i['contractSize'])
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i['symbol'].upper()
                exchange_info.multiplier = float(i['contractSize'])
                exchange_info.tickSize = float(i['filters'][0]['tickSize'])
                exchange_info.stepSize = float(i['filters'][1]['stepSize'])
                self.exchange_info[exchange_info.symbol] = exchange_info
        if error:
            print('获取市场信息错误',error)
        ###
        ticker = await self.get_ticker()
        ###
        res = await self.get_account()
        if res:
            for i in res:
                if self.base == i['asset'].upper():
                    self.data['equity'] = float(i['balance']) * ticker["mp"]
                    self.callback['onEquity'](self.data['equity'])
        if error:
            print('获取账户信息错误',error)
    
    async def universalTransfer(self, _type='UMFUTURE_MAIN', asset='USDT', amount=0):
        params = {}
        params['type'] = _type
        params['asset'] = asset
        params['amount'] = amount
        print('发起提现')
        response = await self._request('POST', '/sapi/v1/asset/transfer', params=params, HOST='https://api.binance.com')
        print(f'提现结果 {response}')
        return response

    async def futuresTransfer(self, _type='2', asset='USDT', amount=0):
        '''
            1: 现货账户向USDT合约账户划转
            2: USDT合约账户向现货账户划转
            3: 现货账户向币本位合约账户划转
            4: 币本位合约账户向现货账户划转
        '''
        params = {}
        params['type'] = _type
        params['asset'] = asset
        params['amount'] = amount
        print('发起转账')
        response = await self._request('POST', '/sapi/v1/futures/transfer', params=params, HOST='https://api.binance.com')
        print(f'转账结果 {response}')
        return response

    async def get_account(self):
        response, error = await self._request('GET','/dapi/v1/balance', params={})
        return response

    async def buy_token(self):
        '''买入平台币'''
        pass


    async def check_position(self, hold_coin=0):
        '''检查是否存在非运行币种的仓位并take平仓'''
        self.logger.info('检查遗漏订单')
        response, error = await self._request('GET', '/dapi/v1/openOrders', params={'symbol':self.symbol})
        self.logger.info(response)
        self.logger.info(error)
        if error:self.logger.error(error)
        if response:
            for i in response:
                symbol = i['symbol']
                order_id = i['orderId']
                if symbol == self.symbol: 
                    res = await self.cancel_order(order_id=i['orderId'], symbol=i['symbol'])
                    await asyncio.sleep(0.1)
                    self.logger.info(res)
        self.logger.info('检查遗漏仓位')
        response, error = await self._request('GET','/dapi/v2/positionRisk', params={})
        if error is not None:self.logger.error(error)
        if response:
            for i in response:
                symbol = i['symbol']
                if symbol == self.symbol:
                    if i['positionSide'] == 'LONG':
                        longPos = abs(float(i['positionAmt']))
                        longAvg = abs(float(i['entryPrice']))
                        if longPos > 0:
                            self.logger.info('发现多头遗留仓位 进行平仓')
                            res, error = await self.take_order(
                                symbol,
                                longPos,
                                'pd',
                                1,
                                utils.get_cid(),
                                order_type='MARKET',
                            )
                            self.logger.info(res)
                            self.logger.info(error)
                    if i['positionSide'] == 'SHORT':
                        shortPos = abs(float(i['positionAmt']))
                        shortAvg = abs(float(i['entryPrice']))
                        if shortPos > 0:
                            self.logger.info('发现空头遗留仓位 进行平仓')
                            res, error = await self.take_order(
                                symbol,
                                shortPos,
                                'pk',
                                1,
                                utils.get_cid(),
                                order_type='MARKET',
                            )
                            self.logger.info(res)
                            self.logger.info(error)
        self.logger.info('遗留仓位检测完毕')
        return

    async def get_position(self):
        '''
            获取仓位信息
        '''
        response, error = await self._request('GET','/dapi/v2/positionRisk', params={'symbol':self.symbol})
        longPos, shortPos = 0, 0
        longAvg, shortAvg = 0, 0
        position = {
            "longPos":abs(longPos),
            "shortPos":abs(shortPos), 
            "longAvg":abs(longAvg), 
            "shortAvg":abs(shortAvg)}
        if response:
            for i in response:
                if i['symbol'] == self.symbol:
                    if i['positionSide'] == 'LONG':
                        longPos = float(i['positionAmt'])
                        longAvg = float(i['entryPrice'])
                    if i['positionSide'] == 'SHORT':
                        shortPos = float(i['positionAmt'])
                        shortAvg = float(i['entryPrice'])
            position = model.Position()
            position.longPos = abs(longPos)
            position.longAvg = abs(longAvg)
            position.shortPos = abs(shortPos)
            position.shortAvg = abs(shortAvg)
            self.callback['onPosition'](position)
        return position

    async def go(self):
        '''
            盘前
            获取市场信息
            获取账户信息
            更改仓位模式（期货）
            清空仓位和挂单
            盘中
            更新账户信息
            更新挂单列表
            更新仓位信息
            更新延迟信息
        '''
        print('Rest循环器启动')
        interval = 60  # 不能太快防止占用限频
        ### beforeTrade
        await self.before_trade()
        await asyncio.sleep(1)
        await self.change_pos_side()
        await asyncio.sleep(1)
        ### onTrade
        loop = 0
        while 1:
            loop += 1
            try:
                # 更新账户
                res = await self.get_account()
                if res:
                    for i in res:
                        if self.quote == i['asset'].upper():
                            self.data['equity'] = float(i['balance'])
                            self.callback['onEquity'](self.data['equity'])
                # 更新仓位
                position = await self.get_position()
                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(30)

    def get_data(self):
        return self.data
    
    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]
                    if cid:
                        asyncio.get_event_loop().create_task(self.cancel_order(client_id=cid))
                    elif 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]:
                        asyncio.get_event_loop().create_task(self.take_order(
                            self.symbol,
                            i[0],
                            i[1],
                            i[2],
                            i[3]
                        ))
            for order_name in orders:
                if 'Check' in order_name:
                    cid = orders[order_name][0]
                    # oid = orders[order_name][1]
                    asyncio.get_event_loop().create_task(self.check_order(client_id=cid))
        except Exception as e:
            traceback.print_exc()
            self.logger.error("执行信号出错"+str(e))
            await asyncio.sleep(0.1)