blame | 历史 | 补丁 | 提交 | 提交对比 | show whitespace
Administrator
2024-03-17 15199f8e93fe48e6261c99eadf6673d788db3a80 
 huaxin_client/l2_client.py


@@ -13,9 +13,10 @@


from huaxin_client import constant


from huaxin_client import l2_data_manager


import lev2mdapi


from huaxin_client.code_queue_distribute_manager import CodeQueueDistributeManager


from huaxin_client.code_queue_distribute_manager import CodeQueueDistributeManager, CodeDataCallbackDistributeManager


from huaxin_client.command_manager import L2ActionCallback


from huaxin_client.l2_data_manager import L2DataUploadManager


from huaxin_client.l2_data_transform_protocol import L2DataCallBack


from log_module import log, async_log_util


from log_module.async_log_util import huaxin_l2_log


from log_module.log import logger_local_huaxin_l2_subscript, logger_system, logger_local_huaxin_l2_transaction, \


@@ -546,7 +547,7 @@






def test_add_codes(queue_r):


    time.sleep(5)


    time.sleep(10)


    # if value:


    #     if type(value) == bytes:


    #         value = value.decode("utf-8")


@@ -565,8 +566,9 @@


                  ("002908", int(50 * 10000 / 12.78), 12.78, 12.00, 200)]




    queue_r.put_nowait(json.dumps({"type": "l2_cmd", "data": [demo_datas[0]]}))


    time.sleep(1)


    time.sleep(10)


    while True:


        try:


        spi.l2_data_upload_manager.add_l2_order_detail(


            {'SecurityID': '603002', 'Price': 6.35, 'Volume': 275000, 'Side': "1", 'OrderType': '0',


             'OrderTime': '13000015',


@@ -575,51 +577,20 @@


            {'SecurityID': '603002', 'Price': 6.35, 'Volume': 200, 'Side': "1", 'OrderType': '0',


             'OrderTime': '13000015',


             'MainSeq': 2, 'SubSeq': 6739148, 'OrderNO': 5512467, 'OrderStatus': 'D'}, 0)


        queue_r.put_nowait(json.dumps({"type": "listen_volume", "data": {"code": "603002", "volume": 100}}))


            # queue_r.put_nowait(json.dumps({"type": "listen_volume", "data": {"code": "603002", "volume": 100}}))


        time.sleep(0.1)


        spi.l2_data_upload_manager.add_l2_order_detail(


            {'SecurityID': '603002', 'Price': 6.35, 'Volume': 100, 'Side': "1", 'OrderType': '0',


             'OrderTime': '13000015',


             'MainSeq': 2, 'SubSeq': 6739148, 'OrderNO': 5512467, 'OrderStatus': 'D'}, 0)


        except Exception as e:


            logging.exception(e)


        finally:


        time.sleep(10)






def run(queue_r: multiprocessing.Queue, order_queues: List[multiprocessing.Queue],


        transaction_queues: List[multiprocessing.Queue], market_queue: multiprocessing.Queue, order_ipc_hosts: list) -> None:


    # def test_add_codes():


    #     time.sleep(5)


    #     # if value:


    #     #     if type(value) == bytes:


    #     #         value = value.decode("utf-8")


    #     #     data = json.loads(value)


    #     #     _type = data["type"]


    #     #     if _type == "listen_volume":


    #     #         volume = data["data"]["volume"]


    #     #         code = data["data"]["code"]


    #     #         spi.set_code_special_watch_volume(code, volume)


    #     #     elif _type == "l2_cmd":


    #     #         l2CommandManager.process_command(command_manager.CLIENT_TYPE_CMD_L2, None, data)


    #


    #     demo_datas = [("603002", int(50 * 10000 / 6.35), 6.35), ("002654", int(50 * 10000 / 15.59), 15.59),


    #                   ("603701", int(50 * 10000 / 14.28), 14.28), ("002908", int(50 * 10000 / 12.78), 12.78)]


    #


    #     queue_r.put_nowait(json.dumps({"type": "l2_cmd", "data": [demo_datas[0]]}))


    #     time.sleep(1)


    #


    #     spi.l2_data_upload_manager.add_l2_order_detail(


    #         {'SecurityID': '603002', 'Price': 6.35, 'Volume': 275000, 'Side': "1", 'OrderType': '0',


    #          'OrderTime': '13000015',


    #          'MainSeq': 2, 'SubSeq': 6739147, 'OrderNO': 5512466, 'OrderStatus': 'D'}, 0)


    #     spi.l2_data_upload_manager.add_l2_order_detail(


    #         {'SecurityID': '603002', 'Price': 6.35, 'Volume': 200, 'Side': "1", 'OrderType': '0',


    #          'OrderTime': '13000015',


    #          'MainSeq': 2, 'SubSeq': 6739148, 'OrderNO': 5512467, 'OrderStatus': 'D'}, 0)


    #     queue_r.put_nowait(json.dumps({"type": "listen_volume", "data": {"code": "603002", "volume": 100}}))


    #     time.sleep(0.1)


    #     spi.l2_data_upload_manager.add_l2_order_detail(


    #         {'SecurityID': '603002', 'Price': 6.35, 'Volume': 100, 'Side': "1", 'OrderType': '0',


    #          'OrderTime': '13000015',


    #          'MainSeq': 2, 'SubSeq': 6739148, 'OrderNO': 5512467, 'OrderStatus': 'D'}, 0)


        transaction_queues: List[multiprocessing.Queue], market_queue: multiprocessing.Queue, order_ipc_hosts: list, data_callbacks:list) -> None:




    logger_system.info("L2进程ID：{}", os.getpid())


    logger_system.info(f"l2_client 线程ID:{tool.get_thread_id()}")


@@ -632,12 +603,13 @@


        # 初始化


        order_queue_distribute_manager = CodeQueueDistributeManager(order_queues)


        transaction_queue_distribute_manager = CodeQueueDistributeManager(transaction_queues)


        data_callback_distribute_manager = CodeDataCallbackDistributeManager(data_callbacks)


        l2_data_upload_manager = L2DataUploadManager(order_queue_distribute_manager,


                                                     transaction_queue_distribute_manager, market_queue, order_ipc_hosts)


                                                     transaction_queue_distribute_manager, market_queue, order_ipc_hosts, data_callback_distribute_manager)


        __init_l2(l2_data_upload_manager)


        l2_data_manager.run_upload_common()


        l2_data_manager.run_log()


        # 测试


        # TODO 测试


        # threading.Thread(target=lambda: test_add_codes(queue_r), daemon=True).start()


        global l2CommandManager


        l2CommandManager = command_manager.L2CommandManager()