信息上传优化

Administrator

2023-08-21 2d27b626b35ec9d62df36daf6405d425557e11ae

信息上传优化

 huaxin_api/trade_client.py

@@ -566,7 +566,7 @@
               pOrderField.LimitPrice, pOrderField.VolumeTotalOriginal, pOrderField.OrderSysID,
               pOrderField.OrderStatus))
        if pOrderField.OrderStatus != traderapi.TORA_TSTP_OST_Unknown:
            set_system_order_id(pOrderField.SInfo, pOrderField.OrderSysID)
            OrderIDManager.set_system_order_id(pOrderField.SecurityID, pOrderField.SInfo, pOrderField.OrderSysID)
            threading.Thread(target=lambda: self.__data_callback(TYPE_ORDER, 0, {"sinfo": pOrderField.SInfo,
                                                                                 "securityId": pOrderField.SecurityID,
                                                                                 "orderLocalId": pOrderField.OrderLocalID,
@@ -734,8 +734,6 @@


# 获取响应发送socket


global req_rid_dict
req_rid_dict = {}

@@ -745,7 +743,8 @@

    def OnTrade(self, client_id, request_id, type_, data):
        if type_ == 1:
            logger_local_huaxin_trade_debug.info(f"---------------------\n请求下单：client_id-{client_id} request_id-{request_id}")
            logger_local_huaxin_trade_debug.info(
                f"---------------------\n请求下单：client_id-{client_id} request_id-{request_id}")
            # 下单
            # 1-买 2-卖
            direction = data["direction"]
@@ -785,10 +784,12 @@
            sinfo = data["sinfo"]
            if direction == 1:
                if not orderSysID and localOrderID:
                    orderSysID = local_order_id_map.get(localOrderID)
                    orderSysID = OrderIDManager.get_system_id_by_local_id(localOrderID)
                # 撤买
                try:
                    if not orderSysID:
                        if localOrderID:
                            OrderIDManager.add_need_cancel_local_order_id(localOrderID)
                        raise Exception("没有找到系统订单号")
                    req_rid_dict[sinfo] = (client_id, request_id)
                    self.__tradeSimpleApi.cancel_buy(code, orderSysID, sinfo)
@@ -913,19 +914,39 @@
    strategy_pipe.send(data)


local_order_id_map = {}
# 订单号管理
class OrderIDManager:
    # 尚未撤单的本地订单号
    not_canceled_local_order_ids = set()
    local_order_id_map = {}

    __TradeSimpleApi = TradeSimpleApi()

# 设置系统订单ID
def set_system_order_id(sinfo, orderSystemId):
    # 获取本地订单ID
    local_order_id = None
    if req_rid_dict and sinfo in req_rid_dict:
        temp_params = req_rid_dict.get(sinfo)
        if len(temp_params) > 2:
            local_order_id = temp_params[2]
    if local_order_id and local_order_id not in local_order_id_map and orderSystemId:
        local_order_id_map[local_order_id] = orderSystemId
    @classmethod
    def get_system_id_by_local_id(cls, local_order_id):
        return cls.local_order_id_map.get(local_order_id)

    # 设置系统订单ID
    @classmethod
    def set_system_order_id(cls, code, sinfo, orderSystemId):
        # 获取本地订单ID
        local_order_id = None
        if req_rid_dict and sinfo in req_rid_dict:
            temp_params = req_rid_dict.get(sinfo)
            if len(temp_params) > 2:
                local_order_id = temp_params[2]
        if local_order_id:
            if local_order_id not in cls.local_order_id_map and orderSystemId:
                cls.local_order_id_map[local_order_id] = orderSystemId
            if local_order_id in cls.not_canceled_local_order_ids:
                # 执行撤单
                cls.not_canceled_local_order_ids.discard(local_order_id)
                cls.__TradeSimpleApi.cancel_buy(code, orderSystemId,
                                                f"local_cancel_buy-{code}-{round(time.time() * 1000)}")

    @classmethod
    def add_need_cancel_local_order_id(cls, local_order_id):
        cls.not_canceled_local_order_ids.add(local_order_id)


# 交易反馈回调

 trade/huaxin/trade_server.py

@@ -352,17 +352,6 @@
        data = data_json["data"]
        datas = data["data"]
        HuaXinL1TargetCodesManager.set_level_1_codes_datas(datas)
        # lp = LineProfiler()
        # lp.enable()
        # lp_wrap = lp(HuaXinL1TargetCodesManager.set_level_1_codes_datas)
        # lp_wrap(datas)
        # output = io.StringIO()
        # lp.print_stats(stream=output)
        # lp.disable()
        # with open(
        #         f"{constant.get_path_prefix()}/logs/profile/set_level_1_codes_datas_{time.time()}.txt",
        #         'w') as f:
        #     f.write(output.getvalue())


def clear_invalid_client():
@@ -375,8 +364,8 @@
            time.sleep(2)


def __recv_pipe_l1(pipe_trade, pipe_l1):
    if pipe_trade is not None and pipe_l1 is not None:
def __recv_pipe_l1(pipe_l1):
    if pipe_l1 is not None:
        while True:
            try:
                val = pipe_l1.recv()
@@ -387,6 +376,40 @@
                    type_ = val["type"]
                    if type_ == "set_target_codes":
                        TradeServerProcessor.set_target_codes(val)
            except Exception as e:
                logging.exception(e)


# 读取交易那边传递过来的数据
def __recv_pipe_trade(pipe_trade):
    if pipe_trade is not None:
        while True:
            try:
                val = pipe_trade.recv()
                if val:
                    data_json = json.loads(val)
                    print("收到来自交易客户端的数据：", data_json["type"])
                    # 处理数据
                    type_ = data_json["type"]
                    if type_ == "response":
                        # 主动触发的响应
                        hx_logger_trade_callback.info(f"response：request_id-{data_json['request_id']}")
                        # 设置响应内容
                        trade_api.set_response(data_json["request_id"], data_json['data'])
                    elif type_ == "trade_callback":
                        try:
                            # 交易回调
                            data_json = data_json["data"]
                            ctype = data_json["type"]
                            # 记录交易反馈日志
                            hx_logger_trade_callback.info(data_json)
                            # 重新请求委托列表与资金
                            huaxin_trade_data_update.add_delegate_list()
                            huaxin_trade_data_update.add_deal_list()
                            huaxin_trade_data_update.add_money_list()
                            # print("响应结果：", data_json['data'])
                        finally:
                            pass
            except Exception as e:
                logging.exception(e)

@@ -680,7 +703,11 @@
    huaxin_trade_api.set_pipe_trade(pipe_trade)

    # 监听l1那边传过来的代码
    t1 = threading.Thread(target=lambda: __recv_pipe_l1(pipe_trade, pipe_l1), daemon=True)
    t1 = threading.Thread(target=lambda: __recv_pipe_l1(pipe_l1), daemon=True)
    t1.start()

    # 监听交易结果
    t1 = threading.Thread(target=lambda: __recv_pipe_trade(pipe_trade), daemon=True)
    t1.start()

    print("create TradeServer")