py-as-zhong/exchange/bybit_usdt_swap_rest.py

import aiohttp
import time
import asyncio
import ujson
import hmac
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

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

def parse_params_to_str(params):
    url = ''
    for key in sorted(params.keys()):
        value = params[key]
        if isinstance(value, bool):
            if params[key]:
                value = "true"
            else:
                value = "false"
        url = url + str(key) + '=' + str(value) + '&'
    return url[0:-1]

class BybitUsdtSwapRest:

    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('不支持colo高速线路')
            self.HOST = 'https://api.bybit.com'
        else:
            self.HOST = 'https://api.bybit.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._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.tickSize = None
        self.stepSize = None
        self.delays = []
        self.max_delay = 0
        self.avg_delay = 0
        self.proxy = None
        if 'win' in sys.platform:
            self.proxy = self.params.proxy
        self.logger = self.get_logger()
        self.mp_from_rest = None
        self.stop_flag = 0
        self.coin_value = 0.0
        self.cash_value = 0.0
        self.min_trade_amount = 0.0
        self.rate_limit_left = 120
        #### 指定发包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]

    def get_sign(self, params, secret_key):
        message = parse_params_to_str(params)
        # mac = hmac.new(bytes(secret_key, encoding='utf-8'), bytes(message, encoding='utf-8'), digestmod='sha256').digest()

        hash = hmac.new(bytes(secret_key, encoding='utf-8'), bytes(message, encoding='utf-8'), hashlib.sha256)
        signature = hash.hexdigest()
        return signature

    async def _request(self, method, uri, body=None, params=None, auth=False):
        url = urljoin(self.HOST, uri)
        headers = {}
        if auth:
            if method in ['GET']:
                timestamp = str(int(time.time()*1000))
                params['timestamp'] = timestamp
                params['api_key'] = self.params.access_key
                params['recv_window'] = "15000"
                params['sign'] = self.get_sign(params, self.params.secret_key)
                headers = {}
            elif method == 'POST':
                timestamp = str(int(time.time()*1000))
                params['timestamp'] = timestamp
                params['api_key'] = self.params.access_key
                params['recv_window'] = "15000"
                params['sign'] = self.get_sign(params, self.params.secret_key)
                headers = {}
        url = url + '?' + parse_params_to_str(params)
        # 发起请求
        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, data=ujson.dumps(params), 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(content_type=None) 
            delay = int(1000*(time.time() - start_time))
            self.delays.append(delay)
            res_msg = msg + f' 回报 {res}'
            self.logger.debug(res_msg)
            #### 检查频率限制
            if 'rate_limit_status' in res:
                self.rate_limit_left = res['rate_limit_status']
                if self.rate_limit_left < 20:
                    self.callback['onExit'](f"{self.name} 即将触发限频封禁 紧急退出")
            ####
            if code not in (200, 201, 202, 203, 204, 205, 206) or res['ret_code'] not in [0]:
                self.logger.error(f'URL:{url} PARAMS:{params} body:{body} ERROR:{res}')
                return None, str(res)
            return res, None
        except Exception as e:
            print(f'{self.name} rest 请求出错', str(e))
            self.logger.error(f'请求错误 {msg}'+str(e))
            self.logger.error(traceback.format_exc())
            return None, e

    async def get_position(self):
        response, error = await self._request('GET', '/private/linear/position/list', params={}, auth=1)
        if error:
            print(error)
        if response:
            position = model.Position()
            for j in response['result']:
                i = j['data']
                if i['symbol'] == self.symbol:
                    side = i['side'].lower()
                    pos = float(i['size'])
                    price = float(i['entry_price'])
                    if side == 'buy':
                        position.longPos = pos
                        position.longAvg = price
                    elif side == 'sell':
                        position.shortPos = pos
                        position.shortAvg = price
            return position

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

    async def check_position(self, hold_coin=0.0):
        '''
            现货交易  已支持全品种
        '''
        try:
            #######################
            self.logger.info("清空挂单")
            params = {
                'symbol':self.symbol,
            }
            response, error = await self._request('POST', '/private/linear/order/cancel-all', params=params, auth=1)
            if error:
                self.logger.info("全部撤单失败")
            if response:
                self.logger.info("全部撤单成功")
            #############################
            self.logger.info("清空仓位")
            res, err = await self._request('GET', '/private/linear/position/list', params={}, auth=1)
            if err:
                self.logger.info(err)
            if res:
                for i in res['result']:
                    if float(i['data']['size']) > 0:
                        ####
                        response ,error = await self._request('GET',f'/v2/public/orderBook/L2', params={"symbol":i['data']['symbol']}, auth=0)
                        if response:
                            bids = []
                            asks = []
                            for j in response['result']:
                                if j['side'].lower() == 'buy':
                                    bids.append(float(j['price']))
                                if j['side'].lower() == 'sell':
                                    asks.append(float(j['price'])) 
                            ap = min(asks)
                            bp = max(bids)
                            mp = (ap+bp)*0.5
                        if error:
                            print(error)
                        ####
                        if self.exchange_info == dict():
                            await self.before_trade()
                        ####
                        if i['data']['side'].lower() == 'buy':
                            _side = "pd"
                            _price = utils.fix_price(mp * 0.999, self.exchange_info[i['data']['symbol']].tickSize)
                        elif i['data']['side'].lower() == 'sell':
                            _side = "pk"
                            _price = utils.fix_price(mp * 1.001, self.exchange_info[i['data']['symbol']].tickSize)
                        ####
                        response = await self.take_order(
                            i['data']['symbol'],
                            float(i['data']['size']),
                            _side,
                            _price,
                            utils.get_cid()
                        )
                        if response:
                            print(response)
            #######################        
        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'):
        ''' 
            下单
            price: 限价时 为 必填  
            marginCoin: 保证金币种
            size: 限价时 为 数量  市价买 为额度 卖为数量
            side: open_long open_short close_long close_short
            orderType: limit(限价)  market(市价)
            timeInForceValue:normal(普通限价订单)   postOnly(只做maker,市价不允许使用这个)  ioc(立即成交并取消剩余)  fok(全部成交或立即取消)
            presetTakeProfitPrice: 预设止盈价格
            presetStopLossPrice: 预设止损价格
        '''
        if origin_side =='kd':
            side = "Buy"
            reduce_only = "false"
        elif origin_side =='pd':
            side = "Sell"
            reduce_only = "true"
        elif origin_side =='kk':
            side = "Sell"
            reduce_only = "false"
        elif origin_side =='pk':
            side = "Buy"
            reduce_only = "true"
        else:
            print("合约不允许此交易方向")
            return None
        if self.rate_limit_left < 40 and origin_side in ['kd','kk']:
            print("即将触发限频 停止开仓单发单")
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['filled_price'] = 0.0                
            order_event['fee'] = 0.0
            order_event['filled'] = 0.0
            order_event['client_id'] = cid
            self.callback["onOrder"](order_event)
            return
        if symbol not in self.exchange_info:
            await self.before_trade()
            # amount = utils.fix_amount(amount, self.exchange_info[symbol].stepSize)
            # price = utils.fix_price(price, self.exchange_info[symbol].tickSize)
            # amount = float(Decimal(str(amount))//Decimal(str(self.exchange_info[symbol].stepSize))*Decimal(str(self.exchange_info[symbol].stepSize)))
            # price = float(int(Decimal(str(price))/Decimal(str(self.exchange_info[symbol].tickSize)))*Decimal(str(self.exchange_info[symbol].tickSize)))
        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['fee'] = 0.0
            order_event['filled'] = 0.0
            order_event['client_id'] = cid
            self.callback["onOrder"](order_event)
            return None
        if float(amount) <=0:
            self.logger.error(f'下单参数错误 amount:{amount}')
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['filled_price'] = 0.0                
            order_event['fee'] = 0.0
            order_event['filled'] = 0.0
            order_event['client_id'] = cid
            self.callback["onOrder"](order_event)
            return None
        params = {
            "order_link_id":cid,
            "symbol": symbol,
            "qty":utils.num_to_str(amount, self.exchange_info[symbol].stepSize),
            "side": side,
            "reduce_only":reduce_only,
            "close_on_trigger":False,
            "time_in_force":"GoodTillCancel",
        }
        if order_type == 'limit':
            params["order_type"] = "Limit"
            params["price"] = utils.num_to_str(price, self.exchange_info[symbol].tickSize)
        elif order_type == 'market':
            params["order_type"] = "Market"
        if self.params.debug == 'True':
            return await asyncio.sleep(0.1)
        else:
            # 发单
            if order_type == 'limit':
                response, error = await self._request('POST', '/private/linear/order/create', params=params, auth=1)
            elif order_type == 'market':
                response, error = await self._request('POST', '/private/linear/order/create', params=params, auth=1)
            # 再更新
            if response:
                # 增加新的
                if 'result' in response:
                    order_event = dict()
                    order_event['status'] = "NEW"
                    order_event['client_id'] = cid
                    order_event['order_id'] = response['result']["order_id"]
                    self.callback["onOrder"](order_event)
            if error:
                order_event = dict()
                order_event['status'] = "REMOVE"
                order_event['filled_price'] = 0.0                
                order_event['fee'] = 0.0
                order_event['filled'] = 0.0
                order_event['client_id'] = cid
                self.callback["onOrder"](order_event)
        return response
    
    async def cancel_order(self, order_id=None, client_id=None):
        '''
            symbol	String	是	产品ID 必须大写
            marginCoin	String	是	保证金币种 必须大写
            orderId	String	是	订单号
        '''
        if order_id:
            response, error = await self._request('POST', f'/private/linear/order/cancel', params={'symbol':self.symbol,'order_id':order_id}, auth=1)
        elif client_id:
            response, error = await self._request('POST', f'/private/linear/order/cancel', params={'symbol':self.symbol,'order_link_id':client_id}, auth=1)
        else:
            raise Exception("撤单出错 没指定订单号")
        if response:
            pass
            # await self.check_order(order_id=response['data']['orderId'])
            # self.logger.debug(f'撤单回报 {response}')
            # order_event = dict()
            # order_event['status'] = "REMOVE"
            # order_event['filled_price'] = float(response['data']['price'])            
            # order_event['fee'] = float(response['data']["deal_fee"])
            # order_event['filled'] = float(response['data']['amount']) - float(response['data']['left'])
            # order_event['client_id'] = response['data']["client_id"]
            # self.callback["onOrder"](order_event)
        if error:
            pass
            # print("撤单失败",error)
            # self.logger.error(error)
        return response
    
    async def check_order(self, order_id=None, client_id=None):
        '''
            symbol	String	是	产品ID 必须大写
            orderId	String	是	订单号
        '''
        if order_id:
            response, error = await self._request('GET', f'/private/linear/order/search', params={'symbol':self.symbol, 'order_id':order_id}, auth=1)
        elif client_id:
            response, error = await self._request('GET', f'/private/linear/order/search', params={'symbol':self.symbol, 'order_link_id':client_id}, auth=1)
        else:
            return
        if response:
            self.logger.debug(f'查单回报 {response}')
            if response['result']:
                if response['result']["order_status"] == 'New':  # 新增订单
                    order_event = dict()
                    order_event['status'] = "NEW"
                    order_event['filled'] = 0
                    order_event['filled_price'] = 0
                    order_event['client_id'] = response['result']["order_link_id"] if "order_link_id" in response['result'] else ""
                    order_event['order_id'] = response['result']['order_id']
                    order_event['fee'] = 0.0
                    self.callback["onOrder"](order_event)
                    # print(order_event)
                elif response['result']["order_status"] in ['Filled','Cancelled']:  # 删除订单
                    # fee 负数是扣手续费 bitget没有返佣
                    order_event = dict() 
                    order_event['status'] = "REMOVE"
                    order_event['client_id'] = response['result']["order_link_id"] if "order_link_id" in response['result'] else ""
                    order_event['order_id'] = response['result']['order_id']
                    order_event['filled'] = float(response['result']["cum_exec_qty"])
                    order_event['filled_price'] = float(response['result']["last_exec_price"]) \
                        if 'last_exec_price' in response['result'] else float(response['result']['price'])
                    order_event['fee'] = float(response['result']['cum_exec_fee'])
                    self.callback["onOrder"](order_event)
                    # print(order_event)
        if error:
            print("查单失败",error)
            self.logger.error(error)
        return response

    async def get_order_list(self):
        params = {
            'market':self.symbol,
            'offset':100,
            "side":0,
            'limit':100,
        }
        response, error = await self._request('GET', '/perpetual/v1/order/pending', params=params, auth=1)
        if response is not None:
            for i in response['data']['records']:
                order_event = dict()
                order_event['symbol'] = self.symbol
                order_event['price'] = float(i["price"])
                order_event['amount'] = float(i["amount"])
                order_event['filled'] = float(i["amount"])-float(i["left"])
                order_event['filled_price'] = float(i["avg_price"])
                order_event['client_id'] = i["clientOid"]
                order_event['order_id'] = i['id']
                asset_fee = float(response['data']["asset_fee"])
                money_fee = float(response['data']["money_fee"])
                stock_fee = float(response['data']["stock_fee"])
                if asset_fee > 0.0: # 非amm品种
                    order_event['fee'] = asset_fee
                else: # amm品种
                    order_event['fee'] = money_fee if money_fee > 0.0 else stock_fee
                if response["data"]['status'] == 'not_deal':
                    order_event['status'] = "NEW"
                elif response["data"]['status'] in ['cancel','done']:
                    order_event['status'] = "REMOVE"
                else:
                    s = response["data"]['status']
                    self.logger.error(f"错误的订单状态 {s}")
                self.callback["onOrder"](order_event)
        if error:
            print(error)
        return response
    
    async def get_server_time(self):
        params = {}
        response = await self._request('GET', '/perpetual/api/v1/timestamp', params=params)
        return response

    async def get_account(self):
        '''
            symbol	String	是	产品ID 必须大写
            marginCoin	String	是	保证金币种
        '''
        return await self._request('GET','/v2/private/wallet/balance', params={}, auth=1)

    async def transfer(self):
        '''
            
        '''
        from uuid import uuid4
        params = {
            "transfer_id":str(uuid4()),
            "coin":"USDT",
            "amount":"200",
            "from_account_type":"SPOT",
            "to_account_type":"CONTRACT",
        }
        return await self._request('POST','/asset/v1/private/transfer', params=params, auth=1)

    async def get_all_sub_account(self):
        '''
            
        '''
        params = {
        }
        return await self._request('GET','/asset/v1/private/sub-member/member-ids', params=params, auth=1)

    async def get_market_details(self):
        return await self._request('GET',f'/v2/public/symbols', params={}, auth=0)

    async def get_ticker(self):
        ####
        response ,error = await self._request('GET',f'/v2/public/orderBook/L2', params={"symbol":self.symbol}, auth=0)
        if response:
            bids = []
            asks = []
            for j in response['result']:
                if j['side'].lower() == 'buy':
                    bids.append(float(j['price']))
                if j['side'].lower() == 'sell':
                    asks.append(float(j['price'])) 
            ap = min(asks)
            bp = max(bids)
            mp = (ap+bp)*0.5
            d = {"name":self.name,'mp': mp, 'bp':bp, 'ap':ap}
            self.callback['onTicker'](d)
            return d
        if error:
            self.logger.error(error)
            return None 

    async def before_trade(self):
        # 切换杠杆
        await self.change_position_side()
        # 获取市场最新价格
        res = await self.get_ticker()
        ticker_price = res["mp"]
        if isinstance(ticker_price, float):
            self.mp_from_rest = ticker_price
        # 获取市场基本情况
        res, error = await self.get_market_details()
        if error:
            pass 
        else:
            for i in res['result']:
                if i['name'] == self.symbol:
                    self.stepSize = float(i['lot_size_filter']['qty_step'])
                    self.tickSize = float(i['price_filter']['tick_size'])
                    self.min_trade_amount = 0.0
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i['name']
                exchange_info.multiplier = 1
                exchange_info.stepSize = float(i['lot_size_filter']['qty_step'])
                exchange_info.tickSize = float(i['price_filter']['tick_size'])
                self.exchange_info[exchange_info.symbol] = exchange_info

    async def get_equity(self):
        # 更新账户
        res, err = await self.get_account()
        if err:print(err)
        if res:
            for i in res['result']:
                if self.quote == i:
                    cash = float(res['result'][i]['equity'])
                    self.callback['onEquity']({
                        self.quote:cash
                    })
                    self.cash_value = cash

    async def change_position_side(self):
        '''切换到全仓'''
        res ,err = await self._request(
            'POST',
            '/private/linear/position/switch-mode',
            params={
                'symbol':self.symbol,
                'mode':"BothSide",},
            auth=1
            )
        if err:print(err)
        if res:print(res)

    async def go(self):
        interval = 60  # 不能太快防止占用限频
        await self.before_trade()
        await asyncio.sleep(1)
        while 1:
            try:
                # 停机信号
                if self.stop_flag:return
                # 更新账户
                res, err = await self.get_account()
                if err:print(err)
                if res:
                    for i in res['result']:
                        if self.quote == i:
                            cash = float(res['result'][i]['equity'])
                            self.callback['onEquity']({
                                self.quote:cash
                            })
                            self.cash_value = cash
                # 更新仓位
                p = await self.get_position()
                self.callback['onPosition'](p)
                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]
                    # 只能用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]:
                        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]
                    # 只能用oid查单
                    if oid:
                        asyncio.get_event_loop().create_task(self.check_order(order_id=oid))
        except Exception as e:
            traceback.print_exc()
            self.logger.error("执行信号出错"+str(e))
            await asyncio.sleep(0.1)