Administrator
2023-10-30 fb47d36048e94b9a506d5c153e3dd19a01e37df1
l2/l2_data_manager_new.py
@@ -1,38 +1,34 @@
import io
import logging
import random
import time as t
from code_attribute import big_money_num_manager, code_volumn_manager, code_data_util, industry_codes_sort, \
    limit_up_time_manager, global_data_loader, gpcode_manager
    limit_up_time_manager, global_data_loader, gpcode_manager, code_nature_analyse
import constant
from db.redis_manager_delegate import RedisUtils
from l2.huaxin import l2_huaxin_util, huaxin_delegate_postion_manager
from l2.l2_sell_manager import L2MarketSellManager
from log_module import async_log_util, log_export
from third_data import kpl_data_manager, block_info
from utils import global_util, ths_industry_util, tool
import l2_data_util
from db import redis_manager_delegate as redis_manager
from third_data.code_plate_key_manager import CodePlateKeyBuyManager
from third_data.code_plate_key_manager import CodePlateKeyBuyManager, KPLCodeJXBlockManager
from trade import trade_manager, trade_queue_manager, l2_trade_factor, l2_trade_util, \
    trade_result_manager, first_code_score_manager, current_price_process_manager
from l2 import safe_count_manager, l2_data_manager, l2_data_log, l2_log, l2_data_source_util, code_price_manager, \
    transaction_progress
from l2.cancel_buy_strategy import SecondCancelBigNumComputer, HourCancelBigNumComputer, L2LimitUpMoneyStatisticUtil, \
    L2LimitUpSellStatisticUtil, DCancelBigNumComputer, LCancelBigNumComputer
from l2.l2_data_manager import L2DataException, TradePointManager
    trade_result_manager, current_price_process_manager, trade_data_manager, trade_huaxin, trade_record_log_util
from l2 import l2_data_manager, l2_log, l2_data_source_util, code_price_manager, \
    transaction_progress, cancel_buy_strategy, l2_data_log
from l2.cancel_buy_strategy import SecondCancelBigNumComputer, HourCancelBigNumComputer, DCancelBigNumComputer, \
    LCancelBigNumComputer, LatestCancelIndexManager, FastCancelBigNumComputer
from l2.l2_data_manager import L2DataException, OrderBeginPosInfo
from l2.l2_data_util import local_today_datas, L2DataUtil, local_today_num_operate_map, local_today_buyno_map, \
    local_latest_datas
    local_latest_datas, local_today_canceled_buyno_map
import l2.l2_data_util
from log_module.log import logger_l2_trade_buy, logger_l2_process, \
    logger_place_order_score, logger_l2_error, logger_profile
from log_module.log import logger_l2_trade_buy, logger_l2_process, logger_l2_error, logger_debug
# TODO l2数据管理
from trade.trade_data_manager import CodeActualPriceProcessor
from line_profiler import LineProfiler
import dask
from trade.trade_manager import TradeTargetCodeModeManager
from trade.trade_manager import TradeTargetCodeModeManager, AccountAvailableMoneyManager
class L2DataManager:
@@ -173,9 +169,9 @@
            # 如果是涨停买撤信号需要看数据位置是否比开始处理时间早
            if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]):
                # 获取买入信号
                buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data(code, total_datas[i],
                                                                                                 local_today_num_operate_map.get(
                                                                                                     code))
                buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i],
                                                                                                    local_today_buyno_map.get(
                                                                                                        code))
                if buy_index is not None and buy_index < begin_pos:
                    continue
@@ -206,15 +202,34 @@
class L2TradeDataProcessor:
    unreal_buy_dict = {}
    volume_rate_info = {}
    l2BigNumForMProcessor = L2BigNumForMProcessor()
    __codeActualPriceProcessor = CodeActualPriceProcessor()
    __ths_l2_trade_queue_manager = trade_queue_manager.thsl2tradequeuemanager()
    __thsBuy1VolumnManager = trade_queue_manager.THSBuy1VolumnManager()
    __buyL2SafeCountManager = safe_count_manager.BuyL2SafeCountManager()
    __l2PlaceOrderParamsManagerDict = {}
    __last_buy_single_dict = {}
    __TradeBuyQueue = transaction_progress.TradeBuyQueue()
    __latest_process_unique_keys = {}
    __latest_process_order_unique_keys = {}
    __latest_process_not_order_unique_keys_count = {}
    __trade_log_placr_order_info_dict = {}  # 下单信息保存
    # 初始化
    __TradePointManager = l2_data_manager.TradePointManager()
    __SecondCancelBigNumComputer = SecondCancelBigNumComputer()
    __HourCancelBigNumComputer = HourCancelBigNumComputer()
    __LCancelBigNumComputer = LCancelBigNumComputer()
    __TradeStateManager = trade_manager.TradeStateManager()
    __CodesTradeStateManager = trade_manager.CodesTradeStateManager()
    __PauseBuyCodesManager = gpcode_manager.PauseBuyCodesManager()
    __Buy1PriceManager = code_price_manager.Buy1PriceManager()
    __AccountAvailableMoneyManager = AccountAvailableMoneyManager()
    __TradeBuyDataManager = trade_data_manager.TradeBuyDataManager()
    __LimitUpTimeManager = limit_up_time_manager.LimitUpTimeManager()
    __BlackListCodeManager = l2_trade_util.BlackListCodeManager()
    __WhiteListCodeManager = l2_trade_util.WhiteListCodeManager()
    __WantBuyCodesManager = gpcode_manager.WantBuyCodesManager()
    __TradeTargetCodeModeManager = TradeTargetCodeModeManager()
    __TradeOrderIdManager = trade_huaxin.TradeOrderIdManager()
    __LatestCancelIndexManager = LatestCancelIndexManager()
    __L2MarketSellManager = L2MarketSellManager()
    # 获取代码评分
    @classmethod
@@ -242,7 +257,7 @@
                if not is_normal:
                    print("历史数据异常:", code)
                    # 数据不正常需要禁止交易
                    l2_trade_util.forbidden_trade(code)
                    l2_trade_util.forbidden_trade(code, msg="L2历史数据异常")
                # 纠正数据
                if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_THS:
                    # 同花顺需要纠正数据,其他渠道不需要
@@ -260,87 +275,98 @@
                    # 保存数据
                    __start_time = round(t.time() * 1000)
                    l2.l2_data_util.save_l2_data(code, datas, add_datas)
                    __start_time = l2_data_log.l2_time(code,
                                                       round(t.time() * 1000) - __start_time,
                                                       "保存数据时间({})".format(len(add_datas)))
                    # __start_time = l2_data_log.l2_time(code,
                    #                                    round(t.time() * 1000) - __start_time,
                    #                                    "保存数据时间({})".format(len(add_datas)))
        finally:
            if code in cls.unreal_buy_dict:
                cls.unreal_buy_dict.pop(code)
    @classmethod
    def set_real_place_order_index(cls, code, index, order_begin_pos:OrderBeginPosInfo):
        trade_record_log_util.add_real_place_order_position_log(code, index, order_begin_pos.buy_single_index)
        if order_begin_pos.mode == OrderBeginPosInfo.MODE_FAST:
            need_cancel = FastCancelBigNumComputer().set_real_order_index(code, index)
            if need_cancel:
                cls.cancel_buy(code, msg="F撤不够2笔触发撤单")
                return
        cancel_buy_strategy.set_real_place_position(code, index, order_begin_pos.buy_single_index)
    # 处理华鑫L2数据
    @classmethod
    def process_huaxin(cls, code, origin_datas):
        print("process_huaxin", code, len(origin_datas))
        origin_start_time = round(t.time() * 1000)
        datas = None
        try:
            l2_data_log.l2_time_log(code, "开始加载历史数据")
            # 加载历史的L2数据
            is_normal = l2.l2_data_util.load_l2_data(code, load_latest=False)
            if not is_normal:
                print("历史数据异常:", code)
                # 数据不正常需要禁止交易
                l2_trade_util.forbidden_trade(code)
                l2_trade_util.forbidden_trade(code, msg="L2历史数据异常")
            # 转换数据格式
            _start_index = 0
            total_datas = local_today_datas.get(code)
            if total_datas:
                _start_index = total_datas[-1]["index"] + 1
            l2_data_log.l2_time_log(code, "开始格式化原始数据")
            datas = l2_huaxin_util.get_format_l2_datas(code, origin_datas,
                                                       gpcode_manager.get_limit_up_price(code), _start_index)
            # 获取下单位置
            place_order_index = huaxin_delegate_postion_manager.get_l2_place_order_position(code, datas)
            if place_order_index:
                logger_l2_process.info("code:{} 获取到下单真实位置:{}", code, place_order_index)
                DCancelBigNumComputer().set_real_order_index(code, place_order_index)
            __start_time = round(t.time() * 1000)
            l2_data_log.l2_time_log(code, "开始处理数据")
            if len(datas) > 0:
                cls.process_add_datas(code, datas, 0, __start_time)
            else:
                pass
                # lp = LineProfiler()
                # lp.enable()
                # lp_wrap = lp(cls.process_add_datas)
                # lp_wrap(code, datas, 0, __start_time)
                # output = io.StringIO()
                # lp.print_stats(stream=output)
                # lp.disable()
                # with open(f"/home/logs/profile/{code}_{datas[0]['index']}_{datas[-1]['index']}.txt", 'w') as f:
                #     f.write(output.getvalue())
            # lp.dump_stats(f"/home/logs/profile/{code}_{round(t.time() * 1000)}.txt")
        except Exception as e:
            print("huaxin L2数据处理异常", code, str(e))
            logging.exception(e)
            logger_l2_error.exception(f"code:{code}")
            logger_l2_error.exception(e)
            async_log_util.error(logger_l2_error, f"code:{code}")
            async_log_util.exception(logger_l2_error, e)
        finally:
            l2_data_log.l2_time(code, round(t.time() * 1000) - origin_start_time,
                                "l2数据处理总耗时",
                                True)
            if datas:
                l2_data_log.l2_time_log(code, "开始保存数据")
                l2.l2_data_util.save_l2_data(code, None, datas)
    @classmethod
    def process_add_datas(cls, code, add_datas, capture_timestamp, __start_time):
        now_time_str = tool.get_now_time_str()
        if len(add_datas) > 0:
            if code not in cls.__trade_log_placr_order_info_dict:
                cls.__trade_log_placr_order_info_dict[code] = trade_record_log_util.PlaceOrderInfo()
            # 拼接数据
            local_today_datas[code].extend(add_datas)
            l2.l2_data_util.load_num_operate_map(local_today_num_operate_map, code, add_datas)
            l2.l2_data_util.load_buy_no_map(local_today_buyno_map, code, add_datas)
            l2.l2_data_util.load_canceled_buy_no_map(local_today_canceled_buyno_map, code, add_datas)
            l2_data_log.l2_time_log(code, "process_add_datas 加载完数据")
            if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_HUAXIN:
                try:
                    if constant.TEST:
                        pass
                        # order_begin_pos = cls.__get_order_begin_pos(code)
                        # if order_begin_pos.buy_exec_index and order_begin_pos.buy_exec_index>=0:
                        #     place_order_index = add_datas[-1]["index"]
                        #     cls.set_real_place_order_index(code, place_order_index, order_begin_pos.buy_single_index)
                    else:
                        # 获取下单位置
                        place_order_index = huaxin_delegate_postion_manager.get_l2_place_order_position(code, float(
                            gpcode_manager.get_limit_up_price(code)), add_datas)
                        if place_order_index:
                            order_begin_pos = cls.__get_order_begin_pos(
                                code)
                            cls.set_real_place_order_index(code, place_order_index, order_begin_pos)
                            async_log_util.info(logger_l2_process, "code:{} 获取到下单真实位置:{}", code, place_order_index)
                except:
                    async_log_util.error(logger_l2_error, f"{code} 处理真实下单位置出错")
            # 第1条数据是否为09:30:00
            if add_datas[0]["val"]["time"] == "09:30:00":
                if global_util.cuurent_prices.get(code):
                    price_data = global_util.cuurent_prices.get(code)
                    if price_data[1]:
                        # 当前涨停价,设置涨停时间
                        logger_l2_process.info("开盘涨停:{}", code)
                        async_log_util.info(logger_l2_process, "开盘涨停:{}", code)
                        # 保存涨停时间
                        limit_up_time_manager.LimitUpTimeManager().save_limit_up_time(code, "09:30:00")
                        cls.__LimitUpTimeManager.save_limit_up_time(code, "09:30:00")
        total_datas = local_today_datas[code]
        __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
                                           "l2数据预处理时间")
        # __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
        #                                    "l2数据预处理时间")
        if len(add_datas) > 0:
            # 是否为首板代码
