bug修复

Administrator

2023-07-24 0e68e24f54db11d340785b17570fff2bc5fc7ac6

 l2/l2_data_manager_new.py

@@ -1,35 +1,34 @@
import logging
import random
import time as t

import big_money_num_manager
import code_data_util
import code_volumn_manager
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
import constant
import global_util
import gpcode_manager
import industry_codes_sort
from l2.huaxin import l2_huaxin_util, huaxin_delegate_postion_manager
from third_data import kpl_data_manager, block_info
from utils import global_util, ths_industry_util, tool
import l2_data_util
import l2_trade_test
import limit_up_time_manager
from db import redis_manager
import ths_industry_util
import tool
from trade import trade_data_manager, trade_manager, trade_queue_manager, l2_trade_factor, l2_trade_util, \
    trade_result_manager, first_code_score_manager
from l2 import safe_count_manager, l2_data_manager, l2_data_log, l2_log, l2_data_source_util, code_price_manager
from third_data.code_plate_key_manager import CodePlateKeyBuyManager
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
    L2LimitUpSellStatisticUtil, DCancelBigNumComputer, LCancelBigNumComputer
from l2.l2_data_manager import L2DataException, TradePointManager
from l2.l2_data_util import local_today_datas, L2DataUtil, load_l2_data, local_today_num_operate_map, local_latest_datas
from l2.l2_data_util import local_today_datas, L2DataUtil, local_today_num_operate_map, local_today_buyno_map, \
    local_latest_datas
import l2.l2_data_util
from log import logger_l2_trade, logger_l2_trade_cancel, logger_l2_trade_buy, logger_l2_process, logger_l2_error, \
    logger_buy_score
from log_module.log import logger_l2_trade_buy, logger_l2_process, \
    logger_place_order_score, logger_l2_error

# TODO l2数据管理
from trade.trade_data_manager import CodeActualPriceProcessor

import dask

from trade.trade_manager import TradeTargetCodeModeManager


class L2DataManager:
@@ -163,6 +162,16 @@
    __buyL2SafeCountManager = safe_count_manager.BuyL2SafeCountManager()
    __l2PlaceOrderParamsManagerDict = {}
    __last_buy_single_dict = {}
    __TradeBuyQueue = transaction_progress.TradeBuyQueue()

    # 获取代码评分
    @classmethod
    def get_code_scores(cls):
        score_dict = {}
        for code in cls.__l2PlaceOrderParamsManagerDict:
            score = cls.__l2PlaceOrderParamsManagerDict[code].score
            score_dict[code] = score
        return score_dict

    @classmethod
    # 数据处理入口
@@ -183,7 +192,9 @@
                    # 数据不正常需要禁止交易
                    l2_trade_util.forbidden_trade(code)
                # 纠正数据
                datas = l2.l2_data_util.L2DataUtil.correct_data(code, local_latest_datas.get(code), datas)
                if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_THS:
                    # 同花顺需要纠正数据，其他渠道不需要
                    datas = l2.l2_data_util.L2DataUtil.correct_data(code, local_latest_datas.get(code), datas)
                _start_index = 0
                if local_today_datas.get(code) is not None and len(
                        local_today_datas[code]) > 0:
@@ -204,6 +215,43 @@
            if code in cls.unreal_buy_dict:
                cls.unreal_buy_dict.pop(code)

    # 处理华鑫L2数据
    @classmethod
    def process_huaxin(cls, code, datas):
        print("process_huaxin", code, len(datas))
        origin_start_time = round(t.time() * 1000)
        try:
            # 加载历史的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)
            # 转换数据格式
            _start_index = 0
            if local_today_datas.get(code) is not None and len(
                    local_today_datas[code]) > 0:
                _start_index = local_today_datas[code][-1]["index"] + 1
            datas = l2_huaxin_util.get_format_l2_datas(code, 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)
            cls.process_add_datas(code, datas, 0, __start_time)
        except Exception as e:
            print("huaxin L2数据处理异常", code, str(e))
            logging.exception(e)
            logger_l2_error.exception(e)
        finally:
            l2_data_log.l2_time(code, round(t.time() * 1000) - origin_start_time,
                                "l2数据处理总耗时",
                                True)
            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()
@@ -212,6 +260,7 @@
            # 拼接数据
            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)

            # 第1条数据是否为09:30:00
            if add_datas[0]["val"]["time"] == "09:30:00":
