bug修复

Administrator

2025-06-10 8ab72ecd563e4bdcb94eac3f8c9d18d0a0a2d502

bug修复

 api/outside_api_callback.py

@@ -43,8 +43,20 @@
        èŽ·å–交易参数
        @return:
        """
        result = strategy_params_settings.settings.to_json_str()
        return {"code": 0, "data": result}
        result = strategy_params_settings.StrategyParamsSettingsManager().get_settings().to_json_str()
        return {"code": 0, "data": json.loads(result)}

    def __on_set_settings(self, data):
        """
        è®¾ç½®äº¤æ˜“参数
        @return:
        """
        settings = strategy_params_settings.StrategyParamsSettingsManager().get_settings()
        for k in data:
            settings.__setattr__(k, data[k])
        strategy_params_settings.StrategyParamsSettingsManager().set_settings(settings)
        # åŒæ­¥å‚数设置
        return {"code": 0, "data": {}}

    def __on_get_env(self, need_hsitory_data):
        """
@@ -64,7 +76,8 @@
            else:
                # å¦‚果在16:00之后采用下一个交易日
                day = TradeDateManager().get_next_trade_day(tool.get_now_date_str())
            fdata["history_data"]["leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count(
            fdata["history_data"][
                "leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count(
                day)

            if tool.get_now_time_str() < '16:00:00':
@@ -103,6 +116,9 @@
        result_json = {}
        if ctype == "get_settings":
            result_json = self.__on_get_settings()
        elif ctype == 'set_settings':
            del data["ctype"]
            result_json = self.__on_set_settings(data)
        elif ctype == 'get_env':
            result_json = self.__on_get_env(data.get("history"))
        elif ctype == 'update_leading_limit_up_datas':

 strategy/data_downloader.py

@@ -114,6 +114,7 @@
                    # ä¿å­˜Tick线
                    with open(tick_path, encoding='utf-8', mode='w') as f:
                        f.write(f"{results}")
                    print("剩余数量：", len(set(batch_codes) - excute_codes))
                finally:
                    excute_codes.add(code)
                    if show_log:

 strategy/strategy_params_settings.py

@@ -3,6 +3,10 @@
"""
import json

from db.mysql_data_delegate import Mysqldb
from utils import tool


class StrategyParamsSettings:
    # ç¦æ­¢ä¹°å…¥
    STATE_FORBIDDEN_BUY = 0
@@ -14,6 +18,8 @@
        self.trade_state = 1
        # ä¹°å…¥é‡‘额
        self.buy_money = 2000
        # æœ€å¤§ä¹°å…¥ç¥¨çš„数量
        self.max_buy_codes_count = 10
        # ä»·æ ¼åŒºé—´
        self.price_range = (3, 60)
        # è€é¢˜ææ¶¨åœæ•°
@@ -49,7 +55,7 @@
        # æ˜¯å¦å¯ä¹°ä»Šæ—¥æ¶¨åœè¿‡çš„票
        self.can_buy_limited_up = False
        # æœ€ä½Žå¼€ç›˜æ¶¨å¹…
        self.min_open_rate = -0.03
        self.min_open_rate = 0.00001
        # å¯ä¹°çš„涨幅比例
        self.avaiable_rates = (-0.03, 0.07)
        # ä»Šæ—¥æ¶¨åœä»·éœ€çªç ´XX日最高价,None表示此条数据不生效
@@ -72,4 +78,28 @@
        return obj


settings: StrategyParamsSettings = StrategyParamsSettings()
@tool.singleton
class StrategyParamsSettingsManager:
    def __init__(self):
        self.musql = Mysqldb()
        self.__settings = StrategyParamsSettings()
        self.__load_data()

    def __load_data(self):
        sql = f"select `value` from config where `key` = 'low_suction_settings'"
        setting_str_list = self.musql.select_one(sql)
        if setting_str_list and setting_str_list[0]:
            self.__settings = StrategyParamsSettings.to_obj(setting_str_list[0])

    def set_settings(self, settings: StrategyParamsSettings):
        sql = f"update config set `value` = '{settings.to_json_str()}'  where `key` = 'low_suction_settings'"
        self.musql.execute(sql)
        self.__settings = settings

    def get_settings(self):
        return self.__settings


if __name__ == "__main__":
    settings = StrategyParamsSettingsManager().get_settings()
    StrategyParamsSettingsManager().set_settings(settings)

 strategy/strategy_variable_factory.py

@@ -203,7 +203,7 @@
    def load_target_plate_and_codes(self):
        """
        åŠ è½½ç›®æ ‡æ¿å—与对应的代码：
        ä»Žæœ€è¿‘120个交易日的真正涨停数据中
        ä»Žæœ€è¿‘60个交易日的真正涨停数据中
        @return: {"板块":[代码]}
        """
        end_date = self.trade_days[:60][-1]
@@ -234,10 +234,11 @@
            if results:
                results = [x for x in results if
                           (tool.is_can_buy_code(x[0]) and x[0] in valid_codes and x[0] not in exclude_codes)]
                max_count = len(results) // 3 if len(results) % 3 == 0 else len(results) // 3 + 1
                results = results[:max_count]
                # å–前10
                results = results[:10]
                # å–前1/3且涨停数是前10
                # max_count = len(results) // 3 if len(results) % 3 == 0 else len(results) // 3 + 1
                # results = results[:max_count]
                # # å–前10
                # results = results[:10]
                codes = [x[0] for x in results]
                fresults[kpl_util.filter_block(b)] = codes
        return fresults

 strategy/time_series_backtest.py

@@ -3,7 +3,7 @@
from strategy.data_analyzer import KPLLimitUpDataAnalyzer
from strategy.data_downloader import DataDownloader
from strategy.low_suction_strategy import LowSuctionOriginDataExportManager
from strategy.strategy_params_settings import StrategyParamsSettings
from strategy.strategy_params_settings import StrategyParamsSettings, StrategyParamsSettingsManager
from strategy.strategy_variable import StockVariables
from strategy.strategy_variable_factory import DataLoader, StrategyVariableFactory
from third_data import kpl_util
@@ -13,7 +13,7 @@

class BackTest:

    def __init__(self, day, script_name="低吸脚本_辨识度_v3.py", settings=StrategyParamsSettings()):
    def __init__(self, day, script_name="低吸脚本_辨识度_v3.py", settings=StrategyParamsSettingsManager().get_settings()):
        self.day = day
        scripts = ""
        with open(script_name, mode='r', encoding='utf-8') as f:
@@ -783,7 +783,7 @@
    #         "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20",  "2025-05-21", "2025-05-22"]
    days = ["2025-05-12", "2025-05-13", "2025-05-14", "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20",
            "2025-05-21", "2025-05-22", "2025-05-23", "2025-05-26", "2025-05-27", "2025-05-28", "2025-05-29",
            "2025-05-30", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06", "2025-06-09"]
            "2025-05-30", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06", "2025-06-09", "2025-06-10"]
    days.reverse()
    for day in days:
        if day not in back_test_dict:

 strategy/µÍÎü½Å±¾_±æʶ¶È_v6.py

@@ -27,6 +27,13 @@
        if target_code.find("60") != 0 and target_code.find("00") != 0 and target_code.find("30") != 0:
            return False, "科创板的票不买"

    if sv.板块成交代码:
        deal_codes = set()
        for p in sv.板块成交代码:
            deal_codes |= set(sv.板块成交代码[p])
        if len(deal_codes) >= settings.max_buy_codes_count:
            return False, f"买入代码数超限({len(deal_codes)}/{settings.max_buy_codes_count})"

    # ç›®æ ‡ç¥¨æ¿å—涨停个数>=2
    can_buy_plates = set()
    for plate in sv.代码板块:
@@ -142,10 +149,9 @@

    limit_up_codes_count = max([(p, len(sv.开盘啦最正板块涨停.get(p, []))) for p in can_buy_plates], key=lambda x: x[1])[1]

    threshold_money *= max(min(10 - limit_up_codes_count + 3, 10), 5)/10
    threshold_money *= max(min(10 - limit_up_codes_count + 1, 10), 5) / 10

    # print(target_code, sv.自由流通市值, threshold_money, limit_up_codes_count)


    # threshold_money = 200e4  # int(sv.昨日成交量 * 0.2 * sv.今日涨停价 * 0.05)
    if big_order_money < threshold_money: