扫入策略修改

Administrator

2024-11-13 d2d5ca80907183f88a5e78aa28c085a746868d6d

扫入策略修改

 l2/l2_data_manager_new.py

@@ -1608,6 +1608,9 @@
                radical_buy_data_manager.BlockPlaceOrderRecordManager().add_record(code, radical_result[2])
            return

        if RadicalBuyDealCodesManager.radical_buy_blocks_dict.get(code):
            return

        # 获取买入信号计算起始位置
        order_begin_pos = cls.__get_order_begin_pos(code)

@@ -2091,7 +2094,8 @@
                    else:
                        # 判断该板块前排是否已经有成交
                        deal_codes = RadicalBuyDealCodesManager().get_deal_codes()
                        buy_blocks = radical_buy_data_manager.is_block_can_radical_buy(code, radical_data[2], deal_codes)
                        buy_blocks = radical_buy_data_manager.is_block_can_radical_buy(code, radical_data[2],
                                                                                       deal_codes)
                        if not buy_blocks:
                            return False, result[1], f"板块代码已有成交：{radical_data[2]}"
            return True, result[1], radical_data[2]

 servers/huaxin_trade_server.py

@@ -994,7 +994,7 @@
def __init():
    def run_pending():
        schedule.every().day.at("15:10:00").do(zyltgb_util.update_all_zylt_volumes)
        schedule.every().day.at("09:10:00").do(__test_pre_place_order)
        schedule.every().day.at("01:01:00").do(__test_pre_place_order)
        schedule.every().day.at("09:10:00").do(__subscript_fixed_codes_l2)
        schedule.every().day.at("08:00:01").do(history_k_data_manager.update_history_k_bars)
        schedule.every().day.at("08:30:01").do(history_k_data_manager.update_history_k_bars)

 third_data/code_plate_key_manager.py

@@ -353,12 +353,25 @@
        @param datas:
        @return:
        """
        # blocks = set()
        # 老版本实现方式
        # for data in datas:
        #     if data[3] <= 0:
        #         break
        #     blocks.add(data[1])
        # cls.__top_jx_blocks = blocks
        blocks = set()
        for data in datas:
            if data[3] <= 0:
                break
            if data[1] in constant.KPL_INVALID_BLOCKS:
                continue
            if data[3] < 1e8:
                continue
            blocks.add(data[1])
        cls.__top_jx_blocks = blocks
            if len(blocks) >= 10:
                break
            blocks.add(kpl_util.filter_block(data[1]))
        # 记录精选流出日志
        cls.__top_jx_blocks = BlockMapManager().filter_blocks(blocks)

    @classmethod
    def set_market_jingxuan_out_blocks(cls, datas):
@@ -368,19 +381,19 @@
        @return:
        """
        blocks = set()
        for i in range(0, len(datas)):
            if i >= 10 and int(tool.get_now_time_str().replace(":", "")) < int("100000"):
                # 10点前看前10，十点后不看前10
                break
            data = datas[i]
            if data[3] > 0 - 5e7:
        for data in datas:
            if data[1] in constant.KPL_INVALID_BLOCKS:
                continue
            if data[3] > -1e8:
                # 过滤5千万以上的
                break
            blocks.add(data[1])
            if len(blocks) >= 10:
                break
            blocks.add(kpl_util.filter_block(data[1]))

        # 记录精选流出日志
        async_log_util.info(logger_kpl_jx_out, f"原数据：{datas[:10]} 板块：{blocks}")
        cls.__top_jx_out_blocks = blocks
        cls.__top_jx_out_blocks = BlockMapManager().filter_blocks(blocks)

    @classmethod
    def get_top_market_jingxuan_blocks(cls):
@@ -1103,5 +1116,6 @@
        cls.__can_buy_compute_result_dict[code] = (
            can_buy_blocks, unique, msg, can_buy_strong_blocks, keys, active_buy_blocks)


if __name__ == "__main__":
    pass

 third_data/kpl_api.py

@@ -125,6 +125,13 @@
    return result


def getMarketJingXuanRealRankingInfoByTimeRange(startTime, endTime, date, orderJingE_DESC=True):
    data = f"Order={1 if orderJingE_DESC else 0}&st=30&a=RealRankingInfo&apiv=w35&Type=5&Index=0&RStart={startTime}&c=ZhiShuRanking&VerSion=5.13.0.0&REnd={endTime}&Date={date}&PhoneOSNew=1&ZSType=7&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9773&"
    result = __base_request("https://apphq.longhuvip.com/w1/api/index.php",
                            data=data)
    return result


# 获取代码的板块
def getStockIDPlate(code):
    data = f"a=GetStockIDPlate_New&apiv=w32&c=StockL2Data&StockID={code}&PhoneOSNew=1&UserID=0&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Token=0&"
@@ -270,11 +277,16 @@


if __name__ == "__main__":
    print(getLimitUpInfoNew())
    # print(getLimitUpInfoNew())

    # __getConceptBK("300564")
    data = (getMarketJingXuanRealRankingInfo(False))
    print(data)
    data = getMarketJingXuanRealRankingInfoByTimeRange("0925", "1000", "2024-11-12")
    data = json.loads(data)
    print(json.loads(data))
    data = getMarketJingXuanRealRankingInfoByTimeRange("0925", "1500", "2024-11-13")
    print(json.loads(data))
    data = getMarketJingXuanRealRankingInfoByTimeRange("0925", "1000", "2024-11-13")
    print(json.loads(data))
    # data=json.loads(data)
    # print(len(data["list"]))
    # data = json.loads(getCodesByPlate("801235"))

 third_data/kpl_data_constant.py

@@ -8,9 +8,12 @@
from utils.kpl_data_db_util import KPLLimitUpDataUtil

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

open_limit_up_code_dict_for_radical_buy = None


@singleton
class LimitUpCodesBlockRecordManager:
    """
    历史涨停代码的板块管理
@@ -26,15 +29,16 @@

    __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
    __day = tool.get_now_date_str()

    def __init__(self, day=tool.get_now_date_str()):
        self.__day = day
        self.__load_data()

    @classmethod
    def __load_data(cls):
        kpl_results = KPLLimitUpDataUtil.get_latest_block_infos()
        kpl_results = KPLLimitUpDataUtil.get_latest_block_infos(min_day=tool.date_sub(cls.__day, 180),
                                                                max_day=cls.__day)
        limit_up_reasons_dict = {}
        limit_up_recommend_block_dict = {}
        for r in kpl_results:
@@ -55,11 +59,10 @@
            cls.__limit_up_recommend_reasons_dict[code] = set(limit_up_recommend_block_dict[code])

        # 加载为扫入买匹配的代码板块
        kpl_results = KPLLimitUpDataUtil.get_latest_block_infos(min_day=tool.date_sub(tool.get_now_date_str(), 365))
        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:
            if r[2] in constant.KPL_INVALID_BLOCKS:
                continue
            code = r[0]
            if code not in kpl_block_dict:
                kpl_block_dict[code] = []
@@ -80,6 +83,7 @@
            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

    def get_limit_up_reasons(self, code):

 third_data/kpl_data_manager.py

@@ -349,7 +349,7 @@

    @classmethod
    # 获取最近几天的数据，根据日期倒序返回
    def get_latest_from_file(cls, type, count):
    def get_latest_from_file(cls, type, count, max_day=tool.get_now_date_str()):
        files = os.listdir(constant.CACHE_PATH)
        file_name_list = []
        for f in files:
@@ -357,7 +357,6 @@
                file_name_list.append((f.split("_")[0], f))

        file_name_list.sort(key=lambda x: x[0], reverse=True)
        file_name_list = file_name_list[:count]
        fresults = []
        for file in file_name_list:
            path = f"{constant.CACHE_PATH}/{file[1]}"
@@ -366,7 +365,10 @@
            with open(path, 'r') as f:
                lines = f.readlines()
                if lines:
                    if int(file[0].replace("-", "")) <= int(max_day.replace("-", "")):
                    fresults.append((file[0], json.loads(lines[0])))
                if len(fresults) >=count:
                    break

        return fresults

@@ -411,18 +413,24 @@

# 获取最近几天的实时涨停信息
# 返回格式（[日期,数据]）
def get_current_limit_up_data_records(count):
def get_current_limit_up_data_records(count, day=tool.get_now_date_str()):
    fresults = []
    day = tool.get_now_date_str()
    datas = []
    if day in __limit_up_list_records_dict:
        datas = __limit_up_list_records_dict[day]
    else:
        logger_debug.info("从文件中获取前几天的实时涨停数据")
        datas = KPLDataManager().get_latest_from_file(KPLDataType.LIMIT_UP, count + 2)
        if datas:
        datas = KPLDataManager().get_latest_from_file(KPLDataType.LIMIT_UP, count + 2, max_day=day)
        # 移除比今天还大的数据
        fdatas = []
        for d in datas:
            if int(d[0].replace("-", "")) > int(day.replace("-", "")):
                continue
            fdatas.append(d)
        if fdatas:
            # 保存数据
            __limit_up_list_records_dict[day] = datas
            __limit_up_list_records_dict[day] = fdatas
    datas = __limit_up_list_records_dict[day]
    for i in range(len(datas)):
        if datas[i][0] == day:
            continue
@@ -447,10 +455,9 @@
__latest_current_limit_up_records = {}


def get_latest_current_limit_up_records():
    day = tool.get_now_date_str()
def get_latest_current_limit_up_records(day=tool.get_now_date_str(), max_day_count=15):
    if day not in __latest_current_limit_up_records:
        fdatas = get_current_limit_up_data_records(15)
        fdatas = get_current_limit_up_data_records(max_day_count)
        __latest_current_limit_up_records[day] = fdatas
    return __latest_current_limit_up_records.get(day)

@@ -599,6 +606,7 @@
        threading.Thread(target=cls.run_market_jingxuan_out, daemon=True).start()


@tool.singleton
class CodeHighLevel:
    """
    代码高度管理
