新版真实下单位寻找

Administrator

2025-01-03 f11cde17628781e516371cc039e98fa884b808e2

 third_data/kpl_data_constant.py

@@ -101,6 +101,7 @@
        # 加载为扫入买匹配的代码板块
        kpl_results = KPLLimitUpDataUtil.get_latest_block_infos(min_day=tool.date_sub(cls.__day, 365),
                                                                max_day=cls.__day)
        # {"代码":[(板块, 日期), (板块, 日期)]}
        kpl_block_dict = {}
        for r in kpl_results:
            # 当日炸板的不计算原因
@@ -109,25 +110,36 @@
            code = r[0]
            if code not in kpl_block_dict:
                kpl_block_dict[code] = []
            kpl_block_dict[code].append((r[2], r[1]))  # (板块, 时间)
            kpl_block_dict[code].append((r[2], r[1]))  # (板块, 日期)
        for code in kpl_block_dict:
            block_infos = kpl_block_dict.get(code)
            block_infos.sort(key=lambda x: x[1], reverse=True)
            temp_dict = {}  # {"板块":[出现次数, 最近出现时间]}
            for b in block_infos:
                if b[0] not in temp_dict:
                    temp_dict[b[0]] = [0, b[1]]
                temp_dict[b[0]][0] += 1
            temp_list = [(k, temp_dict[k][0], temp_dict[k][1]) for k in temp_dict]
            # 按照涨停次数与最近涨停时间排序
            temp_list.sort(key=lambda x: (x[1], x[2]), reverse=True)
            cls.__radical_buy_reasons_origin_data_dict[code] = temp_list
            blocks = {temp_list[0][0]}
            if len(temp_list) > 1:
                if temp_list[1][1] >= 2:
                    blocks.add(temp_list[1][0])
            blocks -= constant.KPL_INVALID_BLOCKS
            cls.__radical_buy_reasons_dict[code] = blocks
            cls.__radical_buy_reasons_dict[code] = cls.__compute_limit_up_reasons(code, block_infos)

    @classmethod
    def __compute_limit_up_reasons(cls, code, block_infos):
        """
        计算涨停原因
        @param code:
        @param block_infos:
        @return:
        """
        # [(板块, 日期)]
        block_infos.sort(key=lambda x: x[1], reverse=True)
        # {"板块":[(出现次数, 最近出现日期)]}
        temp_dict = {}
        for b in block_infos:
            if b[0] not in temp_dict:
                temp_dict[b[0]] = [0, b[1]]
            temp_dict[b[0]][0] += 1
        temp_list = [(k, temp_dict[k][0], temp_dict[k][1]) for k in temp_dict]
        # 按照涨停次数与最近涨停时间排序
        temp_list.sort(key=lambda x: (x[1], x[2]), reverse=True)
        cls.__radical_buy_reasons_origin_data_dict[code] = temp_list
        blocks = {temp_list[0][0]}
        # 取涨停次数最多的和最近涨停的
        blocks.add(block_infos[0][0])
        blocks -= constant.KPL_INVALID_BLOCKS
        return blocks

    def get_limit_up_reasons(self, code):
        """
@@ -239,9 +251,6 @@
                        block_codes[b] = set()
                    block_codes[b].add(code)
        cls.__current_limit_up_block_codes = block_codes




    @classmethod
    def set_history_limit_up_datas(cls, history_limit_up_datas_):