买入金额>=30亿/L后撤只囊括100w以上的

Administrator

2025-06-10 efe62c0c92bee36da5179f34bb73e8ee4db6f814

 third_data/kpl_data_manager.py

@@ -72,7 +72,13 @@
            pass

    @classmethod
    def save_record(cls, day, records):
    def save_record(cls, day, records, set_not_open=False):
        """
        @param day:
        @param records:
        @param set_not_open: 是否需要设置炸板与否
        @return:
        """
        # 统计炸板
        try:
            last_codes = set()
@@ -142,13 +148,17 @@
            result = mysqldb.select_one("select * from kpl_limit_up_record where _id='{}'".format(_id))
            if not result:
                mysqldb.execute(
                    f"insert into kpl_limit_up_record(_id,_day,_hot_block_name,_code,_code_name,_limit_up_time,_blocks,_latest_limit_up_time,_update_time,_create_time,_hot_block_code_count,_limit_up_high_info,_zylt_val) values('{_id}','{day}','{d[5]}','{d[0]}','{d[1]}','{d[2]}','{d[6]}','{d[3]}',now(),now(),{d[10]},'{d[4]}',{d[7]})")
                    f"insert into kpl_limit_up_record(_id,_day,_hot_block_name,_code,_code_name,_limit_up_time,_blocks,_latest_limit_up_time,_update_time,_create_time,_hot_block_code_count,_limit_up_high_info,_zylt_val,_hot_block_code) values('{_id}','{day}','{d[5]}','{d[0]}','{d[1]}','{d[2]}','{d[6]}','{d[3]}',now(),now(),{d[10]},'{d[4]}',{d[7]},{d[9]})")
                cls.__load_hist_and_blocks(code)
            else:
                if _id in cls.latest_datas and json.dumps(cls.latest_datas.get(_id)) != json.dumps(d):
                    mysqldb.execute(
                        f"update kpl_limit_up_record set _latest_limit_up_time='{d[3]}',_limit_up_time='{d[2]}',_hot_block_code_count={d[10]},_limit_up_high_info='{d[4]}' ,_update_time=now() where _id='{_id}'")
                    cls.latest_datas[_id] = d
            if set_not_open:
                # 需要设置不炸板
                mysqldb.execute(f"update kpl_limit_up_record set _open = 0, _update_time = now() where _id='{_id}'")

            cls.latest_datas[_id] = d

            # 获取原来的代码所属板块,删除之前错误的板块
@@ -367,7 +377,7 @@
                if lines:
                    if int(file[0].replace("-", "")) <= int(max_day.replace("-", "")):
                        fresults.append((file[0], json.loads(lines[0])))
                if len(fresults) >=count:
                if len(fresults) >= count:
                    break

        return fresults
@@ -487,17 +497,23 @@
            logger_debug.info("任务修复-开盘啦：涨停列表")
            # 大于20s就需要更新
            threading.Thread(target=cls.run_limit_up_task, daemon=True).start()
        key = "jingxuan_rank"
        if key not in cls.__latest_update_time_dict or time.time() - cls.__latest_update_time_dict[key] > 20:
            logger_debug.info("任务修复-开盘啦：精选流入列表")
            # 大于20s就需要更新
            threading.Thread(target=cls.run_market_jingxuan_in, daemon=True).start()
        # key = "jingxuan_rank"
        # if key not in cls.__latest_update_time_dict or time.time() - cls.__latest_update_time_dict[key] > 20:
        #     logger_debug.info("任务修复-开盘啦：精选流入列表")
        #     # 大于20s就需要更新
        #     threading.Thread(target=cls.run_market_jingxuan_in, daemon=True).start()
        #
        # key = "jingxuan_rank_out"
        # if key not in cls.__latest_update_time_dict or time.time() - cls.__latest_update_time_dict[key] > 20:
        #     logger_debug.info("任务修复-开盘啦：精选流出列表")
        #     # 大于20s就需要更新
        #     threading.Thread(target=cls.run_market_jingxuan_out, daemon=True).start()

        key = "jingxuan_rank_out"
        key = "market_strong"
        if key not in cls.__latest_update_time_dict or time.time() - cls.__latest_update_time_dict[key] > 20:
            logger_debug.info("任务修复-开盘啦：精选流出列表")
            logger_debug.info("任务修复-开盘啦：市场强度")
            # 大于20s就需要更新
            threading.Thread(target=cls.run_market_jingxuan_out, daemon=True).start()
            threading.Thread(target=cls.run_market_strong, daemon=True).start()

    @classmethod
    def run_limit_up_task(cls):
@@ -559,6 +575,23 @@
                time.sleep(3)

    @classmethod
    def run_market_strong(cls):
        """
        精选流出
        @return:
        """
        while True:
            try:
                if tool.is_trade_time():
                    strong_value = kpl_api.getMarketStrong()
                    cls.__upload_data("market_strong", strong_value)
            except:
                pass
            finally:
                cls.__latest_update_time_dict["market_strong"] = time.time()
                time.sleep(3)

    @classmethod
    # 运行拉取任务
    def run_pull_task(cls):
        def get_bidding_money():
@@ -600,10 +633,11 @@
                    time.sleep(3)

        threading.Thread(target=cls.run_limit_up_task, daemon=True).start()
        threading.Thread(target=cls.run_market_strong, daemon=True).start()
        # threading.Thread(target=get_bidding_money, daemon=True).start()
        # threading.Thread(target=get_market_industry, daemon=True).start()
        threading.Thread(target=cls.run_market_jingxuan_in, daemon=True).start()
        threading.Thread(target=cls.run_market_jingxuan_out, daemon=True).start()
        # threading.Thread(target=cls.run_market_jingxuan_in, daemon=True).start()
        # threading.Thread(target=cls.run_market_jingxuan_out, daemon=True).start()


@tool.singleton