redis批量提交数据

Administrator

2023-08-07 20a70116b236a49d68659b451fea0a9f645a0835

 trade/trade_data_manager.py

@@ -236,6 +236,19 @@
        key = "code_current_rate-{}".format(code)
        RedisUtils.setex(self.__get_redis(), key, tool.get_expire(), rate)

    # 批量保存
    def __save_current_rates(self, datas):
        # 变化之后才会持久化
        pipe = self.__get_redis().pipeline()
        for d in datas:
            if self.__code_current_rate_latest.get(d[0]) == d[1]:
                continue
            self.__code_current_rate_latest[d[0]] = d[1]
            tool.CodeDataCacheUtil.set_cache(self.__code_current_rate_cache, d[0], d[1])
            key = "code_current_rate-{}".format(d[0])
            RedisUtils.setex(pipe, key, tool.get_expire(), d[1])
        pipe.execute()

    # 获取当前涨幅
    def __get_current_rate(self, code):
        key = "code_current_rate-{}".format(code)
@@ -274,6 +287,32 @@
            if self.__get_last_down_price_time_cache(code) is None:
                self.__save_down_price_time(code, time_str)

    # datas:[(代码,比例)]
    def process_rates(self, datas, time_str):
        # 9点半之前的数据不处理
        if int(time_str.replace(":", "")) < int("093000"):
            return
        # 保存目前的代码涨幅
        self.__save_current_rates(datas)

        # now_str = tool.get_now_time_str()
        for d in datas:
            code, rate = d[0], d[1]
            if rate >= 0:
                down_start_time = self.__get_last_down_price_time_cache(code)
                if down_start_time is None:
                    continue
                else:
                    # 累计增加时间
                    time_second = tool.trade_time_sub(time_str, down_start_time)
                    self.__increment_down_price_time(code, time_second)
                    # 删除起始时间
                    self.__remove_down_price_time(code)
            else:
                # 记录开始值
                if self.__get_last_down_price_time_cache(code) is None:
                    self.__save_down_price_time(code, time_str)

    # 保存现价
    def save_current_price(self, code, price, is_limit_up):
        global_util.cuurent_prices[code] = (price, is_limit_up, round(time.time()))