@@ -241,11 +290,15 @@
            cls.__l2PlaceOrderParamsManagerDict[code] = l2_trade_factor.L2PlaceOrderParamsManager(code, is_first_code,
                                                                                                  volume_rate,
                                                                                                  volume_rate_index,
                                                                                                  score)
            l2_log.debug(code, "量比：{}，量索引：{}", volume_rate, volume_rate_index)
                                                                                                  score,
                                                                                                  total_datas[-1][
                                                                                                      'val']['time'])
            cls.volume_rate_info[code] = (volume_rate, volume_rate_index)

            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数据准备时间")
            # 时间差不能太大才能处理
            if not l2_trade_util.is_in_forbidden_trade_codes(code):
                # 判断是否已经挂单
@@ -351,7 +404,8 @@
        def h_cancel():
            _start_time = round(t.time() * 1000)
            try:
                b_need_cancel, b_cancel_data = HourCancelBigNumComputer.need_cancel(code, buy_exec_index, start_index,
                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),
@@ -365,6 +419,24 @@
                logging.exception(e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-H撤大单计算")
            return None, ""

        # L撤
        @dask.delayed
        def l_cancel():
            _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)
                if b_need_cancel and b_cancel_data:
                    return b_cancel_data, "L撤销比例触发阈值"
            except Exception as e:
                logging.exception(e)
            finally:
                l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-L撤大单计算")
            return None, ""

        # 板上卖撤
@@ -414,9 +486,10 @@
        f4 = h_cancel()
        f5 = buy_1_cancel()
        f6 = sell_cancel()
        dask_result = is_need_cancel(f1, f2, f3, f4, f5, f6)
        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)
            dask_result = is_need_cancel(f3, f4, f7)

        cancel_data, cancel_msg = dask_result.compute()

@@ -425,6 +498,7 @@

        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,
@@ -461,17 +535,17 @@
        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(
            code)
        if not can:
            l2_log.debug(code, "不可以下单，原因：{}", reason)
            if need_clear_data:
                buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set, buy_volume_rate = cls.__get_order_begin_pos(
                    code)
                trade_result_manager.real_cancel_success(code, buy_single_index, buy_exec_index,
                                                         local_today_datas.get(code))
            return
        else:
            l2_log.debug(code, "可以下单，原因：{}", reason)
            logger_buy_score.info(f"{code}:{cls.__l2PlaceOrderParamsManagerDict[code].score_info}")

            try:
                l2_log.debug(code, "开始执行买入")
                trade_manager.start_buy(code, capture_timestamp, last_data,
@@ -479,7 +553,14 @@
                ################下单成功处理################
                trade_result_manager.real_buy_success(code)
                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)
            except Exception as e:
                logger_l2_error.exception(e)
                l2_log.debug(code, "执行买入异常:{}", str(e))
                pass
            finally:
