修复bug，优化下单

Administrator

2022-11-10 38395204ab04bbca22a5e22be988d826afd5b227

修复bug，优化下单

 alert_util.py

@@ -8,11 +8,11 @@


def alarm():
    if not tool.is_trade_time():
    if not tool.is_alert_time():
        return
    # TODO 暂时关闭报警
    # AlertUtil().stop_audio()
    # AlertUtil().play_audio()
    AlertUtil().stop_audio()
    AlertUtil().play_audio()


class AlertUtil:

 gpcode_manager.py

@@ -120,12 +120,19 @@
    redis_instance.setex("price-pre-{}".format(code), tool.get_expire(), str(price))


__limit_up_price_dict = {}


# 获取涨停价
def get_limit_up_price(code):
    # 读取内存中的值
    if code in __limit_up_price_dict:
        return __limit_up_price_dict[code]
    price = get_price_pre(code)
    if price is None:
        return None
    return tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal("1.1"))
    limit_up_price = tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal("1.1"))
    __limit_up_price_dict[code] = limit_up_price


def get_limit_up_price_by_preprice(price):
@@ -188,6 +195,7 @@

# 同步监听代码位置信息
def __sync_listen_codes_pos():
    redis_instance = __redisManager.getRedis()
    # 获取已经正在监听的代码
    keys = redis_instance.keys("code_listen_pos-*")
    codes_set = set()
@@ -280,6 +288,11 @@
    # return codes.__contains__(code)


def is_listen_old(code):
    codes = get_listen_codes()
    return codes.__contains__(code)


# 监听是否满了
def is_listen_full():
    clients = client_manager.getValidL2Clients()
@@ -307,5 +320,7 @@

if __name__ == '__main__':
    _start = time.time()
    is_listen("002703")
    redis_instance = __redisManager.getRedis()
    val = redis_instance.get("code_listen_pos-{}".format("603786"))
    print(json.loads(val))
    print( (time.time() - _start) * 1000)

 gui.py

@@ -300,13 +300,24 @@
                count = 0
                if codes:
                    count = len(codes)
                if count < 1:
                    normal = False
                    cl_buy_1.configure(text="{}".format(count), foreground="#FF7F27")
                else:
                cl_buy_1.configure(text="{}".format(count), foreground="#008000")
            except:
                pass

            try:
                count = self.codeActualPriceProcessor.get_current_price_codes_count()
                if count is None or int(count) < 1:
                    normal = False
                cl_price_count.configure(
                    text="{}".format(self.codeActualPriceProcessor.get_current_price_codes_count()),
                        text="{}".format(count),
                        foreground="#FF7F27")
                else:
                    cl_price_count.configure(
                        text="{}".format(count),
                    foreground="#008000")
            except:
                pass
@@ -459,8 +470,6 @@
                    showinfo("提示", "修复完成")
                except Exception as e:
                    showerror("修复出错", str(e))



            # 创建界面
            win = Tk()

 juejin.py

@@ -31,7 +31,7 @@

