辨识度修改

Administrator

2025-03-05 ff5037d0d581fa47bb8107e948f4485b1a167984

辨识度修改

 servers/data_server.py

@@ -597,6 +597,10 @@
            fdatas = {"limit_up_list": codes_info}
            # 辨识度票
            fdatas["speical_codes"] =  [(x, gpcode_manager.get_code_name(x)) for x in special_codes]
            forbidden_refer_codes = KPLPlateForbiddenManager().get_watch_high_codes_by_block(plate)
            if forbidden_refer_codes is None:
                forbidden_refer_codes = set()
            fdatas["forbidden_refer_codes"]= [(x, gpcode_manager.get_code_name(x)) for x in forbidden_refer_codes]
            response_data = json.dumps({"code": 0, "data": fdatas})
        elif url.path == "/kpl/get_open_limit_up_count_rank":
            # 获取炸板次数排行

 third_data/code_plate_key_manager.py

@@ -370,6 +370,9 @@
    def get_watch_high_codes(self):
        return self.__watch_high_codes

    def get_watch_high_codes_by_block(self, b):
        return self.__watch_block_high_codes.get(b)

    def compute(self, code_rate_dict: dict):
        """
        根据比例计算需要拉黑的代码

 trade/buy_radical/block_special_codes_manager.py

@@ -1,10 +1,13 @@
"""
板块辨识度票管理
"""
import difflib

import constant
from db import mysql_data_delegate as mysql_data
from third_data import kpl_util
from third_data.history_k_data_util import HistoryKDatasUtils
from third_data.kpl_data_constant import LimitUpCodesBlockRecordManager
from utils import tool


@@ -61,7 +64,7 @@

    def __list_code_blocks(self, min_day, max_day):
        results = self.__mysql.select_all(
            f"SELECT r.`_hot_block_name`, r.`_code`, r.`_day` FROM kpl_limit_up_record  r WHERE r.`_day`>'{min_day}' and r.`_day`<'{max_day}' and r._code not like '68%'")
            f"SELECT r.`_hot_block_name`, r.`_code`, r.`_day`, r.`_open` FROM kpl_limit_up_record  r WHERE r.`_day`>'{min_day}' and r.`_day`<'{max_day}' and r._code not like '68%'")
        return results

    def __get_limit_up_info(self, min_day):
@@ -127,19 +130,80 @@
        获取板块有辨识度的代码
        @return:
        """

        def statistic_continue_target_codes():
            """
            统计连续涨停的辨识度代码代码
            @return:
            """
            dates = HistoryKDatasUtils.get_latest_trading_date_cache(120)
            code_days_map = {}
            for d in code_block_infos:
                if d[3]:
                    # 不看炸板
                    continue
                if d[1] not in code_days_map:
                    code_days_map[d[1]] = set()

                code_days_map[d[1]].add(d[2])
            target_codes_info = set()
            for code in code_days_map:
                # if code != '002400':
                #     continue
                # 且有3天属于连续涨停
                day_list = list(code_days_map[code])
                day_list.sort(reverse=True)
                step = 4
                matcher = difflib.SequenceMatcher(None, dates, day_list)
                match = matcher.find_longest_match(0, len(dates), 0, len(day_list))
                if match.size < step:
                    continue
                target_codes_info.add((code, match.size))
            print(f"大于4板：{target_codes_info}")

            continue_count_dict = {x[0]: x[1] for x in target_codes_info}

            # 统计板块的代码
            block_codes_2 = {}
            for code_info in target_codes_info:
                code = code_info[0]
                bs = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code)
                if bs:
                    for bb in bs:
                        if bb not in block_codes_2:
                            block_codes_2[bb] = set()
                        block_codes_2[bb].add(code)
            for k in block_codes_2:
                block_codes_2[k] = list(block_codes_2[k])
                block_codes_2[k].sort(key=lambda x: continue_count_dict[x], reverse=True)
                block_codes_2[k] = block_codes_2[k][:3]

            return block_codes_2

        trading_dates = HistoryKDatasUtils.get_latest_trading_date(9)
        max_day = trading_dates[0]
        min_day = tool.date_sub(max_day, 180)

        block_map = self.__get_block_map()
        # 统计最近180天涨停数据
        # [(板块名称,代码, 日期)]
        # [(板块名称,代码, 日期, 是否炸板)]
        code_block_infos = self.__list_code_blocks(min_day, max_day)

        # 统计代码的涨停次数
        limit_up_count_dict = {}
        for d in code_block_infos:
            if d[1] not in limit_up_count_dict:
                limit_up_count_dict[d[1]] = set()
            limit_up_count_dict[d[1]].add(d[2])
        limit_up_count_dict = {c: len(limit_up_count_dict[c]) for c in limit_up_count_dict}

        # 拥有连板高度的辨识度：{"板块":[代码1,代码2]}
        continue_target_block_codes = statistic_continue_target_codes()

        # 统计最近8天的涨停数据
        min_day = trading_dates[-1]
        code_block_infos_8 = self.__get_code_blocks(min_day, max_day)
        # 统计涨停次数大于3次的涨停代码
        # 统计最近8天涨停次数大于3次的涨停代码
        # 格式：{"代码":涨停次数}
        count_dict = {}
        filter_codes = set()