@@ -654,23 +735,142 @@
        if not trade_manager.TradeStateManager.is_can_buy():
            return False, True, f"今日已禁止交易"

        if gpcode_manager.PauseBuyCodesManager.is_in(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 constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_HUAXIN:
            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.02001:
                return False, False, f"当前成交价（{trade_price}）尚未在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()
                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)
                        if left_count > 0:
                            buy_index_set.add(total_data[i]["index"])

                if len(buy_index_set) < 5:
                    return False, False, f"成交位置距离当前位置小于5笔"
        else:
            # 判断买1价格档位
            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 >= 200 * 100000000:
                buy1_price = code_price_manager.Buy1PriceManager.get_buy1_price(code)
                if buy1_price is None:
                    return False, True, f"尚未获取到买1价"
                dif = float(limit_up_price) - float(buy1_price)
                # 大于10档
                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)
        price_pre_close = gpcode_manager.get_price_pre(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.get_limit_up_time(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)

        if not gpcode_manager.WantBuyCodesManager.is_in(code):
            # 查看分数等级
            if TradeTargetCodeModeManager.get_mode() == TradeTargetCodeModeManager.MODE_ONLY_BUY_WANT_CODES:
                return False, True, f"只买想买单中的代码"
            score_index = cls.__l2PlaceOrderParamsManagerDict[code].score_index
            score = cls.__l2PlaceOrderParamsManagerDict[code].score
            if score_index < 0:
                return False, True, f"分值：{score}未达到需要买入的分数线"
            # if -1 < score_index < 3 and (0.499 <= cls.volume_rate_info[code][0] <= 0.949):
            #     return True, False, f"分值：{score}达到主动买入的分数线且量足够，买入等级：f{score_index},量比：{cls.volume_rate_info[code][0]}"
            is_limited_up = gpcode_manager.FirstCodeManager.is_limited_up(code)
            gpcode_manager.FirstCodeManager.add_limited_up_record([code])
            if not code_price_manager.Buy1PriceManager.is_can_buy(code):
                return False, True, f"首板代码，没在想要买名单中且未打开涨停板,分数：{score}"
            if not is_limited_up:
                return False, True, f"首板代码，没在想要买名单中且未涨停过,分数：{score}"
            return True, False, ""
            score_info = cls.__l2PlaceOrderParamsManagerDict[code].score_info
            return cls.can_buy_first(code, limit_up_price, score_index, score, score_info, cls.volume_rate_info[code])
        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亿"
        # 判断板块
        plate_can_buy, msg = CodePlateKeyBuyManager.can_buy(code,
                                                            kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
                                                            kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,block_info.get_before_blocks_dict())
        if not plate_can_buy:
            return False, True, msg
        return True, False, msg

        # if volume_rate_info[0] < 0.4:
        #     return False, True, f"量大于40%才下单,量比：{volume_rate_info[0]}"

        # 是否有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, ""

    @classmethod
    def __cancel_buy(cls, code):
@@ -755,8 +955,8 @@
                new_get_single = True
                num = 0
                count = 0
                l2_log.debug(code, "获取到买入信号起始点：{} ,计算范围：{}-{} ，数据：{}", buy_single_index, compute_start_index,
                             compute_end_index, total_datas[buy_single_index])
                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)

@@ -804,7 +1004,7 @@
        if compute_index is not None:
            l2_log.debug(code, "获取到买入执行位置：{} m值：{} 纯买手数：{} 纯买单数：{} 数据：{} ,量比:{} ", compute_index, threshold_money,
                         buy_nums,
                         buy_count, total_datas[compute_index], cls.volume_rate_info[code][0])
                         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,
@@ -908,12 +1108,18 @@
        last_index = None
        count = 0
        start = None

        now_time_s = tool.get_time_as_second(tool.get_now_time_str())
        for i in range(start_index, end_index + 1):
            _val = datas[i]["val"]
            time_s = L2DataUtil.get_time_as_second(_val["time"])
            # 时间要>=09:30:00
            if L2DataUtil.get_time_as_second(_val["time"]) < second_930:
            if time_s < second_930:
                continue

            if not constant.TEST:
                if abs(now_time_s - time_s) > 2:
                    # 正式环境下不处理2s外的数据
                    continue

            if L2DataUtil.is_limit_up_price_buy(_val):

@@ -998,8 +1204,8 @@
            data = total_datas[i]
            _val = total_datas[i]["val"]
            trigger_buy = False
            # 必须为连续3秒内的数据
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds > max_space_time:
            # 必须为连续2秒内的数据
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds + 1 > max_space_time:
                TradePointManager.delete_buy_point(code)
                if i == compute_end_index:
                    # 数据处理完毕
@@ -1055,15 +1261,17 @@
            l2_log.buy_debug(code, "位置-{}，总手数：{}，目标手数：{}", i,
                             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"]
            # 有撤单信号，且小于阈值
            if buy_nums >= threshold_num and buy_count >= get_threshold_count() and trigger_buy and len(
                    max_buy_num_set) >= big_num_count:
            if buy_nums >= threshold_num and buy_count >= get_threshold_count() and trigger_buy and max_buy_num_set_count >= big_num_count:
                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(), len(max_buy_num_set), big_num_count)
                         threshold_num, buy_count, get_threshold_count(), max_buy_num_set_count, big_num_count)

        return None, buy_nums, buy_count, None, max_buy_num_set

@@ -1079,6 +1287,5 @@
    #                                                                  local_today_num_operate_map.get(
    #                                                                      "600213"))
    # print(buy_index, buy_data)
    dict_ = {"code": 0}
    dict_.clear()
    print(dict_)
    volume_rate = code_volumn_manager.get_volume_rate("002343")
    print(volume_rate)