py-as-zhong/exchange/gate_spot_rest.py

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

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


class GateSpotRest:

    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('使用colo高速线路')
            self.HOST = 'https://apiv4-private.gateapi.io'
        else:
            self.HOST = 'https://api.gateio.ws'
        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.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.0
        self.avg_delay = 0.0
        self.proxy = None
        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]

    def generate_signature(self, method, uri, query_param=None, body=None):
        t = time.time()
        m = hashlib.sha512()
        m.update((body or "").encode('utf-8'))
        hashed_payload = m.hexdigest()
        s = '%s\n%s\n%s\n%s\n%s' % (method, uri, query_param or "", hashed_payload, t)
        sign = hmac.new(self.params.secret_key.encode('utf-8'), s.encode('utf-8'), hashlib.sha512).hexdigest()
        return {'KEY': self.params.access_key, 'Timestamp': str(t), 'SIGN': sign}

    async def _request(self, method, uri, body=None, params=None, auth=False):
        url = urljoin(self.HOST, uri)
        if method == "GET":
            headers = {
                "Content-type": "application/x-www-form-urlencoded",
                "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) "
                    "Chrome/39.0.2171.71 Safari/537.36"
            }
        else:
            headers = {
                "Accept": "application/json",
                "Content-type": "application/json"
            }
        if auth:
            if method == "POST":
                body = json.dumps(params)
                query_param = None
                sign_headers = self.generate_signature(method, uri, query_param, body)
                headers.update(sign_headers)
            if method == "GET" or method == "DELETE":
                query_param = ''
                for i in params:
                    query_param += f'{i}={params[i]}&'
                query_param = query_param[:-1]
                sign_headers = self.generate_signature(method, uri, query_param)
                headers.update(sign_headers)
        # 发起请求
        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))
            self.delays.append(delay)
            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}')
                return None, res
            return res, None
        except Exception as e:
            print(f'{self.name} 请求出错', e)
            self.logger.error('请求错误'+str(e))
            self.logger.error(traceback.format_exc())
            return None, e

    async def take_order(self, symbol, amount, origin_side, price, cid="", order_type='limit'):
        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 = 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 amount <= 0.0 or price <= 0.0:
            self.logger.error(f"下单参数错误 amount:{amount} price:{price}")
            order_event = dict()
            order_event['status'] = "REMOVE"
            order_event['client_id'] = cid
            order_event['filled_price'] = 0.0
            order_event['filled'] = 0.0
            order_event['fee'] = 0.0
            self.callback["onOrder"](order_event)
        params = {
            'text':cid,
            'currency_pair': symbol, 
            'amount': utils.num_to_str(amount, self.exchange_info[symbol].stepSize),
            'side': side, 
            'account':"spot",
            '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/v4/spot/orders', params=params, auth=1)
            if response:
                # 增加新的
                order_event = dict()
                order_event['status'] = "NEW"
                order_event['client_id'] = response["text"]
                order_event['order_id'] = response["id"]
                self.callback["onOrder"](order_event)
            if error:
                order_event = dict()
                order_event['status'] = "REMOVE"
                order_event['client_id'] = params["text"]
                order_event['filled_price'] = 0.0
                order_event['filled'] = 0.0
                order_event['fee'] = 0.0
                self.callback["onOrder"](order_event)
                return error
        return response

    async def check_order(self, order_id=None, client_id=None):
        params = {
            "currency_pair": self.symbol
        }
        if order_id:
            response, error = await self._request('GET', f'/api/v4/spot/orders/{order_id}', params=params, auth=1)
        elif client_id:
            response, error = await self._request('GET', f'/api/v4/spot/orders/{client_id}', params=params, auth=1)
        if response:
            if response['status'] in ['cancelled','closed']:  # 已撤销 或 全部成交
                order_event = dict()
                order_event['client_id'] = response["text"]
                order_event['order_id'] = response['id']
                order_event['filled'] = float(response["amount"]) - float(response["left"])
                order_event['filled_price'] = float(response["price"])
                order_event['fee'] = float(response["fee"])
                order_event['status'] = "REMOVE"
                self.callback['onOrder'](order_event)
            else: # 还在挂单中
                order_event = dict()
                order_event['client_id'] = response["text"]
                order_event['order_id'] = response['id']
                order_event['status'] = "NEW"
                self.callback['onOrder'](order_event)
        if error:
            pass
        return response

    async def cancel_order(self, order_id=None, client_id=None):
        params = {
            "currency_pair": self.symbol
        }
        if order_id:
            response, error = await self._request('DELETE', f'/api/v4/spot/orders/{order_id}', params=params, auth=1)
        elif client_id:
            response, error = await self._request('DELETE', f'/api/v4/spot/orders/{client_id}', params=params, auth=1)

        if response:
            # rest cancel 如果有回报 可能会和ws回报 产生重复处理
            self.logger.debug(f'撤单回报 {response}')
            # if response['status'] == 'cancelled':  # 已撤销
            #     order_event = dict()
            #     order_event['price'] = float(response["price"])
            #     order_event['amount'] = float(response["amount"])
            #     order_event['client_id'] = response["text"]
            #     order_event['order_id'] = response['id']
            #     order_event['filled'] = float(response["amount"]) - float(response["left"])
            #     order_event['filled_price'] = float(response["price"])
            #     order_event['fee'] = float(response["fee"])
            #     order_event['status'] = "REMOVE"
            #     self.callback['onOrder'](order_event)
        if error:
            return error
        return response
    
    async def get_order_list(self):
        params = {
            'currency_pair':self.symbol,
            'status':"open",
        }
        response, error = await self._request('GET', '/api/v4/spot/orders', params=params, auth=1)
        orders = [] # 重置本地订单列表
        if response is not None:
            for i in response:
                if i['side'] == 'buy':
                    side = 'kd'
                elif i['side'] == 'sell':
                    side = 'pd'
                else:
                    raise Exception(f"{self.name} wrong side")
                order_event = dict()
                order_event['price'] = float(i["price"])
                order_event['amount'] = float(i["amount"])
                order_event['client_id'] = i["text"]
                order_event['order_id'] = i['id']
                order_event['status'] = "NEW"
                self.callback['onOrder'](order_event)
        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/v4/spot/accounts', params={}, auth=1)

    async def get_position(self):
        '''获取持仓 symbol: BTC-USDT'''
        return await self._request('POST','/linear-swap-api/v1/swap_position_info', params={'contract_code':self.symbol}, auth=1)

    async def get_market_details(self):
        return await self._request('GET',f'/api/v4/spot/currency_pairs', params={}, auth=1)

    async def get_ticker(self):
        res, err = await self._request('GET',f'/api/v4/spot/tickers', params={"currency_pair":self.symbol}, auth=1)
        if res:
            ap = float(res[0]["lowest_ask"])
            bp = float(res[0]["highest_bid"])
            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 before_trade(self):
        res, error = await self.get_market_details()
        if error:
            pass
        else:
            for i in res:
                if i['id'] == self.symbol:
                    self.tickSize = float(Decimal("0.1")**Decimal(i['precision']))
                    self.stepSize = float(Decimal("0.1")**Decimal(i['amount_precision']))
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i['id']
                exchange_info.multiplier = 1
                exchange_info.tickSize = float(Decimal("0.1")**Decimal(i['precision']))
                exchange_info.stepSize = float(Decimal("0.1")**Decimal(i['amount_precision']))
                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:
                if self.quote == i['currency'].upper():
                    cash = float(i['available']) + float(i['locked'])
                    self.callback['onEquity']({
                        self.quote:cash
                        })
                    self.cash_value = cash
                if i['currency'].upper() == self.base:
                    coin = float(i['available']) + float(i['locked'])
                    self.callback['onEquity']({
                        self.base:coin
                        })
                    self.coin_value = coin

    async def buy_token(self):
        '''买入平台币'''
        # 获取u数量 平台币数量
        # 更新账户
        cash, token = 0.0, 0.0
        res, err = await self.get_account()
        if err:self.logger.error(err)
        if res:
            for i in res:
                if 'USDT' == i['currency'].upper():
                    cash = float(i['available']) + float(i['locked'])
                if 'GT' == i['currency'].upper():
                    token = float(i['available']) + float(i['locked'])
        self.logger.info(f"持u{cash} 持GT{token}")
        # 获取平台币价格
        res, err = await self._request('GET',f'/api/v4/spot/tickers', params={"currency_pair":'GT_USDT'}, auth=1)
        if err:print(err)
        if res:
            ap = float(res[0]["lowest_ask"])
            bp = float(res[0]["highest_bid"])
            mp = (ap+bp)*0.5
        # 判断是否需要买入
        token_value = token * mp
        if token_value < 30:
            self.logger.info(f"GT数量过少")
            if cash > 200:
                self.logger.info(f"准备买入GT")
                # 下单买入50uGT
                res = await self.take_order("GT_USDT", 50/mp, "kd", mp*1.001, "t-888", "limit")
                self.logger.info(res)
            else:
                self.logger.warning(f"现金不足 无法买入GT")
        else:
            self.logger.info(f"GT数量充足")

    async def check_position(self, hold_coin=0.0):
        '''
            清空挂单清空仓位
        '''
        try:
            #############################
            self.logger.info("获取挂单")
            params = {
            }
            response, error = await self._request('GET', '/api/v4/spot/open_orders', params=params, auth=1)
            if response is not None:
                for i in response:
                    params = {
                        "currency_pair": i['currency_pair']
                    }
                    for j in i['orders']:
                        oid = j['id']
                        r, e = await self._request('DELETE', f'/api/v4/spot/orders/{oid}', params=params, auth=1)
                        print(r,e)
            if error:
                self.logger.info(error)
            #############################
            res, err = await self.get_account()
            if err:self.logger.info(err)
            if res:
                coin = 0.0
                for i in res:
                    coin_name = i['currency'].upper()
                    if coin_name in ['GT','USDT', 'POINT']:
                        continue
                    symbol = coin_name + '_USDT'
                    if coin_name == self.base:
                        _hold_coin = hold_coin
                    else:
                        _hold_coin = 0
                    coin = float(i['available']) + float(i['locked'])
                    #################
                    ticker, _ = await self._request('GET',f'/api/v4/spot/tickers', params={"currency_pair":symbol}, auth=1)
                    if ticker:
                        ap = float(ticker[0]["lowest_ask"])
                        bp = float(ticker[0]["highest_bid"])
                        mp = (ap+bp)*0.5
                    #################
                    coin_value = coin * mp
                    diff = _hold_coin - coin_value
                    diff *= 0.99 # 避免无法下单
                    self.logger.info(f'{symbol}需要调整现货仓位{diff}usd')
                    if diff > 20.0:
                        res = await self.take_order(symbol, diff/mp, "kd", mp*1.001, "t-123", "limit")
                    elif diff < -20.0:
                        res = await self.take_order(symbol, -diff/mp, "kk", mp*0.999, "t-123", "limit")
            #############################
            params = {
                'currency_pair':self.symbol,
                'status':"open",
            }
            response, error = await self._request('GET', '/api/v4/spot/orders', params=params, auth=1)
            if response is not None:
                for i in response:
                    await self.cancel_order(order_id=i['id'])
                    await self.cancel_order(client_id=i['text'])
            if error:
                self.logger.info(error)        
        except:
            self.logger.error("清仓程序执行出错")
            self.logger.error(traceback.format_exc())
        return 

    async def go(self):
        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:
                        if self.quote == i['currency'].upper():
                            cash = float(i['available']) + float(i['locked'])
                            self.callback['onEquity']({
                                self.quote:cash
                            })
                            self.cash_value = cash
                        if i['currency'].upper() == self.base:
                            coin = float(i['available']) + float(i['locked'])
                            self.callback['onEquity']({
                                self.base:coin
                            })
                            self.coin_value = coin
                # 更新订单
                # res = await self.get_order_list()
                await asyncio.sleep(60)
                # 打印延迟
                self.get_delay_info()
                self.logger.debug(f'{self.name} rest报单延迟 平均{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]
                    # gate优先用oid撤单
                    if oid:
                        asyncio.get_event_loop().create_task(self.cancel_order(order_id=oid))
                    elif cid:
                        asyncio.get_event_loop().create_task(self.cancel_order(client_id=cid))
            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:
                    # gate优先用oid查单
                    cid = orders[order_name][0]
                    oid = orders[order_name][1]
                    if oid:
                        asyncio.get_event_loop().create_task(self.check_order(order_id=oid))
                    elif cid:
                        asyncio.get_event_loop().create_task(self.check_order(client_id=cid))
        except:
            # traceback.print_exc()
            await asyncio.sleep(0.1)