@@ -608,15 +616,13 @@
    __code_level_dict = {}
    __codes = set()

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(CodeHighLevel, cls).__new__(cls, *args, **kwargs)
            cls.__load_data()
        return cls.__instance
    def __init__(self, day=tool.get_now_date_str()):
        self.__day = day
        self.__load_data(day)

    @classmethod
    def __load_data(cls):
        fdatas = get_current_limit_up_data_records(15)
    def __load_data(cls, day):
        fdatas = get_current_limit_up_data_records(15, day=day)
        temp_dict = {d[0]: 2 for d in fdatas[0][1]}
        break_codes = set()
        for i in range(1, len(fdatas)):
@@ -642,5 +648,5 @@


if __name__ == "__main__":
    print(CodeHighLevel().get_high_level("000833"))
    print(CodeHighLevel("2024-11-11").get_high_level("000833"))
    input()

 trade/buy_radical/radical_buy_data_manager.py

@@ -15,7 +15,7 @@
from log_module import async_log_util
from log_module.log import logger_l2_radical_buy, logger_debug, logger_l2_radical_buy_data
from third_data import kpl_data_manager, kpl_block_util, huaxin_l1_data_manager, kpl_util, kpl_data_constant
from third_data.code_plate_key_manager import TargetCodePlateKeyManager, KPLCodeJXBlockManager
from third_data.code_plate_key_manager import TargetCodePlateKeyManager, KPLCodeJXBlockManager, RealTimeKplMarketData
from third_data.kpl_data_constant import LimitUpDataConstant, LimitUpCodesBlockRecordManager
from third_data.kpl_data_manager import CodeHighLevel
from third_data.third_blocks_manager import BlockMapManager
@@ -87,6 +87,13 @@
                big_order_deal_enough = is_big_order_deal_enough(code, volume_rate)
            if not big_order_deal_enough[0] and big_order_deal_enough[2]:
                return False, "交易前两分钟大单不足"

        jx_out_blocks = RealTimeKplMarketData.get_top_market_jingxuan_out_blocks()
        if jx_out_blocks:
            blocks = RadicalBuyBlockManager.get_code_blocks(code)[0]
            same_blocks = blocks & jx_out_blocks
            if same_blocks:
                return False, f"【{same_blocks}】板块精选流出"

        return True, ""

@@ -698,8 +705,14 @@
        # 获取代码的板块
        keys_, info = cls.get_code_blocks(code)
        if not keys_:
            return set(), "没获取到板块交集"
            return set(), "没获取到板块"

        # 获取精选净流入
        jx_in_blocks = RealTimeKplMarketData.get_top_market_jingxuan_blocks()
        keys_in_jx = keys_ & jx_in_blocks
        if not keys_in_jx:
            return set(), f"【{keys_}】板块未在精选流入中"
        keys_ = keys_in_jx
        match_blocks = open_limit_up_blocks & keys_
        can_buy_blocks = set()
        fmsges = []

 utils/tool.py

@@ -21,6 +21,22 @@
    return wrapper


def singleton(cls):
    """
    单例装饰器
    @param cls:
    @return:
    """
    instances = {}

    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance


def get_expire():
    now = int(t.time())
    end = int(t.time()) + 60 * 60 * 24