@@ -189,10 +253,11 @@
                    block_codes_dict[b] = []
                block_codes_dict[b].append((code, code_block_dict[code][b]))

        # 统计最近180天的涨停数据 {"代码": (涨停次数, 名称, 自由流通市值)}
        limit_up_info_map = self.__get_limit_up_info(min_day)
        fdatas = []
        for b in block_codes_dict:
            # if b != '轨道交通':
            # if b != '机器人':
            #     continue

            if b in constant.KPL_INVALID_BLOCKS:
@@ -205,7 +270,7 @@
            zylt_list = code_info_list[:max_count]
            # 取涨停次数最多的
            temp_codes = [x[0] for x in zylt_list]
            code_info_list.sort(key=lambda x: x[1], reverse=True)
            code_info_list.sort(key=lambda x: limit_up_count_dict[x[0]], reverse=True)
            limit_up_count_max = code_info_list[:2]
            rank2_codes = set()
            for d in limit_up_count_max:
@@ -214,10 +279,22 @@
                    if d[0] not in temp_codes:
                        zylt_list.append(d)
                    rank2_codes.add(d[0])
            # 按照涨停次数排序
            zylt_list.sort(key=lambda x: x[1], reverse=True)

            # 添加连板辨识度
            continue_codes = continue_target_block_codes.get(b)
            if continue_codes:
                for c in continue_codes:
                    if int(float(limit_up_info_map[c][2])) < 50:
                        continue
                    zylt_list.append((c, limit_up_count_dict.get(c)))

            # 按照股票实际涨停次数排序
            zylt_list.sort(key=lambda x: limit_up_count_dict[x[0]], reverse=True)
            zylt_list = [x[0] for x in zylt_list]
            index = 0



            # 获取股价，是否是ST
            if not zylt_list:
                continue
@@ -226,8 +303,12 @@
                continue

            juejin_result_dict = {x['sec_id']: (x['sec_id'], x['sec_level'], x['upper_limit']) for x in juejin_results}
            codes = set()
            for code in zylt_list:
                if code_block_dict[code][b] <= 3:
                if code in codes:
                    continue

                if limit_up_count_dict[code] <= 3:
                    # 累计涨停次数小于3次
                    continue

@@ -248,8 +329,9 @@
                    continue
                # [(板块, 代码名称, 代码, 涨停次数, 自由市值)]
                fdatas.append(
                    (kpl_util.filter_block(b), limit_up_info_map[code][1], code, code_block_dict[code][b],
                     int(float(limit_up_info_map[code][2]))))
                    (kpl_util.filter_block(b), limit_up_info_map[code][1], code, limit_up_count_dict[code],
                     int(float(limit_up_info_map[code][2])), limit_up_count_dict.get(code)))
                codes.add(code)
                if index >= 10:
                    break
        # BlockSpecialCodesManager().set_block_codes_list(fdatas)
@@ -267,5 +349,8 @@
if __name__ == "__main__":
    # print(datas)
    datas = AnalysisBlockSpecialCodesManager().get_block_special_codes()
    print(datas)
    # print(datas)
    for d in datas:
        print(d)
    BlockSpecialCodesManager().set_block_codes_list(datas)
    # print(BlockSpecialCodesManager().get_code_blocks("002582"))