辨识度修改/临时大单bug修改

Administrator

2025-01-22 f2dfe1818b6286e4ff25e97c4d72a17bae4eea42

 l2/l2_transaction_data_manager.py

@@ -2,42 +2,158 @@
L2成交数据处理器
"""
import json
import logging
import time

from code_attribute import gpcode_manager
from db import redis_manager
import l2_data_util
from db import redis_manager_delegate as redis_manager
from db.redis_manager_delegate import RedisUtils
from l2 import l2_data_util, l2_data_manager, transaction_progress, l2_data_source_util
from l2.cancel_buy_strategy import LCancelRateManager, LCancelBigNumComputer, \
    SecondCancelBigNumComputer, HourCancelBigNumComputer, \
    GCancelBigNumComputer, FCancelBigNumComputer
from l2.l2_data_manager_new import L2TradeDataProcessor
from l2.l2_data_util import L2DataUtil, local_today_canceled_buyno_map
from log_module import async_log_util
from log_module.log import logger_l2_trade_buy_queue, hx_logger_l2_upload, hx_logger_l2_debug, \
    hx_logger_l2_transaction_desc, logger_debug
from msg import push_msg_manager
from third_data import kpl_data_manager
from trade import current_price_process_manager, trade_manager, l2_trade_factor
from trade.deal_big_money_manager import DealOrderNoManager
from l2 import l2_log
from l2.huaxin import l2_huaxin_util
from l2.l2_data_util import local_today_sellno_map, local_today_datas
from l2.place_order_single_data_manager import L2TradeSingleDataProcessor, L2TradeSingleDataManager

from log_module import async_log_util, log_export
from log_module.log import hx_logger_l2_transaction_desc, hx_logger_l2_transaction_sell_order, hx_logger_l2_active_sell, \
    hx_logger_l2_transaction_big_buy_order, hx_logger_l2_transaction_big_sell_order

from utils import tool


class HuaXinTransactionDatasProcessor:
class BigOrderDealManager:
    """
    成交大单管理
    """
    __total_buy_datas_dict = {}
    __total_sell_datas_dict = {}
    __instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(BigOrderDealManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_datas()
        return cls.__instance

    @classmethod
    def __load_datas(cls):
        # 从日志加载数据
        buy_order_map = log_export.load_huaxin_big_buy_order()
        if buy_order_map:
            for code in buy_order_map:
                cls.__total_buy_datas_dict[code] = buy_order_map[code]

        sell_order_map = log_export.load_huaxin_big_sell_order()
        if sell_order_map:
            for code in sell_order_map:
                cls.__total_sell_datas_dict[code] = sell_order_map[code]

    def add_buy_datas(self, code, datas):
        """
        添加大买单数据
        @param code: 代码
        @param datas:[(买单号,总股数,总成交额,成交开始时间,成交结束时间)]
        @return:
        """
        if not datas:
            return
        if code not in self.__total_buy_datas_dict:
            self.__total_buy_datas_dict[code] = []
        self.__total_buy_datas_dict[code].extend(datas)
        # 将数据保存到日志
        l2_log.info(code, hx_logger_l2_transaction_big_buy_order, f"{datas}")

    def add_sell_datas(self, code, datas):
        """
        添加大卖单数据
        @param code: 代码
        @param datas:[(卖单号,总股数,总成交额)]
        @return:
        """
        if not datas:
            return
        if code not in self.__total_sell_datas_dict:
            self.__total_sell_datas_dict[code] = []
        self.__total_sell_datas_dict[code].extend(datas)
        l2_log.info(code, hx_logger_l2_transaction_big_sell_order, f"{datas}")

    def get_total_buy_money(self, code):
        """
        获取总共的大单买金额
        @param code:
        @return:
        """
        if code not in self.__total_buy_datas_dict:
            return 0
        return int(sum([x[2] for x in self.__total_buy_datas_dict[code]]))

    def get_total_buy_order_ids(self, code):
        """
        获取成交大单的订单号
        @param code:
        @return:
        """
        if code not in self.__total_buy_datas_dict:
            return set()
        return set([x[0] for x in self.__total_buy_datas_dict[code]])

    def get_total_buy_count(self, code):
        """
        获取大单成交的笔数
        @param code:
        @return:
        """
        if code not in self.__total_buy_datas_dict:
            return 0
        return len(self.__total_buy_datas_dict[code])

    def get_total_buy_money_list(self, code):
        """
        获取大单列表
        @param code:
        @return:
        """
        if code not in self.__total_buy_datas_dict:
            return []
        return [x[2] for x in self.__total_buy_datas_dict[code]]

    def get_total_buy_data_list(self, code):
        """
        获取所有的买单数据列表
        @param code:
        @return:[(买单号,总股数,总成交额,成交开始时间,成交结束时间)]
        """
        return self.__total_buy_datas_dict.get(code)

    def get_total_sell_money(self, code, threshold_money=None):
        """
        获取总共的大单卖金额
        @param threshold_money: 大单阈值
        @param code:
        @return:
        """
        if code not in self.__total_sell_datas_dict:
            return 0
        if threshold_money:
            return int(sum([x[2] if x[2] >= threshold_money else 0 for x in self.__total_sell_datas_dict[code]]))
        else:
            return int(sum([x[2] for x in self.__total_sell_datas_dict[code]]))


# 成交数据统计
class HuaXinBuyOrderManager:
    __db = 0
    __instance = None
    __redis_manager = redis_manager.RedisManager(0)

    __TradeBuyQueue = transaction_progress.TradeBuyQueue()
    # 正在成交的订单
    __dealing_order_info_dict = {}

    # 正在成交的主动买的订单
    __dealing_active_buy_order_info_dict = {}
    # 最近成交的订单{"code":(订单号，是否成交完成)}
    __latest_deal_order_info_dict = {}

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(HuaXinTransactionDatasProcessor, cls).__new__(cls, *args, **kwargs)
            cls.__instance = super(HuaXinBuyOrderManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_datas()
        return cls.__instance

@@ -64,163 +180,469 @@
            RedisUtils.setex(self.__get_redis(), f"dealing_order_info-{code}", tool.get_expire(),
                             json.dumps(self.__dealing_order_info_dict[code]))

    # 统计成交的情况
    @classmethod
    def get_dealing_order_info(cls, code):
        """
        获取当前正在成交的数据
        @param code:
        @return: [订单号，总股数，成交金额,成交开始时间,成交结束时间]
        """
        return cls.__dealing_order_info_dict.get(code)

    def __statistic_deal_desc(self, code, data, total_buy_num):
        if code not in self.__dealing_order_info_dict:
            # 数据格式[订单号，总手数，开始成交时间，结束成交时间, 总买]
            self.__dealing_order_info_dict[code] = [data[6], 0, data[3], data[3], total_buy_num]
        if self.__dealing_order_info_dict[code][0] == data[6]:
            # 成交同一个订单号
            self.__dealing_order_info_dict[code][1] += data[2]
            self.__dealing_order_info_dict[code][3] = data[3]
        else:
            # 保存上一条数据
            async_log_util.info(hx_logger_l2_transaction_desc, f"{code}#{self.__dealing_order_info_dict[code]}")
            # 设置最近成交完成的一条数据
            deal_info = (
                self.__dealing_order_info_dict[code][0],
                self.__dealing_order_info_dict[code][4] == self.__dealing_order_info_dict[code][1])
            self.__latest_deal_order_info_dict[code] = deal_info
            # 初始化本条数据
            self.__dealing_order_info_dict[code] = [data[6], data[2], data[3], data[3], total_buy_num]
            return deal_info
        return None
    @classmethod
    def get_dealing_active_order_info(cls, code):
        """
        获取正在主动成交的数据
        @param code:
        @return:[订单号，总股数，成交金额,成交开始时间,成交结束时间]
        """
        return cls.__dealing_active_buy_order_info_dict.get(code)

    # 计算成交进度
    def __compute_latest_trade_progress(self, code, buyno_map, datas):
        buy_progress_index = None
        for i in range(len(datas) - 1, -1, -1):
            d = datas[i]
            buy_no = f"{d[6]}"
            if buyno_map and buy_no in buyno_map:
                async_log_util.info(hx_logger_l2_debug, f"{code}成交进度：{buyno_map[buy_no]['index']}")
                buy_progress_index = buyno_map[buy_no]["index"]
                break
        return buy_progress_index

    def process_huaxin_transaction_datas(self, code, datas):
        # 设置成交价
        current_price_process_manager.set_trade_price(code, datas[-1][1])
        total_datas = l2_data_util.local_today_datas.get(code)
        __start_time = time.time()
        try:
            buyno_map = l2_data_util.local_today_buyno_map.get(code)
            if not buyno_map:
                if trade_manager.CodesTradeStateManager().get_trade_state(
                        code) != trade_manager.TRADE_STATE_NOT_TRADE:
                    l2_data_util.load_l2_data(code)
                    buyno_map = l2_data_util.local_today_buyno_map.get(code)
            if buyno_map is None:
                buyno_map = {}

            order_begin_pos = l2_data_manager.TradePointManager().get_buy_compute_start_data_cache(code)

            if order_begin_pos and order_begin_pos.buy_exec_index and order_begin_pos.buy_exec_index > -1:
                # 已经下单的需要统计F撤
                try:
                    for d in datas:
                        if FCancelBigNumComputer().need_cancel(d)[0]:
                            L2TradeDataProcessor.cancel_buy(code, f"F撤撤单:{d}")
                            order_begin_pos = None
                            break
                except Exception as e:
                    async_log_util.error(hx_logger_l2_debug, str(e))
                try:
                    if LCancelBigNumComputer().add_transaction_datas(code, datas)[0]:
                        L2TradeDataProcessor.cancel_buy(code, f"L后成交太快撤单:{datas[-1]}")
                        order_begin_pos = None
                except Exception as e:
                    async_log_util.error(hx_logger_l2_debug, str(e))


            # 计算已经成交的大单
            big_money_count = 0
            for d in datas:
                data = buyno_map.get(f"{d[6]}")
                buy_num = None
                if data:
                    buy_num = data["val"]["num"] * 100
                # 统计成交单
                deal_info = self.__statistic_deal_desc(code, d, buy_num)
                if deal_info and deal_info[1]:
                    data = buyno_map.get(f"{deal_info[0]}")
                    print("已经成交索引：", data["index"])
                    val = data["val"]
                    if l2_data_util.is_big_money(val) and L2DataUtil.is_limit_up_price_buy(val):
                        big_money_count += 1
                        DealOrderNoManager().add_orderno(code, f"{deal_info[0]}")
                    # L后是否有成交，如果有成交就需要除去当前笔数，然后重新囊括一笔
                    LCancelBigNumComputer().add_deal_index(code, data["index"], order_begin_pos.buy_single_index)
            if big_money_count > 0:
                LCancelRateManager.compute_big_num_deal_rate(code)

            buy_progress_index = self.__compute_latest_trade_progress(code, buyno_map, datas)

            if buy_progress_index is not None:
                buy_progress_index_changed = self.__TradeBuyQueue.set_traded_index(code, buy_progress_index,
                                                                                   total_datas)

                async_log_util.info(logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{}", code,
                                    buy_progress_index)

                GCancelBigNumComputer().set_trade_progress(code, order_begin_pos.buy_single_index, buy_progress_index)

                LCancelBigNumComputer().set_trade_progress(code, order_begin_pos.buy_single_index, buy_progress_index,
                                                           total_datas)
                SecondCancelBigNumComputer().set_transaction_index(
                    code,
                    buy_progress_index)
                if order_begin_pos and order_begin_pos.buy_exec_index and order_begin_pos.buy_exec_index > -1:
                    HourCancelBigNumComputer().set_transaction_index(code, order_begin_pos.buy_single_index,
                                                                     buy_progress_index)
                    # ---------------------------------判断板块是否跟上来了-------------------------------
                    try:
                        pass
                        # order_begin_pos = l2_data_manager.TradePointManager().get_buy_compute_start_data_cache(code)
                        # volume_rate = 0
                        # volume_info = L2TradeDataProcessor.volume_rate_info.get(code)
                        # if volume_info:
                        #     volume_rate = volume_info[0]
                        # need_cancel, msg = UCancelBigNumComputer().need_cancel(code, buy_progress_index, order_begin_pos,
                        #                                                        kpl_data_manager.KPLLimitUpDataRecordManager.get_current_reason_codes_dict(),
                        #                                                        volume_rate)
                        # if need_cancel:
                        #     L2TradeDataProcessor.cancel_buy(code, msg)
                    except Exception as e:
                        logger_debug.exception(e)
                    if buy_progress_index_changed:
                        # 交易进度变化，判断到真实下单位置的距离
                        real_order_index = SecondCancelBigNumComputer().get_real_place_order_index_cache(code)
                        if real_order_index and real_order_index >= buy_progress_index:
                            left_count = 0
                            for i in range(buy_progress_index + 1, real_order_index):
                                val = total_datas[i]["val"]
                                if not L2DataUtil.is_limit_up_price_buy(val):
                                    continue
                                if val["num"] * float(val["price"]) < 5000:
                                    continue
                                left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count_v2(
                                    code, i,
                                    total_datas,
                                    local_today_canceled_buyno_map.get(
                                        code))
                                if left_count > 0:
                                    left_count += 1
                            if left_count <= 3:
                                # 当成交进度距离真实下单位置不足3笔时推送即将成交的消息
                                push_msg_manager.push_order_almost_deal(code, gpcode_manager.get_code_name(code))

    @classmethod
    def statistic_big_buy_data(cls, code, datas, limit_up_price):
        """
        统计大单买
        @param code:
        @param datas:
        @return: 返回数据里面(成交的大单,50w以上的单）
        """
        big_buy_datas = []
        normal_buy_datas = []
        # 大单阈值
        threshold_big_money = l2_data_util.get_big_money_val(limit_up_price, tool.is_ge_code(code))
        for data in datas:
            # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
            #                   data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
            #                   data['SellNo'], data['ExecType']))
            is_limit_up = abs(data[1] - limit_up_price) < 0.0001
            money = data[2] * data[1]
            if code not in cls.__dealing_order_info_dict:
                # 数据格式[订单号，总股数，成交金额,成交开始时间,成交结束时间, 最近的成交价格, 最近的卖单号, 涨停价成交金额]
                cls.__dealing_order_info_dict[code] = [data[6], data[2], money, data[3], data[3], data[1],
                                                       data[7], 0]
                if is_limit_up:
                    cls.__dealing_order_info_dict[code][7] += money
            else:
                pass
            if order_begin_pos and order_begin_pos.buy_exec_index and order_begin_pos.buy_exec_index > -1:
                # 触发L撤上重新计算
                LCancelBigNumComputer().re_compute_l_up_watch_indexes(code, order_begin_pos.buy_single_index)
                if cls.__dealing_order_info_dict[code][0] == data[6]:
                    # 成交同一个订单号
                    cls.__dealing_order_info_dict[code][1] += data[2]
                    cls.__dealing_order_info_dict[code][2] += money
                    cls.__dealing_order_info_dict[code][4] = data[3]
                    cls.__dealing_order_info_dict[code][5] = data[1]
                    cls.__dealing_order_info_dict[code][6] = data[7]
                    if is_limit_up:
                        cls.__dealing_order_info_dict[code][7] += money
                else:
                    # 保存上一条数据
                    async_log_util.info(hx_logger_l2_transaction_desc, f"{code}#{cls.__dealing_order_info_dict[code]}")
                    # 设置最近成交完成的一条数据
                    deal_info = cls.__dealing_order_info_dict[code]
                    cls.__latest_deal_order_info_dict[code] = deal_info
                    # 是否为大买单
                    if deal_info[7] >= threshold_big_money:
                        # # 如果最后一笔成交价格不是涨停价就不算
                        # if abs(deal_info[5] - limit_up_price) < 0.0001:
                        big_buy_datas.append(deal_info)
                    if deal_info[2] >= 500000:
                        normal_buy_datas.append(deal_info)

        except Exception as e:
            logging.exception(e)
            hx_logger_l2_debug.exception(e)
        finally:
            use_time = int((time.time() - __start_time) * 1000)
            if use_time > 10:
                async_log_util.info(hx_logger_l2_upload, f"{code}处理成交用时：{use_time}")
                    # 初始化本条数据
                    cls.__dealing_order_info_dict[code] = [data[6], data[2], money, data[3], data[3],
                                                           data[1], data[7], 0]
                    if is_limit_up:
                        cls.__dealing_order_info_dict[code][7] += money

            # 统计主动买(买单号大于卖单号)
            try:
                if data[6] > data[7]:
                    if code not in cls.__dealing_active_buy_order_info_dict:
                        # 数据格式[订单号，总股数，成交金额,成交开始时间,成交结束时间]
                        cls.__dealing_active_buy_order_info_dict[code] = [data[6], data[2], data[2] * data[1], data[3],
                                                                          data[3]]
                    else:
                        if cls.__dealing_active_buy_order_info_dict[code][0] == data[6]:
                            # 成交同一个订单号
                            cls.__dealing_active_buy_order_info_dict[code][1] += data[2]
                            cls.__dealing_active_buy_order_info_dict[code][2] += data[2] * data[1]
                            cls.__dealing_active_buy_order_info_dict[code][4] = data[3]
                        else:
                            # 初始化本条数据
                            cls.__dealing_active_buy_order_info_dict[code] = [data[6], data[2], data[2] * data[1],
                                                                              data[3], data[3]]
            except:
                pass

        return big_buy_datas, normal_buy_datas


# 卖单统计数据
class HuaXinSellOrderStatisticManager:
    # 最近的大卖单成交，格式:{code:[卖单信息,...]}
    __latest_sell_order_info_list_dict = {}

    # 大单卖单号的集合，格式：{code:{卖单号}}
    __big_sell_order_ids_dict = {}

    # 大卖单的卖单号->卖单信息映射
    __big_sell_order_info_dict = {}

    # 大单列表
    __big_sell_order_info_list_dict = {}

    # 最近的卖单, 格式{code:[卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）]}
    __latest_sell_order_dict = {}
    # 最近所有的卖单
    __latest_all_sell_orders_dict = {}

    # 保存最近成交的价格
    __latest_trade_price_dict = {}

    __last_trade_data_dict = {}

    # 用来暂时存放统计结果
    __dealing_order_info_dict = {}

    @classmethod
    def statistic_big_sell_data(cls, code, datas):
        """
        统计大卖单
        @param code:
        @param datas:
        @return: 返回数据里面成交的大单
        """
        big_sell_datas = []
        for data in datas:
            # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
            #                   data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
            #                   data['SellNo'], data['ExecType']))
            if code not in cls.__dealing_order_info_dict:
                # 数据格式[订单号，总股数，成交金额]
                cls.__dealing_order_info_dict[code] = [data[7], data[2], data[2] * data[1]]
            if cls.__dealing_order_info_dict[code][0] == data[7]:
                # 成交同一个订单号
                cls.__dealing_order_info_dict[code][1] += data[2]
                cls.__dealing_order_info_dict[code][2] += data[2] * data[1]
            else:
                # 保存上一条数据
                async_log_util.info(hx_logger_l2_transaction_desc, f"{code}#{cls.__dealing_order_info_dict[code]}")
                # 设置最近成交完成的一条数据
                deal_info = cls.__dealing_order_info_dict[code]
                # 是否为大买单
                if deal_info[2] >= 2990000:
                    big_sell_datas.append(deal_info)
                # 初始化本条数据
                cls.__dealing_order_info_dict[code] = [data[7], data[2], data[2] * data[1]]
        return big_sell_datas

    # 统计所有的成交量
    __deal_volume_list_dict = {}
    # 统计涨停主动买的成交量
    __deal_active_buy_volume_list_dict = {}

    @classmethod
    def statistic_total_deal_volume(cls, code, datas, limit_up_price):
        # 只统计被动买
        if code not in cls.__deal_volume_list_dict:
            cls.__deal_volume_list_dict[code] = []

        time_dict = {}
        for d in datas:
            # 只统计被动买
            if d[7] < d[6]:
                continue
            time_str = ''
            if d[3] in time_dict:
                time_str = time_dict[d[3]]
            else:
                time_dict[d[3]] = l2_huaxin_util.convert_time(d[3])
                time_str = time_dict[d[3]]
            if cls.__deal_volume_list_dict[code]:
                if cls.__deal_volume_list_dict[code][-1][0] == time_str:
                    # 如果是同一秒
                    cls.__deal_volume_list_dict[code][-1][1] += d[2]
                else:
                    # 不是同一秒
                    cls.__deal_volume_list_dict[code].append([time_str, d[2]])
            else:
                cls.__deal_volume_list_dict[code].append([time_str, d[2]])
        # 删除超过5条数据
        if len(cls.__deal_volume_list_dict[code]) > 5:
            cls.__deal_volume_list_dict[code] = cls.__deal_volume_list_dict[code][-5:]

        try:
            # 统计主动买的成交量
            if code not in cls.__deal_active_buy_volume_list_dict:
                cls.__deal_active_buy_volume_list_dict[code] = []
            for d in datas:
                # 只统计主动买
                if d[7] > d[6]:
                    continue
                # 只统计涨停买
                if d[1] != limit_up_price:
                    continue
                if d[3] in time_dict:
                    time_str = time_dict[d[3]]
                else:
                    time_dict[d[3]] = l2_huaxin_util.convert_time(d[3])
                    time_str = time_dict[d[3]]
                if cls.__deal_active_buy_volume_list_dict[code]:
                    if cls.__deal_active_buy_volume_list_dict[code][-1][0] == time_str:
                        # 如果是同一秒
                        cls.__deal_active_buy_volume_list_dict[code][-1][1] += d[2]
                    else:
                        # 不是同一秒
                        cls.__deal_active_buy_volume_list_dict[code].append([time_str, d[2]])
                else:
                    cls.__deal_active_buy_volume_list_dict[code].append([time_str, d[2]])
            # 删除超过10条数据
            if len(cls.__deal_active_buy_volume_list_dict[code]) > 10:
                cls.__deal_active_buy_volume_list_dict[code] = cls.__deal_active_buy_volume_list_dict[code][-10:]
        except:
            pass

        time_dict.clear()

    @classmethod
    def get_latest_3s_continue_deal_volumes(cls, code):
        """
        获取最近3s的成交量分布
        @param code:
        @return: [(时间,量)]
        """
        deal_list = cls.__deal_volume_list_dict.get(code)
        if not deal_list:
            return []
        fdatas = [deal_list[-1]]
        # 从倒数第二个数据计算
        for i in range(len(deal_list) - 2, -1, -1):
            if tool.trade_time_sub(fdatas[0][0], deal_list[i][0]) < 3:
                fdatas.append(deal_list[i])
        return fdatas

    @classmethod
    def get_latest_2s_continue_deal_volumes(cls, code):
        """
        获取最近2s的成交量分布
        @param code:
        @return: [(时间,量)]
        """
        deal_list = cls.__deal_volume_list_dict.get(code)
        if not deal_list:
            return []
        fdatas = [deal_list[-1]]
        # 从倒数第二个数据计算
        for i in range(len(deal_list) - 1, -1, -1):
            if tool.trade_time_sub(fdatas[0][0], deal_list[i][0]) < 2:
                fdatas.append(deal_list[i])
        return fdatas

    @classmethod
    def get_latest_6s_active_buy_deal_volumes(cls, code):
        """
        获取最近6s的主动买成交
        @param code:
        @return: [(时间,量)]
        """
        deal_list = cls.__deal_active_buy_volume_list_dict.get(code)
        if not deal_list:
            return []
        latest_time = deal_list[-1][0]
        fdatas = []
        # 从倒数第二个数据计算
        for i in range(len(deal_list) - 1, -1, -1):
            if tool.trade_time_sub(latest_time, deal_list[i][0]) < 6:
                fdatas.append(deal_list[i])
        return fdatas

    @classmethod
    def clear_latest_deal_volume(cls, code):
        if code in cls.__deal_volume_list_dict:
            cls.__deal_volume_list_dict.pop(code)

    # 返回最近1s的大单卖：(总卖金额,[(卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）),...])
    @classmethod
    def add_transaction_datas(cls, code, datas, limit_up_price=None):
        # 是否为主动卖
        def is_active_sell(sell_no, buy_no):
            return sell_no > buy_no

        # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
        #           data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
        #           data['SellNo'], data['ExecType']))
        if code not in cls.__latest_sell_order_info_list_dict:
            cls.__latest_sell_order_info_list_dict[code] = []
        if code not in cls.__big_sell_order_ids_dict:
            cls.__big_sell_order_ids_dict[code] = set()
        if code not in cls.__big_sell_order_info_dict:
            cls.__big_sell_order_info_dict[code] = {}

        if code not in cls.__big_sell_order_info_list_dict:
            cls.__big_sell_order_info_list_dict[code] = []

        if code not in cls.__latest_all_sell_orders_dict:
            cls.__latest_all_sell_orders_dict[code] = []

        sell_no_map = local_today_sellno_map.get(code)
        total_datas = local_today_datas.get(code)
        if not sell_no_map:
            sell_no_map = {}

        # 保存最近的成交价格:(价格,成交时间)
        cls.__latest_trade_price_dict[code] = (datas[-1][1], datas[-1][3])

        L2TradeSingleDataProcessor.process_passive_limit_up_sell_data(code, datas, limit_up_price)

        for d in datas:
            # 获取当前是否为主动买
            try:
                _is_active_sell = is_active_sell(d[7], d[6])
                if not _is_active_sell:
                    continue

                if d[1] == limit_up_price:
                    # 涨停主动卖
                    L2TradeSingleDataProcessor.add_active_limit_up_sell_data(d)

                # 判断是否是涨停被动变主动
                last_trade_data = cls.__last_trade_data_dict.get(code)
                if last_trade_data and not is_active_sell(last_trade_data[7], last_trade_data[6]) and last_trade_data[
                    1] == limit_up_price:
                    if d[1] == limit_up_price:
                        # 涨停被动变主动
                        L2TradeSingleDataManager.set_sell_passive_to_active_datas(code, last_trade_data, d)
                cls.__latest_sell_order_info_list_dict[code].append(d)
                if code not in cls.__latest_sell_order_dict:
                    cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])]
                else:
                    if cls.__latest_sell_order_dict[code][0] == d[7]:
                        cls.__latest_sell_order_dict[code][1] += d[2]
                        cls.__latest_sell_order_dict[code][2] = d[1]
                        cls.__latest_sell_order_dict[code][4] = (d[3], d[6])
                    else:
                        info = cls.__latest_sell_order_dict[code]

                        # 上个卖单成交完成
                        # 封存数据，计算新起点
                        # 大于50w的卖单才会保存
                        # 大于50w加入卖单
                        money = info[1] * info[2]
                        if money >= 500000:
                            cls.__big_sell_order_ids_dict[code].add(info[0])
                            cls.__big_sell_order_info_dict[code][info[0]] = info
                            cls.__big_sell_order_info_list_dict[code].append(info)
                        # 只保留10w以上的单
                        if money > 100000:
                            cls.__latest_all_sell_orders_dict[code].append(info)
                            l2_log.info(code, hx_logger_l2_transaction_sell_order,
                                        f"{info}")
                        if limit_up_price == info[2]:
                            # 将涨停主动卖记入日志
                            l2_log.info(code, hx_logger_l2_active_sell, f"{info}")

                        cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])]
            finally:
                cls.__last_trade_data_dict[code] = d

        latest_time = l2_huaxin_util.convert_time(datas[-1][3], with_ms=True)
        min_time = tool.trade_time_add_millionsecond(latest_time, -1000)
        min_time_int = int(min_time.replace(":", "").replace(".", ""))
        # 计算最近1s的大单成交
        total_big_sell_datas = cls.__big_sell_order_info_list_dict.get(code)

        total_sell_info = [0, None]  # 总资金,开始成交信息,结束成交信息
        latest_sell_order_info = cls.__latest_sell_order_dict.get(code)
        if latest_sell_order_info:
            # 不是第一次非主动卖上传
            big_sell_order_ids = cls.__big_sell_order_ids_dict[code]
            # print("大卖单", big_sell_order_ids)
            big_sell_orders = []
            temp_sell_order_ids = set()
            # 统计已经结算出的大单
            for i in range(len(total_big_sell_datas) - 1, -1, -1):
                bd = total_big_sell_datas[i]
                if min_time_int > int(
                        l2_huaxin_util.convert_time(bd[3][0], with_ms=True).replace(":", "").replace(".", "")):
                    break
                if bd[0] != latest_sell_order_info[0]:
                    # 不是最近的成交且不是大单直接过滤
                    if bd[0] not in big_sell_order_ids:
                        continue
                    else:
                        if bd[0] not in temp_sell_order_ids:
                            big_sell_orders.append(cls.__big_sell_order_info_dict[code].get(bd[0]))
                            temp_sell_order_ids.add(bd[0])
                else:
                    # 是最近的但不是大单需要过滤
                    if latest_sell_order_info[1] * latest_sell_order_info[2] < 500000:
                        continue
                    else:
                        if latest_sell_order_info[0] not in temp_sell_order_ids:
                            big_sell_orders.append(latest_sell_order_info)
                            temp_sell_order_ids.add(latest_sell_order_info[0])

                # 统计最近1s的大卖单数据
                total_sell_info[0] += int(bd[1] * bd[2])
            # 统计最近的大单
            if latest_sell_order_info[1] * latest_sell_order_info[2] >= 500000:
                if latest_sell_order_info[0] not in temp_sell_order_ids:
                    # if is_active_sell(latest_sell_order_info[0], latest_sell_order_info[3][1]):
                    big_sell_orders.append(latest_sell_order_info)
                    temp_sell_order_ids.add(latest_sell_order_info[0])
                    total_sell_info[0] += int(latest_sell_order_info[1] * latest_sell_order_info[2])
            big_sell_orders.reverse()
            total_sell_info[1] = big_sell_orders
        # ----------------统计涨停主动买-----------------
        try:
            limit_up_active_buy_datas = []
            for d in datas:
                if is_active_sell(d[7], d[6]):
                    # 被动买
                    continue
                # 是否是涨停
                if d[1] == limit_up_price:
                    # 有涨停主动买
                    limit_up_active_buy_datas.append(d)
            L2TradeSingleDataManager.set_limit_up_active_buy(code, limit_up_active_buy_datas)
        except:
            pass

        return total_sell_info

    # 获取最近成交数据
    @classmethod
    def get_latest_transaction_datas(cls, code, min_sell_order_no=None, min_deal_time=None, min_sell_money=None):
        """
        获取最近的主动卖成交信息
        @param code:
        @param min_sell_order_no:
        @param min_deal_time:
        @param min_sell_money:
        @return:
        """
        total_orders = []
        sell_orders = cls.__latest_all_sell_orders_dict.get(code)
        if sell_orders:
            for i in range(len(sell_orders) - 1, -1, -1):
                if min_deal_time and tool.trade_time_sub(min_deal_time,
                                                         l2_huaxin_util.convert_time(sell_orders[i][3][0])) > 0:
                    break

                if min_sell_order_no and min_sell_order_no > sell_orders[i][0]:
                    continue
                if min_sell_money and sell_orders[i][1] * sell_orders[i][2] < min_sell_money:
                    # 过滤小金额
                    continue

                total_orders.append(sell_orders[i])
        if code in cls.__latest_sell_order_dict:
            if min_sell_order_no:
                if cls.__latest_sell_order_dict[code][0] >= min_sell_order_no:
                    total_orders.append(cls.__latest_sell_order_dict[code])
            else:
                total_orders.append(cls.__latest_sell_order_dict[code])
        return total_orders

    # 获取最近成交价格信息, 返回格式：（价格,时间）
    @classmethod
    def get_latest_trade_price_info(cls, code):
        return cls.__latest_trade_price_dict.get(code)