扫入参考代码板块优化

Administrator

2024-11-04 8ffb521f738254038d85bbade839b2e30456e525

 third_data/kpl_data_constant.py

@@ -1,9 +1,119 @@
# 当前涨停数据
import constant
from third_data import kpl_block_util
from db import redis_manager_delegate as redis_manager
from db.redis_manager_delegate import RedisUtils

# 用于计算激进买开1的板数：{"代码":(几版,{板块})}
from utils import tool
from utils.kpl_data_db_util import KPLLimitUpDataUtil

open_limit_up_code_dict_for_radical_buy = None


class LimitUpCodesBlockRecordManager:
    """
    历史涨停代码的板块管理
    """
    # 涨停原因
    __limit_up_reasons_dict = {}
    # 涨停推荐原因
    __limit_up_recommend_reasons_dict = {}

    __instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(LimitUpCodesBlockRecordManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_data()
        return cls.__instance

    @classmethod
    def __load_data(cls):
        kpl_results = KPLLimitUpDataUtil.get_latest_block_infos()
        limit_up_reasons_dict = {}
        limit_up_recommend_block_dict = {}
        for r in kpl_results:
            code = r[0]
            if code not in limit_up_reasons_dict:
                limit_up_reasons_dict[code] = []
            if code not in limit_up_recommend_block_dict:
                limit_up_recommend_block_dict[code] = []
            if len(limit_up_reasons_dict[code]) >= 2:
                continue
            limit_up_reasons_dict[code].append(r[2])
            if r[3]:
                limit_up_recommend_block_dict[code].extend(r[3].split("、"))
        for code in limit_up_reasons_dict:
            cls.__limit_up_reasons_dict[code] = set(limit_up_reasons_dict[code])

        for code in limit_up_recommend_block_dict:
            cls.__limit_up_recommend_reasons_dict[code] = set(limit_up_recommend_block_dict[code])

    def get_limit_up_reasons(self, code):
        """
        获取涨停原因
        @param code:
        @return:
        """
        if code in self.__limit_up_reasons_dict:
            return set(self.__limit_up_reasons_dict[code])
        return set()

    def get_limit_up_reasons_with_recommend(self, code):
        """
        获取涨停原因与推荐原因
        @param code:
        @return:
        """
        blocks = set()
        if code in self.__limit_up_reasons_dict:
            blocks |= self.__limit_up_reasons_dict[code]
        if code in self.__limit_up_recommend_reasons_dict:
            blocks |= self.__limit_up_recommend_reasons_dict[code]
        return blocks


class TodayLimitUpReasonChangeManager:
    """
    今日涨停原因变化
    """
    # 涨停原因
    __today_change_reasons_dict = {}
    # 涨停
    __instance = None
    __db = 1
    __redisManager = redis_manager.RedisManager(1)

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(TodayLimitUpReasonChangeManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_data()
        return cls.__instance

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

    @classmethod
    def __load_data(cls):
        keys = RedisUtils.keys(cls.__get_redis(), "kpl_limit_up_reason_his-*")
        if keys:
            for key in keys:
                code = key.split("-")[1]
                reasons = RedisUtils.smembers(cls.__get_redis(), key)
                cls.__today_change_reasons_dict[code] = reasons

    def set_today_limit_up_reason_change(self, code, from_reason, to_reason):
        if code not in self.__today_change_reasons_dict:
            self.__today_change_reasons_dict[code] = set()
        self.__today_change_reasons_dict[code].add(from_reason)
        RedisUtils.sadd_async(self.__db, f"kpl_limit_up_reason_his-{code}", from_reason)
        RedisUtils.expire_async(self.__db, f"kpl_limit_up_reason_his-{code}", tool.get_expire())

    def get_changed_reasons(self, code):
        if code in self.__today_change_reasons_dict:
            return self.__today_change_reasons_dict[code]
        return set()


class LimitUpDataConstant:
@@ -25,14 +135,22 @@
            return
        cls.history_limit_up_datas = history_limit_up_datas_
        for d in cls.history_limit_up_datas:
            # 参考原因：当前涨停原因+当日变化之前的原因+180天最近2条涨停原因
            code = d[3]
            blocks = {d[2]}
            history_reasons = LimitUpCodesBlockRecordManager().get_limit_up_reasons(code)
            if history_reasons:
                blocks |= history_reasons
            today_changed_reasons = TodayLimitUpReasonChangeManager().get_changed_reasons(code)
            if today_changed_reasons:
                blocks |= today_changed_reasons
            # 开1才能包含推荐原因
            if d[6] and kpl_block_util.open_limit_up_time_range[0] <= int(d[5]) < \
                    kpl_block_util.open_limit_up_time_range[1]:
                blocks |= set(d[6].split("、"))
            # if d[6] and kpl_block_util.open_limit_up_time_range[0] <= int(d[5]) < \
            #         kpl_block_util.open_limit_up_time_range[1]:
            #     blocks |= set(d[6].split("、"))
            blocks -= constant.KPL_INVALID_BLOCKS
            cls.__history_code_blocks_dict[d[3]] = blocks
            cls.__history_code_data_dict[d[3]] = d
            cls.__history_code_blocks_dict[code] = blocks
            cls.__history_code_data_dict[code] = d

    @classmethod
    def get_blocks_with_history(cls, code):