py-as-zhong/exchange/binance_spot_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 utils
import model
from decimal import Decimal
from decimal import ROUND_HALF_UP, ROUND_FLOOR

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


class BinanceSpotRest:

    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('不支持colo高速线路')
            self.HOST = 'https://api.binance.com'
        else:
            self.HOST = 'https://api.binance.com'
        self.params = params
        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}"
        self.name = self.params.name
        self._SESSIONS = dict()
        self.callback = {
            "onMarket":empty_call,
            "onPosition":empty_call,
            "onOrder":empty_call,
            "onEquity":empty_call,
            "onTicker":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.stop_flag = 0
        self.coin_value = 0.0
        self.cash_value = 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_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

        if params != None:
            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() 
            res_msg = msg + f' 回报 {res}'
            self.logger.debug(res_msg)
            if code not in (200, 201, 202, 203, 204, 205, 206):
                print(f'URL:{url} PARAMS:{params} body:{body} ERROR:{res}')
                if code == 429 or code == 418:
                    self.callback['onExit'](f"{self.name} 即将触发限频封禁 紧急退出")
                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'):
        '''
            下单接口
        '''
        if symbol not in self.exchange_info:
            await self.before_trade()
            # amount = float(Decimal(str(amount//self.exchange_info[symbol].stepSize))*Decimal(str(self.exchange_info[symbol].stepSize)))
            # price = float(Decimal(str(price//self.exchange_info[symbol].tickSize))*Decimal(str(self.exchange_info[symbol].tickSize)))
            amount = utils.fix_amount(amount, self.exchange_info[symbol].stepSize)
            price = utils.fix_price(price, self.exchange_info[symbol].tickSize)
        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:
            raise Exception(f'下单参数错误 side:{origin_side}')
        if float(amount) <= 0.0:
            self.logger.error(f'下单参数错误 amount:{amount}')
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['filled_price'] = 0.0
            order_event['filled'] = 0.0
            order_event['client_id'] = cid
            self.callback["onOrder"](order_event)
            return None, 'amount error'
        if float(price) <= 0.0:
            self.logger.error(f'下单参数错误 price:{price}')
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['filled_price'] = 0.0
            order_event['filled'] = 0.0
            order_event['client_id'] = cid
            self.callback["onOrder"](order_event)
            return None, 'price error'
        params = {
            'symbol': symbol, 
            'quantity': utils.num_to_str(amount, self.exchange_info[symbol].stepSize),
            'side': side, 
            'type':order_type,
            'newClientOrderId':cid,
        }
        if order_type in ['LIMIT','STOP','TAKE_PROFIT']:
            params['price'] = utils.num_to_str(price, self.exchange_info[symbol].tickSize)
            params['timeInForce'] = 'GTC'
        if self.params.debug == 'True':
            await asyncio.sleep(0.1)
            return None, None
        else:
            # 再报单
            response, error = await self._request('POST', '/api/v3/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'/api/v3/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:
                pass
                # if 'status' in response:
                #     if response['status'] in ['CANCELED','EXPIRED']:  # 已撤销 删除本地订单表
                #         order_event = dict()
                #         order_event['status'] = "REMOVE"
                #         order_event['client_id'] = response["origClientOrderId"]
                #         order_event['order_id'] = response["orderId"]
                #         order_event['filled'] = float(response["executedQty"])
                #         order_event['filled_price'] = float(response["cummulativeQuoteQty"])/float(response["executedQty"]) if float(response["executedQty"]) > 0 else 0
                #         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'/api/v3/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['fee'] = 0.0   # 查询订单信息中没有手续费信息
                        order_event['filled'] = float(response["executedQty"])
                        order_event['filled_price'] = float(response["cummulativeQuoteQty"])/float(response["executedQty"]) if float(response["executedQty"]) > 0 else 0
                        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', '/api/v3/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_history_order(self):
        params = {
            "symbol":self.symbol,
            "limit":1000,
            "startTime":1635815135000,
            "endTime":1635901535000,
        }
        response, error = await self._request('GET', '/api/v3/allOrders', params=params)
        import json
        fp = open("123.csv", "w")
        json.dump(response,fp)
        return response
    
    async def get_server_time(self):
        params = {}
        response = await self._request('GET', '/api/v3/time', params=params)
        return response

    async def before_trade(self):
        response, error = await self._request('GET', '/api/v3/exchangeInfo', params=None)
        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'])
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i['symbol'].upper()
                exchange_info.multiplier = 1
                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)

    async def get_equity(self):
        res, err = await self.get_account()
        if res:
            for i in res['balances']:
                if self.quote == i['asset'].upper():
                    cash = float(i['free']) + float(i['locked'])
                    self.callback['onEquity']({
                        self.quote:cash
                    })
                    self.cash_value = cash
                if i['asset'].upper() == self.base:
                    coin = float(i['free']) + float(i['locked'])
                    self.callback['onEquity']({
                        self.base:coin
                    })
                    self.coin_value = coin
        if err:
            print('获取账户信息错误',err)
        await asyncio.sleep(1)
    
    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/v3/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/v3/futures/transfer', params=params, HOST='https://api.binance.com')
        print(f'转账结果 {response}')
        return response

    async def get_account(self):
        return await self._request('GET','/api/v3/account', params={})

    async def get_ticker(self):
        res ,err = await self._request('GET', '/api/v3/ticker/bookTicker', params=None)
        if res:
            for i in res:
                if i['symbol'] == self.symbol:
                    ap = float(i['bidPrice'])
                    bp = float(i['askPrice'])
                    mp = (ap+bp)*0.5
                    d = {"name":self.name,'mp': mp, 'bp':bp, 'ap':ap}
                    self.callback['onTicker'](d)
                    return d
        if err:
            self.logger.error(err)
            return None 

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

    async def check_position(self, hold_coin=0.0):
        '''
            重置账户挂单和仓位 已支持全品种
        '''
        try:
            self.logger.info('检查遗漏订单')
            response, error = await self._request('GET', '/api/v3/openOrders', params={})
            self.logger.info(response)
            self.logger.info(error)
            if error:self.logger.error(error)
            if response:
                for i in response:
                    res = await self.cancel_order(order_id=i['orderId'], symbol=i['symbol'])
                    await asyncio.sleep(0.1)
                    self.logger.info(res)
            self.logger.info('检查遗漏仓位')
            ###########
            res ,err = await self._request('GET', '/api/v3/ticker/bookTicker', params=None)
            tickers_mp = dict()
            if res:
                for i in res:
                    ap = float(i['bidPrice'])
                    bp = float(i['askPrice'])
                    mp = (ap+bp)*0.5
                    tickers_mp[i['symbol']] = mp 
            if err:
                self.logger.error(err)
            ###########
            if self.exchange_info == dict():
                await self.before_trade()
            ###########
            response, error = await self._request('GET','/api/v3/account', params={})
            if error is not None:self.logger.error(error)
            if response:
                for i in response['balances']:
                    asset = i['asset']
                    if asset in ['BNB','USDT', 'TUSD']:
                        continue
                    symbol = asset + 'USDT'
                    if symbol not in tickers_mp:
                        continue
                    coin = abs(float(i['free']))+abs(float(i['locked']))
                    if coin == 0.0:
                        continue
                    mp = tickers_mp[symbol]
                    coin_value = coin * mp
                    if symbol == self.symbol:
                        _hold_coin = hold_coin
                    else:
                        _hold_coin = 0                
                    diff = _hold_coin - coin_value
                    diff *= 0.99 # 避免无法下单
                    self.logger.info(f'需要调整现货仓位{diff}usd')
                    if diff > 20.0:
                        self.logger.info('买入现货')
                        res, err = await self.take_order(
                            symbol,
                            diff/mp,
                            'kd',
                            1,
                            utils.get_cid(),
                            'MARKET'
                        )
                        self.logger.info(res)
                        self.logger.info(err)
                    elif diff < -20.0:
                        self.logger.info('卖出现货')
                        res, err = await self.take_order(
                            symbol,
                            -diff/mp,
                            'kk',
                            1,
                            utils.get_cid(),
                            'MARKET'
                        )
                        self.logger.info(res)
                        self.logger.info(err)
            self.logger.info('遗留仓位检测完毕')
        except:
            self.logger.error("清仓程序执行出错")
            self.logger.error(traceback.format_exc())
        return

    async def go(self):
        '''
            盘前
            获取市场信息
            获取账户信息
            更改仓位模式（期货）
            清空仓位和挂单
            盘中
            更新账户信息
            更新挂单列表
            更新仓位信息
            更新延迟信息
        '''
        print('Rest循环器启动')
        interval = 60  # 不能太快防止占用限频
        ### beforeTrade
        await self.before_trade()
        await asyncio.sleep(1)
        ### onTrade
        loop = 0
        while 1:
            loop += 1
            try:
                # 停机信号
                if self.stop_flag:
                    return
                # 更新账户
                res, err = await self.get_account()
                if res:
                    for i in res['balances']:
                        if self.quote == i['asset'].upper():
                            self.callback['onEquity']({
                                self.quote: float(i['free']) + float(i['locked'])
                            })
                        if i['asset'].upper() == self.base:
                            self.callback['onEquity']({
                                self.base: float(i['free']) + float(i['locked'])
                            })
                await asyncio.sleep(interval)
                # 打印延迟
                self.get_delay_info()
                self.logger.debug(f'报单延迟 平均{self.avg_delay}ms 最大{self.max_delay}ms')
            except asyncio.CancelledError:
                return
            except:
                traceback.print_exc()
                await asyncio.sleep(30)
    
    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)