import trade_gui
from l2_code_operate import L2CodeOperate
from l2_data_manager import L2LimitUpMoneyStatisticUtil, L2DataUtil
from l2_data_manager import L2DataUtil
from log import logger_juejin_tick, logger_system
from trade_data_manager import CodeActualPriceProcessor
from trade_queue_manager import JueJinBuy1VolumnManager
@@ -232,13 +232,13 @@
        rate = round((price - pricePre) * 100 / pricePre, 1)
        if rate >= 7:
            logger_juejin_tick.info("{}-{}-{}", code, price, rate)
            if not gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(
            if not gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(
                    code) and not gpcode_manager.is_listen_full():
                L2CodeOperate.get_instance().add_operate(1, code, "现价变化,rate-{} from-{}".format(rate, price_from))
            # 进入监控
        elif rate < 5:
            # 移除监控
            if gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(code):
            if gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(code):
                L2CodeOperate.get_instance().add_operate(0, code, "现价变化,rate-{} from-{}".format(rate, price_from))


@@ -315,12 +315,12 @@
        # 后面的代码数量
        # 先删除应该删除的代码
        for code in del_list:
            if gpcode_manager.is_listen(code):
            if gpcode_manager.is_listen_old(code):
                # 判断是否在监听里面
                L2CodeOperate.get_instance().add_operate(0, code, "现价变化")
        # 增加应该增加的代码
        for code in add_code_list:
            if not gpcode_manager.is_listen(code):
            if not gpcode_manager.is_listen_old(code):
                if not l2_trade_util.is_in_forbidden_trade_codes(code):
                    L2CodeOperate.get_instance().add_operate(1, code, "现价变化")
            else:

 l2_code_operate.py

@@ -53,6 +53,7 @@
                L2CodeOperate.set_operate_code_state(client_id, position, 1)

        except Exception as e:
            logging.exception(e)
            logger_code_operate.error("setGPCode出错：{}", str(e))
        finally:
            gpcode_manager.rm_operate(gpcode)
@@ -78,20 +79,20 @@
                # print("读取操作队列", data, redis.llen("code_operate_queue"))
                if data is not None:
                    data = json.loads(data)
                    logger_code_operate.info("读取操作队列：{}", data)
                    # logger_code_operate.info("读取操作队列：{}", data)
                    type, code = data["type"], data["code"]
                    create_time = data.get("create_time")
                    if create_time is not None:
                        # 设置10s超时时间
                        if round(time.time() * 1000) - create_time > 20 * 1000:
                            logger_code_operate.debug("读取操作超时：{}", data)
                        if round(time.time() * 1000) - create_time > 15 * 1000:
                            # logger_code_operate.debug("读取操作超时：{}", data)
                            continue

                    if type == 0:
                        # 是否在固定库
                        if l2_data_manager.is_in_l2_fixed_codes(code):
                            continue
                        if gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(code):
                        if gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(code):
                            client_id, pos = gpcode_manager.get_listen_code_pos(code)
                            if client_id is not None and pos is not None:
                                L2CodeOperate.setGPCode(client_id, pos, "")
@@ -99,7 +100,7 @@
                        if l2_trade_util.is_in_forbidden_trade_codes(code):
                            continue

                        if not gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(
                        if not gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(
                                code) and not gpcode_manager.is_listen_full():
                            client_id, pos = gpcode_manager.get_can_listen_pos()
                            if pos is not None and client_id is not None:
@@ -120,13 +121,13 @@
                        L2CodeOperate.setGPCode(client_id, pos, code)
                    # 修复l2的数据错误
                    elif type == 3:
                        if tool.is_set_code_time():
                        if tool.is_repaire_time():
                            client = data["client"]
                            data = data["data"]
                            result = server.send_msg(client, data)
                            print("L2數據修復結果：", result)
                        else:
                            print("非交易时间，放弃修复L2")
                            print("非修复时间，放弃修复L2")
                    elif type == 4:
                        # 清理监听位置
                        client = data["client"]
@@ -172,7 +173,7 @@
    # 移除监控
    def remove_l2_listen(self, code, msg):
        # 是否正在监听
        if gpcode_manager.is_listen(code):
        if gpcode_manager.is_listen_old(code):
            self.add_operate(0, code, msg=msg)

    # 设置代码操作状态，服务器保存的代码是否与实际设置的代码保持一致
@@ -213,8 +214,8 @@
            __reset_code_dict[key] = round(time.time() * 1000)
            if code_ is None:
                code_ = ""
            if tool.is_set_code_time():
                L2CodeOperate.repaire_operate(int(client), int(channel), code_)
            if tool.is_repaire_time():
                L2CodeOperate().repaire_operate(int(client), int(channel), code_)
    else:
        key = "{}-{}".format(client, channel)
        if key not in __set_operate_code_state_dict or round(
@@ -249,8 +250,8 @@
            for index in range(0, 8):
                code = gpcode_manager.get_listen_code_by_pos(client_id, index)
                if code is not None and len(code) > 0 and index_codes.get(index) != code:
                    # 交易时间才修复代码
                    if tool.is_set_code_time():
                    # 修复时间才修复代码
                    if tool.is_repaire_time():
                        L2CodeOperate().repaire_operate(client_id, index, code)
                elif code is None or len(code) == 0 and index_codes.get(index) is not None:
                    # 删除前端代码位

 l2_data_manager.py

@@ -251,7 +251,6 @@
    data = data["data"]
    limit_up_price = gpcode_manager.get_limit_up_price(code)


    datas = L2DataUtil.format_l2_data(data, code, limit_up_price)
    # 获取涨停价
    return day, client, channel, code, capture_time, process_time, datas,data
@@ -260,6 +259,8 @@
# 保存l2数据
def save_l2_data(code, datas, add_datas):
    redis = _redisManager.getRedis()
    # 只有有新曾数据才需要保存
    if len(add_datas) > 0:
    # 保存最近的数据
    __start_time = round(t.time()* 1000)
    redis.setex("l2-data-latest-{}".format(code), tool.get_expire(), json.dumps(datas))
@@ -267,7 +268,6 @@
    # 设置进内存
    local_latest_datas[code] = datas
    __set_l2_data_latest_count(code, len(datas))
    if len(add_datas) > 0:
        saveL2Data(code, add_datas)


@@ -398,8 +398,6 @@
        # TODO 测试的时候开启，方便记录大单数据
        #l2_data_util.save_big_data(code, same_time_num, data)
        return datas



    @classmethod
    def get_time_as_second(cls, time_str):
@@ -1915,223 +1913,6 @@
            process_index = end_index
        cls.__save_recod(code, process_index, count)


# 涨停封单额统计
class L2LimitUpMoneyStatisticUtil:
    _redisManager = redis_manager.RedisManager(1)

    @classmethod
    def __get_redis(cls):
        return cls._redisManager.getRedis()

    # 设置l2的每一秒涨停封单额数据
    @classmethod
    def __set_l2_second_money_record(cls, code, time, num, from_index, to_index):
        old_num, old_from, old_to = cls.__get_l2_second_money_record(code, time)
        if old_num is None:
            old_num = num
            old_from = from_index
            old_to = to_index
        else:
            old_num += num
            old_to = to_index

        key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", ""))

        cls.__get_redis().setex(key, tool.get_expire(), json.dumps((old_num, old_from, old_to)))

    @classmethod
    def __get_l2_second_money_record(cls, code, time):
        key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", ""))
        val = cls.__get_redis().get(key)
        return cls.__format_second_money_record_val(val)

    @classmethod
    def __format_second_money_record_val(cls, val):
        if val is None:
            return None, None, None
        val = json.loads(val)
        return val[0], val[1], val[2]

    @classmethod
    def __get_l2_second_money_record_keys(cls, code, time_regex):
        key = "l2_limit_up_second_money-{}-{}".format(code, time_regex)
        keys = cls.__get_redis().keys(key)
        return keys

    # 设置l2最新的封单额数据
    @classmethod
    def __set_l2_latest_money_record(cls, code, index, num):
        key = "l2_limit_up_money-{}".format(code)
        cls.__get_redis().setex(key, tool.get_expire(), json.dumps((num, index)))

    # 返回数量,索引
    @classmethod
    def __get_l2_latest_money_record(cls, code):
        key = "l2_limit_up_money-{}".format(code)
        result = cls.__get_redis().get(key)
        if result:
            result = json.loads(result)
            return result[0], result[1]
        else:
            return 0, -1

    # 矫正数据
    # 矫正方法为取矫正时间两侧的秒分布数据，用于确定计算结束坐标
    @classmethod
    def verify_num(cls, code, num, time_str):
        time_ = time_str.replace(":", "")
        key = None
        for i in range(4, -2, -2):
            # 获取本(分钟/小时/天)内秒分布数据
            time_regex = "{}*".format(time_[:i])
            keys_ = cls.__get_l2_second_money_record_keys(code, time_regex)
            if keys_ and len(keys_) > 1:
                # 需要排序
                keys = []
                for k in keys_:
                    keys.append(k)
                keys.sort(key=lambda tup: int(tup.split("-")[-1]))
                # 有2个元素
                for index in range(0, len(keys) - 1):
                    time_1 = keys[index].split("-")[-1]
                    time_2 = keys[index + 1].split("-")[-1]
                    if int(time_1) <= int(time_) <= int(time_2):
                        # 在此时间范围内
                        if time_ == time_2:
                            key = keys[index + 1]
                        else:
                            key = keys[index]
                        break
            if key:
                val = cls.__get_redis().get(key)
                old_num, old_from, old_to = cls.__format_second_money_record_val(val)
                end_index = old_to
                # 保存最近的数据
                cls.__set_l2_latest_money_record(code, end_index, num)
                break

    # 计算量，用于涨停封单量的计算
    @classmethod
    def __compute_num(cls, code, data, buy_single_data):
        if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]) or L2DataUtil.is_sell(data["val"]):
            # 涨停买撤与卖
            return 0 - int(data["val"]["num"]) * data["re"]
        else:
            # 卖撤
            if L2DataUtil.is_sell_cancel(data["val"]):
                # 卖撤的买数据是否在买入信号之前，如果在之前就不计算，不在之前就计算
                if l2_data_util.is_sell_index_before_target(data, buy_single_data,
                                                            local_today_num_operate_map.get(code)):
                    return 0

            return int(data["val"]["num"]) * data["re"]

    @classmethod
    def clear(cls, code):
        key = "l2_limit_up_money-{}".format(code)
        cls.__get_redis().delete(key)

    # 返回取消的标志数据
    # with_cancel 是否需要判断是否撤销
    @classmethod
    def process_data(cls, code, start_index, end_index, buy_single_begin_index, with_cancel=True):
        start_time = round(t.time() * 1000)
        total_datas = local_today_datas[code]
        time_dict_num = {}
        # 记录计算的坐标
        time_dict_num_index = {}
        num_dict = {}
        # 统计时间分布
        time_dict = {}
        for i in range(start_index, end_index + 1):
            data = total_datas[i]
            val = data["val"]
            time_ = val["time"]
            if time_ not in time_dict:
                time_dict[time_] = i

        for i in range(start_index, end_index + 1):
            data = total_datas[i]
            val = data["val"]
            time_ = val["time"]
            if time_ not in time_dict_num:
                time_dict_num[time_] = 0
                time_dict_num_index[time_] = {"s": i, "e": i}
            time_dict_num_index[time_]["e"] = i
            num = cls.__compute_num(code, data, total_datas[buy_single_begin_index])
            num_dict[i] = num
            time_dict_num[time_] = time_dict_num[time_] + num
        for t_ in time_dict_num:
            cls.__set_l2_second_money_record(code, t_, time_dict_num[t_], time_dict_num_index[t_]["s"],
                                             time_dict_num_index[t_]["e"])

        print("保存涨停封单额时间：", round(t.time() * 1000) - start_time)

        # 累计最新的金额
        total_num, index = cls.__get_l2_latest_money_record(code)
        if index == -1:
            # 没有获取到最新的矫正封单额，需要从买入信号开始点计算
            index = buy_single_begin_index - 1
            total_num = 0
        # TODO 待优化计算
        cancel_index = None
        cancel_msg = None
        # 待计算量
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        min_volumn = round(10000000 / (limit_up_price * 100))
        # 不同时间的数据开始坐标
        time_start_index_dict = {}
        # 数据时间分布
        time_list = []
        # 到当前时间累积的买1量
        time_total_num_dict = {}
        for i in range(index + 1, end_index + 1):
            data = total_datas[i]
            time_ = data["val"]["time"]
            if time_ not in time_start_index_dict:
                # 记录每一秒的开始位置
                time_start_index_dict[time_] = i
                # 记录时间分布
                time_list.append(time_)
                # 上一段时间的总数
                time_total_num_dict[time_] = total_num

            val = num_dict.get(i)
            if val is None:
                val = cls.__compute_num(code, data, total_datas[buy_single_begin_index])
            total_num += val
            # 如果是减小项，且在处理数据的范围内，就需要判断是否要撤单了
            if val < 0 and start_index <= i <= end_index:
                # 累计封单金额小于1000万
                if total_num < min_volumn:
                    cancel_index = i
                    cancel_msg = "封单金额小于1000万"
                    break
                # 相邻2s内的数据减小50%
                # 上1s的总数
                last_second_total_volumn = time_total_num_dict.get(time_list[-1])
                if last_second_total_volumn > 0 and (
                        last_second_total_volumn - total_num) / last_second_total_volumn >= 0.5:
                    # 相邻2s内的数据减小50%
                    cancel_index = i
                    cancel_msg = "相邻2s({})内的封单量减小50%({}->{})".format(time_, last_second_total_volumn,
                                                                     total_num)
                    break
        if not with_cancel:
            cancel_index = None

        print("封单额计算时间：", round(t.time() * 1000) - start_time)
        process_end_index = end_index
        if cancel_index:
            process_end_index = cancel_index
        # 保存最新累计金额
        # cls.__set_l2_latest_money_record(code, process_end_index, total_num)
        l2_data_log.l2_time(code, round(t.time() * 1000) - start_time, "l2数据封单额计算时间",
                            False)
        if cancel_index:
            return total_datas[cancel_index], cancel_msg
        return None, None


def __get_time_second(time_str):

 l2_data_manager_new.py

@@ -1,4 +1,5 @@
import datetime
import json
import logging
import random
import time as t
@@ -18,8 +19,8 @@
import tool
import trade_manager
from l2_data_manager import L2DataException, TradePointManager, local_today_datas, L2DataUtil, load_l2_data, \
    local_today_num_operate_map, L2LimitUpMoneyStatisticUtil
from log import logger_l2_trade, logger_l2_trade_cancel, logger_l2_trade_buy, logger_l2_process
    local_today_num_operate_map
from log import logger_l2_trade, logger_l2_trade_cancel, logger_l2_trade_buy, logger_l2_process, logger_buy_1_volumn

# TODO l2数据管理
from trade_data_manager import CodeActualPriceProcessor
@@ -205,8 +206,6 @@
                total_datas = local_today_datas[code]
                __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "l2数据预处理时间")
                if len(add_datas) > 0:

                    _start_time = round(t.time() * 1000)
                    latest_time = add_datas[len(add_datas) - 1]["val"]["time"]
                    # 时间差不能太大才能处理
                    # TODO 暂时关闭处理
