l2数据计算优化

Administrator

2022-09-23 21d753614ea7bbe936b8560cbf466c4e438821b2

 l2_data_manager.py

@@ -18,8 +18,9 @@
import redis_manager
import tool
import trade_manager
from log import logger_l2_trade, logger_l2_trade_cancel
from log import logger_l2_trade, logger_l2_trade_cancel, logger_l2_trade_buy
from trade_data_manager import TradeBuyDataManager
import limit_up_time_manager

_redisManager = redis_manager.RedisManager(1)
# l2数据管理
@@ -94,38 +95,52 @@
    # 获取撤买入开始计算的信息
    # 返回数据的内容为：撤销点索引 撤买纯买额 计算的数据索引
    @staticmethod
    def get_buy_cancel_compute_start_data(code):
    def get_buy_cancel_single_pos(code):
        redis = TradePointManager.__get_redis()
        info = redis.get("buy_cancel_compute_info-{}".format(code))
        info = redis.get("buy_cancel_single_pos-{}".format(code))
        if info is None:
            return None, None, None
            return None
        else:
            info = json.loads(info)
            return info[0], info[1], info[2]
            return int(info)

    # 设置买撤点信息
    # buy_num 纯买额  computed_index计算到的下标  index撤买信号起点

    @classmethod
    def set_buy_cancel_compute_start_data(cls, code, buy_num, computed_index, index):
    def set_buy_cancel_single_pos(cls, code, index):
        redis = TradePointManager.__get_redis()
        expire = tool.get_expire()
        redis.setex("buy_cancel_compute_info-{}".format(code), expire, json.dumps((index, buy_num, computed_index)))

    # 增加撤买的纯买额
    @classmethod
    def add_buy_nums_for_cancel(cls, code, num_add, computed_index):
        cancel_index, nums, c_index = cls.get_buy_cancel_compute_start_data(code)
        if cancel_index is None:
            raise Exception("无撤买信号记录")
        nums += num_add
        cls.set_buy_cancel_compute_start_data(code, nums, computed_index)
        redis.setex("buy_cancel_single_pos-{}".format(code), expire, index)

    # 删除买撤点数据
    @staticmethod
    def delete_buy_cancel_point(code):
    @classmethod
    def delete_buy_cancel_point(cls, code):
        redis = TradePointManager.__get_redis()
        redis.delete("buy_cancel_compute_info-{}".format(code))
        redis.delete("buy_cancel_single_pos-{}".format(code))

    # 设置买撤纯买额
    @classmethod
    def set_compute_info_for_cancel_buy(cls, code, index, nums):
        redis = TradePointManager.__get_redis()
        expire = tool.get_expire()
        redis.setex("compute_info_for_cancel_buy-{}".format(code), expire, json.dumps((index, nums)))
        logger_l2_trade_buy.info("{}保存撤单纯买额信息：{}，{}", code, index, nums)

    # 获取买撤纯买额计算信息
    @classmethod
    def get_compute_info_for_cancel_buy(cls, code):
        redis = TradePointManager.__get_redis()
        info = redis.get("compute_info_for_cancel_buy-{}".format(code))
        if info is None:
            return None, 0
        else:
            info = json.loads(info)
            return info[0], info[1]

    @classmethod
    def delete_compute_info_for_cancel_buy(cls, code):
        redis = TradePointManager.__get_redis()
        redis.delete("compute_info_for_cancel_buy-{}".format(code))


def load_l2_data(code, force=False):
@@ -398,6 +413,11 @@
            ("thread-id={} code={}  ".format(cls.random_key[code], code) + content).format(*args))

    @classmethod
    def buy_debug(cls, code, content, *args):
        logger_l2_trade_buy.debug(
            ("thread-id={} code={}  ".format(cls.random_key[code], code) + content).format(*args))

    @classmethod
    # 数据处理入口
    # datas: 本次截图数据
    # capture_timestamp:截图时间戳
