代码异常保护

Administrator

2024-11-04 65afea1ba534b51f947cbe7989d7f4d650bbc9e6

 third_data/kpl_limit_up_data_manager.py

@@ -35,6 +35,77 @@
    return kpl_data_manager.KPLLimitUpDataRecordManager.total_datas


class CodeLimitUpSequenceManager:
    """
    代码身位管理
    """
    # 首板身位
    __first_block_sequence_dict = {}

    @classmethod
    def set_current_limit_up_datas(cls, current_limit_up_datas):
        """
        设置目前的涨停代码
        @param current_limit_up_datas:
        @return:
        """
        records = get_today_history_limit_up_datas_cache()
        # 按代码排序
        # {"代码":（代码，涨停原因, 涨停时间, 几版）}
        current_code_block_dict = {x[0]: (x[0], x[5], x[2], x[4]) for x in current_limit_up_datas}
        record_code_block_dict = {x[3]: (x[3], x[2], x[5], x[12]) for x in records}
        # 根据涨停原因统计
        # {"板块":{代码}}
        block_codes = {}
        limit_up_codes = set()
        for code in current_code_block_dict:
            b = current_code_block_dict[code][1]
            if b not in block_codes:
                block_codes[b] = set()
            block_codes[b].add(code)
            limit_up_codes.add(code)
        for code in record_code_block_dict:
            b = record_code_block_dict[code][1]
            if b not in block_codes:
                block_codes[b] = set()
            block_codes[b].add(code)
        # 获取上个交易日涨停的代码
        yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
        if yesterday_codes is None:
            yesterday_codes = set()

        for code in limit_up_codes:
            # 计算身位
            b = current_code_block_dict[code][1]
            codes = block_codes[b]
            total_count = len(codes)
            # 统计真正涨停数
            limit_up_count = 0
            limit_up_codes_list = []
            for c in codes:
                if c in limit_up_codes:
                    limit_up_count += 1
                    if c not in yesterday_codes:
                        limit_up_codes_list.append((c, current_code_block_dict[c][2]))
            # 获取首板代码的排位
            limit_up_codes_list.sort(key=lambda x: x[1])
            index = 1
            for i in range(0, len(limit_up_codes_list)):
                if limit_up_codes_list[i][0] == code:
                    index = i + 1
                    break
            cls.__first_block_sequence_dict[code] = (b, index, total_count, limit_up_count)

    @classmethod
    def get_current_limit_up_sequence(cls, code):
        """
        获取代码当前的板块身位
        @param code:
        @return:（板块名称,身位,总涨停数量，目前涨停数量）
        """
        return cls.__first_block_sequence_dict.get(code)


class LatestLimitUpBlockManager:
    """
    最近涨停的板块管理
@@ -157,7 +228,7 @@

        block_count_list = [(k, block_count_dict[k]) for k in block_count_dict]
        block_count_list.sort(key=lambda x: x[1], reverse=True)
        block_count_list = block_count_list[:20]
        block_count_list = block_count_list[:50]
        # [(涨停原因,累计涨停次数,连续次数)]
        fdatas = []
        today_records_code_dict = {d[3]: d for d in self.__history_limit_up_day_datas.get(now_day)}
@@ -184,7 +255,7 @@
            max_rate_info = None
            for code in block_codes_dict[d[0]]:
                if max_rate_info is None:
                    max_rate_info = (code, self.__k_max_rate.get(code),  self.__code_name_dict.get(code))
                    max_rate_info = (code, self.__k_max_rate.get(code), self.__code_name_dict.get(code))
                if max_rate_info[1] < self.__k_max_rate.get(code):
                    max_rate_info = (code, self.__k_max_rate.get(code), self.__code_name_dict.get(code))
            fdata.append(max_rate_info)
@@ -234,5 +305,3 @@
        rate = int((volumes_data[0]["close"] - min_price) * 100 / min_price)
        cls.__k_max_rate[code] = rate
        return cls.__k_datas.get(code)