新版深证下单完善

Administrator

2024-04-12 b92c5507bc73f043882ccf39fa5376686b4684d6

 l2/place_order_single_data_manager.py

@@ -2,10 +2,12 @@
下单信号管理
"""
from l2 import l2_log
from l2.huaxin import l2_huaxin_util
from log_module.log import logger_l2_trade_buy, logger_debug
from utils import tool


class L2TradeSingleManager:
class L2TradeSingleDataProcessor:
    """
    L2逐笔成交数据信号管理
    """
@@ -23,7 +25,6 @@
        @param data:
        @return:
        """
        l2_log.info(code, logger_l2_trade_buy, f"涨停卖数据：{data['val']['orderNo']}")
        if code not in cls.__latest_limit_up_sell_list_dict:
            cls.__latest_limit_up_sell_list_dict[code] = []
        cls.__latest_limit_up_sell_list_dict[code].append(data)
@@ -69,15 +70,104 @@
                else:
                    cls.__latest_sell_data[code] = [sell_no, data[2]]
            # 判断是否是最后一笔卖单
            l2_log.info(code, logger_l2_trade_buy, f"被动卖数据：{data}")

            # 判断这个订单号是否成交完
            sell_list = cls.__latest_limit_up_sell_list_dict.get(code)
            if not sell_list:
                return
            sell_info = sell_list[-1]
            l2_log.info(code, logger_l2_trade_buy, f"最近涨停卖：{sell_info['val']['orderNo']}")
            if sell_no == sell_info['val']['orderNo'] and sell_info["val"]["num"] == cls.__latest_sell_data[code][
                1] // 100:
                # 成交完成
                l2_log.info(code, logger_l2_trade_buy, f"找到最近的被动涨停卖单数据：{data['val']['orderNo']}, 可以触发下单")
                L2TradeSingleDataManager.set_latest_sell_data(code, data)
                logger_l2_trade_buy.info(f"{code}#找到最近的被动涨停卖单数据：{data['val']['orderNo']}, 可以触发下单")
                # l2_log.info(code, logger_l2_trade_buy, f"找到最近的被动涨停卖单数据：{data['val']['orderNo']}, 可以触发下单")
        except Exception as e:
            logger_debug.exception(e)


class L2TradeSingleDataManager:
    TYPE_PASSIVE = 0
    TYPE_ACTIVE = 1

    """
    买入信号管理
    """
    # 最近的涨停卖被动成交数据
    __latest_sell_data_dict = {}  # 格式为：{code:(逐笔成交数据,生效时间（带ms）)}

    # 由被动向主动卖成交转变的数据
    __latest_sell_active_deal_data_dict = {}  # 格式为：{code:(逐笔成交数据,生效时间（带ms）)}

    @classmethod
    def set_latest_sell_data(cls, code, data):
        """
        设置最近成交的涨停卖被动成交数据
        @param code: 代码
        @param data: L2逐笔成交数据  数据格式：(data['SecurityID'], data['TradePrice'], data['TradeVolume'],
        #           data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
        #           data['SellNo'], data['ExecType'])
        @return:
        """
        deal_time = l2_huaxin_util.convert_time(data[3], True)
        # 生效时间在1s以内
        cls.__latest_sell_data_dict[code] = (data, tool.trade_time_add_millionsecond(deal_time, 1000))

    @classmethod
    def set_sell_passive_to_active_datas(cls, code, passive_data, active_data):
        """
        设置被动卖成交向主动卖成交的转变数据
        @param code: 代码
        @param passive_data: 被动卖成交逐笔
        @param active_data: 主动卖成交逐笔
        @return:
        """
        deal_time = l2_huaxin_util.convert_time(passive_data[3], True)
        # 生效时间在1s以内
        cls.__latest_sell_active_deal_data_dict[code] = (
            active_data, tool.trade_time_add_millionsecond(deal_time, 1000))

    @classmethod
    def get_valid_trade_single(cls, code, latest_time_with_ms):
        """
        获取有效的成交下单信号
        @param code:
        @param latest_time_with_ms:
        @return: （逐笔成交数据, 类型）
        """
        # 如果有最近卖涨停成交完就用最近的数据
        data = cls.__latest_sell_data_dict.get(code)
        if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0:
            return data, cls.TYPE_PASSIVE
        data = cls.__latest_sell_active_deal_data_dict.get(code)
        if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0:
            return data, cls.TYPE_ACTIVE
        return None

    @classmethod
    def is_can_place_order(cls, code, buy_data):
        """
        是否可以下单
        @param code: 代码
        @param buy_data: L2逐笔委托大于50w的买入数据
        @return: (是否可以下单, 原因)
        """
        buy_data_time_with_ms = tool.to_time_with_ms(buy_data['val']['time'], buy_data['val']['tms'])
        single = cls.get_valid_trade_single(code, buy_data_time_with_ms)
        if not single:
            return False, "没有找到可用信号"
        if single[0][6] < buy_data['val']['orderNo']:
            # 信号位置的成交买单号要小于买入数据的买单号才行
            return True, f"找到信号：{single}"
        else:
            return False, f"买单没有在信号位之后 信号：{single}"

    @classmethod
    def clear_data(cls, code):
        if code in cls.__latest_sell_data_dict:
            cls.__latest_sell_data_dict.pop(code)

        if code in cls.__latest_sell_active_deal_data_dict:
            cls.__latest_sell_active_deal_data_dict.pop(code)