@@ -236,9 +235,12 @@
    # 处理未挂单
    @classmethod
    def __process_not_order(cls, code, start_index, end_index, capture_time):
        _start_time = t.time()
        __start_time = t.time()
        # 获取阈值
        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值数据耗时")
        cls.__start_compute_buy(code, start_index, end_index, threshold_money, capture_time)

    # 处理已挂单
@@ -304,6 +306,11 @@
    @classmethod
    def __can_buy(cls, code):

        # 量比超过1.3的不能买
        volumn_rate = l2_trade_factor.L2TradeFactorUtil.get_volumn_rate_by_code(code)
        if volumn_rate >= 1.3:
            return False, "最大量比超过1.3不能买"

        limit_up_time = limit_up_time_manager.get_limit_up_time(code)
        if limit_up_time is not None and l2_data_manager.L2DataUtil.get_time_as_second(
                limit_up_time) >= l2_data_manager.L2DataUtil.get_time_as_second(
@@ -320,7 +327,8 @@

        if cls.__codeActualPriceProcessor.is_under_water(code):
            # 水下捞且板块中的票小于21不能买
            if global_util.industry_hot_num.get(industry) <= 16:
            if global_util.industry_hot_num.get(industry) is not None and global_util.industry_hot_num.get(
                    industry) <= 16:
                return False, "水下捞,板块中的票小于2只,为{}".format(global_util.industry_hot_num.get(industry))

            if codes_index.get(code) != 0:
@@ -418,19 +426,19 @@
                # 如果是今天第一次有下单开始信号，需要设置大单起始点
                cls.l2BigNumForMProcessor.set_begin_pos(code, buy_single_index)

        _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "下单信号计算时间")

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

        _start_time = t.time()

        # 计算m值大单
        cls.l2BigNumForMProcessor.process(code, max(buy_single_index, compute_start_index), compute_end_index,
                                          gpcode_manager.get_limit_up_price(code))
        _start_time = t.time()

        _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "计算m值大单")

        threshold_money, msg = cls.__get_threshmoney(code)
        _start_time = t.time()
        # 买入纯买额统计
        compute_index, buy_nums, buy_count, rebegin_buy_pos = cls.__sum_buy_num_for_order_3(code, max(buy_single_index,
                                                                                                      compute_start_index),
@@ -438,10 +446,10 @@
                                                                                            count, threshold_money,
                                                                                            buy_single_index,
                                                                                            capture_time)
        _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "纯买额统计时间")

        cls.debug(code, "m值-{} m值因子-{}", threshold_money, msg)

        _start_time = t.time()
        # 买入信号位与计算位置间隔2s及以上了
        if rebegin_buy_pos is not None:
            # 需要重新计算纯买额
