gp_trade_low_suction.git

			@@ -1,9 +1,11 @@
			import logging

			import constant
			from code_attribute import gpcode_manager, code_nature_analyse
			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 +15,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:
			@@ -120,15 +122,15 @@
			:return: 按时间排序的数据列表
			"""
			if self.day >= '2025-05-26':
			IS_BY_BIG_ORDER = True
			else:
			IS_BY_BIG_ORDER = False
			else:
			IS_BY_BIG_ORDER = True
			day = self.day
			fdata = {}
			__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(day)
			all_limit_up_list = __LowSuctionOriginDataExportManager.export_limit_up_list()
			fdata["limit_up_list"] = {d[0][:8]: d[1] for d in all_limit_up_list}
			big_order_deals = __LowSuctionOriginDataExportManager.export_big_order_deal(BIG_ORDER_MONEY_THRESHOLD)
			big_order_deals = __LowSuctionOriginDataExportManager.export_big_order_deal(BIG_ORDER_MONEY_THRESHOLD, max_deal_space=3)
			if not big_order_deals or IS_BY_BIG_ORDER:
			big_order_deals = __LowSuctionOriginDataExportManager.export_big_order_deal_by(BIG_ORDER_MONEY_THRESHOLD)
			# 转换格式为：{时间: [("代码", (买单号, 量, 金额, 时间, 最终成交价))]
			@@ -298,6 +300,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_),
			timeline_data["limit_up_record_data"].get(code_), timeline_data["trade_days"])
			@@ -353,7 +358,7 @@
			# self.fcodes, self.head_rise_code_blocks = self.__get_target_codes_v3() # __filter_codes(current_data, timeline_data)
			self.fcodes, self.head_rise_code_blocks = self.__get_target_codes_v4(), {}

			print(len(self.fcodes), self.fcodes)
			print(len(self.fcodes))
			if not self.current_tick_data:
			try:
			self.current_tick_data = self.load_current_tick_datas(self.data_loader)
			@@ -582,12 +587,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 +668,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)
			@@ -677,6 +700,8 @@
			code = big_order[0]
			if code not in self.fcodes:
			continue
			if DEBUG_CODES and code not in DEBUG_CODES:
			continue
			self.init_stock_variables(code, self.timeline_data, self.current_data)
			stock_variables: StockVariables = self.stock_variables_dict.get(code)
			if plate_limit_up_codes_info is not None:
			@@ -693,10 +718,16 @@

			if block_in_datas:
			stock_variables.资金流入板块 = block_in_datas
			compute_result = self.__run_backtest(code, stock_variables)
			# print(compute_result)
			self.__process_test_result(code, stock_variables, next_trade_day, big_order[1][4],
			huaxin_util.convert_time(big_order[1][3]), compute_result)

			stock_variables.当前价 = big_order[1][4]
			try:
			compute_result = self.__run_backtest(code, stock_variables)
			# print(compute_result)
			self.__process_test_result(code, stock_variables, next_trade_day, big_order[1][4],
			huaxin_util.convert_time(big_order[1][3]), compute_result)
			except Exception as e:
			print(time_str)
			logging.exception(e)

			print("可买题材：", all_new_plates)

			@@ -766,7 +797,9 @@
			stock_variables.板块成交代码 = self.deal_block_codes


			# DEBUG_CODES = ['002365', '000953', '002907', '002688', '003020', '002900', '002082', '000566', '300204', '002317']
			# 锂电池 ['002882', '002667', '002846', '300530', '002074', '301662', '002580', '300584', '603399', '601515']
			# 化工 ['600610', '002427', '002165', '002809', '000565', '002365', '603192', '600370', '600800', '603188']
			# DEBUG_CODES = ['600610', '002427', '002165', '002809', '000565', '002365', '603192', '600370', '600800', '603188']
			DEBUG_CODES = []

			VOLUME_LOG_ENABLE = False
			@@ -779,16 +812,20 @@

			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-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"]
			days = ["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", "2025-06-18", "2025-06-19",
			"2025-06-20","2025-06-23"]

			# days = ["2025-05-23"]

			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()