@@ -349,11 +375,11 @@
            volume_rate = code_volumn_manager.get_volume_rate(code)
            volume_rate_index = code_volumn_manager.get_volume_rate_index(volume_rate)
            # 计算分值
            limit_up_time = limit_up_time_manager.LimitUpTimeManager().get_limit_up_time_cache(code)
            limit_up_time = cls.__LimitUpTimeManager.get_limit_up_time_cache(code)
            if limit_up_time is None:
                limit_up_time = tool.get_now_time_str()
            score = first_code_score_manager.get_score(code, volume_rate, limit_up_time, True)
            # score = first_code_score_manager.get_score(code, volume_rate, limit_up_time, True)
            score = None
            cls.__l2PlaceOrderParamsManagerDict[code] = l2_trade_factor.L2PlaceOrderParamsManager(code, is_first_code,
                                                                                                  volume_rate,
                                                                                                  volume_rate_index,
@@ -364,41 +390,27 @@
            latest_time = add_datas[len(add_datas) - 1]["val"]["time"]
            __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
                                               "l2数据准备时间")
            # __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
            #                                    "l2数据准备时间")
            # 时间差不能太大才能处理
            if not l2_trade_util.is_in_forbidden_trade_codes(code):
                # 判断是否已经挂单
                state = trade_manager.CodesTradeStateManager().get_trade_state_cache(code)
                state = cls.__CodesTradeStateManager.get_trade_state_cache(code)
                start_index = len(total_datas) - len(add_datas)
                end_index = len(total_datas) - 1
                l2_data_log.l2_time_log(code, "process_add_datas 开始处理")
                if state == trade_manager.TRADE_STATE_BUY_DELEGATED or state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or state == trade_manager.TRADE_STATE_BUY_SUCCESS:
                    # 已挂单
                    if True:  # len(add_datas) < 10:
                        cls.__process_order(code, start_index, end_index, capture_timestamp, is_first_code)
                    else:
                        pass
                        # lp = LineProfiler()
                        # lp.enable()
                        # lp_wrap = lp(cls.__process_order)
                        # lp_wrap(code, start_index, end_index, capture_timestamp, is_first_code)
                        # output = io.StringIO()
                        # lp.print_stats(stream=output)
                        # lp.disable()
                        # with open(
                        #         f"/home/logs/profile/{code}_process_order_{add_datas[0]['index']}_{add_datas[-1]['index']}.txt",
                        #         'w') as f:
                        #     f.write(output.getvalue())
                    cls.__process_order(code, start_index, end_index, capture_timestamp, is_first_code)
                else:
                    # 未挂单,时间相差不大才能挂单
                    if l2.l2_data_util.L2DataUtil.is_same_time(now_time_str, latest_time):
                        cls.__process_not_order(code, start_index, end_index, capture_timestamp, is_first_code)
            logger_l2_process.info("code:{} 处理数据范围: {}-{} 处理时间:{} 截图时间戳:{}", code, add_datas[0]["index"],
                                   add_datas[-1]["index"], round(t.time() * 1000) - __start_time,
                                   capture_timestamp)
            __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
                                               "l2数据处理时间")
            async_log_util.info(logger_l2_process, "code:{} 处理数据范围: {}-{} 处理时间:{} 截图时间戳:{}", code,
                                add_datas[0]["index"],
                                add_datas[-1]["index"], round(t.time() * 1000) - __start_time,
                                capture_timestamp)
    # 处理未挂单
    @classmethod
@@ -406,21 +418,7 @@
        __start_time = round(t.time() * 1000)
        # 获取阈值
        threshold_money, msg = cls.__get_threshmoney(code)
        if round(t.time() * 1000) - __start_time > 10:
            __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
                                               "获取m值数据耗时")
        if True:  # end_index - start_index < 10:
            cls.__start_compute_buy(code, start_index, end_index, threshold_money, capture_time, is_first_code)
        else:
            # pass
            lp = LineProfiler()
            lp.enable()
            lp_wrap = lp(cls.__start_compute_buy)
            lp_wrap(code, start_index, end_index, threshold_money, capture_time, is_first_code)
            output = io.StringIO()
            lp.print_stats(stream=output)
            lp.disable()
            logger_profile.info(output.getvalue())
        cls.__start_compute_buy(code, start_index, end_index, threshold_money, capture_time, is_first_code)
    # 测试专用
    @classmethod