@@ -463,6 +471,8 @@

            # 涨停封单额计算
            L2LimitUpMoneyStatisticUtil.process_data(code, buy_single_index, compute_index, buy_single_index, False)

            _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "记录执行买入数据")

            # 数据是否处理完毕
            if compute_index >= compute_end_index:
@@ -542,12 +552,42 @@
    def __get_threshmoney(cls, code):
        return l2_trade_factor.L2TradeFactorUtil.compute_m_value(code)

    # 是否为万手哥
    @classmethod
    def __is_big_money(cls, limit_up_price, val):
        if int(val["num"]) >= 7888:
            return True
        if int(val["num"]) * limit_up_price >= 29900:
            return True
        return False

    # 计算万手哥笔数
    @classmethod
    def __compute_big_money_count(cls, total_datas, start_index, end_index):
        count = 0
        for i in range(start_index, end_index + 1):
            if L2DataUtil.is_limit_up_price_buy(total_datas[i]["val"]):
                count += total_datas[i]["re"]
            elif L2DataUtil.is_limit_up_price_buy_cancel(total_datas[i]["val"]):
                count -= total_datas[i]["re"]
        return count

    # 统计买入净买量，不计算在买入信号之前的买撤单
    @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, capture_time):
        def get_threshold_count():
            count = threshold_count - sub_threshold_count
            if count < 3:
                count = 3
            return count
        _start_time = t.time()
        total_datas = local_today_datas[code]
        # 计算从买入信号开始到计算开始位置的大单数量
        sub_threshold_count = cls.__compute_big_money_count(total_datas, buy_single_index, compute_start_index - 1)
        if sub_threshold_count < 0:
            sub_threshold_count = 0

        buy_nums = origin_num
        buy_count = origin_count
        limit_up_price = gpcode_manager.get_limit_up_price(code)
@@ -563,7 +603,8 @@
        for i in range(compute_start_index, compute_end_index + 1):
            data = total_datas[i]
            _val = total_datas[i]["val"]
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds > 1:
            # 必须为连续3秒内的数据
            if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds > 2:
                TradePointManager.delete_buy_point(code)
                if i == compute_end_index:
                    # 数据处理完毕
