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 import sys from urllib.parse import urlparse import logging import logging.handlers import utils import logging import logging.handlers import model from decimal import Decimal def empty_call(msg): print(f'空的回调函数 {msg}') class CoinExSpotRest: def __init__(self, params: model.ClientParams, colo=0): if colo: print('不支持colo高速线路') self.HOST = 'https://api.coinex.com' else: self.HOST = 'https://api.coinex.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( 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): sort_params = sorted(params) data = [] for item in sort_params: data.append(item + '=' + str(params[item])) str_params = "{0}&secret_key={1}".format('&'.join(data), secret_key) token = hashlib.md5(str_params.encode("utf8")).hexdigest().upper() return token async def _request(self, method, uri, body=None, params=None, auth=False): url = urljoin(self.HOST, uri) headers = {} if auth: if method in ['GET', 'DELETE']: params['access_id'] = self.params.access_key params['tonce'] = int(time.time()*1000) headers = { 'AUTHORIZATION': self.get_sign(params, self.params.secret_key), 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'application/json', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36' } elif method == 'POST': body['access_id'] = self.params.access_key body['tonce'] = int(time.time()*1000) headers = { 'AUTHORIZATION': self.get_sign(body, self.params.secret_key), 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'application/json', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36' } # 发起请求 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 async def buy_token(self): '''买入平台币''' pass async def check_position(self, hold_coin=0.0): ''' 现货交易 已支持全品种 ''' try: ####################### self.logger.info("清空挂单") params = { 'access_id': self.params.access_key, 'market': self.symbol, } response, error = await self._request('DELETE', '/v1/order/pending', params=params, auth=1) if error: self.logger.info(error) if response: self.logger.info(error) ############################# res, err = await self.get_account() if err: self.logger.info(err) if res: for i in res['data']: if i in ['CET','USDT']: continue symbol = i + 'USDT' ####################### ticker, error = await self._request('GET', f'/v1/market/depth', params={"market": symbol, 'merge': '0.00000001'}, auth=0) if ticker: ap = float(ticker["data"]['asks'][0][0]) bp = float(ticker["data"]['bids'][0][0]) mp = (ap+bp)*0.5 if error: self.logger.error('ger ticker failed!') continue coin = float(res['data'][i]['available']) + \ float(res['data'][i]['frozen']) coin_value = coin * mp if i == self.base: _hold_coin = hold_coin else: _hold_coin = 0 diff = _hold_coin - coin_value diff *= 0.99 # 避免无法下单 self.logger.info(f'{symbol}需要调整现货仓位{diff}usd') if diff > 20.0: ####################### self.logger.info("清空挂单") params = { 'access_id': self.params.access_key, 'market': symbol, } response, error = await self._request('DELETE', '/v1/order/pending', params=params, auth=1) if error: self.logger.info(error) if response: self.logger.info(error) ############################# info = await self.take_order(symbol, diff/mp, "kd", mp*1.001, "123", "limit") self.logger.info(info) elif diff < -20.0: ####################### self.logger.info("清空挂单") params = { 'access_id': self.params.access_key, 'market': symbol, } response, error = await self._request('DELETE', '/v1/order/pending', params=params, auth=1) if error: self.logger.info(error) if response: self.logger.info(error) ############################# info = await self.take_order(symbol, -diff/mp, "kk", mp*0.999, "123", "limit") self.logger.info(info) ####################### 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'): 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 params = { 'access_id': self.params.access_key, 'client_id': cid, 'market': 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', '/v1/order/limit', 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): if order_id: response, error = await self._request('DELETE', f'/v1/order/pending', params={'market': self.symbol, 'id': order_id}, auth=1) elif client_id: response, error = await self._request('DELETE', f'/v1/order/pending', params={'market': self.symbol, 'id': client_id}, auth=1) else: raise Exception("撤单出错 没指定订单号") if response: self.logger.debug(f'撤单回报 {response}') # if response["data"]['status'] in ['cancel','done']: # order_event = dict() # order_event['status'] = "REMOVE" # order_event['filled_price'] = float(response['data']['avg_price']) # 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 # 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: print("撤单失败", error) self.logger.error(error) # if client_id:await self.check_order(client_id=client_id) # if order_id:await self.check_order(order_id=order_id) return response async def check_order(self, order_id=None, client_id=None): if order_id: response, error = await self._request('GET', f'/v1/order/status', params={'market': self.symbol, 'id': order_id}, auth=1) elif client_id: response, error = await self._request('GET', f'/v1/order/status', params={'market': self.symbol, 'id': 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 = { 'market': self.symbol, 'page': 1, 'limit': 100, } response, error = await self._request('GET', '/v1/order/pending', 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_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', '/v1/balance/info', params={"access_id": self.params.access_key}, auth=1) async def get_market_details(self): return await self._request('GET', f'/v1/market/info', params={}, auth=0) async def get_ticker(self): res, err = await self._request('GET', f'/v1/market/depth', params={"market": 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 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 go(self): await self.before_trade() await asyncio.sleep(1) ### 检查是否为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 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] 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)