@@ -432,138 +430,95 @@
    def __process_order(cls, code, start_index, end_index, capture_time, is_first_code, new_add=True):
        # 增加推出机制
        unique_key = f"{start_index}-{end_index}"
        if cls.__latest_process_unique_keys.get(code) == unique_key:
            logger_l2_error.error(f"重复处理数据:code-{code} start_index-{start_index} end_index-{end_index}")
        if cls.__latest_process_order_unique_keys.get(code) == unique_key:
            async_log_util.error(logger_l2_error, f"重复处理数据:code-{code} start_index-{start_index} end_index-{end_index}")
            return
        cls.__latest_process_unique_keys[code] = unique_key
        # 计算安全笔数
        @dask.delayed
        def compute_safe_count():
            _start_time = round(t.time() * 1000)
            # 处理安全笔数
            cls.__buyL2SafeCountManager.compute_left_rate(code, start_index, end_index, total_data,
                                                          local_today_num_operate_map.get(code))
            l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time,
                                "已下单-获取买入信息耗时")
            return None, ""
        @dask.delayed
        # m值大单计算
        def compute_m_big_num():
            _start_time = round(t.time() * 1000)
            # 计算m值大单
            cls.l2BigNumForMProcessor.process(code, max(buy_single_index, start_index), end_index,
                                              gpcode_manager.get_limit_up_price(code))
            l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time,
                                "已下单-m值大单计算")
            return None, ""
        # 买1撤计算
        @dask.delayed
        def buy_1_cancel():
            _start_time = round(t.time() * 1000)
            # 撤单计算,只看买1
            cancel_data, cancel_msg = L2LimitUpMoneyStatisticUtil().process_data(code, start_index,
                                                                                 end_index,
                                                                                 buy_single_index, buy_exec_index)
            l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time,
                                "已下单-买1统计耗时")
            # 买1不会触发撤单
            return None, ""
            # return cancel_data, cancel_msg
        cls.__latest_process_order_unique_keys[code] = unique_key
        # S撤
        @dask.delayed
        def s_cancel():
        def s_cancel(_buy_single_index, _buy_exec_index):
            _start_time = round(t.time() * 1000)
            # S撤单计算,看秒级大单撤单
            try:
                b_need_cancel, b_cancel_data = SecondCancelBigNumComputer().need_cancel(code, buy_single_index,
                                                                                        buy_exec_index, start_index,
                                                                                        end_index, total_data,
                                                                                        code_volumn_manager.get_volume_rate_index(
                                                                                            buy_volume_rate),
                                                                                        cls.volume_rate_info[code][1],
                                                                                        is_first_code)
                b_need_cancel, b_cancel_data = cls.__SecondCancelBigNumComputer.need_cancel(code, _buy_single_index,
                                                                                            _buy_exec_index,
                                                                                            start_index,
                                                                                            end_index, total_data,
                                                                                            code_volumn_manager.get_volume_rate_index(
                                                                                                buy_volume_rate),
                                                                                            cls.volume_rate_info[code][
                                                                                                1],
                                                                                            is_first_code)
                if b_need_cancel:
                    return b_cancel_data, "S大单撤销比例触发阈值"
            except Exception as e:
                logging.exception(e)
                async_log_util.error(logger_l2_error,
                                     f"S撤出错 参数:buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index}")
                async_log_util.exception(logger_l2_error, e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time,
                                    "已下单-s级大单估算")
                # l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time,
                #                     "已下单-s级大单估算")
                pass
            return None, ""
        # H撤
        @dask.delayed
        def h_cancel():
        def h_cancel(_buy_single_index, _buy_exec_index):
            _start_time = round(t.time() * 1000)
            try:
                b_need_cancel, b_cancel_data = HourCancelBigNumComputer().need_cancel(code, buy_single_index,
                                                                                      buy_exec_index, start_index,
                                                                                      end_index, total_data,
                                                                                      local_today_num_operate_map.get(
                                                                                          code),
                                                                                      code_volumn_manager.get_volume_rate_index(
                                                                                          buy_volume_rate),
                                                                                      cls.volume_rate_info[code][1],
                                                                                      is_first_code)
                b_need_cancel, b_cancel_data = cls.__HourCancelBigNumComputer.need_cancel(code, _buy_single_index,
                                                                                          _buy_exec_index, start_index,
                                                                                          end_index, total_data,
                                                                                          code_volumn_manager.get_volume_rate_index(
                                                                                              order_begin_pos.buy_volume_rate),
                                                                                          cls.volume_rate_info[code][1],
                                                                                          is_first_code)
                if b_need_cancel and b_cancel_data:
                    return b_cancel_data, "H撤销比例触发阈值"
            except Exception as e:
                logging.exception(e)
                if constant.TEST:
                    logging.exception(e)
                async_log_util.error(logger_l2_error,
                                     f"H撤出错 参数:buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} {str(e)}")
                async_log_util.exception(logger_l2_error, e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-H撤大单计算")
                # l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-H撤大单计算")
                pass
            return None, ""
        # F撤
        def f_cancel(_buy_single_index, _buy_exec_index):
            try:
                b_need_cancel, b_cancel_data = FastCancelBigNumComputer().need_cancel(code, start_index, end_index,
                                                                                      order_begin_pos)
                if b_need_cancel and b_cancel_data:
                    return b_cancel_data, f"F撤"
            except Exception as e:
                if constant.TEST:
                    logging.exception(e)
                async_log_util.error(logger_l2_error,
                                     f"F撤出错 参数:buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因:{str(e)}")
                async_log_util.exception(logger_l2_error, e)
            return None, ""
        # L撤
        @dask.delayed
        def l_cancel():
        def l_cancel(_buy_single_index, _buy_exec_index):
            _start_time = round(t.time() * 1000)
            try:
                b_need_cancel, b_cancel_data = LCancelBigNumComputer().need_cancel(code,
                                                                                   buy_exec_index, start_index,
                                                                                   end_index, total_data,
                                                                                   local_today_num_operate_map.get(
                                                                                       code), is_first_code)
                b_need_cancel, b_cancel_data, extra_msg = cls.__LCancelBigNumComputer.need_cancel(code,
                                                                                                  _buy_exec_index,
                                                                                                  start_index,
                                                                                                  end_index, total_data,
                                                                                                  is_first_code)
                if b_need_cancel and b_cancel_data:
                    return b_cancel_data, "L撤销比例触发阈值"
                    return b_cancel_data, f"L撤销比例触发阈值({extra_msg})"
            except Exception as e:
                logging.exception(e)
                async_log_util.error(logger_l2_error,
                                     f"L撤出错 参数:buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因:{str(e)}")
                logger_l2_error.exception(e)
                async_log_util.exception(logger_l2_error, e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-L撤大单计算")
            return None, ""
        # 板上卖撤
        @dask.delayed
        def sell_cancel():
            _start_time = round(t.time() * 1000)
            # 统计板上卖
            try:
                cancel_data, cancel_msg = L2LimitUpSellStatisticUtil().process(code, start_index,
                                                                               end_index,
                                                                               buy_exec_index)
                return cancel_data, cancel_msg
            except Exception as e:
                logging.exception(e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-板上卖耗时")
            return None, ""
        # 是否需要撤销
        @dask.delayed
        def is_need_cancel(*args):
            try:
                for i in range(0, len(args)):
                    _cancel_data, _cancel_msg = args[i]
                    if _cancel_data:
                        return _cancel_data, _cancel_msg
            except Exception as e:
                logging.exception(e)
            finally:
                # l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-L撤大单计算")
                pass
            return None, ""
@@ -575,60 +530,32 @@
        total_data = local_today_datas.get(code)
        _start_time = tool.get_now_timestamp()
        # 获取买入信号起始点
        buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
        order_begin_pos = cls.__get_order_begin_pos(
            code)
        # 默认量为0.2
        if order_begin_pos.buy_volume_rate is None:
            buy_volume_rate = 0.2
        cancel_data, cancel_msg = None, ""
        if order_begin_pos.mode == OrderBeginPosInfo.MODE_FAST:
            cancel_data, cancel_msg = f_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
        f1 = compute_safe_count()
        f2 = compute_m_big_num()
        f3 = s_cancel()
        f4 = h_cancel()
        f5 = buy_1_cancel()
        f6 = sell_cancel()
        f7 = l_cancel()
        dask_result = is_need_cancel(f1, f2, f3, f4, f5, f6, f7)
        if is_first_code:
            dask_result = is_need_cancel(f3, f4, f7)
        cancel_data, cancel_msg = dask_result.compute()
        _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                          "已下单-撤单 判断是否需要撤单")
        # 依次处理
        if not cancel_data:
            cancel_data, cancel_msg = l_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
        # 暂时取消S撤
        # if not cancel_data:
        #     cancel_data, cancel_msg = s_cancel(buy_single_index, buy_exec_index)
        if not cancel_data:
            cancel_data, cancel_msg = h_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
        if cancel_data:
            l2_log.debug(code, "触发撤单,撤单位置:{} ,撤单原因:{}", cancel_data["index"], cancel_msg)
            l2_log.trade_record(code, "撤单", "'index':{} , 'msg':'{}'", cancel_data["index"], cancel_msg)
            # 撤单
            if cls.cancel_buy(code, cancel_msg):
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  "已下单-撤单 耗时")
                # 撤单成功,继续计算下单
                cls.__process_not_order(code, cancel_data["index"] + 1, end_index, capture_time, is_first_code)
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  "处理剩余数据 耗时")
            else:
                # 撤单尚未成功
                pass
        else:
            # 如果有虚拟下单需要真实下单
            unreal_buy_info = cls.unreal_buy_dict.get(code)
            if unreal_buy_info is not None:
                l2_log.debug(code, "有虚拟下单,无买撤信号,开始执行买入,执行位置:{},截图时间:{}", unreal_buy_info[0], capture_time)
                # unreal_buy_info 的内容格式为:(触法买操作下标,截图时间)
                # 真实下单
                cls.__buy(code, unreal_buy_info[1], local_today_datas[code][unreal_buy_info[0]],
                          unreal_buy_info[0], is_first_code)
                # lp = LineProfiler()
                # lp.enable()
                # lp_wrap = lp(cls.__buy)
                # lp_wrap(code, unreal_buy_info[1], local_today_datas[code][unreal_buy_info[0]],
                #         unreal_buy_info[0], is_first_code)
                # output = io.StringIO()
                # lp.print_stats(stream=output)
                # lp.disable()
                # logger_profile.info(output.getvalue())
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  "已虚拟下单-执行真实下单 外部耗时")
            # 撤单
            cls.cancel_buy(code, cancel_msg, cancel_index=cancel_data["index"])
            # 撤单成功,继续计算下单
            cls.__process_not_order(code, cancel_data["index"] + 1, end_index, capture_time, is_first_code)
        else:
            pass
    @classmethod
    def __buy(cls, code, capture_timestamp, last_data, last_data_index, is_first_code):