@@ -573,16 +614,19 @@
                    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

            # 涨停买
            if L2DataUtil.is_limit_up_price_buy(_val):
                if cls.__is_big_money(limit_up_price,_val):
                    sub_threshold_count += int(total_datas[i]["re"])
                # 涨停买
                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 >= threshold_count:
                    logger_l2_trade_buy.info("{}获取到买入执行点：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{} 目标纯买单数：{}", code, i, buy_nums,
                                             threshold_num, buy_count, threshold_count)
                if buy_nums >= threshold_num and buy_count >= get_threshold_count():
                    logger_l2_trade_buy.info("{}获取到买入执行点：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{} 目标纯买单数：{}, 大单数量：{}", code, i, buy_nums,
                                             threshold_num, buy_count, get_threshold_count(),sub_threshold_count)
            elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
                if cls.__is_big_money(limit_up_price, _val):
                    sub_threshold_count -= int(total_datas[i]["re"])
                # 涨停买撤
                # 判断买入位置是否在买入信号之前
                buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(total_datas[i],
@@ -608,12 +652,12 @@
            cls.buy_debug(code, "位置-{}，总手数：{}，目标手数：{}", i,
                          buy_nums, threshold_num)
            # 有撤单信号，且小于阈值
            if buy_nums >= threshold_num and buy_count >= threshold_count:
            if buy_nums >= threshold_num and buy_count >= get_threshold_count():
                return i, buy_nums, buy_count, None

        cls.buy_debug(code, "尚未获取到买入执行点，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{}  统计纯买单数：{} 目标纯买单数：{}", compute_start_index,
        cls.buy_debug(code, "尚未获取到买入执行点，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{}  统计纯买单数：{} 目标纯买单数：{} 大单数量：{}", compute_start_index,
                      buy_nums,
                      threshold_num, buy_count, threshold_count)
                      threshold_num, buy_count, get_threshold_count(),sub_threshold_count)

        return None, buy_nums, buy_count, None

@@ -682,8 +726,290 @@
        print("时间花费:", round((t.time() - _start) * 1000))
        pass

    @classmethod
    def test2(cls):
        code = "002864"
        load_l2_data(code)
        limit_up_time_manager.load_limit_up_time()
        limit_up_time = limit_up_time_manager.get_limit_up_time(code)
        if limit_up_time is not None and l2_data_manager.L2DataUtil.get_time_as_second(
                limit_up_time) >= l2_data_manager.L2DataUtil.get_time_as_second(
            "14:30:00"):
            return False, "14:30后涨停的不能买，涨停时间为{}".format(limit_up_time)

        # 同一板块中老二后面的不能买
        industry, codes = ths_industry_util.get_same_industry_codes(code, gpcode_manager.get_gp_list())
        if industry is None:
            return True, "没有获取到行业"
        codes_index = limit_up_time_manager.sort_code_by_limit_time(codes)
        if codes_index is not None and codes_index.get(code) is not None and codes_index.get(code) > 1:
            return False, "同一板块中老三,老四,...不能买"

        if cls.__codeActualPriceProcessor.is_under_water(code):
            # 水下捞且板块中的票小于21不能买
            if global_util.industry_hot_num.get(industry) is not None and global_util.industry_hot_num.get(
                    industry) <= 16:
                return False, "水下捞,板块中的票小于2只,为{}".format(global_util.industry_hot_num.get(industry))

            if codes_index.get(code) != 0:
                return False, "水下捞,不是老大，是老{}".format(codes_index.get(code))

        # 13:30后涨停，本板块中涨停票数<29不能买
        limit_up_time = limit_up_time_manager.get_limit_up_time(code)
        if limit_up_time is not None:
            if int(limit_up_time.replace(":", "")) >= 133000 and global_util.industry_hot_num.get(industry) is not None:
                if global_util.industry_hot_num.get(industry) < 16:
                    return False, "13:30后涨停，本板块中涨停票数<16不能买"

        if codes_index.get(code) is not None and codes_index.get(code) == 1:
            # 如果老大已经买成功了，老二就不需要买了
            first_codes = []
            for key in codes_index:
                if codes_index.get(key) == 0:
                    first_codes.append(key)

            for key in first_codes:
                state = trade_manager.get_trade_state(key)
                if state == trade_manager.TRADE_STATE_BUY_SUCCESS:
                    # 老大已经买成功了
                    return False, "老大{}已经买成功，老二无需购买".format(key)

            # 有9点半涨停的老大才能买老二，不然不能买
            # 获取老大的涨停时间
            for key in first_codes:
                # 找到了老大
                time_ = limit_up_time_manager.get_limit_up_time(key)
                if time_ == "09:30:00":
                    return True, "9:30涨停的老大，老二可以下单"
            return False, "老大非9:30涨停，老二不能下单"


# 涨停封单额统计
class L2LimitUpMoneyStatisticUtil:
    _redisManager = redis_manager.RedisManager(1)

    @classmethod
    def __get_redis(cls):
        return cls._redisManager.getRedis()

    # 设置l2的每一秒涨停封单额数据
    @classmethod
    def __set_l2_second_money_record(cls, code, time, num, from_index, to_index):
        old_num, old_from, old_to = cls.__get_l2_second_money_record(code, time)
        if old_num is None:
            old_num = num
            old_from = from_index
            old_to = to_index
        else:
            old_num += num
            old_to = to_index

        key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", ""))

        cls.__get_redis().setex(key, tool.get_expire(), json.dumps((old_num, old_from, old_to)))

    @classmethod
    def __get_l2_second_money_record(cls, code, time):
        key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", ""))
        val = cls.__get_redis().get(key)
        return cls.__format_second_money_record_val(val)

    @classmethod
    def __format_second_money_record_val(cls, val):
        if val is None:
            return None, None, None
        val = json.loads(val)
        return val[0], val[1], val[2]

    @classmethod
    def __get_l2_second_money_record_keys(cls, code, time_regex):
        key = "l2_limit_up_second_money-{}-{}".format(code, time_regex)
        keys = cls.__get_redis().keys(key)
        return keys

    # 设置l2最新的封单额数据
    @classmethod
    def __set_l2_latest_money_record(cls, code, index, num):
        key = "l2_limit_up_money-{}".format(code)
        cls.__get_redis().setex(key, tool.get_expire(), json.dumps((num, index)))

    # 返回数量,索引
    @classmethod
    def __get_l2_latest_money_record(cls, code):
        key = "l2_limit_up_money-{}".format(code)
        result = cls.__get_redis().get(key)
        if result:
            result = json.loads(result)
            return result[0], result[1]
        else:
            return 0, -1

    # 矫正数据
    # 矫正方法为取矫正时间两侧的秒分布数据，用于确定计算结束坐标
    @classmethod
    def verify_num(cls, code, num, time_str):
        # 记录买1矫正日志
        logger_buy_1_volumn.info("涨停封单量矫正：代码-{} 量-{} 时间-{}", code, num, time_str)
        time_ = time_str.replace(":", "")
        key = None
        for i in range(4, -2, -2):
            # 获取本(分钟/小时/天)内秒分布数据
            time_regex = "{}*".format(time_[:i])
            keys_ = cls.__get_l2_second_money_record_keys(code, time_regex)
            if keys_ and len(keys_) > 1:
                # 需要排序
                keys = []
                for k in keys_:
                    keys.append(k)
                keys.sort(key=lambda tup: int(tup.split("-")[-1]))
                # 有2个元素
                for index in range(0, len(keys) - 1):
                    time_1 = keys[index].split("-")[-1]
                    time_2 = keys[index + 1].split("-")[-1]
                    if int(time_1) <= int(time_) <= int(time_2):
                        # 在此时间范围内
                        if time_ == time_2:
                            key = keys[index + 1]
                        else:
                            key = keys[index]
                        break
            if key:
                val = cls.__get_redis().get(key)
                old_num, old_from, old_to = cls.__format_second_money_record_val(val)
                end_index = old_to
                # 保存最近的数据
                cls.__set_l2_latest_money_record(code, end_index, num)
                logger_buy_1_volumn.info("涨停封单量矫正结果：代码-{} 位置-{} 量-{}", code, end_index, num)
                break

    # 计算量，用于涨停封单量的计算
    @classmethod
    def __compute_num(cls, code, data, buy_single_data):
        if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]) or L2DataUtil.is_sell(data["val"]):
            # 涨停买撤与卖
            return 0 - int(data["val"]["num"]) * data["re"]
        else:
            # 卖撤
            if L2DataUtil.is_sell_cancel(data["val"]):
                # 卖撤的买数据是否在买入信号之前，如果在之前就不计算，不在之前就计算
                if l2_data_util.is_sell_index_before_target(data, buy_single_data,
                                                            local_today_num_operate_map.get(code)):
                    return 0

            return int(data["val"]["num"]) * data["re"]

    @classmethod
    def clear(cls, code):
        key = "l2_limit_up_money-{}".format(code)
        cls.__get_redis().delete(key)

    # 返回取消的标志数据
    # with_cancel 是否需要判断是否撤销
    @classmethod
    def process_data(cls, code, start_index, end_index, buy_single_begin_index, with_cancel=True):
        start_time = round(t.time() * 1000)
        total_datas = local_today_datas[code]
        time_dict_num = {}
        # 记录计算的坐标
        time_dict_num_index = {}
        num_dict = {}
        # 统计时间分布
        time_dict = {}
        for i in range(start_index, end_index + 1):
            data = total_datas[i]
            val = data["val"]
            time_ = val["time"]
            if time_ not in time_dict:
                time_dict[time_] = i

        for i in range(start_index, end_index + 1):
            data = total_datas[i]
            val = data["val"]
            time_ = val["time"]
            if time_ not in time_dict_num:
                time_dict_num[time_] = 0
                time_dict_num_index[time_] = {"s": i, "e": i}
            time_dict_num_index[time_]["e"] = i
            num = cls.__compute_num(code, data, total_datas[buy_single_begin_index])
            num_dict[i] = num
            time_dict_num[time_] = time_dict_num[time_] + num
        for t_ in time_dict_num:
            cls.__set_l2_second_money_record(code, t_, time_dict_num[t_], time_dict_num_index[t_]["s"],
                                             time_dict_num_index[t_]["e"])

        print("保存涨停封单额时间：", round(t.time() * 1000) - start_time)

        # 累计最新的金额
        total_num, index = cls.__get_l2_latest_money_record(code)
        record_msg = f"同花顺买1信息 {total_num},{index}"

        if index == -1:
            # 没有获取到最新的矫正封单额，需要从买入信号开始点计算
            index = buy_single_begin_index - 1
            total_num = 0

        cancel_index = None
        cancel_msg = None
        # 待计算量
        limit_up_price = gpcode_manager.get_limit_up_price(code)
        min_volumn = round(10000000 / (limit_up_price * 100))
        # 不同时间的数据开始坐标
        time_start_index_dict = {}
        # 数据时间分布
        time_list = []
        # 到当前时间累积的买1量
        time_total_num_dict = {}
        for i in range(index + 1, end_index + 1):
            data = total_datas[i]
            time_ = data["val"]["time"]
            if time_ not in time_start_index_dict:
                # 记录每一秒的开始位置
                time_start_index_dict[time_] = i
                # 记录时间分布
                time_list.append(time_)
                # 上一段时间的总数
                time_total_num_dict[time_] = total_num

            val = num_dict.get(i)
            if val is None:
                val = cls.__compute_num(code, data, total_datas[buy_single_begin_index])
            total_num += val
            # 如果是减小项，且在处理数据的范围内，就需要判断是否要撤单了
            if val < 0 and start_index <= i <= end_index:
                # 累计封单金额小于1000万
                if total_num < min_volumn:
                    cancel_index = i
                    cancel_msg = "封单金额小于1000万"
                    break
                # 相邻2s内的数据减小50%
                # 上1s的总数
                last_second_total_volumn = time_total_num_dict.get(time_list[-1])
                if last_second_total_volumn > 0 and (
                        last_second_total_volumn - total_num) / last_second_total_volumn >= 0.5:
                    # 相邻2s内的数据减小50%
                    cancel_index = i
                    cancel_msg = "相邻2s({})内的封单量减小50%({}->{})".format(time_, last_second_total_volumn,
                                                                     total_num)
                    break
        if not with_cancel:
            cancel_index = None

        print("封单额计算时间：", round(t.time() * 1000) - start_time)
        process_end_index = end_index
        if cancel_index:
            process_end_index = cancel_index
        # 保存最新累计金额
        # cls.__set_l2_latest_money_record(code, process_end_index, total_num)
        l2_data_log.l2_time(code, round(t.time() * 1000) - start_time, "l2数据封单额计算时间",
                            False)
        if cancel_index:
            L2TradeDataProcessor.cancel_debug(code, "数据处理位置：{}-{}，{}，最终买1为：{}", start_index, end_index, record_msg,
                                              total_num)
            return total_datas[cancel_index], cancel_msg
        return None, None


