redis批量提交数据

Administrator

2023-08-07 8e163037d6334645ac0ab88f3a2f70841ce4ec55

 l2/code_price_manager.py

@@ -51,6 +51,20 @@
        cls.__current_buy_1_price[code] = buy_1_price
        RedisUtils.setex(cls.__get_redis(), f"buy1_price-{code}", tool.get_expire(), buy_1_price)

    # datas:[(code, buy_1_price)]
    @classmethod
    def __save_buy1_prices(cls, datas):
        pipe = cls.__get_redis().pipeline()
        for d in datas:
            code = d[0]
            buy_1_price = d[1]
            # 不保存重复的数据
            if code in cls.__current_buy_1_price and cls.__current_buy_1_price[code] == buy_1_price:
                continue
            cls.__current_buy_1_price[code] = buy_1_price
            RedisUtils.setex(pipe, f"buy1_price-{code}", tool.get_expire(), buy_1_price)
        pipe.execute()

    @classmethod
    def __get_buy1_price(cls, code):
        return RedisUtils.get(cls.__get_redis(), f"buy1_price-{code}")
@@ -113,6 +127,43 @@
            # 之前涨停过且现在尚未涨停
            cls.set_open_limit_up_lowest_price(code, buy_1_price)

    # datas:[ (code, buy_1_price, time_str, limit_up_price, sell_1_price, sell_1_volumn)  ]
    @classmethod
    def processes(cls, datas):
        temp_buy1_prices = []
        for d in datas:
            code, buy_1_price, time_str, limit_up_price, sell_1_price, sell_1_volumn = d
            data_str = f"{buy_1_price},{time_str},{limit_up_price},{sell_1_price},{sell_1_volumn}"
            if cls.__latest_data.get(code) == data_str:
                continue
            cls.__latest_data[code] = data_str
            # 保存买1价格
            temp_buy1_prices.append((code, buy_1_price))

            # 记录日志
            logger_trade_queue_price_info.info(
                f"code={code} data: time_str-{time_str}, buy_1_price-{buy_1_price},limit_up_price-{limit_up_price},sell_1_price-{sell_1_price},sell_1_volumn-{sell_1_volumn}")
            # 买1价格不能小于1块
            if float(buy_1_price) < 1.0:
                continue

            is_limit_up = abs(float(limit_up_price) - float(buy_1_price)) < 0.01
            old_limit_up_time, old_open_limit_up_time = cls.__get_buy1_price_info_cache(code)
            if old_limit_up_time and old_open_limit_up_time:
                continue
            if is_limit_up and old_limit_up_time is None and float(sell_1_price) < 0.1 and int(sell_1_volumn) <= 0:
                # 卖1消失，买1为涨停价则表示涨停
                cls.__save_buy1_price_info(code, time_str, None)
            elif old_limit_up_time and not is_limit_up and old_open_limit_up_time is None:
                # 有涨停时间，当前没有涨停，之前没有打开涨停
                cls.__save_buy1_price_info(code, old_limit_up_time, time_str)

            if old_limit_up_time and not is_limit_up:
                # 之前涨停过且现在尚未涨停
                cls.set_open_limit_up_lowest_price(code, buy_1_price)
        if temp_buy1_prices:
            cls.__save_buy1_prices(temp_buy1_prices)

    # 是否可以下单
    @classmethod
    def is_can_buy(cls, code):