@@ -639,48 +566,70 @@
        else:
            can, need_clear_data, reason = cls.__can_buy_first(code)
        __start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - __start_time, "最后判断是否能下单", force=True)
        # __start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - __start_time, "最后判断是否能下单", force=True)
        # 删除虚拟下单
        if code in cls.unreal_buy_dict:
            cls.unreal_buy_dict.pop(code)
        buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
        order_begin_pos = cls.__get_order_begin_pos(
            code)
        if not can:
            l2_log.debug(code, "不可以下单,原因:{}", reason)
            trade_record_log_util.add_cant_place_order_log(code, reason)
            if need_clear_data:
                trade_result_manager.real_cancel_success(code, buy_single_index, buy_exec_index,
                trade_result_manager.real_cancel_success(code, order_begin_pos.buy_single_index,
                                                         order_begin_pos.buy_exec_index,
                                                         local_today_datas.get(code))
            return
            return False
        else:
            l2_log.debug(code, "可以下单,原因:{}", reason)
            try:
                l2_log.debug(code, "开始执行买入")
                trade_manager.start_buy(code, capture_timestamp, last_data,
                                        last_data_index)
                ################下单成功处理################
                trade_result_manager.real_buy_success(code)
                                        last_data_index, order_begin_pos.mode)
                l2_log.debug(code, "执行买入成功")
                ################下单成功处理################
                trade_result_manager.real_buy_success(code, cls.__TradePointManager)
                cancel_buy_strategy.set_real_place_position(code, local_today_datas.get(code)[-1]["index"],
                                                            order_begin_pos.buy_single_index)
                l2_log.debug(code, "处理买入成功")
                params_desc = cls.__l2PlaceOrderParamsManagerDict[code].get_buy_rank_desc()
                l2_log.debug(code, params_desc)
                l2_log.trade_record(code, "下单",
                                    "'buy_start_index':{} ,'buy_exec_index':{},'volume_reate':{},'score':{},'desc':'{}'",
                                    buy_single_index, buy_exec_index, cls.volume_rate_info[code][0],
                                    cls.__l2PlaceOrderParamsManagerDict[code].score, params_desc)
                ############记录下单时的数据############
                try:
                    jx_blocks, jx_blocks_by = KPLCodeJXBlockManager().get_jx_blocks_cache(
                        code), KPLCodeJXBlockManager().get_jx_blocks_cache(code, by=True)
                    info = cls.__trade_log_placr_order_info_dict[code]
                    info.mode = order_begin_pos.mode
                    info.set_buy_index(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
                    if jx_blocks:
                        info.set_kpl_blocks(list(jx_blocks))
                    elif jx_blocks_by:
                        info.set_kpl_blocks(list(jx_blocks_by))
                    else:
                        info.set_kpl_blocks([])
                    trade_record_log_util.add_place_order_log(code, info)
                except Exception as e:
                    async_log_util.error(logger_l2_error, f"加入买入记录日志出错:{str(e)}")
            except Exception as e:
                logger_l2_error.exception(e)
                async_log_util.exception(logger_l2_error, e)
                l2_log.debug(code, "执行买入异常:{}", str(e))
                pass
            finally:
                l2_log.debug(code, "m值影响因子:{}", l2_trade_factor.L2TradeFactorUtil.factors_to_string(code))
                # l2_log.debug(code, "m值影响因子:{}", l2_trade_factor.L2TradeFactorUtil.factors_to_string(code))
                pass
            return True
    # 是否可以取消
    @classmethod
    def __can_cancel(cls, code):
        if constant.TEST:
            return True, ""
        if l2_trade_util.WhiteListCodeManager().is_in_cache(code):
        if cls.__WhiteListCodeManager.is_in_cache(code):
            return False, "代码在白名单中"
        # 暂时注释掉
@@ -714,7 +663,7 @@
    @classmethod
    def __can_buy(cls, code):
        __start_time = t.time()
        if not trade_manager.TradeStateManager().is_can_buy_cache():
        if not cls.__TradeStateManager.is_can_buy_cache():
            return False, True, f"今日已禁止交易"
        # 之前的代码
        # 首板代码且尚未涨停过的不能下单
@@ -748,10 +697,9 @@
                if sell1_time is not None and sell1_volumn > 0:
                    # 获取执行位信息
                    buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
                        code)
                    buy_nums = num
                    for i in range(buy_exec_index + 1, total_datas[-1]["index"] + 1):
                    order_begin_pos = cls.__get_order_begin_pos(code)
                    buy_nums = order_begin_pos.num
                    for i in range(order_begin_pos.buy_exec_index + 1, total_datas[-1]["index"] + 1):
                        _val = total_datas[i]["val"]
                        # 涨停买
                        if L2DataUtil.is_limit_up_price_buy(_val):
@@ -769,7 +717,7 @@
            if volumn_rate >= 1.3:
                return False, False, "最大量比超过1.3不能买"
            limit_up_time = limit_up_time_manager.LimitUpTimeManager().get_limit_up_time_cache(code)
            limit_up_time = cls.__LimitUpTimeManager.get_limit_up_time_cache(code)
            if limit_up_time is not None:
                limit_up_time_seconds = l2.l2_data_util.L2DataUtil.get_time_as_second(
                    limit_up_time)
@@ -837,47 +785,74 @@
            # 可以下单
            return True, False, None
        finally:
            l2_data_log.l2_time(code, round((t.time() - __start_time) * 1000), "是否可以下单计算")
            # l2_data_log.l2_time(code, round((t.time() - __start_time) * 1000), "是否可以下单计算")
            pass
    @classmethod
    def __can_buy_first(cls, code):
        if not trade_manager.TradeStateManager().is_can_buy_cache():
        if not cls.__TradeStateManager.is_can_buy_cache():
            return False, True, f"今日已禁止交易"
        if gpcode_manager.PauseBuyCodesManager().is_in_cache(code):
        if cls.__PauseBuyCodesManager.is_in_cache(code):
            return False, True, f"该代码被暂停交易"
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        if float(limit_up_price) >= 40:
            return False, True, "股价大于40块"
        if float(limit_up_price) >= constant.MAX_CODE_PRICE:
            return False, True, f"股价大于{constant.MAX_CODE_PRICE}块"
        if code_nature_analyse.LatestMaxVolumeManager().is_latest_max_volume(code):
            # 最近几天有最大量,判断量比是否大于60%
            if cls.volume_rate_info[code][0] < 0.6:
                return False, True, f"近日出现最大量,当日量比({cls.volume_rate_info[code][0]})小于0.6"
        if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_HUAXIN:
            total_data = local_today_datas.get(code)
            trade_price = current_price_process_manager.get_trade_price(code)
            if trade_price is None:
                return False, True, f"尚未获取到当前成交价"
            if float(limit_up_price) - float(trade_price) > 0.04001:
                return False, False, f"当前成交价({trade_price})尚未在4档及以内"
            if float(limit_up_price) - float(trade_price) > 0.00001:
                # 计算信号起始位置到当前的手数
                order_begin_pos = cls.__get_order_begin_pos(
                    code)
                num_operate_map = local_today_num_operate_map.get(code)
                total_num = 0
                for i in range(order_begin_pos.buy_single_index, total_data[-1]["index"] + 1):
                    data = total_data[i]
                    val = data["val"]
                    if not L2DataUtil.is_limit_up_price_buy(val):
                        continue
                    left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count_v2(code,
                                                                                                             data[
                                                                                                                 "index"],
                                                                                                             total_data,
                                                                                                             local_today_canceled_buyno_map.get(
                                                                                                                 code))
                    total_num += left_count * val["num"]
                m_base_val = l2_trade_factor.L2PlaceOrderParamsManager.get_base_m_val(code)
                thresh_hold_num = m_base_val // (float(gpcode_manager.get_limit_up_price(code)) * 100)
                if total_num < thresh_hold_num * 2:
                    return False, False, f"当前成交价({trade_price})尚未在0档及以内 且 纯买额({total_num})小于2倍M值({thresh_hold_num * 2})"
            # 判断成交进度是否距离我们的位置很近
            total_data = local_today_datas.get(code)
            trade_index, is_default = cls.__TradeBuyQueue.get_traded_index(code)
            if not is_default and trade_index:
                buy_index_set = set()
            if False and not is_default and trade_index:
                not_cancel_num = 0
                num_operate_map = local_today_num_operate_map.get(code)
                for i in range(trade_index + 1, total_data[-1]["index"] + 1):
                    if L2DataUtil.is_limit_up_price_buy(total_data[i]["val"]):
                        left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count(code,
                                                                                                              total_data[
                                                                                                                  i][
                                                                                                                  "index"],
                                                                                                              total_data,
                                                                                                              num_operate_map)
                        left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count_v2(code,
                                                                                                                 total_data[
                                                                                                                     i][
                                                                                                                     "index"],
                                                                                                                 total_data,
                                                                                                                 local_today_canceled_buyno_map.get(
                                                                                                                     code))
                        if left_count > 0:
                            buy_index_set.add(total_data[i]["index"])
                if len(buy_index_set) < 5:
                    return False, False, f"成交位置距离当前位置小于5笔"
                            not_cancel_num += total_data[i]["val"]["num"]
                m_base_val = l2_trade_factor.L2PlaceOrderParamsManager.get_base_m_val(code)
                not_cancel_money = not_cancel_num * 100 * float(gpcode_manager.get_limit_up_price(code))
                if m_base_val > not_cancel_money:
                    return False, False, f"成交位置距离当前位置纯买额({not_cancel_money})小于m值({m_base_val})"
        else:
            # 判断买1价格档位
            zyltgb = global_util.zyltgb_map.get(code)
@@ -886,7 +861,7 @@
                zyltgb = global_util.zyltgb_map.get(code)
            if zyltgb >= 200 * 100000000:
                buy1_price = code_price_manager.Buy1PriceManager().get_buy1_price(code)
                buy1_price = cls.__Buy1PriceManager.get_buy1_price(code)
                if buy1_price is None:
                    return False, True, f"尚未获取到买1价"
                dif = float(limit_up_price) - float(buy1_price)