if __name__ == "__main__":
    L2TradeDataProcessor.test()
    L2TradeDataProcessor.test2()
    print("----------------------")
    L2TradeDataProcessor.test()
    # L2TradeDataProcessor.test()

 l2_trade_factor.py

@@ -29,39 +29,51 @@
        if t < 0.9:
            return 0
        elif t <= 1.1:
            return 0.2
            return 0.7
        elif t <= 1.6:
            return 0
            return 0.4
        else:
            rate = 0
            for i in range(0, 30):
                if t <= 2.1 + 0.5 * i:
                    rate = 0.03 * (i + 1)
                    rate = 0.18 + 0.08 * i
                    break
            if rate > 0.9:
                rate = 0.9
            return rate
            return round(rate,4)

    # 获取量影响比例
    @classmethod
    def get_volumn_rate(cls, day60_max, yest, today):
        old_volumn = yest
        if day60_max > yest:
            old_volumn = day60_max
        r = round(today / old_volumn, 2)
        if day60_max is None:
            return 0
        if yest is None:
            return 0
        if today is None:
            return 0
        old_volumn = int(yest)
        if int(day60_max) > int(yest):
            old_volumn = int(day60_max)
        r = round(int(today) / old_volumn, 2)
        if r < 0.01:
            r = 0.01
        print("比例：", r)
        rate = 0
        if r < 0.5:
            rate = 0.3 - (r - 0.01)
        elif r <= 0.75:
            rate = -0.2 + (r - 0.5) * 2
        elif r <= 1.35:
            rate = 0.3 - (r - 0.75)
        if r <= 0.5:
            rate = 0.6 - (r - 0.01) * 2
        elif r <= 0.85:
            rate = -0.38 + (r - 0.5) * 2.8
        elif r <= 1.15:
            rate = 0.6 - (r - 0.85) * 4
        else:
            rate = -0.3
            rate = -0.6
        return round(rate, 4)

    @classmethod
    def get_volumn_rate_by_code(cls, code):
        volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code)
        rate = cls.get_volumn_rate(volumn_day60_max, volumn_yest, volumn_today)
        return rate

    # 当前股票首次涨停时间的影响比例
    @classmethod
