""" 交易API """ import copy import json import logging import multiprocessing import queue import random import threading import time import concurrent.futures import zmq from code_attribute import gpcode_manager from huaxin_client import constant as huaxin_client_constant, trade_client from log_module import async_log_util from log_module.log import hx_logger_trade_debug, hx_logger_trade_loop, hx_logger_trade_callback, \ logger_system from trade.huaxin import huaxin_trade_data_update, huaxin_trade_record_manager from trade.huaxin.huaxin_trade_record_manager import TradeOrderIdManager from trade.huaxin.huaxin_trade_order_processor import CancelOrderManager, HuaxinOrderEntity, TradeResultProcessor from utils import socket_util, huaxin_util, tool __response_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=15) __save_data_queue = queue.Queue(maxsize=1000) def __run_save_data(): while True: try: data = __save_data_queue.get() huaxin_trade_record_manager.DelegateRecordManager.add_one(data) except: pass finally: time.sleep(0.1) def __run_recv_queue_trade(queue: multiprocessing.Queue): def __cancel_order(code, order_ref): # 2s没成交就撤单 time.sleep(2) order_: HuaxinOrderEntity = TradeResultProcessor.get_huaxin_order_by_order_ref(order_ref) if order_ is not None: if huaxin_util.is_can_cancel(order_.orderStatus): cancel_order(TRADE_DIRECTION_SELL, code, order_.orderSysID) def __process_order(data): # 更新委托队列 push_msg_manager.push_delegate_queue_update() code = data["securityID"] accountID = data["accountID"] orderStatus = data["orderStatus"] orderRef = data["orderRef"] orderSysID = data["orderSysID"] insertTime = data.get("insertTime") acceptTime = data.get("acceptTime") insertDate = data.get("insertDate") direction = data.get("direction") limitPrice = data.get("limitPrice") volume = data.get("volume") is_shadow_order = False # 获取涨停价 limit_up_price = gpcode_manager.get_limit_up_price(code) if limit_up_price and volume == huaxin_client_constant.SHADOW_ORDER_VOLUME: if abs(float(limitPrice) - float(limit_up_price)) >= 0.001: is_shadow_order = True order = HuaxinOrderEntity(code, orderStatus, orderRef, accountID, orderSysID, insertTime=insertTime, acceptTime=acceptTime, insertDate=insertDate, direction=direction, is_shadow_order=is_shadow_order) try: if str(order.direction) == str(huaxin_util.TORA_TSTP_D_Sell): # 刷新持仓列表 huaxin_trade_data_update.add_position_list() if huaxin_util.is_deal(order.orderStatus): # 如果成交了需要刷新委托列表 huaxin_trade_data_update.add_delegate_list("卖成交") else: if huaxin_util.is_deal(order.orderStatus): # 如果成交了需要刷新委托列表 huaxin_trade_data_update.add_delegate_list("买成交") need_cancel = TradeResultProcessor.process_buy_order(order) # if need_cancel: # # 需要撤买单 # threading.Thread(target=lambda: cancel_order(TRADE_DIRECTION_SELL, order.code, order.orderSysID), # daemon=True).start() need_watch_cancel = TradeResultProcessor.process_sell_order(order) # if need_watch_cancel: # # 需要撤卖单 # threading.Thread(target=lambda: __cancel_order(order.code, order.orderRef), daemon=True).start() finally: try: # 加入2次，增大加入成功率 __save_data_queue.put_nowait(data) except Exception as e: hx_logger_trade_debug.exception(e) if not is_shadow_order: # 订单相关回调 # 重新请求委托列表与资金 # huaxin_trade_data_update.add_delegate_list("来自交易管道") huaxin_trade_data_update.add_deal_list() huaxin_trade_data_update.add_money_list() # 设置结果 def __set_response(data_json): if 'request_id' not in data_json: return # 处理数据 async_log_util.info(hx_logger_trade_callback, f"response：request_id-{data_json['request_id']}") # 设置响应内容 set_response(data_json["request_id"], data_json['data']) logger_system.info(f"huaxin_trade_api __run_recv_pipe_trade 线程ID:{tool.get_thread_id()}") if queue is not None: while True: try: val = queue.get() if val: data_json = json.loads(val) # 处理数据 type_ = data_json["type"] if type_ == "response": # 主动触发的响应 request_id = data_json['request_id'] async_log_util.info(hx_logger_trade_callback, f"response：request_id-{request_id}") __response_thread_pool.submit(__set_response, data_json) if type(data_json.get("data")) == dict: data = data_json["data"].get("data") if type(data) == dict and "orderRef" in data: __response_thread_pool.submit(__process_order, data) elif type_ == "trade_callback": try: # 交易回调 data_json = data_json["data"] ctype = data_json["type"] # 记录交易反馈日志 async_log_util.info(hx_logger_trade_callback, f"{data_json}") if ctype == 0: data = data_json.get("data") # 获取订单状态 __process_order(data) finally: pass except: pass def __create_trade_ipc_context(trade_ipc_addr): """ 创建IPC发送端口 @param trade_ipc_addr:（下单地址,撤单地址） @return: """ context = zmq.Context() global order_socket, cancel_order_socket order_socket = context.socket(zmq.REQ) order_socket.connect(trade_ipc_addr[0]) cancel_order_socket = context.socket(zmq.REQ) cancel_order_socket.connect(trade_ipc_addr[1]) # while True: # try: # # datas = ('000990', 7.52, 400, 93000030, 2012, 380422, 380421, 375477, '1') * 150 # # L2SharedMemoryDataUtil.set_data(datas, shared_memory) # socket.send_json({'data': [], "time": time.time()}) # response = socket.recv_string() # except Exception as e: # logging.exception(e) # 下单ZMQ通信锁 __order_zmq_lock = threading.Lock() def __order_by_zmq(data_json): """ 通过zmq发送下单信息 @param data_json: @return: """ with __order_zmq_lock: order_socket.send_json(data_json) response = order_socket.recv_string() def __cancel_order_by_zmq(data_json): cancel_order_socket.send_json(data_json) response = cancel_order_socket.recv_string() def __test_order(): time.sleep(60) for i in range(20): time.sleep(30) order_ref = huaxin_util.create_order_ref() order(1, "000333", 100, 1.00, price_type=2, blocking=False, order_ref=order_ref, shadow_price=0.99) time.sleep(30) cancel_order(1, "000333", '123123', orderRef=order_ref, blocking=False) queue_strategy_w_trade_r_for_read = None queue_strategy_w_trade_r = None def run(): global queue_strategy_w_trade_r_for_read, queue_strategy_w_trade_r queue_strategy_r_trade_w, queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_read = multiprocessing.Queue(), multiprocessing.Queue(), multiprocessing.Queue() order_ipc_addr, cancel_order_ipc_addr = "ipc://trade_order_ls.ipc", "ipc://trade_order_ls_cancel.ipc" trade_process = multiprocessing.Process( target=trade_client.run, args=(order_ipc_addr, cancel_order_ipc_addr, queue_strategy_r_trade_w, queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_read)) trade_process.start() t1 = threading.Thread(target=lambda: __run_recv_queue_trade(queue_strategy_r_trade_w), daemon=True) t1.start() t1 = threading.Thread(target=lambda: __run_save_data(), daemon=True) t1.start() t1 = threading.Thread(target=lambda: CancelOrderManager().run(cancel_order), daemon=True) t1.start() # 创建IPC发送端口 __create_trade_ipc_context((order_ipc_addr, cancel_order_ipc_addr)) # 交易通道的错误次数 trade_pipe_channel_error_count = 0 # pipe的交易通道是否正常 def is_pipe_channel_normal(): return True # 测试交易通道 def test_trade_channel(): global trade_pipe_channel_error_count sid = random.randint(0, 1000000) result = __test_trade_channel(sid) if result["code"] == 0 and result["data"]["data"]["sid"] == sid: trade_pipe_channel_error_count = 0 return True trade_pipe_channel_error_count += 1 if trade_pipe_channel_error_count > 100: trade_pipe_channel_error_count = 100 return False class ClientSocketManager: # 客户端类型 CLIENT_TYPE_TRADE = "trade" CLIENT_TYPE_DELEGATE_LIST = "delegate_list" CLIENT_TYPE_DEAL_LIST = "deal_list" CLIENT_TYPE_POSITION_LIST = "position_list" CLIENT_TYPE_MONEY = "money" CLIENT_TYPE_DEAL = "deal" CLIENT_TYPE_CMD_L2 = "l2_cmd" socket_client_dict = {} socket_client_lock_dict = {} active_client_dict = {} @classmethod def list_client(cls, _type): if _type == cls.CLIENT_TYPE_TRADE: if _type in cls.socket_client_dict: return cls.socket_client_dict.get(_type) else: if _type in cls.socket_client_dict: return [cls.socket_client_dict.get(_type)] return [] @classmethod def add_client(cls, _type, rid, sk): if _type == cls.CLIENT_TYPE_TRADE: # 交易列表 if _type not in cls.socket_client_dict: cls.socket_client_dict[_type] = [] cls.socket_client_dict[_type].append((rid, sk)) cls.socket_client_lock_dict[rid] = threading.Lock() hx_logger_trade_debug.info(f"add_client:{rid}") else: cls.socket_client_dict[_type] = (rid, sk) cls.socket_client_lock_dict[rid] = threading.Lock() # 是否已经被锁住 @classmethod def is_client_locked(cls, rid): if rid in cls.socket_client_lock_dict: return cls.socket_client_lock_dict[rid].locked() return None @classmethod def acquire_client(cls, _type): if _type == cls.CLIENT_TYPE_TRADE: if _type in cls.socket_client_dict: # 根据排序活跃时间排序 client_list = sorted(cls.socket_client_dict[_type], key=lambda x: cls.active_client_dict.get(x[0]) if x[ 0] in cls.active_client_dict else 0, reverse=True) hx_logger_trade_debug.info(f"acquire_client client_list数量：{len(client_list)}") hx_logger_trade_debug.info( f"acquire_client socket_client_lock_dict数量：{len(cls.socket_client_lock_dict.keys())}") for d in client_list: if d[0] in cls.socket_client_lock_dict: try: if cls.socket_client_lock_dict[d[0]].acquire(blocking=False): hx_logger_trade_debug.info(f"acquire_client success:{d[0]}") return d except threading.TimeoutError: hx_logger_trade_debug.error("acquire_client TimeoutError") else: if _type in cls.socket_client_dict: try: d = cls.socket_client_dict[_type] if d[0] in cls.socket_client_lock_dict: if cls.socket_client_lock_dict[d[0]].acquire(blocking=False): return d except threading.TimeoutError: pass return None @classmethod def release_client(cls, client_id): sucess = False if client_id in cls.socket_client_lock_dict: sucess = True # 释放锁 if cls.socket_client_lock_dict[client_id].locked(): cls.socket_client_lock_dict[client_id].release() if sucess: hx_logger_trade_debug.info(f"release_client success:{client_id}") else: hx_logger_trade_debug.info(f"release_client fail:{client_id}") @classmethod def del_client(cls, rid): # 删除线程锁 if rid in cls.socket_client_lock_dict: cls.socket_client_lock_dict.pop(rid) # 删除sk for t in cls.socket_client_dict: if type(cls.socket_client_dict[t]) == list: for d in cls.socket_client_dict[t]: if d[0] == rid: try: # 关闭socket d[1].close() except: pass cls.socket_client_dict[t].remove(d) break elif type(cls.socket_client_dict[t]) == tuple: if cls.socket_client_dict[t][0] == rid: try: # 关闭socket cls.socket_client_dict[t][1].close() except: pass cls.socket_client_dict.pop(t) break # 心跳信息 @classmethod def heart(cls, rid): cls.active_client_dict[rid] = time.time() @classmethod def del_invalid_clients(cls): # 清除长时间无心跳的客户端通道 for k in cls.active_client_dict.keys(): if time.time() - cls.active_client_dict[k] > 20: # 心跳时间间隔20s以上视为无效 cls.del_client(k) TRADE_DIRECTION_BUY = 1 TRADE_DIRECTION_SELL = 2 # 超时时间2s TIMEOUT = 2.0 # 等待响应的request_id __request_response_dict = {} def __get_request_id(type): return f"r_{type}_{round(time.time() * 10000)}_{random.randint(0, 100000)}" # 网络请求 def __request(_type, data, request_id=None, log_enable=True, is_trade=False): """ 请求，将交易（包含下单/撤单）与查询（包含查持仓/账户可用金额/委托列表/成交列表）队列分离 @param _type: @param data: @param request_id: @param log_enable: @param is_trade: @return: """ if not request_id: request_id = __get_request_id(_type) try: if log_enable: async_log_util.info(hx_logger_trade_loop, "请求发送开始：client_id-{} request_id-{}", 0, request_id) root_data = {"type": _type, "data": data, "request_id": request_id, "time": time.time() } root_data = socket_util.encryp_client_params_sign(root_data) start_time = time.time() if is_trade: # queue_strategy_w_trade_r.put_nowait(root_data) # 采用zmq通信 if data['trade_type'] == 1: __order_by_zmq(root_data) elif data['trade_type'] == 2: __cancel_order_by_zmq(root_data) else: queue_strategy_w_trade_r.put_nowait(root_data) else: queue_strategy_w_trade_r_for_read.put_nowait(root_data) use_time = int((time.time() - start_time) * 1000) if use_time > 10: async_log_util.info(hx_logger_trade_loop, f"发送耗时：request_id-{request_id} 耗时时间：{use_time}") if log_enable: async_log_util.info(hx_logger_trade_loop, "请求发送成功：request_id-{}", request_id) except BrokenPipeError as e: async_log_util.info(hx_logger_trade_loop, "请求发送异常：request_id-{} error-{}", request_id, str(e)) raise e except Exception as e: async_log_util.info(hx_logger_trade_loop, "请求发送异常： request_id-{} error-{}", request_id, str(e)) logging.exception(e) raise e return request_id def __read_response(request_id, blocking, timeout=TIMEOUT, log_enable=True): if blocking: start_time = time.time() try: while True: time.sleep(0.005) if request_id in __request_response_dict: # 获取到了响应内容 result = __request_response_dict.pop(request_id) if log_enable: async_log_util.info(hx_logger_trade_loop, "请求读取成功： request_id-{}", request_id) return result if time.time() - start_time > timeout: if log_enable: async_log_util.info(hx_logger_trade_loop, "请求读取超时： request_id-{}", request_id) # 读取内容超时才会释放 raise Exception(f"读取内容超时: request_id={request_id}") finally: pass return None __TradeOrderIdManager = TradeOrderIdManager() def set_response(request_id, response): if request_id: async_log_util.info(hx_logger_trade_loop, f"请求响应: request_id-{request_id} 内容-{response}") # 主动触发 __request_response_dict[request_id] = response else: # 被动触发 pass def order(direction, code, volume, price, price_type=2, blocking=False, sinfo=None, request_id=None, order_ref=None, shadow_price=None, shadow_volume=100): """ 下单委托 @param shadow_volume: 影子单的量 @param direction: 1-买 2-卖 @param code: @param volume:交易量 @param price:价格（如果是卖时不传价格就按照5挡价卖） @param price_type: @param blocking:是否阻塞进程 @param sinfo: @param request_id: @param order_ref: @param shadow_price: @return: """ timestamp = round(time.time() * 1000) if not sinfo: sinfo = f"ba_{code}_{timestamp}" if not order_ref: order_ref = huaxin_util.create_order_ref() if not request_id: request_id = __get_request_id(ClientSocketManager.CLIENT_TYPE_TRADE) for i in range(1): cancel_shadow = True if int(tool.get_now_time_str().replace(":", "")) < int("091500"): # 预埋单不能撤影子单 cancel_shadow = False request_id = __request(ClientSocketManager.CLIENT_TYPE_TRADE, {"type": ClientSocketManager.CLIENT_TYPE_TRADE, "trade_type": 1, "direction": direction, "code": code, "order_ref": order_ref, "volume": volume, "price_type": price_type, "price": price, "shadow_price": shadow_price, "shadow_volume": shadow_volume, "sinfo": sinfo, "blocking": blocking, "cancel_shadow": cancel_shadow}, request_id=request_id, is_trade=True) try: if blocking: return __read_response(request_id, blocking) else: return {"order_ref": order_ref} finally: # huaxin_trade_data_update.add_delegate_list("下单", delay=0.2) huaxin_trade_data_update.add_money_list() def order_new(direction, code, order_info_list, price_type=2, blocking=False, sinfo=None, request_id=None): """ 下单委托 @param direction: @param code: @param order_info_list: 下单信息：[(量,价, order_ref),(量,价, order_ref)] @param price_type: @param blocking: 是否阻塞进程 @param sinfo: @param request_id: @return: """ timestamp = round(time.time() * 1000) if not sinfo: sinfo = f"ba_{code}_{timestamp}" if not request_id: request_id = __get_request_id(ClientSocketManager.CLIENT_TYPE_TRADE) for i in range(1): request_id = __request(ClientSocketManager.CLIENT_TYPE_TRADE, {"type": ClientSocketManager.CLIENT_TYPE_TRADE, "trade_type": 1, "direction": direction, "code": code, "order_info_list": order_info_list, "price_type": price_type, "sinfo": sinfo, "blocking": blocking, "cancel_shadow": False}, request_id=request_id, is_trade=True) try: if blocking: return __read_response(request_id, blocking) else: return {"order_ref_list": [x[2] for x in order_info_list]} finally: # huaxin_trade_data_update.add_delegate_list("下单", delay=0.2) huaxin_trade_data_update.add_money_list() __canceling_order_dict = {} def cancel_order(direction, code, orderSysID, orderRef=None, blocking=False, sinfo=None, request_id=None, recancel=False): """ 测单 @param direction: 1-买 2-卖 @param code: @param orderSysID: @param orderRef: @param blocking: @param sinfo: @param request_id: @param recancel: @return: """ if tool.trade_time_sub(tool.get_now_time_str(), "14:57:00") >= 0 and tool.trade_time_sub(tool.get_now_time_str(), "15:00:01") <= 0: # 集合竞价不撤单 return if not recancel: CancelOrderManager().start_cancel(code, orderRef, orderSysID) if not sinfo: sinfo = f"cb_{code}_{round(time.time() * 1000)}_{random.randint(0, 10000)}" order_action_ref = huaxin_util.create_order_ref() if not request_id: request_id = __get_request_id(ClientSocketManager.CLIENT_TYPE_TRADE) # 加入撤单记录，用于校验最后的撤单是否成功 if code not in __canceling_order_dict: __canceling_order_dict[code] = set() __canceling_order_dict[code].add(json.dumps((orderRef, orderSysID))) # 执行2次撤单，防止没有撤到 for i in range(2): request_id = __request(ClientSocketManager.CLIENT_TYPE_TRADE, {"type": ClientSocketManager.CLIENT_TYPE_TRADE, "trade_type": 2, "direction": direction, "code": code, "orderRef": orderRef, "orderActionRef": order_action_ref, "orderSysID": orderSysID, "sinfo": sinfo}, request_id=request_id, is_trade=True) try: return __read_response(request_id, blocking) finally: huaxin_trade_data_update.add_money_list() def batch_cancel_order(direction, code, orderInfos: list, blocking=False, request_id=None, recancel=False): """ 测单 @param direction: 1-买 2-卖 @param code: @param orderInfos:[(orderRef, orderSysID)] @param blocking: @param request_id: @param recancel: @return: """ if tool.trade_time_sub(tool.get_now_time_str(), "14:57:00") >= 0 and tool.trade_time_sub(tool.get_now_time_str(), "15:00:01") <= 0: # 集合竞价不撤单 return if not recancel: for orderInfo in orderInfos: CancelOrderManager().start_cancel(code, orderInfo[0], orderInfo[1]) sinfos = [] for i in range(len(orderInfos)): sinfo = f"cb_{code}_{round(time.time() * 1000)}_{random.randint(0, 10000)}_{i}" sinfos.append(sinfo) order_action_refs = [] for i in range(len(orderInfos)): order_action_ref = huaxin_util.create_order_ref() order_action_refs.append(order_action_ref) if not request_id: request_id = __get_request_id(ClientSocketManager.CLIENT_TYPE_TRADE) # 加入撤单记录，用于校验最后的撤单是否成功 if code not in __canceling_order_dict: __canceling_order_dict[code] = set() for orderInfo in orderInfos: __canceling_order_dict[code].add(json.dumps((orderInfo[0], orderInfo[1]))) # 执行2次撤单，防止没有撤到 for i in range(2): request_id = __request(ClientSocketManager.CLIENT_TYPE_TRADE, {"type": ClientSocketManager.CLIENT_TYPE_TRADE, "trade_type": 2, "direction": direction, "code": code, "orderInfos": orderInfos, "orderActionRefs": order_action_refs, "sinfos": sinfos}, request_id=request_id, is_trade=True) try: return __read_response(request_id, blocking) finally: huaxin_trade_data_update.add_money_list() # CLIENT_TYPE_DELEGATE_LIST = "delegate_list" # CLIENT_TYPE_DEAL_LIST = "deal_list" # CLIENT_TYPE_POSITION_LIST = "position_list" # CLIENT_TYPE_MONEY = "money" # CLIENT_TYPE_DEAL = "deal" # 获取委托列表 # can_cancel:是否可以撤 def get_delegate_list(can_cancel=True, blocking=True, timeout=TIMEOUT): request_id = __request(ClientSocketManager.CLIENT_TYPE_DELEGATE_LIST, {"type": ClientSocketManager.CLIENT_TYPE_DELEGATE_LIST, "can_cancel": 1 if can_cancel else 0}) return __read_response(request_id, blocking, timeout=timeout) # 获取成交列表 def get_deal_list(blocking=True, timeout=TIMEOUT): request_id = __request(ClientSocketManager.CLIENT_TYPE_DEAL_LIST, {"type": ClientSocketManager.CLIENT_TYPE_DEAL_LIST}) return __read_response(request_id, blocking, timeout=timeout) # 获取持仓列表 def get_position_list(blocking=True): request_id = __request(ClientSocketManager.CLIENT_TYPE_POSITION_LIST, {"type": ClientSocketManager.CLIENT_TYPE_POSITION_LIST}) return __read_response(request_id, blocking) # 获取账户资金状况 def get_money(blocking=True): request_id = __request(ClientSocketManager.CLIENT_TYPE_MONEY, {"type": ClientSocketManager.CLIENT_TYPE_MONEY}) return __read_response(request_id, blocking) # 设置L2订阅数据 def set_l2_codes_data(codes_data, blocking=True): request_id = __request(ClientSocketManager.CLIENT_TYPE_CMD_L2, {"type": ClientSocketManager.CLIENT_TYPE_CMD_L2, "data": codes_data}) return __read_response(request_id, blocking) # 设置L2订阅数据 def __test_trade_channel(sid): request_id = __request("test", {"type": "test", "data": {"sid": sid}}, log_enable=False) return __read_response(request_id, True, log_enable=False) def parseResponse(data_str): if not data_str: raise Exception("反馈内容为空") res = data_str if type(res) == str: res = json.loads(data_str) res = res['data'] if res['code'] != 0: raise Exception(res['msg']) return res['data'] if __name__ == "__main__": pass