@@ -894,46 +869,46 @@
                if dif > 0.10001:
                    return False, True, f"自由流通200亿以上,买1剩余档数大于10档,买一({buy1_price})涨停({limit_up_price})"
        open_limit_up_lowest_price = code_price_manager.Buy1PriceManager().get_open_limit_up_lowest_price(code)
        open_limit_up_lowest_price = cls.__Buy1PriceManager.get_open_limit_up_lowest_price(code)
        price_pre_close = gpcode_manager.CodePrePriceManager.get_price_pre_cache(code)
        if open_limit_up_lowest_price and (
                float(open_limit_up_lowest_price) - price_pre_close) / price_pre_close < 0.05:
            return False, True, f"炸板后最低价跌至5%以下"
        limit_up_info = code_price_manager.Buy1PriceManager().get_limit_up_info(code)
        if limit_up_info[0] is None and False:
            total_data = local_today_datas.get(code)
            buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
                code)
            # 之前没有涨停过
            # 统计买入信号位到当前位置没有撤的大单金额
            min_money_w = l2_data_util.get_big_money_val(float(total_data[buy_single_index]["val"]["price"])) // 10000
            left_big_num = l2.cancel_buy_strategy.SecondCancelBigNumComputer().compute_left_big_num(code,
                                                                                                    buy_single_index,
                                                                                                    buy_exec_index,
                                                                                                    total_data[-1][
                                                                                                        "index"],
                                                                                                    total_data,
                                                                                                    0, min_money_w)
            if left_big_num > 0:
                # 重新获取分数与分数索引
                limit_up_time = limit_up_time_manager.LimitUpTimeManager().get_limit_up_time_cache(code)
                if limit_up_time is None:
                    limit_up_time = tool.get_now_time_str()
                score = first_code_score_manager.get_score(code, cls.volume_rate_info[code][0], limit_up_time, True,
                                                           left_big_num)
                cls.__l2PlaceOrderParamsManagerDict[code].set_score(score)
        # limit_up_info = cls.__Buy1PriceManager.get_limit_up_info(code)
        # if limit_up_info[0] is None and False:
        #     total_data = local_today_datas.get(code)
        #     buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
        #         code)
        #     # 之前没有涨停过
        #     # 统计买入信号位到当前位置没有撤的大单金额
        #     min_money_w = l2_data_util.get_big_money_val(float(total_data[buy_single_index]["val"]["price"])) // 10000
        #     left_big_num = cls.__SecondCancelBigNumComputer.compute_left_big_num(code,
        #                                                                          buy_single_index,
        #                                                                          buy_exec_index,
        #                                                                          total_data[-1][
        #                                                                              "index"],
        #                                                                          total_data,
        #                                                                          0, min_money_w)
        #     if left_big_num > 0:
        #         # 重新获取分数与分数索引
        #         limit_up_time = cls.__LimitUpTimeManager.get_limit_up_time_cache(code)
        #         if limit_up_time is None:
        #             limit_up_time = tool.get_now_time_str()
        #         score = first_code_score_manager.get_score(code, cls.volume_rate_info[code][0], limit_up_time, True,
        #                                                    left_big_num)
        #         cls.__l2PlaceOrderParamsManagerDict[code].set_score(score)
        logger_place_order_score.info("code={},data='score_index':{},'score_info':{}", code,
                                      cls.__l2PlaceOrderParamsManagerDict[code].score_index,
                                      cls.__l2PlaceOrderParamsManagerDict[code].score_info)
        # logger_place_order_score.info("code={},data='score_index':{},'score_info':{}", code,
        #                               cls.__l2PlaceOrderParamsManagerDict[code].score_index,
        #                               cls.__l2PlaceOrderParamsManagerDict[code].score_info)
        if not gpcode_manager.WantBuyCodesManager().is_in_cache(code):
            if TradeTargetCodeModeManager().get_mode_cache() == TradeTargetCodeModeManager.MODE_ONLY_BUY_WANT_CODES:
        if not cls.__WantBuyCodesManager.is_in_cache(code):
            if cls.__TradeTargetCodeModeManager.get_mode_cache() == TradeTargetCodeModeManager.MODE_ONLY_BUY_WANT_CODES:
                return False, True, f"只买想买单中的代码"
            score_index = cls.__l2PlaceOrderParamsManagerDict[code].score_index
            score = cls.__l2PlaceOrderParamsManagerDict[code].score
            score_info = cls.__l2PlaceOrderParamsManagerDict[code].score_info
            score_index = None  # cls.__l2PlaceOrderParamsManagerDict[code].score_index
            score = None  # cls.__l2PlaceOrderParamsManagerDict[code].score
            score_info = None  # cls.__l2PlaceOrderParamsManagerDict[code].score_info
            # lp = LineProfiler()
            # lp.enable()
@@ -945,56 +920,32 @@
            # with open(f"{constant.get_path_prefix()}/logs/profile/{code}_can_buy_first.txt", 'w') as f:
            #     f.write(output.getvalue())
            # return results
            return cls.can_buy_first(code, limit_up_price, score_index, score, score_info, cls.volume_rate_info[code])
            return cls.can_buy_first(code, limit_up_price)
        else:
            return True, False, "在想买名单中"
    @classmethod
    def can_buy_first(cls, code, limit_up_price, score_index, score, score_info, volume_rate_info):
        def is_has_k_format(score_info):
            # (15个交易日涨幅是否大于24.9%,是否破前高,是否超跌,是否接近前高,是否N,是否V,是否有形态,天量大阳信息,是否具有辨识度)
            if score_info[1][3][6][0] and not score_info[1][3][3][0]:
                return True
            if score_info[1][3][7][0]:
                return True
            return False
        if float(limit_up_price) >= 40:
            return False, True, "股价大于40块"
        # 9:35之前买大市值(>=80亿)票
        if int(tool.get_now_date_str("%Y%m%d")) < int("093500"):
            zyltgb = global_util.zyltgb_map.get(code)
            if zyltgb is None:
                global_data_loader.load_zyltgb()
                zyltgb = global_util.zyltgb_map.get(code)
            if zyltgb >= 80 * 100000000:
                return True, False, "{9:30:00-9:35:00}自由市值≥80亿"
    def can_buy_first(cls, code, limit_up_price):
        # 判断板块
        yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
        plate_can_buy, msg = CodePlateKeyBuyManager.can_buy(code,
                                                            kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
                                                            kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,
                                                            yesterday_codes,
                                                            block_info.get_before_blocks_dict())
        if not plate_can_buy:
            return False, True, msg
        return True, False, msg
        can_buy_result = CodePlateKeyBuyManager.can_buy(code)
        if can_buy_result is None:
            async_log_util.warning(logger_debug, "没有获取到板块缓存,将获取板块")
            yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
            CodePlateKeyBuyManager.update_can_buy_blocks(code,
                                                         kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
                                                         kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,
                                                         yesterday_codes,
                                                         block_info.get_before_blocks_dict())
            can_buy_result = CodePlateKeyBuyManager.can_buy(code)
        # if volume_rate_info[0] < 0.4:
        #     return False, True, f"量大于40%才下单,量比:{volume_rate_info[0]}"
        if can_buy_result is None:
            return False, True, "尚未获取到板块信息"
        # 是否有K线形态(有K线形态或者天量大阳),10点后才需要判断是否有K线形态与分值
        if int(tool.get_now_time_str().replace(":", "")) > int("100000"):
            has_k_format = score_info[1][3][6][0] or score_info[1][3][7][0]
            if not has_k_format:
                return False, True, f"无K线形态"
            if score_index < 0:
                return False, True, f"分值:{score}未达到需要买入的分数线"
        return True, False, ""
        # 板块不可买,且没有挂过单的就不能买
        if not can_buy_result[0] and trade_manager.CodesTradeStateManager().get_trade_state_cache(
                code) == trade_manager.TRADE_STATE_NOT_TRADE:
            return False, True, can_buy_result[1]
        return True, False, can_buy_result[1]
    @classmethod
    def __cancel_buy(cls, code):
@@ -1007,24 +958,27 @@
            logging.exception(e)
            l2_log.debug(code, "执行撤单异常:{}", str(e))
            return False
        finally:
            pass
    @classmethod
    def cancel_buy(cls, code, msg=None, source="l2"):
    def cancel_buy(cls, code, msg=None, source="l2", cancel_index=None):
        # 是否是交易队列触发
        buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
        order_begin_pos = cls.__get_order_begin_pos(
            code)
        total_datas = local_today_datas[code]
        if source == "trade_queue":
            # 交易队列触发的需要下单后5s
            if buy_exec_index is not None and buy_exec_index > 0:
            if order_begin_pos.buy_exec_index is not None and order_begin_pos.buy_exec_index > 0:
                now_time_str = tool.get_now_time_str()
                if tool.trade_time_sub(now_time_str, total_datas[buy_exec_index]["val"]["time"]) < 5:
                if tool.trade_time_sub(now_time_str, total_datas[order_begin_pos.buy_exec_index]["val"]["time"]) < 5:
                    return False
        if code in cls.unreal_buy_dict:
            cls.unreal_buy_dict.pop(code)
            # 取消买入标识
            trade_result_manager.virtual_cancel_success(code, buy_single_index, buy_exec_index, total_datas)
            trade_result_manager.virtual_cancel_success(code, order_begin_pos.buy_single_index,
                                                        order_begin_pos.buy_exec_index, total_datas)
        else:
            can_cancel, reason = cls.__can_cancel(code)
            if not can_cancel:
@@ -1032,9 +986,15 @@
                l2_log.cancel_debug(code, "撤单中断,原因:{}", reason)
                l2_log.debug(code, "撤单中断,原因:{}", reason)
                return False
            if cancel_index is None:
                cancel_index = total_datas[-1]["index"]
            cls.__LatestCancelIndexManager.set_latest_cancel_index(code, cancel_index)
            # 添加撤单日志记录
            trade_record_log_util.add_cancel_msg_log(code, msg)
            cancel_result = cls.__cancel_buy(code)
            if cancel_result:
                trade_result_manager.real_cancel_success(code, buy_single_index, buy_exec_index, total_datas)
                trade_result_manager.real_cancel_success(code, order_begin_pos.buy_single_index,
                                                         order_begin_pos.buy_exec_index, total_datas)
        l2_log.debug(code, "执行撤单结束,原因:{}", msg)
        return True
