import aiohttp import time import asyncio import json, ujson import zlib import hashlib import hmac import base64 import traceback import random, csv, sys import logging, logging.handlers from datetime import datetime from urllib.parse import urlparse import urllib from decimal import Decimal import utils import model def empty_call(msg): pass def sort_num(n): if n.isdigit(): return int(n) else: return float(n) class OkexUsdtSwapRest: """""" def __init__(self, params:model.ClientParams, colo=0): if colo: print('不支持colo高速线路') self.REST = 'https://www.okx.com' # hk # REST = 'https://aws.okx.com' # aws else: self.REST = 'https://www.okx.com' # hk # REST = 'https://aws.okx.com' # aws self.params = params self.name = self.params.name self.base = params.pair.split('_')[0].upper() self.quote = params.pair.split('_')[1].upper() self.symbol = f"{self.base}-{self.quote}-SWAP" if len(self.params.pair.split('_')) > 2: self.delivery = self.params.pair.split('_')[2] # 210924 self.symbol += f"-{self.delivery}" self.data = {} self._SESSIONS = dict() self.callback = { "onMarket":empty_call, "onPosition":empty_call, "onOrder":empty_call, "onEquity":empty_call, "onExit":empty_call, } self.exchange_info = dict() self.stepSize = None self.tickSize = None self.ctVal = None self.ctMult = 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 self.getheader = self.make_header() self.postheader = self.make_header() #### 指定发包ip iplist = utils.get_local_ip_list() self.ip = iplist[int(self.params.ip)] 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) amount = int(amount/self.ctVal) # 这里把币转成张 后续用张来下单 # 似乎有了num_to_str就不再需要下面两行 if origin_side =='kd': side = 'buy' positionSide = 'long' elif origin_side =='pd': side = 'sell' positionSide = 'long' elif origin_side =='kk': side = 'sell' positionSide = 'short' elif origin_side =='pk': side = 'buy' positionSide = 'short' else: raise Exception(f'下单参数错误 side:{origin_side}') if amount <= 0.0: # okex 因为 数量单位为张 很容易出现这个问题 避免频繁写入日志 # self.logger.error(f'下单参数错误 amount:{amount} side:{origin_side}') 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, None if price <= 0.0: self.logger.error(f'下单参数错误 price:{price} side:{origin_side}') 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, None params = { 'instId': symbol, 'tdMode': "cross", 'sz': amount, 'side': side, 'posSide': positionSide, 'ordType': "limit" if order_type!="MARKET" else 'market', 'clOrdId': cid, } if params['ordType'] == 'limit': params['px'] = utils.num_to_str(price, self.exchange_info[symbol].tickSize) if self.params.debug == 'True': await asyncio.sleep(0.1) return None, None else: # 再报单 response, error = await self.http_post_request('/api/v5/trade/order', params) # 再更新 if response is not None: if response: data = response['data'][0] order_event = dict() order_event['status'] = "NEW" order_event['client_id'] = params['clOrdId'] order_event['order_id'] = data['ordId'] 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['clOrdId'] self.callback["onOrder"](order_event) return response, error async def cancel_order(self, order_id=None, client_id=None, symbol=None): """注意，ok撤单不能更新订单状态，撤单成功也仅仅代表交易所收到了撤单请求""" params = { "instId": self.symbol if symbol==None else symbol, } if order_id: params["ordId"] = order_id if client_id: params["clOrdId"] = client_id if self.params.debug == 'True': await asyncio.sleep(0.1) return None else: response, error = await self.http_post_request('/api/v5/trade/cancel-order', params) if error: # print("撤单失败",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 return response async def check_order(self, order_id=None, client_id=None, symbol=None): params = { "instId": self.symbol if symbol==None else symbol, } if order_id: params["ordId"] = order_id if client_id: params["clOrdId"] = client_id if self.params.debug == 'True': await asyncio.sleep(0.1) return None else: response, error = await self.http_get_request('/api/v5/trade/order', params) if error: print(f"{self.name} 查单失败 {error}") return error if response['code']: for order in response['data']: if order['state'] in ['canceled', 'filled']: order_event = dict() order_event['status'] = 'REMOVE' order_event['client_id'] = order['clOrdId'] order_event['order_id'] = order['ordId'] order_event['filled'] = float(order['accFillSz'])*self.ctVal if order['accFillSz'] != '' else 0.0 # usdt永续需要考虑每张的单位 order_event['filled_price'] = float(order['avgPx']) if order['avgPx'] != '' else 0.0 self.callback['onOrder'](order_event) else: order_event = dict() order_event['status'] = "NEW" order_event['client_id'] = order['clOrdId'] order_event['order_id'] = order['ordId'] self.callback['onOrder'](order_event) return response async def get_order_list(self): ''' 获取挂单表 ''' response, error = await self.http_get_request('/api/v5/trade/orders-pending', {'instId':self.symbol}) # print(response) orders = [] # 查询当前挂单 只可能出现 new 和 partfill 默认成交为0 只有 done状态的订单才考虑是否有成交 if response and response['code']: for i in response['data']: order_event = dict() order_event['status'] = "NEW" order_event['filled'] = 0 order_event['filled_price'] = 0 order_event['client_id'] = i["clOrdId"] order_event['order_id'] = i['ordId'] self.callback["onOrder"](order_event) orders.append(order_event) if error: print('查询列表出错',error) return orders async def get_server_time(self): response = await self.http_get_request('/api/v5/public/time') return response async def get_equity(self): ########## res, err = await self.get_account() if res: for data in res['data']: for i in data['details']: if self.quote == i['ccy']: self.data['equity'] = float(i['eq']) self.callback['onEquity']({self.quote:self.data['equity']}) if err: print('获取账户信息错误', err) ########## async def universalTransfer(self, _type='UMFUTURE_MAIN', asset='USDT', amount=0): """okex现在统一账户，没有钱包这个概念了，不实现""" pass async def futuresTransfer(self, _type='2', asset='USDT', amount=0): """okex现在统一账户，没有钱包这个概念了，不实现""" pass async def buy_token(self): '''买入平台币''' pass async def check_position(self, hold_coin=0.0): ''' 检查是否存在非运行币种的仓位并take平仓 已支持全品种 ''' try: ### self.logger.info(f'{self.name} 检查遗漏订单') response, error = await self.http_get_request('/api/v5/trade/orders-pending', {}) if response: for order in response['data']: params = { "instId": order['instId'], "ordId": order['ordId'] } res, err = await self.http_post_request('/api/v5/trade/cancel-order', params) await asyncio.sleep(0.1) self.logger.info(f"{self.name} 清理遗漏订单 {res} {err}") ### if self.exchange_info == dict(): await self.before_trade() ### self.logger.info(f'{self.name} 检查遗漏仓位') # 清空全部仓位 response, error = await self.http_get_request('/api/v5/account/positions', {"instType":"SWAP"}) if response: for i in response['data']: symbol = i['instId'] pos = float(i['pos']) posSide = i['posSide'] ### ticker, err = await self.http_get_request('/api/v5/market/ticker', {'instId':symbol}) if err: print(err) if ticker: ap = float(ticker['data'][0]['askPx']) bp = float(ticker['data'][0]['bidPx']) ### 每个品种都要获取各自的精度 trade_side = 'sell' if posSide == 'long' else "buy" trade_pos = abs(pos) trade_pos_side = posSide params = { 'instId': symbol, 'tdMode': "cross", 'sz': int(trade_pos), 'px': ap*1.001 if trade_side == 'buy' else bp*0.999, 'side' : trade_side, 'posSide': trade_pos_side, 'ordType': "limit", } response, error = await self.http_post_request('/api/v5/trade/order', params) print("下单结果",response,error) self.logger.info('遗留仓位检测完毕') except: self.logger.error("清仓程序执行出错") self.logger.error(traceback.format_exc()) return async def before_trade(self): """""" response, error = await self.get_instruments() if error: print('获取市场信息错误',error) else: self.update_instruments(response) print(f'before_trade get_instruments successed. {self.symbol} {self.stepSize} {self.tickSize} {self.ctVal} {self.ctMult}') # 更新账户 res, err = await self.get_account() if err: print(err) else: for data in res['data']: for i in data['details']: if self.quote == i['ccy']: self.data['equity'] = float(i['eq']) self.callback['onEquity']({self.quote:self.data['equity']}) self.cash_value = float(i['eq']) print(f"{self.name} on_go {self.symbol} equity {self.quote} {self.data['equity']}") await self.change_pos_side() await asyncio.sleep(1) await self.set_leverage(10) await asyncio.sleep(1) await self.get_position() async def get_all_position(self): ''' 获取仓位信息 ''' response, error = await self.http_get_request('/api/v5/account/positions', {}) print(f'查看此账号全部仓位 {response} {error}') async def get_position(self): ''' 获取仓位信息 ''' response, error = await self.http_get_request('/api/v5/account/positions', {'instId':self.symbol}) if error: print(f"{self.name} get_position error {error}") return None longPos, shortPos = 0, 0 longAvg, shortAvg = 0, 0 for i in response['data']: if i['instId'] == self.symbol and i['pos'] and i['avgPx']: if i['posSide'] == 'long': longPos = float(i['pos'])*self.ctVal longAvg = float(i['avgPx']) elif i['posSide'] == 'short': shortPos = float(i['pos'])*self.ctVal shortAvg = float(i['avgPx']) position = model.Position() position.longPos = abs(longPos) position.longAvg = abs(longAvg) position.shortPos = abs(shortPos) position.shortAvg = abs(shortAvg) self.callback['onPosition'](position) return position async def get_ticker(self): res, err = await self.http_get_request('/api/v5/market/ticker', {'instId':self.symbol}) if res: ap = float(res['data'][0]['askPx']) bp = float(res['data'][0]['bidPx']) mp = (ap+bp)/2 ticker = {"name":self.name,'mp': mp, 'bp':bp, 'ap':ap} return ticker if err: self.logger.debug(err) return None async def get_account(self): response, error = await self.http_get_request('/api/v5/account/balance', {'ccy':self.quote}) return response, error 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 err: print(err) else: for data in res['data']: for i in data['details']: if self.quote == i['ccy']: self.data['equity'] = float(i['eq']) self.callback['onEquity']({self.quote:self.data['equity']}) # print(f"{self.name} on_go {self.symbol} equity {self.quote} {self.data['equity']}") # 更新仓位 await self.get_position() # print(f"{self.name} on_go {self.symbol} position {position}") 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 '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)) 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] )) except Exception as e: traceback.print_exc() self.logger.error("执行信号出错"+str(e)) await asyncio.sleep(0.1) async def set_leverage(self, lever=10): params = {'instId':self.symbol, 'lever':utils.num_to_str(lever, 1), 'mgnMode':'cross'} res, error = await self.http_post_request('/api/v5/account/set-leverage', params) if error: print(f"{self.name} 设置杠杆倍数 {params} failed. -->{error}") return None, error if res['code'] == '0': print(f"{self.name} 设置杠杆倍数 {params} success. -->{res}") else: print(f"{self.name} 设置杠杆倍数 {params} 暂时不能设置. -->{res}") return res, error async def change_pos_side(self, dual='true'): '''''' params = {'posMode': "long_short_mode"} res, error = await self.http_post_request('/api/v5/account/set-position-mode', params) if error: print(f"{self.name} 设置双向持仓 failed. -->{error}") return res, error if res['code'] == '0': print(f"{self.name} 设置双向持仓 success. -->{res}") else: print(f"{self.name} 设置双向持仓 暂时不能设置. -->{res}") return res, error async def get_instruments(self): """从rest获取合约信息""" params = {'instType': 'SWAP'} res, error = await self.http_get_request('/api/v5/public/instruments', params) return res, error def update_instruments(self, data): """根据信息调整合约信息""" for info in data['data']: if info['instId'] == self.symbol: self.ctVal = sort_num(info['ctVal']) self.ctMult = sort_num(info['ctMult']) self.tickSize = sort_num(info['tickSz']) self.stepSize = sort_num(info['minSz'])*self.ctVal #### 保存交易规则信息 exchange_info = model.ExchangeInfo() exchange_info.symbol = info['instId'] exchange_info.multiplier = sort_num(info['ctMult']) exchange_info.tickSize = sort_num(info['tickSz']) exchange_info.stepSize = sort_num(info['minSz'])*sort_num(info['ctMult']) self.exchange_info[exchange_info.symbol] = exchange_info def __sign(self, timestamp, method, path, data): if data: message = timestamp + method + path + data else: message = timestamp + method + path digest = hmac.new(bytes(self.params.secret_key.encode('utf8')), bytes(message.encode('utf8')), digestmod=hashlib.sha256).digest() return base64.b64encode(digest).decode('utf-8') 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 async def http_post_request(self, method, query=None, **args): """""" params = dict() params.update(**args) if query is not None: params.update(query) data = json.dumps(params) timestamp = self.timestamp() sign = self.__sign(timestamp, 'POST', method, data) self.postheader['OK-ACCESS-SIGN'] = sign self.postheader['OK-ACCESS-TIMESTAMP'] = timestamp url = f"{self.REST}{method}" res, error = await self._request('POST', url, data) return res, error async def http_get_request(self, method, query=None, **args): """""" params = dict() params.update(**args) if query is not None: params.update(query) timestamp = self.timestamp() if params: path = '{method}?{params}'.format(method=method, params=urllib.parse.urlencode(params)) else: path = method sign = self.__sign(timestamp, 'GET', path, None) self.getheader['OK-ACCESS-SIGN'] = sign self.getheader['OK-ACCESS-TIMESTAMP'] = timestamp url = f"{self.REST}{path}" res, error = await self._request('GET', url) return res, error async def _request(self, method, url, params=None): """""" try: ###### msg = f"rest请求记录 {method} {url} {params}" self.logger.debug(msg) ###### session = self._get_session(url) start_time = time.time() if method == 'GET': headers = self.getheader response = await session.get( url, headers=headers, timeout=10, proxy=self.proxy ) elif method == 'POST': headers = self.postheader response = await session.post( url, data=params, headers=headers, timeout=10, proxy=self.proxy ) code = response.status res = await response.json() msg = f"rest请求记录 {method} {url} {headers} {params}" res_msg = msg + f' {res}' self.logger.debug(res_msg) if code not in (200, 201, 202, 203, 204, 205, 206): self.logger.error(f'METHOD:{method} URL:{url} PARAMS:{params} ERROR:{res}') return None, str(res) if 'code' in res: if int(res['code']) not in (0,): if '51401' in str(res): pass else: self.logger.error(f'METHOD:{method} URL:{url} PARAMS:{params} ERROR:{res}') 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) async def get_history(self): params = { 'instType': "SWAP", 'limit':"100" } res, error = await self.http_get_request('/api/v5/trade/fills', params) b_id=res['data'][0]['billId'] ### data = [] while 1: params = { 'instType': "SWAP", 'limit':"100", 'after':b_id } await asyncio.sleep(0.3) res, _ = await self.http_get_request('/api/v5/trade/fills', params) if res: if len(res['data']) == 0: break for i in res['data']: data.append(i) b_id = res['data'][-1]['billId'] # b_id_s = [] # for i in data: # if i['billId'] in b_id_s: # print(i['billId']) # b_id_s.append(i['billId']) with open("data5.json", 'w+') as f: f.write(json.dumps(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) def timestamp(self): return datetime.utcnow().isoformat("T")[:-3] + 'Z' def login_params(self): """生成login字符""" timestamp = str(time.time()) message = timestamp + 'GET' + '/users/self/verify' mac = hmac.new(bytes(self.params.secret_key.encode('utf8')), bytes(message.encode('utf8')), digestmod=hashlib.sha256).digest() sign = base64.b64encode(mac) login_dict = {} login_dict['apiKey'] = self.params.access_key login_dict['passphrase'] = self.params.pass_key login_dict['timestamp'] = timestamp login_dict['sign'] = sign.decode('utf-8') login_param = {'op': 'login', 'args': [login_dict]} login_str = ujson.dumps(login_param) return login_str def make_header(self): """""" headers = {} headers['Content-Type'] = 'application/json' headers['OK-ACCESS-KEY'] = self.params.access_key headers['OK-ACCESS-SIGN'] = None headers['OK-ACCESS-TIMESTAMP'] = None headers['OK-ACCESS-PASSPHRASE'] = self.params.pass_key return headers 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]