| | |
| | | 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 |
| | |
| | | |
| | | 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: |
| | |
| | | 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 |
| | | |
| | |
| | | |
| | | if not stock_variables.今日最低价 or tick["price"] < stock_variables.今日最低价: |
| | | stock_variables.今日最低价 = tick["price"] |
| | | |
| | | 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 most_real_kpl_plate_limit_up_codes_info: |
| | | stock_variables.开盘啦最正板块涨停 = most_real_kpl_plate_limit_up_codes_info |
| | | # 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) |
| | |
| | | stock_variables.板块成交代码 = self.deal_block_codes |
| | | |
| | | |
| | | # DEBUG_CODES = ['002365', '000953', '002907', '002688', '003020', '002900', '002082', '000566', '300204', '002317'] |
| | | # DEBUG_CODES = ['603040', '603758', '603286', '603586', '605255', '002048', '605208', '002806', '603266', '603788'] |
| | | DEBUG_CODES = [] |
| | | |
| | | VOLUME_LOG_ENABLE = False |
| | |
| | | |
| | | DEBUG_BLOCKS = [] |
| | | |
| | | BIG_ORDER_MONEY_THRESHOLD = 200e4 |
| | | BIG_ORDER_MONEY_THRESHOLD = 100e4 |
| | | |
| | | 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", "2025-05-19", "2025-05-20"] |
| | | # "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", |
| | | "2025-06-11", "2025-06-12"] |
| | | |
| | | # days = ["2025-06-09"] |
| | | |
| | | 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() |