@@ -1050,72 +1010,102 @@
                            new_add=True):
        if compute_end_index < compute_start_index:
            return
        unique_key = f"{code}-{compute_start_index}-{compute_end_index}"
        if cls.__latest_process_not_order_unique_keys_count.get(
                unique_key) and cls.__latest_process_not_order_unique_keys_count.get(unique_key) > 2:
            async_log_util.error(logger_l2_error,
                                 f"重复处理数据:code-{code} start_index-{compute_start_index} end_index-{compute_end_index}")
            return
        if unique_key not in cls.__latest_process_not_order_unique_keys_count:
            cls.__latest_process_not_order_unique_keys_count[unique_key] = 0
        cls.__latest_process_not_order_unique_keys_count[unique_key] += 1
        _start_time = tool.get_now_timestamp()
        total_datas = local_today_datas[code]
        # 处理安全笔数
        cls.__buyL2SafeCountManager.compute_left_rate(code, compute_start_index, compute_end_index, total_datas,
                                                      local_today_num_operate_map.get(code))
        # 获取买入信号计算起始位置
        buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
        order_begin_pos = cls.__get_order_begin_pos(
            code)
        # 是否为新获取到的位置
        new_get_single = False
        buy_single_index = order_begin_pos.buy_single_index
        if buy_single_index is None:
            continue_count = cls.__l2PlaceOrderParamsManagerDict[code].get_begin_continue_buy_count()
            # 有买入信号
            has_single, _index = cls.__compute_order_begin_pos(code, max(
                (compute_start_index - continue_count - 1) if new_add else compute_start_index, 0), continue_count,
                                                               compute_end_index)
            # 尝试计算快速成交信号
            has_single, _index, sell_info = cls.__compute_fast_order_begin_pos(code, compute_start_index,
                                                                               compute_end_index)
            if has_single:
                order_begin_pos.mode = OrderBeginPosInfo.MODE_FAST
                order_begin_pos.sell_info = sell_info
            elif _index is not None and _index < 0:
                continue_count = cls.__l2PlaceOrderParamsManagerDict[code].get_begin_continue_buy_count()
                # 有买入信号
                has_single, _index = cls.__compute_order_begin_pos(code, max(
                    (compute_start_index - continue_count - 1) if new_add else compute_start_index, 0), continue_count,
                                                                   compute_end_index)
                order_begin_pos.mode = OrderBeginPosInfo.MODE_NORMAL
            # 如果买入信号与上次的买入信号一样就不能算新的信号
            if cls.__last_buy_single_dict.get(code) == _index:
                has_single = None
                _index = None
            buy_single_index = _index
            if has_single:
                cls.__last_buy_single_dict[code] = buy_single_index
                new_get_single = True
                num = 0
                count = 0
                l2_log.debug(code, "获取到买入信号起始点:{} ,计算范围:{}-{} ,量比:{},数据:{}", buy_single_index, compute_start_index,
                             compute_end_index, cls.volume_rate_info[code], total_datas[buy_single_index])
                # 如果是今天第一次有下单开始信号,需要设置大单起始点
                cls.l2BigNumForMProcessor.set_begin_pos(code, buy_single_index)
                order_begin_pos.num = 0
                order_begin_pos.count = 0
                order_begin_pos.buy_single_index = buy_single_index
        _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "下单信号计算时间")
                if sell_info:
                    order_begin_pos.threshold_money = sell_info[1]
                l2_log.debug(code, "获取到买入信号起始点:{} ,计算范围:{}-{} ,量比:{},数据:{} 模式:{}", buy_single_index,
                             compute_start_index,
                             compute_end_index, cls.volume_rate_info[code], total_datas[buy_single_index],
                             order_begin_pos.mode)
        if buy_single_index is None:
        # _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "下单信号计算时间")
        if order_begin_pos.buy_single_index is None:
            # 未获取到买入信号,终止程序
            return None
        # 开始计算的位置
        start_process_index = max(buy_single_index, compute_start_index)
        start_process_index = max(order_begin_pos.buy_single_index, compute_start_index)
        if new_get_single:
            start_process_index = buy_single_index
            start_process_index = order_begin_pos.buy_single_index
        # 计算m值大单
        cls.l2BigNumForMProcessor.process(code, start_process_index,
                                          compute_end_index,
                                          gpcode_manager.get_limit_up_price(code))
        _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "计算m值大单")
        # _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "计算m值大单")
        threshold_money, msg = cls.__get_threshmoney(code)
        _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "m值阈值计算")
        # _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "m值阈值计算")
        # 买入纯买额统计
        compute_index, buy_nums, buy_count, rebegin_buy_pos, max_num_set_new = cls.__sum_buy_num_for_order_3(code,
                                                                                                             start_process_index,
                                                                                                             compute_end_index,
                                                                                                             num, count,
                                                                                                             threshold_money,
                                                                                                             buy_single_index,
                                                                                                             max_num_set)
        _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "纯买额统计时间")
        new_buy_exec_index, buy_nums, buy_count, rebegin_buy_pos, max_num_set_new = None, None, None, None, []
        if order_begin_pos.mode == OrderBeginPosInfo.MODE_FAST:
            threshold_money = order_begin_pos.threshold_money
            new_buy_exec_index, buy_nums, buy_count, rebegin_buy_pos, threshold_money_new = cls.__sum_buy_num_for_order_fast(
                code,
                start_process_index,
                compute_end_index,
                order_begin_pos.num,
                order_begin_pos.count,
                threshold_money,
                order_begin_pos.buy_single_index)
            threshold_money = threshold_money_new
            order_begin_pos.threshold_money = threshold_money
        else:
            new_buy_exec_index, buy_nums, buy_count, rebegin_buy_pos, max_num_set_new = cls.__sum_buy_num_for_order_3(
                code,
                start_process_index,
                compute_end_index,
                order_begin_pos.num,
                order_begin_pos.count,
                threshold_money,
                order_begin_pos.buy_single_index,
                order_begin_pos.max_num_set)
        l2_log.debug(code, "m值-{} 量比:{} rebegin_buy_pos:{}", threshold_money, cls.volume_rate_info[code][0],
                     rebegin_buy_pos)