@@ -147,17 +159,23 @@
        #         total_industry_limit_percent -= global_util.limit_up_codes_percent[code]
        return total_industry_limit_percent

    @classmethod
    def __get_rate_factors(cls, code):
        zyltgb = global_util.zyltgb_map.get(code)
        total_industry_limit_percent = cls.__get_industry_limit_percent(code)
        # 获取量
    @classmethod
    def __get_volumns(cls, code):
        volumn_day60_max, volumn_yest, volumn_today = global_util.max60_volumn.get(
            code), global_util.yesterday_volumn.get(code), global_util.today_volumn.get(code)
        if volumn_day60_max is None or volumn_yest is None:
            global_data_loader.load_volumn()
            volumn_day60_max, volumn_yest, volumn_today = global_util.max60_volumn.get(
                code), global_util.yesterday_volumn.get(code), global_util.today_volumn.get(code)
        return volumn_day60_max, volumn_yest, volumn_today

    @classmethod
    def __get_rate_factors(cls, code):
        zyltgb = global_util.zyltgb_map.get(code)
        total_industry_limit_percent = cls.__get_industry_limit_percent(code)
        # 获取量
        volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code)
        # 首次涨停时间
        limit_up_time = global_util.limit_up_time.get(code)
        if limit_up_time is None:
@@ -204,24 +222,21 @@
        if not gb:
            # 默认10笔
            return 8
        count = gb // 100000000
        count = gb // 100000000 - 2
        if count > 30:
            count = 30
        if count < 5:
            count = 5
        elif count < 3:
            count = 3

        big_money_num = global_util.big_money_num.get(code)
        if big_money_num is None:
            big_money_num = big_money_num_manager.get_num(code)
        rate = 0
        if big_money_num is not None:
            rate = cls.get_big_money_rate(big_money_num)

        # 获取行业热度对应的比例
        total_industry_limit_percent = cls.__get_industry_limit_percent(code)
        industry_rate = cls.get_industry_rate(total_industry_limit_percent)
        volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code)
        rate = cls.get_volumn_rate(volumn_day60_max, volumn_yest, volumn_today)
        # 取大单影响值与行业影响值的较大值
        return round(count * (1 - max(rate, industry_rate)))
        count = round(count * (1 - rate))
        if count < 3:
            count = 3
        elif count > 30:
            count = 30
        return count


# l2因子归因数据
@@ -241,14 +256,15 @@


if __name__ == "__main__":
    print(L2TradeFactorUtil.get_industry_rate(10))
    # print(L2TradeFactorUtil.get_rate_factors("003004"))
    # print(L2TradeFactorUtil.factors_to_string("003004"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("09:30:30"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("11:30:00"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("13:00:00"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("13:48:00"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("13:53:23"))
    print(L2TradeFactorUtil.get_limit_up_time_rate("14:23:23"))
    # print(L2TradeFactorUtil.get_safe_buy_count("002864"))
    # print(L2TradeFactorUtil.get_limit_up_time_rate("11:30:00"))
    # print(L2TradeFactorUtil.get_limit_up_time_rate("13:00:00"))
    # print(L2TradeFactorUtil.get_limit_up_time_rate("13:48:00"))
    # print(L2TradeFactorUtil.get_limit_up_time_rate("13:53:23"))
    # print(L2TradeFactorUtil.get_limit_up_time_rate("14:23:23"))

    # print(L2TradeFactorUtil.get_big_money_rate(2))
    # print(L2TradeFactorUtil.get_big_money_rate(3))

 log.py

