扫入板块调整

Administrator

2024-09-30 950c4677c35738d9c77ac449f836e5785fc5bf69

扫入板块调整

 api/outside_api_command_callback.py

@@ -35,7 +35,8 @@
    logger_real_place_order_position, logger_device
from output import l2_output_util
from third_data import kpl_util, history_k_data_manager, huaxin_l1_data_manager, third_blocks_manager, kpl_data_manager
from third_data.code_plate_key_manager import CodePlateKeyBuyManager, KPLCodeJXBlockManager, RealTimeKplMarketData
from third_data.code_plate_key_manager import CodePlateKeyBuyManager, KPLCodeJXBlockManager, RealTimeKplMarketData, \
    RadicalBuyBlockManager
from third_data.history_k_data_manager import HistoryKDataManager
from third_data.history_k_data_util import JueJinApi, HistoryKDatasUtils
from third_data.kpl_data_manager import KPLDataManager
@@ -1234,16 +1235,14 @@
                    kpl_blocks = set()
                if not kpl_blocks:
                    KPLCodeJXBlockManager().load_jx_blocks_radical(code)
                match_blocks, info = CodeThirdBlocksManager().get_intersection_blocks_info(code, kpl_blocks)
                if match_blocks:
                    match_blocks = set(match_blocks)
                    match_blocks &= RealTimeKplMarketData.get_top_market_jingxuan_blocks()

                filter_blocks, match_blocks = RadicalBuyBlockManager.get_code_blocks(code)
                source_origin_dict[SOURCE_TYPE_KPL] = kpl_blocks
                source_dict[SOURCE_TYPE_KPL] = BlockMapManager().filter_blocks(kpl_blocks)
                data = {
                    "blocks": {},
                    "origin_blocks": {},
                    "match_blocks": list(match_blocks)
                    "match_blocks": [list(filter_blocks), list(match_blocks)]
                }
                for s in source_origin_dict:
                    data["origin_blocks"][s] = list(source_origin_dict[s])

 third_data/code_plate_key_manager.py

@@ -1422,6 +1422,25 @@
        return True, ""

    @classmethod
    def get_code_blocks(cls, code):
        """
        获取目标代码的板块
        @param code:
        @return: 过滤后的板块,过滤前的板块
        """
        blocks = KPLCodeJXBlockManager().get_jx_blocks_radical(code)
        if blocks is None:
            blocks = set()
        # 将获取涨停原因与涨停推荐
        keys = TargetCodePlateKeyManager().get_plate_keys(code, contains_today=False)
        if keys and keys[0]:
            blocks |= set(keys[0])
        match_blocks, info = CodeThirdBlocksManager().get_intersection_blocks_info(code, blocks)
        match_blocks -= constant.KPL_INVALID_BLOCKS
        fblocks = match_blocks & RealTimeKplMarketData.get_top_market_jingxuan_blocks()
        return fblocks, match_blocks

    @classmethod
    def is_radical_buy(cls, code, yesterday_limit_up_codes):
        """
        是否是激进买
@@ -1436,15 +1455,7 @@
            for c in open_limit_up_code_dict:
                open_limit_up_blocks |= open_limit_up_code_dict[c][1]
        # 获取代码的板块
        keys_ = KPLCodeJXBlockManager().get_jx_blocks_radical(code)
        keys_ &= RealTimeKplMarketData.get_top_market_jingxuan_blocks()
        if not keys_:
            return set(), "没获取到板块"
        keys_ = BlockMapManager().filter_blocks(keys_)
        if not keys_:
            return set(), "过滤后没获取到板块"
        # 获取交集
        keys_, info = CodeThirdBlocksManager().get_intersection_blocks_info(code, keys_)
        keys_, info = cls.get_code_blocks(code)
        if not keys_:
            return set(), "没获取到板块交集"


 third_data/kpl_data_manager.py

@@ -530,7 +530,7 @@

        def get_market_jingxuan():
            while True:
                if tool.is_trade_time():
                if tool.is_trade_time() or True:
                    try:
                        results = kpl_api.getMarketJingXuanRealRankingInfo()
                        result = json.loads(results)

 third_data/third_blocks_manager.py

@@ -191,10 +191,10 @@
        for block in blocks:
            if block.endswith("概念"):
                block = block[:-2]
            # b = self.get_map_blocks_cache(block)
            # if b:
            #     fbs |= b
            fbs.add(block)
            b = self.get_map_blocks_cache(block)
            if b:
                fbs |= b
            # fbs.add(block)
        return fbs

    def get_all_blocks(self):