拆单初步规划/禁止交易板块管理

Administrator

2025-01-02 d957e0dc6dcfb15304a6499e1b9ddee4e18be531

 third_data/code_plate_key_manager.py

@@ -222,7 +222,10 @@

# 开盘啦禁止交易板块管理
class KPLPlateForbiddenManager:
    __redisManager = redis_manager.RedisManager(3)
    """
    不能买的板块管理
    """
    __redis_manager = redis_manager.RedisManager(3)
    __kpl_forbidden_plates_cache = set()

    __instance = None
@@ -237,17 +240,22 @@
    def __load_datas(cls):
        __redis = cls.__get_redis()
        try:
            __kpl_forbidden_plates_cache = RedisUtils.smembers(__redis, "kpl_forbidden_plates")
            cls.__kpl_forbidden_plates_cache = RedisUtils.smembers(__redis, "kpl_forbidden_plates")
        finally:
            RedisUtils.realse(__redis)

    @classmethod
    def __get_redis(cls):
        return cls.__redisManager.getRedis()
        return cls.__redis_manager.getRedis()

    def save_plate(self, plate):
        self.__kpl_forbidden_plates_cache.add(plate)
        RedisUtils.sadd(self.__get_redis(), "kpl_forbidden_plates", plate)
        RedisUtils.expire(self.__get_redis(), "kpl_forbidden_plates", tool.get_expire())

    def delete_plate(self, plate):
        self.__kpl_forbidden_plates_cache.discard(plate)
        RedisUtils.srem(self.__get_redis(), "kpl_forbidden_plates", plate)
        RedisUtils.expire(self.__get_redis(), "kpl_forbidden_plates", tool.get_expire())

    def list_all(self):
@@ -255,6 +263,11 @@

    def list_all_cache(self):
        return self.__kpl_forbidden_plates_cache

    def is_in_cache(self, plate):
        if self.__kpl_forbidden_plates_cache and plate in self.__kpl_forbidden_plates_cache:
            return True
        return False


class LimitUpCodesPlateKeyManager:
@@ -352,19 +365,38 @@
    __top_jx_out_blocks = []
    # 精选板块流入金额
    __jx_blocks_in_money_dict = {}
    # 市场行情热度，默认为60
    __market_strong = 60

    @classmethod
    def get_jingxuan_in_block_threshold_count(cls):
        """
        获取买精选流入前几
        @return:
        """
        score = 60
        if cls.__market_strong is not None:
            score = int(cls.__market_strong)
        for info in constant.RADICAL_BUY_TOP_IN_COUNT_BY_MARKET_STRONG:
            if info[0] <= score < info[1]:
                return info[2]
        return 10

    @classmethod
    def set_market_jingxuan_blocks(cls, datas):
        """
        设置精选流入数据
        @param datas:
        @param datas:[(板块编号,板块名称,涨幅, 板块流入金额)]
        @return:
        """
        # 流入阈值
        THRESHOLD_MONEY = 100 * (tool.trade_time_sub(tool.get_now_time_str(), "09:30:00") // 60) + 1000
        THRESHOLD_MONEY = min(THRESHOLD_MONEY, 10000)
        THRESHOLD_MONEY = max(THRESHOLD_MONEY, 1000)
        THRESHOLD_MONEY = THRESHOLD_MONEY * 10000
        # THRESHOLD_MONEY = 50 * (tool.trade_time_sub(tool.get_now_time_str(), "09:30:00") // 60) + 1000
        # THRESHOLD_MONEY = min(THRESHOLD_MONEY, 10000)
        # THRESHOLD_MONEY = max(THRESHOLD_MONEY, 1000)
        # THRESHOLD_MONEY = THRESHOLD_MONEY * 10000
        THRESHOLD_MONEY = 0
        # 最大数量
        # MAX_COUNT = cls.get_jingxuan_in_block_threshold_count()

        cls.top_in_list_cache = datas
        blocks = set()
@@ -381,7 +413,6 @@
            if blocks & fb:
                continue


            for b in fb:
                fblock_money[b] = data[3]
            blocks |= fb
@@ -394,11 +425,15 @@
                    break
            if has_code:
                count += 1

            if count >= 10:
                break
                if count == 10:
                    strong = cls.get_market_strong()
                    if strong is None:
                        strong = 60
                    THRESHOLD_MONEY = int((1 - strong / 200) * data[3])
            # if count >= MAX_COUNT:
            #     break
        # 记录精选流出日志
        async_log_util.info(logger_kpl_jx_in, f"原数据：{datas[:20]} 板块：{blocks}")
        async_log_util.info(logger_kpl_jx_in, f"原数据：{datas[:50]} 板块：{blocks}")
        blocks = list(blocks)
        blocks.sort(key=lambda x: fblock_money.get(x), reverse=True)
        cls.__top_jx_blocks = blocks
@@ -433,12 +468,37 @@
        cls.__top_jx_out_blocks = list(blocks)

    @classmethod
    def set_market_strong(cls, strong):
        """
        设置市场行情强度
        @param strong:
        @return:
        """
        cls.__market_strong = strong

    @classmethod
    def get_market_strong(cls):
        return cls.__market_strong

    @classmethod
    def get_top_market_jingxuan_blocks(cls):
        return cls.__top_jx_blocks

    @classmethod
    def get_top_market_jingxuan_out_blocks(cls):
        return cls.__top_jx_out_blocks

    @classmethod
    def get_block_info_at_block_in(cls, b):
        """
        获取板块的净流入情况
        @param b:
        @return: (板块名称,身位,流入金额)
        """
        for i in range(0, len(cls.top_in_list_cache)):
            if cls.top_in_list_cache[i][1] == b:
                return b, i, cls.top_in_list_cache[i][3]
        return b, -1, 0

    @classmethod
    def set_top_5_industry(cls, datas):
@@ -1008,7 +1068,8 @@
    @classmethod
    def __compute_can_buy_blocks(cls, code, current_limit_up_datas, limit_up_record_datas,
                                 yesterday_current_limit_up_codes, before_blocks_dict,
                                 current_limit_up_block_codes_dict, high_level_general_code_blocks, codes_delegate, codes_success):
                                 current_limit_up_block_codes_dict, high_level_general_code_blocks, codes_delegate,
                                 codes_success):
        # 根据代码泛化板块获取泛化板块的代码集合
        high_level_general_block_codes = {}
        for c in high_level_general_code_blocks:
@@ -1137,11 +1198,14 @@
                                                                                                                   yesterday_current_limit_up_codes,
                                                                                                                   before_blocks_dict,
                                                                                                                   current_limit_up_block_codes_dict,
                                                                                                                   high_level_general_code_blocks, delegate_codes, deal_codes)
                                                                                                                   high_level_general_code_blocks,
                                                                                                                   delegate_codes,
                                                                                                                   deal_codes)
        # 保存板块计算结果
        cls.__can_buy_compute_result_dict[code] = (
            can_buy_blocks, unique, msg, can_buy_strong_blocks, keys, active_buy_blocks)


if __name__ == "__main__":
    pass
    RealTimeKplMarketData.set_market_strong(120)
    print(RealTimeKplMarketData.get_jingxuan_in_block_threshold_count())