‘功能完善

Administrator

5 天以前 0ce5457b060bddc2ab65e4ff094159781cd15deb

 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:
@@ -297,6 +297,9 @@
        """
        if code_ in self.stock_variables_dict:
            return

        if code_ == '002907':
            print("")

        stock_variables = StrategyVariableFactory.create_from_history_data(
            timeline_data["kline_data"].get(code_), timeline_data["minute_data"].get(code_),
@@ -582,12 +585,27 @@
                            if p not in most_real_kpl_plate_limit_up_codes_info:
                                most_real_kpl_plate_limit_up_codes_info[p] = []
                            most_real_kpl_plate_limit_up_codes_info[p].append(code)
                # print(time_str, "涨停数大于3个", [p for p in most_real_kpl_plate_limit_up_codes_info if
                #                             len(most_real_kpl_plate_limit_up_codes_info[p]) >= 3])

                # ---------测试--------
                # test_plate = "化工"
                # if len(most_real_kpl_plate_limit_up_codes_info.get(test_plate, [])) >= 3:
                #     print("测试开始=========")
                #     code_plates_for_buy = self.current_data["code_plates_for_buy"]
                #     plate_codes = [c for c in code_plates_for_buy if test_plate in code_plates_for_buy[c]]
                #     print(f"{test_plate}满足", time_str, plate_codes)
                #     for c in plate_codes:
                #         sv: StockVariables = self.stock_variables_dict.get(c)
                #         if sv and sv.当前价 > sv.昨日收盘价:
                #             print(c)
                #     print("测试完毕=========")

            if ticks:
                for tick in ticks:
                    code = tick["symbol"][-6:]
                    # if code not in self.fcodes:
                    #     continue
                    if code not in self.fcodes:
                        continue
                    if DEBUG_CODES and code not in DEBUG_CODES:
                        continue

@@ -648,12 +666,15 @@

                    if not stock_variables.今日最低价 or tick["price"] < stock_variables.今日最低价:
                        stock_variables.今日最低价 = tick["price"]
                    if most_real_kpl_plate_limit_up_codes_info:
                        stock_variables.开盘啦最正板块涨停 = most_real_kpl_plate_limit_up_codes_info

                    stock_variables.开盘啦最正板块涨停 = most_real_kpl_plate_limit_up_codes_info

                    # compute_result = self.__run_backtest(code, stock_variables)
                    # self.__process_test_result(code, stock_variables, next_trade_day, stock_variables.当前价,
                    #                            time_str, compute_result)
                    # if time_str >= '09:30:00':
                    #     if stock_variables.今日大单数据 and stock_variables.开盘啦最正板块涨停 and max(
                    #             [len(stock_variables.开盘啦最正板块涨停.get(x, [])) for x in stock_variables.代码板块]) >= 3:
                    #         compute_result = self.__run_backtest(code, stock_variables)
                    #         self.__process_test_result(code, stock_variables, next_trade_day, stock_variables.当前价,
                    #                                    time_str, compute_result)

                    # if len(real_codes) >= 2 and time_str > '09:30:00':
                    #     # print(time_str, plate)
@@ -766,7 +787,7 @@
            stock_variables.板块成交代码 = self.deal_block_codes


# DEBUG_CODES = ['002365', '000953', '002907', '002688', '003020', '002900', '002082', '000566', '300204', '002317']
# DEBUG_CODES = ['603579', '300884']
DEBUG_CODES = []

VOLUME_LOG_ENABLE = False
@@ -779,16 +800,21 @@

if __name__ == "__main__":
    back_test_dict = {}
    # days = ["2025-05-06", "2025-05-07", "2025-05-08", "2025-05-09", "2025-05-12", "2025-05-13", "2025-05-14",
    #         "2025-05-15", "2025-05-16"]
    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"]
    # 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-06-10",
    #         "2025-06-11", "2025-06-12", "2025-06-13", "2025-06-16", "2025-06-17"]

    # days = ["2025-06-13"]

    days.reverse()
    for day in days:
        if day not in back_test_dict:
            # back_test_dict[day] = BackTest(day, "今日量是否足够.py")
            back_test_dict[day] = BackTest(day, "低吸脚本_辨识度_v6.py")
            back_test_dict[day] = BackTest(day, "strategy_script_v6.py")
        print("=========================", day)
        # back_test_dict[day].run_volume()
        back_test_dict[day].run()