py-as-zhong/exchange/mexc_spot_rest.py

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

def empty_call(msg):
    print(f'空的回调函数 {msg}')
class MexcSpotRest:
    def __init__(self, params:model.ClientParams, colo=0):
        if colo:
            print('不支持colo高速线路')
            self.HOST = 'https://api.mexc.com'
        else:
            self.HOST = 'https://api.mexc.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.data = {}
        self._SESSIONS = dict()
        self.logger = self.get_logger()
        self.data['account'] = {}
        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.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.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(
            "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_sign(self, params, secret_key):
        key = secret_key.encode('utf-8')
        params = params.encode('utf-8')
        sign = base64.b64encode(hmac.new(key, params, digestmod=sha256).digest()).decode()
        return sign

    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, auth=False):
        url = urljoin(HOST, uri)
        headers = {'X-MEXC-APIKEY': self.access_key, 'Content-Type': 'application/json'}
        if params != None:
            params['timestamp'] = int(time.time())*1000
            query_string = "&".join(["{}={}".format(k, params[k]) for k in params.keys()])
            params['signature'] = self.get_sign(query_string, self.secret_key)
        
        if auth:
            if method == 'GET':
                headers = {
                    'X-MEXC-APIKEY': self.access_key,
                    'Content-Type': 'application/json'
                    }
            else:
                headers = {
                    'X-MEXC-APIKEY': self.access_key,
                    'Content-Type': 'application/x-www-form-urlencoded'
                }
        # 发起请求
        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=json.dumps(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) or res['code'] not in (0, 200):
                print(f'URL:{url} PARAMS:{params} body:{body} ERROR:{res}')
                return None, res
            return res, None
        except Exception as e:
            print(f'{self.name} rest 请求出错', str(e))
            self.logger.error('请求错误'+str(e))
            self.logger.error(traceback.format_exc())
            return None, e

    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 origin_side == 'kd':
            side = 'buy'
        elif origin_side == 'pd':
            side = 'sell'
        elif origin_side == 'kk':
            side = 'sell'
        elif origin_side == 'pk':
            side = 'buy'
        else:
            print("现货不允许此交易方向")
            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)))
        if 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
        if price <= 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
        # Mexc 参数可能有变 Galois
        params = {
            'access_id': self.params.access_key,
            'client_id': cid,
            'symbol': symbol,
            'amount': utils.num_to_str(amount, self.exchange_info[symbol].stepSize),
            'type': side,
            'price': utils.num_to_str(price, self.exchange_info[symbol].tickSize),
        }
        # logger.info(f'下单指令 {params}')
        if self.params.debug == 'True':
            return await asyncio.sleep(0.1)
        else:
            # 发单
            response, error = await self._request('POST', '/api/v3/order', body=params, auth=1)
            # 再更新
            if response:
                # logger.info(f'下单回报 {response}')
                # 增加新的
                if 'data' in response:
                    order_event = dict()
                    order_event['status'] = "NEW"
                    order_event['client_id'] = params["client_id"]
                    order_event['order_id'] = response['data']["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'] = params["client_id"]
                self.callback["onOrder"](order_event)
                return error
        return response
    
    async def cancel_order(self, order_id=None, client_id=None, symbol=None):
        if order_id:
            response, error = await self._request('DELETE', f'/api/v3/margin/order', params={'symbol': self.symbol, 'orderId': order_id}, auth=1)
        elif client_id:
            response, error = await self._request('DELETE', f'/api/v3/margin/order', params={'symbol': self.symbol, 'orderId': client_id}, auth=1)
        else:
            raise Exception("撤单出错 没指定订单号")
        if response:
            self.logger.debug(f'撤单回报 {response}')
        if error:
            print("撤单失败", error)
            self.logger.error(error)
        return response
    
    async def check_order(self, order_id=None, client_id=None):
        if order_id:
            response, error = await self._request('GET', f'/api/v3/order', params={'symbol': self.symbol, 'orderId': order_id}, auth=1)
        elif client_id:
            response, error = await self._request('GET', f'/api/v3/order', params={'symbol': self.symbol, 'orderId': client_id}, auth=1)
        else:
            return
        if response:
            self.logger.debug(f'查单回报 {response}')
            order_event = dict()
            if response["data"]['status'] in ['not_deal', 'part_deal']:
                order_event['status'] = "NEW"
            elif response["data"]['status'] in ['cancel', 'done']:
                order_event['status'] = "REMOVE"
            else:
                self.logger.error("错误的订单状态")
            order_event['price'] = float(response["data"]["price"])
            order_event['amount'] = float(response["data"]["amount"])
            order_event['filled'] = float(
                response["data"]["amount"])-float(response["data"]["left"])
            order_event['filled_price'] = float(response["data"]["avg_price"])
            order_event['client_id'] = response["data"]["client_id"]
            order_event['order_id'] = response["data"]['id']
            asset_fee = float(response['data']["asset_fee"])
            money_fee = float(response['data']["money_fee"])
            stock_fee = float(response['data']["stock_fee"])
            # 非amm品种 优先扣cet 其次u 再次b
            # amm品种 买入收b 卖出收u
            if response['data']['type'] == "sell":
                # 卖出
                order_event['fee'] = money_fee
            elif response['data']['type'] == "buy":
                # 买入
                order_event['fee'] = stock_fee
            self.callback["onOrder"](order_event)
        if error:
            print("查单失败", error)
            self.logger.error(error)
        return response
    
    async def get_order_list(self):
        params = {
            'symbol': self.symbol,
            'limit': 100,
        }
        response, error = await self._request('GET', '/api/v3/allOrders', params=params, auth=1)
        orders = []  # 重置本地订单列表
        if response is not None:
            for i in response['data']['data']:
                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["client_id"] if 'client_id' in i else ""
                order_event['order_id'] = i['id']
                asset_fee = float(i["asset_fee"])
                money_fee = float(i["money_fee"])
                stock_fee = float(i["stock_fee"])
                # 非amm品种 优先扣cet 其次u 再次b
                # amm品种 买入收b 卖出收u
                if i['type'] == "sell":
                    # 卖出
                    order_event['fee'] = money_fee
                elif i['type'] == "buy":
                    # 买入
                    order_event['fee'] = stock_fee
                if i['status'] in ['not_deal', 'part_deal']:
                    order_event['status'] = "NEW"
                elif i['status'] in ['cancel', 'done']:
                    order_event['status'] = "REMOVE"
                else:
                    self.logger.error("错误的订单状态")
                self.callback["onOrder"](order_event)
        if error:
            print(error)
        return response

    async def get_history_order(self):
        pass

    async def get_server_time(self):
        params = {}
        response = await self._request('GET', '/api/v3/time', params=params)
        return response
    
    async def before_trade(self):
        # 获取市场基本情况
        res, error = await self.get_market_details()
        if error:
            pass
        else:
            for i in res['data']:
                if res['data'][i]['name'] == self.symbol:
                    self.stepSize = float(Decimal("0.1")**Decimal(res['data'][i]["trading_decimal"]))
                    self.tickSize = float(Decimal("0.1")**Decimal(res['data'][i]["pricing_decimal"]))
                #### 保存交易规则信息
                exchange_info = model.ExchangeInfo()
                exchange_info.symbol = i
                exchange_info.multiplier = 1
                exchange_info.stepSize = float(Decimal("0.1")**Decimal(res['data'][i]["trading_decimal"]))
                exchange_info.tickSize = float(Decimal("0.1")**Decimal(res['data'][i]["pricing_decimal"]))
                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["data"]:
                if self.quote == i:
                    self.data['equity'] = float(
                        res['data'][i]['available'])+float(res['data'][i]['frozen'])
                    self.callback['onEquity']({
                        self.quote: self.data['equity']
                    })
                    self.cash_value = self.data['equity']
                elif self.base == i:
                    coin = float(res['data'][i]['available']) + \
                        float(res['data'][i]['frozen'])
                    self.callback['onEquity']({
                        self.base: coin
                    })
                    self.coin_value = coin
    async def universalTransfer(self, _type='UMFUTURE_MAIN', asset='USDT', amount=0):
        pass
    async def futuresTransfer(self, _type='2', asset='USDT', amount=0):
        pass
    async def get_account(self):
        return await self._request('GET', '/api/v3/account', params={"access_id": self.params.access_key}, auth=1)
    async def get_market_details(self):
        return await self._request('GET', f'/api/v3/exchangeInfo', params={}, auth=0)
    async def get_ticker(self):
        ## 'merge' 参数可能需要去掉 Galois
        res, err = await self._request('GET', f'/api/v3/depth', params={"symbol": self.symbol, 'merge': '0.00000001'}, auth=0)
        if res:
            ap = float(res["data"]['asks'][0][0])
            bp = float(res["data"]['bids'][0][0])
            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 go(self):
        await self.before_trade()
        await asyncio.sleep(1)
        ### Mexc无法检查是否为AMMM品种
        # try:
        #     async with aiohttp.ClientSession(connector = aiohttp.TCPConnector(
        #                         limit=50,
        #                         keepalive_timeout=120,
        #                         verify_ssl=False,
        #                         local_addr=(self.ip,0)
        #                     )) as session:
        #         response = await session.get(
        #             "https://api.coinex.com/v1/amm/market",
        #             proxy=self.proxy
        #         )
        #         res = await response.json()
        #         amm_list = res['data']
        #         print(f'AMM列表{amm_list}')
        #         if self.symbol in amm_list:
        #             self.callback['onExit'](f"{self.name} coinex spot 禁止跑AMM品种")
        #         else:
        #             print(f'不是AMM品种 正常运行')
        # except:
        #     self.logger.error(traceback.format_exc())
        #     self.callback['onExit'](f"{self.name} coinex spot AMM列表获取失败")
        while 1:
            try:
                # 停机信号
                if self.stop_flag:
                    return
                # 更新账户
                res, err = await self.get_account()
                if err:
                    print(err)
                if res:
                    for i in res["data"]:
                        if self.quote == i:
                            self.data['equity'] = float(
                                res['data'][i]['available']) + float(res['data'][i]['frozen'])
                            self.callback['onEquity']({
                                self.quote: self.data['equity']
                            })
                        elif self.base == i:
                            coin = float(res['data'][i]['available']) + \
                                float(res['data'][i]['frozen'])
                            self.callback['onEquity']({
                                self.base: coin
                            })
                # 更新订单
                # res = await self.get_order_list()
                await asyncio.sleep(60)
                # 打印延迟
                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_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]
                    # 只能用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]
                    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)