@@ -1126,116 +1116,63 @@
                                    is_first_code, False)
            return
        if compute_index is not None:
            l2_log.debug(code, "获取到买入执行位置:{} m值:{} 纯买手数:{} 纯买单数:{} 数据:{} ,量比:{} ", compute_index, threshold_money,
        if new_buy_exec_index is not None:
            l2_log.debug(code, "获取到买入执行位置:{} m值:{} 纯买手数:{} 纯买单数:{} 数据:{} ,量比:{} ,下单模式:{}", new_buy_exec_index,
                         threshold_money,
                         buy_nums,
                         buy_count, total_datas[compute_index], cls.volume_rate_info[code])
            f1 = dask.delayed(cls.__save_order_begin_data)(code, buy_single_index, compute_index, compute_index,
                                                           buy_nums, buy_count, max_num_set_new,
                                                           cls.volume_rate_info[code][0])
            f2 = dask.delayed(limit_up_time_manager.LimitUpTimeManager().save_limit_up_time)(code,
                                                                                             total_datas[compute_index][
                                                                                                 "val"]["time"])
            f3 = dask.delayed(cls.__virtual_buy)(code, buy_single_index, compute_index, capture_time)
            f4 = dask.delayed(l2_data_manager.TradePointManager().delete_buy_cancel_point)(code)
            # 暂时不需要
            # f5 = dask.delayed(L2LimitUpMoneyStatisticUtil.process_data)(code, buy_single_index,
            #                                                             compute_index,
            #                                                             buy_single_index,
            #                                                             buy_exec_index, False)
            dask.compute(f1, f2, f3, f4)
            # 已被并行处理
            # # 记录买入信号位置
            # cls.__save_order_begin_data(code, buy_single_index, compute_index, compute_index, buy_nums, buy_count,
            #                             max_num_set_new)
            # # 如果是今天第一次有下单执行信号,涨停时间(买入执行位时间)
            # limit_up_time_manager.save_limit_up_time(code, total_datas[compute_index]["val"]["time"])
            # # 虚拟下单
            # cls.__virtual_buy(code, buy_single_index, compute_index, capture_time)
            # # 删除之前的所有撤单信号
            # l2_data_manager.TradePointManager.delete_buy_cancel_point(code)
            #
            # # 涨停封单额计算
            # L2LimitUpMoneyStatisticUtil.process_data(cls.random_key[code], code, buy_single_index, compute_index,
            #                                          buy_single_index,
            #                                          buy_exec_index, False)
            _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                              "记录执行买入数据", force=True)
                         buy_count, total_datas[new_buy_exec_index], cls.volume_rate_info[code], order_begin_pos.mode)
            cls.__save_order_begin_data(code, OrderBeginPosInfo(buy_single_index=buy_single_index,
                                                                buy_exec_index=new_buy_exec_index,
                                                                buy_compute_index=new_buy_exec_index,
                                                                num=buy_nums, count=buy_count,
                                                                max_num_set=max_num_set_new,
                                                                buy_volume_rate=cls.volume_rate_info[code][0],
                                                                mode=order_begin_pos.mode,
                                                                sell_info=order_begin_pos.sell_info,
                                                                threshold_money=threshold_money))
            cls.__LimitUpTimeManager.save_limit_up_time(code, total_datas[new_buy_exec_index]["val"]["time"])
            l2_log.debug(code, "delete_buy_cancel_point")
            # 直接下单
            ordered = cls.__buy(code, capture_time, total_datas[-1], total_datas[-1]["index"], is_first_code)
            # 数据是否处理完毕
            if compute_index >= compute_end_index:
                need_cancel, cancel_data = SecondCancelBigNumComputer().need_cancel(code, buy_single_index,
                                                                                    compute_index,
                                                                                    buy_single_index, compute_index,
                                                                                    total_datas, is_first_code,
                                                                                    cls.volume_rate_info[code][1],
                                                                                    cls.volume_rate_info[code][1],
                                                                                    True)
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  "S级大单处理耗时", force=True)
                l2_log.debug(code, "数据处理完毕,下单, 数据截图时间-{}", capture_time)
                # 数据已经处理完毕,如果还没撤单就实际下单
                if need_cancel:
                    if cls.cancel_buy(code, "S级大单撤销"):
                        # 执行撤单成功
                        pass
            if new_buy_exec_index < compute_end_index:
                if ordered:
                    cls.__process_order(code, new_buy_exec_index + 1, compute_end_index, capture_time, is_first_code,
                                        False)
                else:
                    cls.__buy(code, capture_time, total_datas[compute_index], compute_index, is_first_code)
            else:
                SecondCancelBigNumComputer().need_cancel(code, buy_single_index, compute_index, buy_single_index,
                                                         compute_index, total_datas, is_first_code,
                                                         cls.volume_rate_info[code][1],
                                                         cls.volume_rate_info[code][1], False)
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  "S级大单处理耗时", force=True)
                # 数据尚未处理完毕,进行下一步处理
                l2_log.debug(code, "数据尚未处理完毕,进行下一步处理,处理进度:{} end_index:{}", compute_index, compute_end_index)
                # 处理撤单步骤
                cls.__process_order(code, compute_index + 1, compute_end_index, capture_time, is_first_code, False)
                # lp = LineProfiler()
                # lp.enable()
                # lp_wrap = lp(cls.__process_order)
                # lp_wrap(code, compute_index + 1, compute_end_index, capture_time, is_first_code, False)
                # output = io.StringIO()
                # lp.print_stats(stream=output)
                # lp.disable()
                # logger_profile.info(output.getvalue())
                _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time,
                                                  f"处理撤单步骤耗时,范围:{compute_index + 1}-{compute_end_index}", force=True)
                    cls.__start_compute_buy(code, new_buy_exec_index + 1, compute_end_index, threshold_money,
                                            capture_time,
                                            is_first_code, False)
        else:
            # 未达到下单条件,保存纯买额,设置纯买额
            # 记录买入信号位置
            cls.__save_order_begin_data(code, buy_single_index, -1, compute_end_index, buy_nums, buy_count,
                                        max_num_set_new, None)
            print("保存大单时间", round((t.time() - _start_time) * 1000))
            cls.__save_order_begin_data(code, OrderBeginPosInfo(buy_single_index=buy_single_index, buy_exec_index=-1,
                                                                buy_compute_index=compute_end_index, num=buy_nums,
                                                                count=buy_count,
                                                                max_num_set=max_num_set_new, mode=order_begin_pos.mode,
                                                                sell_info=order_begin_pos.sell_info,
                                                                threshold_money=threshold_money))
            _start_time = t.time()
        pass
        l2_data_log.l2_time_log(code, "__start_compute_buy 结束")
    # 获取下单起始信号
    @classmethod
    def __get_order_begin_pos(cls, code):
        buy_single_index, buy_exec_index, compute_index, num, count, max_num_set, volume_rate = l2_data_manager.TradePointManager().get_buy_compute_start_data_cache(
    def __get_order_begin_pos(cls, code) -> OrderBeginPosInfo:
        order_begin_pos = cls.__TradePointManager.get_buy_compute_start_data_cache(
            code)
        return buy_single_index, buy_exec_index, compute_index, num, count, max_num_set, volume_rate
        return order_begin_pos
    # 保存下单起始信号
    @classmethod
    def __save_order_begin_data(self, code, buy_single_index, buy_exec_index, compute_index, num, count, max_num_set,
                                volume_rate):
        TradePointManager().set_buy_compute_start_data(code, buy_single_index, buy_exec_index, compute_index, num,
                                                       count,
                                                       max_num_set, volume_rate)
    def __save_order_begin_data(cls, code, info: OrderBeginPosInfo):
        cls.__TradePointManager.set_buy_compute_start_data_v2(code, info)
    # 计算下单起始信号
    # compute_data_count 用于计算的l2数据数量
    @classmethod
    def __compute_order_begin_pos(cls, code, start_index, continue_count, end_index):
        second_930 = 9 * 3600 + 30 * 60 + 0
        # 倒数100条数据查询
        datas = local_today_datas[code]
@@ -1260,7 +1197,12 @@
                    continue
            if L2DataUtil.is_limit_up_price_buy(_val):
                # 金额要大于50万
                if _val["num"] * float(_val["price"]) < 5000:
                    continue
                # 寻找前面continue_count-1个涨停买
                # for j in range(start_index - 1, -1, -1):
                #     if  datas[j]["val"]
                if last_index is None or (datas[last_index]["val"]["time"] == datas[i]["val"]["time"]):
                    if start is None:
                        start = i
@@ -1282,9 +1224,70 @@
        return False, None
    # 快速买入法的信号位置查找
    @classmethod
    def __compute_fast_order_begin_pos(cls, code, start_index, end_index):
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        # if float(limit_up_price) >= 10:
        #     return False, -1, "股价大于10块"
        total_datas = local_today_datas[code]
        start_time_str = total_datas[start_index]["val"]["time"]
        # if tool.trade_time_sub(start_time_str, "13:00:00") > 0:
        #     return False, -1, "超过规定时间"
        refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code, start_time_str)
        if refer_sell_data is None:
            return False, -1, "总卖为空"
        if cls.__L2MarketSellManager.is_refer_sell_time_used(code, refer_sell_data[0]):
            return False, -1, "总卖统计时间已被使用"
        # 是否大于500万
        if refer_sell_data[1] <= 500 * 10000:
            return False, -1, "总卖小于指定金额"
        # 统计之前的卖
        threshold_money = refer_sell_data[1]
        for i in range(start_index - 1, -1, -1):
            val = total_datas[i]["val"]
            if tool.compare_time(val["time"], refer_sell_data[0]) <= 0:
                break
            if L2DataUtil.is_sell(val):
                threshold_money += val["num"] * int(float(val["price"]) * 100)
            elif L2DataUtil.is_sell_cancel(val):
                threshold_money -= val["num"] * int(float(val["price"]) * 100)
        # 是否为本秒的第一个涨停买
        for i in range(start_index, end_index + 1):
            data = total_datas[i]
            val = data['val']
            if not L2DataUtil.is_limit_up_price_buy(val):
                # 要统计卖与卖撤
                if L2DataUtil.is_sell(val):
                    threshold_money += val["num"] * int(float(val["price"]) * 100)
                elif L2DataUtil.is_sell_cancel(val):
                    threshold_money -= val["num"] * int(float(val["price"]) * 100)
                continue
            # 50 万以下的不需要
            if val["num"] * float(val["price"]) < 5000:
                continue
            # 是否为本s的第一次涨停
            is_first_limit_up = True
            for j in range(i - 1, -1, -1):
                temp_val = total_datas[j]["val"]
                if temp_val["time"] == val["time"]:
                    if L2DataUtil.is_limit_up_price_buy(temp_val) and temp_val["num"] * float(
                            temp_val["price"]) >= 5000:
                        is_first_limit_up = True
                        break
                else:
                    break
            if is_first_limit_up:
                return True, i, [refer_sell_data[0], threshold_money]
        return False, None, None
    @classmethod
    def __get_threshmoney(cls, code):
        return cls.__l2PlaceOrderParamsManagerDict[code].get_m_val()
        m, msg = cls.__l2PlaceOrderParamsManagerDict[code].get_m_val()
        if trade_manager.CodesTradeStateManager().get_trade_state_cache(code) == trade_manager.TRADE_STATE_NOT_TRADE:
            # 首次下单m值扩大1.5倍
            m = int(m * 1.5)
        return m, msg
    # 计算万手哥笔数
    @classmethod
@@ -1301,13 +1304,9 @@
    @classmethod
    def __sum_buy_num_for_order_3(cls, code, compute_start_index, compute_end_index, origin_num, origin_count,
                                  threshold_money, buy_single_index, max_num_set):
        def get_threshold_count():
            count = threshold_count
            return count
        _start_time = t.time()
        total_datas = local_today_datas[code]
        is_first_code = gpcode_manager.FirstCodeManager().is_in_first_record_cache(code)
        # is_first_code = gpcode_manager.FirstCodeManager().is_in_first_record_cache(code)
        buy_nums = origin_num
        buy_count = origin_count
@@ -1316,6 +1315,8 @@
            raise Exception("涨停价无法获取")
        # 目标手数
        threshold_num = round(threshold_money / (limit_up_price * 100))
        # 大目标手数(满足这个就不需要看安全笔数)
        threshold_max_num = int(threshold_num * 1.2)
        # place_order_count = trade_data_manager.PlaceOrderCountManager().get_place_order_count(code)
        # 目标订单数量
@@ -1336,7 +1337,7 @@
        big_num_count = cls.__l2PlaceOrderParamsManagerDict[code].get_big_num_count()
        # 较大单的手数
        bigger_num = round(5900 / limit_up_price)
        bigger_num = round(5000 / limit_up_price)
        for i in range(compute_start_index, compute_end_index + 1):
            data = total_datas[i]
@@ -1344,7 +1345,7 @@
            trigger_buy = False
            # 必须为连续2秒内的数据
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds + 1 > max_space_time:
                TradePointManager().delete_buy_point(code)
                cls.__TradePointManager.delete_buy_point(code)
                if i == compute_end_index:
                    # 数据处理完毕
                    return None, buy_nums, buy_count, None, max_buy_num_set