@@ -64,6 +64,9 @@
        logger.add(self.get_path("system", "system"), filter=lambda record: record["extra"].get("name") == "system",
                   rotation="00:00", compression="zip", enqueue=True)

        logger.add(self.get_path("ths", "buy_1_volumn"), filter=lambda record: record["extra"].get("name") == "buy_1_volumn",
                   rotation="00:00", compression="zip", enqueue=True)

    def get_path(self, dir_name, log_name):
        return "D:/logs/gp/{}/{}".format(dir_name, log_name) + ".{time:YYYY-MM-DD}.log"

@@ -89,6 +92,8 @@
logger_code_operate = __mylogger.get_logger("code_operate")
logger_device = __mylogger.get_logger("device")
logger_system = __mylogger.get_logger("system")

logger_buy_1_volumn = __mylogger.get_logger("buy_1_volumn")


class LogUtil:
@@ -123,6 +128,18 @@
                                 "{}/l2_trade_cancel_{}.log".format(dir, date))


def __analyse_pricess_time():
    date = datetime.datetime.now().strftime("%Y-%m-%d")
    file_path = f"D:/logs/gp/l2/l2_process.{date}.log"
    with open(file_path, encoding="utf-8") as f:
        line = f.readline()
        while line:
            time_ = line.split(":")[-1]
            if int(time_) > 150:
                print(line)
            line = f.readline()


def export_l2_log(code):
    if len(code) < 6:
        return
@@ -139,5 +156,6 @@
    # logger_l2_process_time.info("test123")

    date = datetime.datetime.now().strftime("%Y-%m-%d")
    LogUtil.extract_log_from_key("002846", "D:/logs/gp/l2/l2_process_time.{}.log".format(date),
                                 "D:/logs/gp/l2/l2_process_time{}.{}.log".format("002846", date))
    LogUtil.extract_log_from_key("002383", "D:/logs/gp/l2/l2_process_time.{}.log".format(date),
                                 "D:/logs/gp/l2/l2_process_time{}.{}.log".format("002383", date))
    # __analyse_pricess_time()

 server.py

@@ -1,7 +1,7 @@
"""
接受客户端数据的服务器
"""

import datetime
import json
import logging
import socketserver
@@ -22,6 +22,7 @@
import l2_data_manager
import l2_data_manager_new
import l2_data_util
import limit_up_time_manager
import ths_industry_util
import ths_util
import tool
@@ -129,7 +130,7 @@
                                        self.l2_data_error_dict[key] = round(time.time() * 1000)

                            except Exception as e:
                                print("异常", str(e))
                                print("异常", str(e), code)
                                logging.exception(e)
                                logger_l2_error.error("出错：{}".format(str(e)))
                                logger_l2_error.error("内容：{}".format(_str))
@@ -168,6 +169,22 @@
                    if dataList:
                        global_data_loader.add_limit_up_codes(dataList)
                    ths_industry_util.set_industry_hot_num(dataList)
                    # 保存涨停时间
                    gp_list = gpcode_manager.get_gp_list()
                    gp_code_set = set(gp_list)
                    now_str = datetime.datetime.now().strftime("%H:%M:%S")
                    for d in dataList:
                        if d["time"] == "00:00:00" or tool.get_time_as_second(now_str) < tool.get_time_as_second(
                                d["time"]):
                            continue
                        if d["code"] not in gp_code_set:
                            continue

                        # 获取是否有涨停时间
                        if limit_up_time_manager.get_limit_up_time(d["code"]) is None:
                            limit_up_time_manager.save_limit_up_time(d["code"], d["time"])


                elif type == 3:
                    # 交易成功信息
                    dataList = data_process.parseList(_str)
@@ -245,7 +262,7 @@
                                l2_data_manager_new.L2TradeDataProcessor.cancel_buy(code, cancel_msg)
                            if need_sync:
                                # 同步数据
                                l2_data_manager.L2LimitUpMoneyStatisticUtil.verify_num(code, volumn, time_)
                                l2_data_manager_new.L2LimitUpMoneyStatisticUtil.verify_num(code, volumn, time_)

                elif type == 30:
                    # 心跳信息

 tool.py

@@ -47,14 +47,40 @@

    relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
    start1 = 60 * 60 * 9 + 24 * 60;
    end1 = 60 * 60 * 11 + 35 * 60;
    start2 = 60 * 60 * 12 + 50 * 60;
    end2 = 60 * 60 * 15 + 5 * 60;
    end1 = 60 * 60 * 11 + 31 * 60;
    start2 = 60 * 60 * 12 + 58 * 60;
    end2 = 60 * 60 * 15 + 1 * 60;
    if start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2:
        return True
    else:
        return False

# 是否为报警时间
def is_alert_time():
    relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
    start1 = 60 * 60 * 9 + 29 * 60
    end1 = 60 * 60 * 11 + 29 * 60
    start2 = 60 * 60 * 13
    end2 = 60 * 60 * 14 + 54 * 60
    if start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2:
        return True
    else:
        return False


# 是否为修复时间
def is_repaire_time():
    relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
    start1 = 60 * 60 * 9 + 29 * 60
    end1 = 60 * 60 * 11 + 29 * 60
    start2 = 60 * 60 * 13
    end2 = 60 * 60 * 15
    if start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2:
        return True
    else:
        return False



def is_set_code_time():
    # 测试

 trade_queue_manager.py

@@ -73,7 +73,7 @@
        if limit_up_price != tool.to_price(decimal.Decimal(price)):
            # 非涨停价
            volumn = 0
        last_volumn = self.__last_data[code]
        last_volumn = self.__last_data.get(code)
        # 不保存和上一次相同的数据
        if code in self.__last_data and last_volumn == volumn:
            return False, False, None