# -*- coding: utf-8 -*- import json import logging import marshal import multiprocessing import queue import threading import time import msgpack import constant from huaxin_client import socket_util from huaxin_client.client_network import SendResponseSkManager # 活动时间 from huaxin_client.code_queue_distribute_manager import CodeDataChannelDistributeManager from log_module import async_log_util from log_module.async_log_util import huaxin_l2_log from log_module.log import logger_local_huaxin_l2_error, logger_system, logger_local_huaxin_l2_subscript, \ logger_local_huaxin_l2_special_volume, logger_debug, logger_local_huaxin_l2_orderdetail from utils import tool, shared_memery_util import collections import zmq order_detail_upload_active_time_dict = {} transaction_upload_active_time_dict = {} # 临时数据 tmep_order_detail_queue_dict = {} tmep_transaction_queue_dict = {} target_codes = set() target_codes_add_time = {} common_queue = queue.Queue(maxsize=1000) # L2上传数据管理器 class L2DataUploadManager: """ L2逐笔委托/L2逐笔成交：通过共享内存+ZMQ上传数据 L2市场行情： 通过普通数据上传队列写入 L2订阅的代码： 通过普通数据上传队列写入 """ TYPE_DELEGATE = 1 TYPE_TRANSACTION = 2 TYPE_MARKET = 3 def __init__(self, data_channel_distribute_manager: CodeDataChannelDistributeManager, common_data_upload_queue: multiprocessing.Queue): self.data_channel_distribute_manager = data_channel_distribute_manager self.common_data_upload_queue = common_data_upload_queue # 代码分配的对象 self.temp_order_queue_dict = {} self.temp_transaction_queue_dict = {} self.temp_log_queue_dict = {} self.filter_order_condition_dict = {} self.upload_l2_data_task_dict = {} self.l2_order_codes = set() self.l2_transaction_codes = set() # 设置订单过滤条件 # special_price:过滤的1手的价格 def set_order_fileter_condition(self, code, min_volume, limit_up_price, shadow_price, buy_volume, special_volumes): if not special_volumes: special_volumes = set() # if code not in self.filter_order_condition_dict: try: # (最小的量, 涨停价格, 影子单价格, 买的量, 废弃使用, 特殊的量集合) self.filter_order_condition_dict[code] = [(min_volume, limit_up_price, shadow_price, buy_volume, int(min_volume) // 50, set(special_volumes))] # logger_local_huaxin_l2_subscript.info(f"({code})常规过滤条件设置：{self.filter_order_condition_dict[code]}") except Exception as e: logger_debug.error(f"{str(e)} - min_volume-{min_volume}") # 过滤订单 def __filter_order(self, item): filter_condition = self.filter_order_condition_dict.get(item[0]) if filter_condition: # item[2]为量 if item[2] >= filter_condition[0][0]: return item # 1手的买单满足价格 # if item[2] == 100 and abs(filter_condition[0][2] - item[1]) < 0.001: # return item # 买量 if item[2] == filter_condition[0][3]: return item # 所有的涨停卖 if item[3] != '1': # 卖与卖撤 if abs(item[1] - filter_condition[0][1]) < 0.001: # 涨停价 return item else: if item[2] in filter_condition[0][5]: # 特殊手数 return item return None return item # 过滤订单 def __filter_transaction(self, item): filter_condition = self.filter_order_condition_dict.get(item[0]) if filter_condition: # item[2]为量 if abs(item[1] - filter_condition[0][1]) < 0.201: return item return None return item # 添加委托详情 def add_l2_order_detail(self, data, start_time=0, istransaction=False): code = data["SecurityID"] # 不直接加入 # queue_info = self.order_queue_distribute_manager.get_distributed_queue(code) # if not queue_info: # return # queue_info[1].put_nowait( # (data['SecurityID'], data['Price'], data['Volume'], data['Side'], data['OrderType'], data['OrderTime'], # data['MainSeq'], data['SubSeq'], data['OrderNO'], data['OrderStatus'], time.time(), start_time)) # if data['Volume'] == 100: # log_queue = self.temp_log_queue_dict.get(code) # if log_queue: # log_queue.put_nowait(data) q: collections.deque = self.temp_order_queue_dict.get(code) if q is not None: q.append( (data['SecurityID'], data['Price'], data['Volume'], data['Side'], data['OrderType'], data['OrderTime'], data['MainSeq'], data['SubSeq'], data['OrderNO'], data['OrderStatus'], time.time(), start_time)) # 添加逐笔成交 def add_transaction_detail(self, data): code = data["SecurityID"] # 不直接加入 # queue_info = self.transaction_queue_distribute_manager.get_distributed_queue(code) # if not queue_info: # return # # 判断是否为大单成交 # queue_info[1].put_nowait((data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType'])) q: collections.deque = self.temp_transaction_queue_dict.get(code) if q is not None: q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'], data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], data['SellNo'], data['ExecType'], time.time())) def add_market_data(self, data): code = data['securityID'] # 改为队列回调发送 self.common_data_upload_queue.put_nowait({"type": "l2_market", "data": (code, data)}) # 分配上传队列 def distribute_upload_queue(self, code, _target_codes=None): """ 分配上传队列 @param code: 代码 @param _target_codes: 所有的目标代码 @return: """ if not self.data_channel_distribute_manager.get_distributed_channel(code): self.data_channel_distribute_manager.distribute_channel(code, _target_codes) if code not in self.temp_order_queue_dict: self.temp_order_queue_dict[code] = collections.deque() if code not in self.temp_transaction_queue_dict: self.temp_transaction_queue_dict[code] = collections.deque() if code not in self.temp_log_queue_dict: self.temp_log_queue_dict[code] = queue.Queue(maxsize=1000) if code not in self.upload_l2_data_task_dict: t1 = threading.Thread(target=lambda: self.__run_upload_order_task(code), daemon=True) t1.start() t2 = threading.Thread(target=lambda: self.__run_upload_transaction_task(code), daemon=True) t2.start() # t3 = threading.Thread(target=lambda: self.__run_log_task(code), daemon=True) # t3.start() self.upload_l2_data_task_dict[code] = (t1, t2) # 释放已经分配的队列 def release_distributed_upload_queue(self, code): self.data_channel_distribute_manager.release_distribute_channel(code) if code in self.temp_order_queue_dict: self.temp_order_queue_dict[code].clear() self.temp_order_queue_dict.pop(code) if code in self.temp_transaction_queue_dict: self.temp_transaction_queue_dict[code].clear() self.temp_transaction_queue_dict.pop(code) if code in self.temp_log_queue_dict: self.temp_log_queue_dict.pop(code) if code in self.upload_l2_data_task_dict: self.upload_l2_data_task_dict.pop(code) # 处理订单数据并上传 def __run_upload_order_task(self, code): q: collections.deque = self.temp_order_queue_dict.get(code) temp_list = [] while True: try: while len(q) > 0: data = q.popleft() # 前置数据处理，过滤掉无用的数据 data = self.__filter_order(data) if data: temp_list.append(data) if temp_list: # 上传数据 # self.__upload_l2_data(code, upload_queue, temp_list) # self.__upload_l2_order_data(code, temp_list) __start_time = time.time() last_data = temp_list[-1] shared_memery_number, m_array, zmq_host = \ self.data_channel_distribute_manager.get_distributed_channel(code)[0] # 数据填充到内存 shared_memery_util.set_datas(m_array, (code, temp_list, time.time())) # 通知获取数据 _socket = self.data_channel_distribute_manager.get_zmq_socket(zmq_host) _socket.send( msgpack.packb({"type": self.TYPE_DELEGATE, "data": {"memery_number": shared_memery_number}})) _socket.recv_string() use_time = time.time() - __start_time if use_time > 0.01: # 记录10ms以上的数据 huaxin_l2_log.info(logger_local_huaxin_l2_error, f"耗时:{use_time}s 结束数据：{last_data} 数据数量：{len(temp_list)}") temp_list = [] else: if code not in self.temp_order_queue_dict: self.l2_order_codes.discard(code) break self.l2_order_codes.add(code) time.sleep(0.001) except Exception as e: logging.exception(e) logger_local_huaxin_l2_error.exception(e) finally: temp_list.clear() # 处理成交数据并上传 def __run_upload_transaction_task(self, code): q: collections.deque = self.temp_transaction_queue_dict.get(code) temp_list = [] while True: try: while len(q) > 0: data = q.popleft() data = self.__filter_transaction(data) if data: temp_list.append(data) if temp_list: # 上传数据 # self.__upload_l2_data(code, upload_queue, temp_list) shared_memery_number, m_array, zmq_host = \ self.data_channel_distribute_manager.get_distributed_channel(code)[1] # 数据填充到内存 shared_memery_util.set_datas(m_array, (code, temp_list)) # 通知获取数据 _socket = self.data_channel_distribute_manager.get_zmq_socket(zmq_host) _socket.send( msgpack.packb({"type": self.TYPE_TRANSACTION, "data": {"memery_number": shared_memery_number}})) _socket.recv_string() temp_list = [] else: if code not in self.temp_transaction_queue_dict: self.l2_transaction_codes.discard(code) break self.l2_transaction_codes.add(code) time.sleep(0.001) except Exception as e: logger_local_huaxin_l2_error.exception(e) finally: temp_list.clear() def __run_log_task(self, code): q: queue.Queue = self.temp_log_queue_dict.get(code) while True: try: temp = q.get(timeout=10) huaxin_l2_log.info(logger_local_huaxin_l2_special_volume, f"{temp}") except: time.sleep(0.02) finally: if code not in self.temp_log_queue_dict: break class L2DataUploadProtocolManager: # ipchosts IPC协议 def __init__(self, ipchosts): self.ipchosts = ipchosts # 所有的client self.socket_client_dict = {} # 保存代码分配的client 格式：{code:(host, socket)} self.code_socket_client_dict = {} self.rlock = threading.RLock() context = zmq.Context() if constant.is_windows(): return for host in self.ipchosts: socket = context.socket(zmq.REQ) socket.connect(host) self.socket_client_dict[host] = socket # 获取 def __get_available_ipchost(self): if len(self.code_socket_client_dict) >= len(self.socket_client_dict): raise Exception("无可用host") used_hosts = set([self.code_socket_client_dict[k][0] for k in self.code_socket_client_dict]) for host in self.socket_client_dict: if host not in used_hosts: return host, self.socket_client_dict[host] raise Exception("无可用host") # 分配HOST def distribute_upload_host(self, code): if code in self.code_socket_client_dict: return self.rlock.acquire() try: host_info = self.__get_available_ipchost() if host_info: self.code_socket_client_dict[code] = host_info finally: self.rlock.release() def release_distributed_upload_host(self, code): if code not in self.code_socket_client_dict: return self.rlock.acquire() try: if code in self.code_socket_client_dict: self.code_socket_client_dict.pop(code) finally: self.rlock.release() def upload_data_as_json(self, code, data): if code not in self.code_socket_client_dict: raise Exception("尚未分配host") host, socket = self.code_socket_client_dict[code] socket.send(marshal.dumps(data)) socket.recv_string() def add_target_code(code): target_codes.add(code) # 记录代码加入时间 target_codes_add_time[code] = time.time() def del_target_code(code): target_codes.discard(code) if code in target_codes_add_time: target_codes_add_time.pop(code) def add_subscript_codes(codes): # print("add_subscript_codes", codes) # 加入上传队列 pid = multiprocessing.current_process().pid common_queue.put(('', "l2_subscript_codes_v2", (pid, list(codes)))) def __send_response(sk, msg): msg = socket_util.load_header(msg) sk.sendall(msg) result, header_str = socket_util.recv_data(sk) if result: result_json = json.loads(result) if result_json.get("code") == 0: return True return False # 发送消息 def send_response(type, msg): try: sk = SendResponseSkManager.get_send_response_sk(type) if __send_response(sk, msg): return True else: # 再次发送 # print("再次发送") return __send_response(sk, msg) except ConnectionResetError as e: SendResponseSkManager.del_send_response_sk(type) sk = SendResponseSkManager.get_send_response_sk(type) return __send_response(sk, msg) except BrokenPipeError as e: SendResponseSkManager.del_send_response_sk(type) sk = SendResponseSkManager.get_send_response_sk(type) return __send_response(sk, msg) # 上传数据 def upload_data(code, _type, datas, new_sk=False): key = f"{_type}_{code}" fdata = json.dumps( {"type": _type, "data": {"code": code, "data": datas, "time": round(time.time() * 1000)}}) result = None try: if new_sk: sk = SendResponseSkManager.create_send_response_sk() result = __send_response(sk, fdata.encode('utf-8')) else: result = send_response(key, fdata.encode('utf-8')) except Exception as e: logging.exception(e) finally: pass def __run_upload_common(): # print("__run_upload_common") logger_system.info(f"l2_client __run_upload_common 线程ID:{tool.get_thread_id()}") while True: try: while not common_queue.empty(): temp = common_queue.get() upload_data(temp[0], temp[1], temp[2]) except Exception as e: logger_local_huaxin_l2_error.exception(e) logger_local_huaxin_l2_error.error(f"上传普通数据出错：{str(e)}") finally: time.sleep(0.01) def __run_log(): # print("__run_log") logger_system.info(f"l2_client __run_log 线程ID:{tool.get_thread_id()}") async_log_util.huaxin_l2_log.run_sync() # 采用socket传输数据 def run_upload_common(): t = threading.Thread(target=lambda: __run_upload_common(), daemon=True) t.start() def run_log(): t = threading.Thread(target=lambda: __run_log(), daemon=True) t.start() def __test(): # 分配数据 pass def run_test(): t = threading.Thread(target=lambda: __test(), daemon=True) t.start() def test(): ipclist = [] for i in range(0, 70): ipclist.append(f"ipc://l2order{i}.ipc") manager = L2DataUploadProtocolManager(ipclist) code = "000333" manager.distribute_upload_host(code) manager.upload_data_as_json(code, {"test": "test"})