@@ -474,38 +494,45 @@
        if start_index < 0:
            start_index = 0
        # 获取之前是否有记录的撤买信号
        cancel_index, buy_num_for_cancel, computed_index = cls.__has_order_cancel_begin_pos(code)
        cancel_index = TradePointManager.get_buy_cancel_single_pos(code)

        cancel_computed_index, cancel_buy_num = TradePointManager.get_compute_info_for_cancel_buy(code)
        if cancel_computed_index is None:
            logger_l2_trade.error("{} 未获取到买撤纯买额,起始计算位：{}", code, start_index)

        buy_single_index, buy_exec_index, buy_compute_index, buy_num = cls.__get_order_begin_pos(code)
        if cancel_index is None:
            # 无撤单信号起始点记录
            cancel_index = cls.__compute_order_cancel_begin_single(code, max(start_index - 3, 0), 3)
            buy_num_for_cancel = buy_num
            computed_index = buy_single_index
            if cancel_index is not None:
                cls.debug(code, "找到撤单信号，数据处理起始点：{} 数据：{}", start_index, local_today_datas[code][start_index])
        if cancel_index is not None:
                cls.debug(code, "找到撤单信号-{}，买入信号为 ，数据处理起始点：{}", cancel_index, start_index)
                # 保存撤单信号
                TradePointManager.set_buy_cancel_single_pos(code, cancel_index)
            # 获取阈值 有买撤信号，统计撤买纯买额
            threshold_money = cls.__get_threshmoney(code)
            cls.__start_compute_cancel(code, cancel_index, max(computed_index, buy_exec_index + 1), buy_num_for_cancel,
                                       threshold_money,
                                       capture_time)
        else:
            # 无买撤信号,是否有虚拟下单
            unreal_buy_info = cls.unreal_buy_dict.get(code)
            if unreal_buy_info is not None:
                cls.debug(code, "有虚拟下单，无买撤信号，开始执行买入")
                # unreal_buy_info 的内容格式为：(触法买操作下标,截图时间)
                # 真实下单
                cls.__buy(code, unreal_buy_info[1], local_today_datas[code][unreal_buy_info[0]],
                          unreal_buy_info[0])
                pass
        threshold_money = cls.__get_threshmoney(code)
        cls.__start_compute_cancel(code, cancel_index, cancel_computed_index + 1,
                                   cancel_buy_num,
                                   threshold_money,
                                   capture_time)

        # 是否有虚拟下单
        unreal_buy_info = cls.unreal_buy_dict.get(code)
        if unreal_buy_info is not None:
            cls.debug(code, "有虚拟下单，无买撤信号，开始执行买入,截图时间：{}", capture_time)
            # unreal_buy_info 的内容格式为：(触法买操作下标,截图时间)
            # 真实下单
            cls.__buy(code, unreal_buy_info[1], local_today_datas[code][unreal_buy_info[0]],
                      unreal_buy_info[0])

    # 开始计算撤的信号
    @classmethod
    def __start_compute_cancel(cls, code, cancel_index, compute_start_index, origin_num, threshold_money, capture_time):
        # sure_type 0-虚拟挂买位  1-真实挂买位
        cancel_single = cancel_index is not None
        computed_index, buy_num_for_cancel, sure_type = cls.__sum_buy_num_for_cancel_order(code, compute_start_index,
                                                                                           origin_num, threshold_money)
                                                                                           origin_num, threshold_money,
                                                                                           cancel_single)

        total_datas = local_today_datas[code]
        if computed_index is not None:
            cls.debug(code, "获取到撤单执行信号，信号位置：{}，m2:{} 数据：{}", computed_index, threshold_money,
@@ -528,10 +555,10 @@
                cls.__start_compute_buy(code, computed_index + 1, threshold_money, capture_time)
                pass
        else:
            cls.debug(code, "未获取到撤单执行信号，计算开始位置：{}， 纯买额：{}", compute_start_index, buy_num_for_cancel)
            # 无需撤买，记录撤买信号
            TradePointManager.set_buy_cancel_compute_start_data(code, buy_num_for_cancel, len(total_datas) - 1,
                                                                cancel_index)
            cls.debug(code, "撤买纯买额计算，计算位置：{}-{}，目前为止纯买手数：{}", compute_start_index, total_datas[-1]["index"],
                      buy_num_for_cancel)
            # 无需撤买，设置计算信息
            TradePointManager.set_compute_info_for_cancel_buy(code, int(total_datas[-1]["index"]), buy_num_for_cancel)
            # 判断是否有虚拟下单
            unreal_buy_info = cls.unreal_buy_dict.get(code)
            if unreal_buy_info is not None:
@@ -547,6 +574,9 @@

    @classmethod
    def __buy(cls, code, capture_timestamp, last_data, last_data_index):
        # 删除虚拟下单
        if code in cls.unreal_buy_dict:
            cls.unreal_buy_dict.pop(code)
        cls.debug(code, "开始执行买入")
        try:
            trade_manager.start_buy(code, capture_timestamp, last_data,
@@ -565,6 +595,7 @@
            # 取消买入标识
            TradePointManager.delete_buy_point(code)
            TradePointManager.delete_buy_cancel_point(code)
            TradePointManager.delete_compute_info_for_cancel_buy(code)
            cls.debug(code, "执行撤单成功")
        except Exception as e:
            cls.debug(code, "执行撤单异常：{}", str(e))
@@ -585,22 +616,29 @@
                num = 0
                new_get_pos = True
                cls.debug(code, "获取到买入信号起始点：{}  数据：{}", buy_single_index, total_datas[buy_single_index])
                limit_up_time_manager.save_limit_up_time(code, total_datas[buy_single_index]["val"]["time"])

        if buy_single_index is None:
            # 未获取到买入信号，终止程序
            return None

        # 买入纯买额统计
        compute_index, buy_nums = cls.__sum_buy_num_for_order(code, max(buy_single_index, compute_start_index), num,
                                                              threshold_money)
        compute_index, buy_nums = cls.__sum_buy_num_for_order_2(code, max(buy_single_index, compute_start_index), num,threshold_money,buy_single_index)

        #cls.__sum_buy_num_for_order(code, max(buy_single_index, compute_start_index), num,threshold_money)
        if compute_index is not None:
            cls.debug(code, "获取到买入执行位置：{} m值：{} 数据：{}", compute_index, threshold_money, total_datas[compute_index])
            cls.debug(code, "获取到买入执行位置：{} m值：{} 纯买手数：{} 数据：{}", compute_index, threshold_money, buy_nums,
                      total_datas[compute_index])
            # 记录买入信号位置
            cls.__save_order_begin_data(code, buy_single_index, compute_index, compute_index, buy_nums)
            # 虚拟下单
            cls.unreal_buy_dict[code] = (compute_index, capture_time)
            # 删除之前的所有撤单信号
            TradePointManager.delete_buy_cancel_point(code)
            TradePointManager.delete_compute_info_for_cancel_buy(code)
            # 为买撤保存基础纯买额
            TradePointManager.set_compute_info_for_cancel_buy(code, compute_index, buy_nums)

            # 数据是否处理完毕
            if L2DataUtil.is_index_end(code, compute_index):
                cls.debug(code, "数据处理完毕，下单, 数据截图时间-{}", capture_time)
@@ -626,15 +664,6 @@
    @classmethod
    def __save_order_begin_data(self, code, buy_single_index, buy_exec_index, compute_index, num):
        TradePointManager.set_buy_compute_start_data(code, buy_single_index, buy_exec_index, compute_index, num)

    # 获取撤单起始位置
    @classmethod
    def __has_order_cancel_begin_pos(cls, code):
        # cancel_index:撤单信号起点
        # buy_num_for_cancel：从挂入点计算的纯买额
        # computed_index 计算的最后位置
        cancel_index, buy_num_for_cancel, computed_index = TradePointManager.get_buy_cancel_compute_start_data(code)
        return cancel_index, buy_num_for_cancel, computed_index

    # 计算下单起始信号
    # compute_data_count 用于计算的l2数据数量
@@ -776,11 +805,66 @@
                # 涨停买
                buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
                if buy_nums >= threshold_num:
                    logger_l2_trade_buy.info("{}获取到买入执行点：{} 统计纯买手数：{} 目标纯买手数：{}", code, i, buy_nums, threshold_num)
                    return i, buy_nums
            elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
                # 涨停买撤
                buy_nums -= int(_val["num"]) * int(total_datas[i]["re"])
        logger_l2_trade_buy.info("{}尚未获取到买入执行点，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{}", code, compute_start_index, buy_nums,
                                 threshold_num)
        return None, buy_nums

    # 统计买入净买量，不计算在买入信号之前的买撤单
    @classmethod
    def __sum_buy_num_for_order_2(cls, code, compute_start_index, origin_num, threshold_money, buy_single_index):
        total_datas = local_today_datas[code]
        buy_nums = origin_num
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        if limit_up_price is None:
            raise Exception("涨停价无法获取")
        threshold_num = threshold_money / (limit_up_price * 100)
        property_buy_num_count = 0
        same_time_property = cls.__get_same_time_property(code)
        for i in range(compute_start_index, len(total_datas)):
            data = total_datas[i]
            _val = total_datas[i]["val"]
            # 有连续4个涨停买就标记计算起始点
            if L2DataUtil.is_limit_up_price_buy(_val):
                # 涨停买
                buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
                if buy_nums >= threshold_num:
                    logger_l2_trade_buy.info("{}获取到买入执行点：{} 统计纯买手数：{} 目标纯买手数：{}", code, i, buy_nums, threshold_num)
            elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
                # 涨停买撤
                # 判断买入位置是否在买入信号之前
                buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(total_datas[i],
                                                                                 local_today_num_operate_map.get(code))
                if buy_index is not None:
                    # 找到买撤数据的买入点
                    if buy_index >= buy_single_index:
                        buy_nums -= int(_val["num"]) * int(data["re"])
                        cls.buy_debug(code, "{}数据在买入信号之后 撤买纯买手数：{} 目标手数：{}", i, buy_nums, threshold_num)
                    else:
                        cls.buy_debug(code, "{}数据在买入信号之前，买入位：{}", i, buy_index)
                        if total_datas[buy_single_index]["val"]["time"] == buy_data["val"]["time"]:
                            # 同一秒,而且还在预估买入位之后按概率计算
                            property_buy_num_count -= int(_val["num"]) * int(data["re"])
                            cls.buy_debug(code, "{}数据买入位与预估买入位在同一秒", i)
                else:
                    # 未找到买撤数据的买入点
                    cls.cancel_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data)
                    buy_nums -= int(_val["num"]) * int(total_datas[i]["re"])
            property_buy_num = round(property_buy_num_count * same_time_property)
            cls.buy_debug(code, "买入信号点之前同一秒买入手数-{}，位置-{}，总手数：{}，目标手数：{}", property_buy_num, i,
                          buy_nums + property_buy_num, threshold_num)
            # 有撤单信号，且小于阈值
            if buy_nums + property_buy_num >= threshold_num:
                return i, buy_nums + property_buy_num

        cls.buy_debug(code, "尚未获取到买入执行点，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{}", compute_start_index,
                      buy_nums + property_buy_num,
                      threshold_num)
        return None, buy_nums + property_buy_num

    # 同一时间买入的概率计算
    @classmethod
@@ -795,7 +879,7 @@

    # 统计买撤净买量
    @classmethod
    def __sum_buy_num_for_cancel_order(cls, code, start_index, origin_num, threshold_money):
    def __sum_buy_num_for_cancel_order(cls, code, start_index, origin_num, threshold_money, cancel_single=True):
        buy_nums = origin_num
        total_datas = local_today_datas[code]
        limit_up_price = gpcode_manager.get_limit_up_price(code)
@@ -807,7 +891,8 @@
        same_time_property = cls.__get_same_time_property(code)
        # 同一秒，在预估买入位之后的数据之和
        property_buy_num_count = 0
        cls.cancel_debug(code, "撤单纯买额计算位置：{}-{} 预估挂买位：{}", start_index, len(total_datas) - 1, sure_pos)
        cls.cancel_debug(code, "撤单纯买额计算位置：{}-{} 预估挂买位：{} 是否有撤单信号：{}", start_index, len(total_datas) - 1, sure_pos,
                         cancel_single)
        for i in range(start_index, len(total_datas)):
            data = total_datas[i]
            _val = data["val"]
@@ -828,7 +913,7 @@
                    # 找到买撤数据的买入点
                    if buy_index < sure_pos:
                        buy_nums -= int(_val["num"]) * int(data["re"])
                        cls.cancel_debug(code, "{}数据在预估买入位之前 撤买纯买额：{}", i, buy_nums * limit_up_price)
                        cls.cancel_debug(code, "{}数据在预估买入位之前 撤买纯买手数：{} 目标手数：{}", i, buy_nums, threshold_num)
                    else:
                        cls.cancel_debug(code, "{}数据在预估买入位之后，买入位：{}", i, buy_index)
                        if sure_data["val"]["time"] == buy_data["val"]["time"]:
@@ -840,10 +925,14 @@
                    cls.cancel_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data)

            property_buy_num = round(property_buy_num_count * same_time_property)
            cls.cancel_debug(code, "预估买入点之后同一秒买入手数-{}，位置-{}，总手数：{}", property_buy_num, i, buy_nums + property_buy_num)
            if buy_nums + property_buy_num <= threshold_num:
            cls.cancel_debug(code, "预估买入点之后同一秒买入手数-{}，位置-{}，总手数：{}，目标手数：{}", property_buy_num, i,
                             buy_nums + property_buy_num, threshold_num)
            # 有撤单信号，且小于阈值
            if buy_nums + property_buy_num <= threshold_num and cancel_single:
                return i, buy_nums + property_buy_num, sure_type
        return None, buy_nums + round(property_buy_num_count * same_time_property), sure_type
        buy_num_news = buy_nums + round(property_buy_num_count * same_time_property)
        cls.cancel_debug(code, "处理起始位置：{} 最终纯买额：{}", start_index, buy_num_news)
        return None, buy_num_news, sure_type

    @classmethod
    def test(cls):