redis批量提交数据

Administrator

2023-08-07 e8ea6340db725a3b36a78e090fc6f97a90900264

 l2/safe_count_manager.py

@@ -11,6 +11,9 @@
from utils import tool
from l2.l2_data_util import L2DataUtil

latest_place_order_info_cache = {}
safe_count_l2_cache = {}


class BuyL2SafeCountManager(object):
    __redis_manager = redis_manager.RedisManager(0)
@@ -24,8 +27,10 @@
    # 记录每一次的处理进度
    def __save_compute_progress(self, code, last_buy_single_index, process_index, buy_num, cancel_num):
        key = "safe_count_l2-{}-{}".format(code, last_buy_single_index)
        RedisUtils.setex(self.__getRedis(), key, tool.get_expire(),
                         json.dumps((last_buy_single_index, process_index, buy_num, cancel_num)))
        tool.CodeDataCacheUtil.set_cache(safe_count_l2_cache, f"{code}-{last_buy_single_index}",
                                         (last_buy_single_index, process_index, buy_num, cancel_num))
        RedisUtils.setex_async(self.__getRedis(), key, tool.get_expire(),
                               json.dumps((last_buy_single_index, process_index, buy_num, cancel_num)))

    # 返回数据与更新时间
    def __get_compute_progress(self, code, last_buy_single_index):
@@ -36,8 +41,18 @@
        val = json.loads(val)
        return val[0], val[1], val[2], val[3]

    def __get_compute_progress_cache(self, code, last_buy_single_index):
        cache_result = tool.CodeDataCacheUtil.get_cache(safe_count_l2_cache, f"{code}-{last_buy_single_index}")
        if cache_result[0]:
            return cache_result[1]
        val = self.__get_compute_progress(code, last_buy_single_index)
        tool.CodeDataCacheUtil.set_cache(safe_count_l2_cache, f"{code}-{last_buy_single_index}", val)
        return val

    # 保存最近的下单信息
    def __save_latest_place_order_info(self, code, buy_single_index, buy_exec_index, cancel_index):
        tool.CodeDataCacheUtil.set_cache(latest_place_order_info_cache, code,
                                         (buy_single_index, buy_exec_index, cancel_index))
        key = "latest_place_order_info-{}".format(code)
        RedisUtils.setex(self.__getRedis(), key, tool.get_expire(),
                         json.dumps((buy_single_index, buy_exec_index, cancel_index)))
@@ -49,6 +64,14 @@
            return None, None, None
        val = json.loads(val)
        return val[0], val[1], val[2]

    def __get_latest_place_order_info_cache(self, code):
        cache_result = tool.CodeDataCacheUtil.get_cache(latest_place_order_info_cache, code)
        if cache_result[0]:
            return cache_result[1]
        val = self.__get_latest_place_order_info(code)
        tool.CodeDataCacheUtil.set_cache(latest_place_order_info_cache, code, val)
        return val

    def __get_all_compute_progress(self, code):
        key_regex = f"safe_count_l2-{code}-*"
@@ -66,8 +89,9 @@
        for k in keys:
            RedisUtils.delete(self.__getRedis(), k)

        tool.CodeDataCacheUtil.clear_cache(latest_place_order_info_cache, code)
        key = f"latest_place_order_info-{code}"
        RedisUtils.delete(self.__getRedis(), key)
        RedisUtils.delete_async(self.__getRedis(), key)

    # 获取基础的安全笔数
    def __get_base_save_count(self, code, is_first):
@@ -98,15 +122,15 @@
    # end_index 数据结束位置
    def compute_left_rate(self, code, start_index, end_index, total_datas,
                          local_today_num_operate_map):
        last_buy_single_index, buy_exec_index, cancel_index = self.__get_latest_place_order_info(code)
        last_buy_single_index, buy_exec_index, cancel_index = self.__get_latest_place_order_info_cache(code)
        if last_buy_single_index is None:
            return
        cancel_time = None
        if cancel_index is not None:
            cancel_time = total_datas[cancel_index]["val"]["time"]
        # 获取处理的进度
        last_buy_single_index_, process_index, buy_num, cancel_num = self.__get_compute_progress(code,
                                                                                                 last_buy_single_index)
        last_buy_single_index_, process_index, buy_num, cancel_num = self.__get_compute_progress_cache(code,
                                                                                                       last_buy_single_index)

        break_index = -1
        for i in range(start_index, end_index):