@@ -1362,18 +1363,17 @@
                    # 只统计59万以上的金额
                    buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
                    buy_count += int(total_datas[i]["re"])
                    if buy_nums >= threshold_num and buy_count >= get_threshold_count():
                        logger_l2_trade_buy.info(
                            f"{code}获取到买入执行点:{i} 统计纯买手数:{buy_nums} 目标纯买手数:{threshold_num} 统计纯买单数:{buy_count} 目标纯买单数:{get_threshold_count()}, 大单数量:{len(max_buy_num_set)}")
                    if (buy_nums >= threshold_num and buy_count >= threshold_count) or buy_nums >= threshold_max_num:
                        async_log_util.info(logger_l2_trade_buy,
                                            f"{code}获取到买入执行点:{i} 统计纯买手数:{buy_nums} 目标纯买手数:{threshold_num}/{threshold_max_num} 统计纯买单数:{buy_count} 目标纯买单数:{threshold_count}, 大单数量:{len(max_buy_num_set)}")
            elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
                if _val["num"] >= bigger_num:
                    # 只统计59万以上的金额
                    # 涨停买撤
                    # 判断买入位置是否在买入信号之前
                    buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data(code,
                                                                                                     total_datas[i],
                                                                                                     local_today_num_operate_map.get(
                                                                                                         code))
                    buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i],
                                                                                                        local_today_buyno_map.get(
                                                                                                            code))
                    if buy_index is not None:
                        # 找到买撤数据的买入点
                        if buy_index >= buy_single_index:
@@ -1400,21 +1400,160 @@
                             buy_nums, threshold_num)
            max_buy_num_set_count = 0
            for i in max_buy_num_set:
                max_buy_num_set_count += total_datas[i]["re"]
            for i1 in max_buy_num_set:
                max_buy_num_set_count += total_datas[i1]["re"]
            # 有撤单信号,且小于阈值
            if buy_nums >= threshold_num and buy_count >= get_threshold_count() and trigger_buy and max_buy_num_set_count >= big_num_count:
            if buy_nums >= threshold_num and buy_count >= threshold_count and trigger_buy and max_buy_num_set_count >= big_num_count:
                try:
                    info = cls.__trade_log_placr_order_info_dict[code]
                    info.set_trade_factor(threshold_money, threshold_count, list(max_buy_num_set))
                except Exception as e:
                    async_log_util.error(logger_l2_error, f"记录交易因子出错:{str(e)}")
                return i, buy_nums, buy_count, None, max_buy_num_set
        l2_log.buy_debug(code, "尚未获取到买入执行点,起始计算位置:{} 统计纯买手数:{} 目标纯买手数:{}  统计纯买单数:{} 目标纯买单数:{} 大单数量:{} 目标大单数量:{}",
                         compute_start_index,
                         buy_nums,
                         threshold_num, buy_count, get_threshold_count(), max_buy_num_set_count, big_num_count)
                         threshold_num, buy_count, threshold_count, max_buy_num_set_count, big_num_count)
        return None, buy_nums, buy_count, None, max_buy_num_set
    # 返回(买入执行点, 总手, 总笔数, 从新计算起点, 纯买额阈值)
    # 计算快速买入
    @classmethod
    def __sum_buy_num_for_order_fast(cls, code, compute_start_index, compute_end_index, origin_num, origin_count,
                                     threshold_money_origin, buy_single_index):
        _start_time = t.time()
        total_datas = local_today_datas[code]
        # is_first_code = gpcode_manager.FirstCodeManager().is_in_first_record_cache(code)
        buy_nums = origin_num
        buy_count = origin_count
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        if limit_up_price is None:
            raise Exception("涨停价无法获取")
        limit_up_price = float(limit_up_price)
        threshold_money = threshold_money_origin
        # 目标手数
        threshold_num = round(threshold_money / (limit_up_price * 100))
        buy_single_time_seconds = L2DataUtil.get_time_as_second(total_datas[buy_single_index]["val"]["time"])
        # 可以触发买,当有涨停买信号时才会触发买
        trigger_buy = True
        # 间隔最大时间为3s
        max_space_time = 3
        for i in range(compute_start_index, compute_end_index + 1):
            data = total_datas[i]
            _val = total_datas[i]["val"]
            trigger_buy = False
            # 必须为连续2秒内的数据
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds + 1 > max_space_time:
                cls.__TradePointManager.delete_buy_point(code)
                if i == compute_end_index:
                    # 数据处理完毕
                    return None, buy_nums, buy_count, None, threshold_money
                else:
                    # 计算买入信号,不能同一时间开始计算
                    for ii in range(buy_single_index + 1, compute_end_index + 1):
                        if total_datas[buy_single_index]["val"]["time"] != total_datas[ii]["val"]["time"]:
                            return None, buy_nums, buy_count, ii, threshold_money
            if L2DataUtil.is_sell(_val):
                threshold_money += _val["num"] * int(float(_val["price"]) * 100)
                threshold_num = round(threshold_money / (limit_up_price * 100))
            elif L2DataUtil.is_sell_cancel(_val):
                threshold_money -= _val["num"] * int(float(_val["price"]) * 100)
                threshold_num = round(threshold_money / (limit_up_price * 100))
            # 涨停买
            elif L2DataUtil.is_limit_up_price_buy(_val):
                trigger_buy = True
                # 只统计59万以上的金额
                buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
                buy_count += int(total_datas[i]["re"])
                if buy_nums >= threshold_num:
                    async_log_util.info(logger_l2_trade_buy,
                                        f"{code}获取到买入执行点(快速买入):{i} 统计纯买手数:{buy_nums} 目标纯买手数:{threshold_num} 统计纯买单数:{buy_count}")
            elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
                # 判断买入位置是否在买入信号之前
                buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i],
                                                                                                    local_today_buyno_map.get(
                                                                                                        code))
                if buy_index is not None:
                    # 找到买撤数据的买入点
                    if buy_index >= buy_single_index:
                        buy_nums -= int(_val["num"]) * int(data["re"])
                        buy_count -= int(data["re"])
                        l2_log.buy_debug(code, "{}数据在买入信号之后 撤买纯买手数:{} 目标手数:{}", i, buy_nums, threshold_num)
                    else:
                        l2_log.buy_debug(code, "{}数据在买入信号之前,买入位:{}", i, buy_index)
                        if total_datas[buy_single_index]["val"]["time"] == total_datas[buy_index]["val"]["time"]:
                            # 同一秒,当作买入信号之后处理
                            buy_nums -= int(_val["num"]) * int(data["re"])
                            buy_count -= int(data["re"])
                            # 大单撤销
                            l2_log.buy_debug(code, "{}数据买入位与预估买入位在同一秒", i)
                else:
                    # 未找到买撤数据的买入点
                    l2_log.buy_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data)
                    buy_nums -= int(_val["num"]) * int(total_datas[i]["re"])
                    buy_count -= int(total_datas[i]["re"])
            l2_log.buy_debug(code, "位置-{},总手数:{},目标手数:{}", i,
                             buy_nums, threshold_num)
            # 有撤单信号,且小于阈值
            if buy_nums >= threshold_num and trigger_buy:
                try:
                    info = cls.__trade_log_placr_order_info_dict[code]
                    info.set_trade_factor(threshold_money, 0, [])
                except Exception as e:
                    async_log_util.error(logger_l2_error, f"记录交易因子出错:{str(e)}")
                return i, buy_nums, buy_count, None, threshold_money
        l2_log.buy_debug(code, "尚未获取到买入执行点(快速买入),起始计算位置:{} 统计纯买手数:{} 目标纯买手数:{}  统计纯买单数:{}",
                         compute_start_index,
                         buy_nums,
                         threshold_num, buy_count)
        return None, buy_nums, buy_count, None, threshold_money
def test_trade_record():
    code = "000333"
    __trade_log_placr_order_info_dict = {code: trade_record_log_util.PlaceOrderInfo()}
    try:
        jx_blocks, jx_blocks_by = KPLCodeJXBlockManager().get_jx_blocks_cache(
            code), KPLCodeJXBlockManager().get_jx_blocks_cache(code, by=True)
        info = __trade_log_placr_order_info_dict[code]
        info.set_buy_index(0, 1)
        if jx_blocks:
            info.set_kpl_blocks(list(jx_blocks))
        elif jx_blocks_by:
            info.set_kpl_blocks(list(jx_blocks_by))
        else:
            info.set_kpl_blocks([])
        trade_record_log_util.add_place_order_log(code, info)
    except:
        pass
if __name__ == "__main__":
    yesterday_limit_up_data_records = kpl_data_manager.get_current_limit_up_data_records(1)[0][1]
    yesterday_codes = set([x[0] for x in yesterday_limit_up_data_records])
    print(yesterday_codes)
    # test_trade_record()
    # yesterday_limit_up_data_records = kpl_data_manager.get_current_limit_up_data_records(1)[0][1]
    # yesterday_codes = set([x[0] for x in yesterday_limit_up_data_records])
    # print(yesterday_codes)
    code = "603003"
    datas = log_export.load_l2_from_log()
    datas = datas.get(code)
    if datas is None:
        datas = []
    l2.l2_data_util.local_today_datas[code] = datas[:191]
    l2.l2_data_util.load_buy_no_map(l2.l2_data_util.local_today_buyno_map, code,
                                    l2.l2_data_util.local_today_datas[code])
    l2.l2_data_util.load_canceled_buy_no_map(l2.l2_data_util.local_today_canceled_buyno_map, code,
                                             l2.l2_data_util.local_today_datas[code])
    start_index = 73
    end_index = 190
    LCancelBigNumComputer().compute_watch_index(code, start_index, end_index)