""" GUI管理 """ import math import psutil import constant from code_attribute import gpcode_manager from l2.subscript import l2_subscript_manager from log_module import log from log_module.log import logger_l2_trade, logger_system import logging import multiprocessing import os import threading from task import task_manager from third_data import hx_qc_value_util from third_data.code_plate_key_manager import KPLPlateForbiddenManager from utils import shared_memery_util logger_system.info("程序启动Pre：{}", os.getpid()) import huaxin_client.trade_client import huaxin_client.l2_client import huaxin_client.l1_client from huaxin_client import l2_market_client, l2_client_v2 from servers import server_util, huaxin_trade_server, server def run_strategy(queue_strategy_r_trade_w_: multiprocessing.Queue, queue_l1_w_strategy_r_: multiprocessing.Queue, queue_strategy_w_trade_r_: multiprocessing.Queue, queue_strategy_w_trade_r_for_read_: multiprocessing.Queue, trade_ipc_addr): """ 策略进程 @param queue_strategy_r_trade_w_: @param queue_l1_w_strategy_r_: @param queue_strategy_w_trade_r_: @param queue_strategy_w_trade_r_for_read_: @param trade_ipc_addr: 交易ipc地址(下单地址, 撤单地址) @return: """ logger_system.info("策略进程ID：{}", os.getpid()) log.close_print() # 初始化参数 server.global_data_loader.init() # 开启数据服务器 threading.Thread(target=server_util.run_data_server, name="createDataServer", daemon=True).start() # 运行数据监听服务 threading.Thread(target=task_manager.run_data_listener, name="task_manager", args=(queue_other_w_l2_r, queue_l1_w_strategy_r_), daemon=True).start() # # 启动华鑫交易服务 huaxin_trade_server.run(queue_strategy_r_trade_w_, queue_strategy_w_trade_r_, queue_strategy_w_trade_r_for_read_, trade_ipc_addr) # 主服务 def createServer(pipe): logger_system.info("create Server") laddr = "", 9001 try: tcpserver = server.MyThreadingTCPServer(laddr, server.MyBaseRequestHandle, pipe_trade=pipe) # 注意：参数是MyBaseRequestHandle tcpserver.serve_forever() except Exception as e: logger_system.exception(e) logger_system.error(f"端口服务器：{laddr[1]} 启动失败") if __name__ == '__main__1': huaxin_client.l2_client.test() def __create_l2_subscript(): channel_list = [] for i in range(constant.HUAXIN_L2_MAX_CODES_COUNT): # 创建委托/成交的共享数组和ZMQ通信通道 delegate_ipc_addr = f"ipc://order_{i}.ipc", deal_ipc_addr = f"ipc://deal_{i}.ipc", delegate = [0, shared_memery_util.create_array(), delegate_ipc_addr] delegate[0] = shared_memery_util.get_number(delegate[1]) deal = [0, shared_memery_util.create_array(), deal_ipc_addr] deal[0] = shared_memery_util.get_number(deal[1]) channel_list.append((delegate, deal)) # L2进程数量 l2_process_count = 8 base_channel_count = len(channel_list) // l2_process_count left_count = len(channel_list) % l2_process_count index = 0 # ======分组====== # 记录每个分组的数量 channel_count_list = [] # 数据回调队列 data_callback_queue_list = [] # 消息传递队列 sub_single_queue_list = [] for i in range(l2_process_count): channel_count = base_channel_count + (1 if i < left_count else 0) channel_count_list.append(channel_count) # 该进程下的通道 channels = channel_list[index:index + channel_count] index += channel_count # 订阅信号队列, 数据回调队列（回调频次小的数据通过这种回调） sub_single_queue, data_callback_queue = multiprocessing.Queue(), multiprocessing.Queue() sub_single_queue_list.append(sub_single_queue) data_callback_queue_list.append(data_callback_queue) l2_process = multiprocessing.Process(target=l2_client_v2.run, args=(sub_single_queue, data_callback_queue, channels, i, )) l2_process.start() l2_subscript_manager.process_manager = l2_subscript_manager.TargetCodeProcessManager(sub_single_queue_list, channel_count_list) # 监听L2市场行情数据 huaxin_trade_server.run_l2_market_info_reciever(data_callback_queue_list) # 启动ZMQserver，针对委托队列与成交队列进行监听 l2_subscript_manager.L2DataListener(channel_list).create_data_listener(huaxin_trade_server.my_l2_data_callback) if __name__ == '__main__': # 可绑定16-31之间的核 try: logger_l2_trade.info("启动程序") logger_system.info("启动程序--------") log.close_print() # L2读其他写 queue_other_w_l2_r = multiprocessing.Queue(maxsize=1000) # l1 queue_l1_w_strategy_r = multiprocessing.Queue(maxsize=1000) queue_l1_r_strategy_w = multiprocessing.Queue(maxsize=1000) # 交易读策略写 queue_strategy_w_trade_r = multiprocessing.Queue(maxsize=1000) queue_strategy_w_trade_r_for_read = multiprocessing.Queue(maxsize=1000) # 策略读交易写 queue_strategy_r_trade_w = multiprocessing.Queue(maxsize=1000) # 下单,撤单ipc地址 order_ipc_addr, cancel_order_ipc_addr = "ipc://trade_order.ipc", "ipc://trade_cancel_order.ipc" logger_system.info("主进程ID：{}", os.getpid()) fix_codes = set() open_buy_codes = gpcode_manager.BuyOpenLimitUpCodeManager().get_codes() if open_buy_codes: fix_codes |= set(open_buy_codes) # 要监控的高标 high_codes = KPLPlateForbiddenManager().get_watch_high_codes() if high_codes: fix_codes |= set(high_codes) # L1订阅数据 l1Process = multiprocessing.Process(target=huaxin_client.l1_client.run, args=(queue_l1_w_strategy_r, queue_l1_r_strategy_w, fix_codes,)) l1Process.start() l2MarketProcess = multiprocessing.Process(target=l2_market_client.run, args=(queue_l1_w_strategy_r,)) l2MarketProcess.start() # 交易进程 tradeProcess = multiprocessing.Process( target=huaxin_client.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)) tradeProcess.start() # 此处将L2的进程与策略进程合并 # 测试L2单独进程 if True: __create_l2_subscript() else: # 将L2的进程改为线程执行 threading.Thread(target=huaxin_client.l2_client.run, args=( queue_other_w_l2_r, huaxin_trade_server.my_l2_data_callbacks), daemon=True).start() # 运行华鑫增值服务进程 threading.Thread(target=hx_qc_value_util.run, daemon=True).start() # 绑核运行 # psutil.Process(l1Process.pid).cpu_affinity([20]) # psutil.Process(tradeProcess.pid).cpu_affinity([21, 22]) # cpu_indexes = [i for i in range(23, 30)] # psutil.Process(os.getpid()).cpu_affinity(cpu_indexes) # 主进程 run_strategy(queue_strategy_r_trade_w, queue_l1_w_strategy_r, queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_read, (order_ipc_addr, cancel_order_ipc_addr)) # 将tradeServer作为主进程 l1Process.join() # l2Process.join() tradeProcess.join() except Exception as e: logging.exception(e) logger_system.exception(e)