gp-server.git - Gitblit

			@@ -33,13 +33,22 @@
			class HuaXinTransactionDatasProcessor:
			__statistic_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=constant.HUAXIN_L2_MAX_CODES_COUNT + 2)
			__TradeBuyQueue = transaction_progress.TradeBuyQueue()
			# 非涨停成交时间
			__not_limit_up_time_dict = {}

			# 计算成交进度
			@classmethod
			def __compute_latest_trade_progress(cls, code, fdatas):
			def __compute_latest_trade_progress(cls, code, fdatas, buy_exec_index=None):
			"""
			计算真实下单位置
			@param code:
			@param fdatas:
			@param buy_exec_index:
			@return:真实下单位置, 是否是近似计算
			"""
			buyno_map = l2_data_util.local_today_buyno_map.get(code)
			if not buyno_map:
			return None
			return None, False
			buy_progress_index = None
			for i in range(len(fdatas) - 1, -1, -1):
			d = fdatas[i]
			@@ -49,7 +58,37 @@
			if L2DataUtil.is_limit_up_price_buy(buyno_map[buy_no]["val"]):
			buy_progress_index = buyno_map[buy_no]["index"]
			break
			return buy_progress_index
			if buy_progress_index is None and buy_exec_index is not None and buy_exec_index >= 0:
			# 没有找到真实成交进度位且有买入执行位置
			# 根据最近的成交买单号计算真实成交位置
			try:
			latest_buy_order_no = fdatas[-1][0][6]
			total_datas = l2_data_util.local_today_datas.get(code)
			if tool.trade_time_sub(total_datas[-1]["val"]["time"], total_datas[buy_exec_index]["val"]["time"]) < 60:
			# 下单60s内才这样计算
			# 取最新的成交买单号，在两个涨停委托买之间的后一个数据
			index_1 = None
			max_index = total_datas[-1]["index"]
			for i in range(max_index, -1, -1):
			data = total_datas[i]
			val = data['val']
			if not L2DataUtil.is_limit_up_price_buy(val):
			continue
			if val['orderNo'] < latest_buy_order_no:
			index_1 = i
			break
			if index_1 is not None:
			for i in range(index_1 + 1, max_index + 1):
			data = total_datas[i]
			val = data['val']
			if not L2DataUtil.is_limit_up_price_buy(val):
			continue
			if val['orderNo'] > latest_buy_order_no:
			buy_progress_index = i
			return buy_progress_index, True
			except Exception as e:
			async_log_util.info(logger_debug, f"计算真实成交进度位出错：{str(e)}")
			return buy_progress_index, False

			@classmethod
			def statistic_big_order_infos(cls, code, fdatas, order_begin_pos: OrderBeginPosInfo):
			@@ -61,7 +100,10 @@
			"""

			def statistic_big_buy_data():
			use_time_list = []
			__start_time = time.time()
			buy_datas, bigger_buy_datas = HuaXinBuyOrderManager.statistic_big_buy_data(code, fdatas, limit_up_price)
			use_time_list.append((time.time() - __start_time, "买单统计"))
			if buy_datas:
			BigOrderDealManager().add_buy_datas(code, buy_datas)
			active_big_buy_orders = []
			@@ -71,6 +113,7 @@
			# (买单号, 成交金额, 最后成交时间)
			active_big_buy_orders.append((x[0], x[2], x[4]))
			EveryLimitupBigDealOrderManager.add_big_buy_order_deal(code, active_big_buy_orders)
			use_time_list.append((time.time() - __start_time, "买单统计结果处理"))
			try:
			is_placed_order = l2_data_manager.TradePointManager.is_placed_order(order_begin_pos)
			if is_placed_order:
			@@ -88,12 +131,22 @@
			order_begin_pos.buy_single_index)
			except Exception as e:
			logger_debug.exception(e)
			if use_time_list and use_time_list[-1][0] > 0.005:
			l2_log.info(code, hx_logger_l2_upload,
			f"买单统计+处理耗时：{use_time_list[-1][0]} 详情:{use_time_list}")

			return buy_datas

			def statistic_big_sell_data():
			use_time_list = []
			__start_time = time.time()
			sell_datas = HuaXinSellOrderStatisticManager.statistic_big_sell_data(code, fdatas)
			if sell_datas:
			BigOrderDealManager().add_sell_datas(code, sell_datas)
			use_time_list.append((time.time() - __start_time, "卖单统计"))
			if use_time_list and use_time_list[-1][0] > 0.005:
			l2_log.info(code, hx_logger_l2_upload,
			f"卖单统计+处理耗时：{use_time_list[-1][0]} 详情:{use_time_list}")
			return sell_datas

			def statistic_big_data(f1_, f2_):
			@@ -102,7 +155,7 @@

			limit_up_price = gpcode_manager.get_limit_up_price_as_num(code)

			if len(fdatas) > 100:
			if False and len(fdatas) > 100:
			# 并行处理买单与卖单
			# 超过100条数据才需要并行处理
			f1 = dask.delayed(statistic_big_buy_data)()
			@@ -137,16 +190,23 @@
			temp_time_dict.clear()

			__start_time = time.time()
			limit_up_price = gpcode_manager.get_limit_up_price_as_num(code)
			# 设置成交价
			try:
			current_price_process_manager.set_trade_price(code, fdatas[-1][0][1])
			if limit_up_price > fdatas[-1][0][1]:
			# 没有涨停
			EveryLimitupBigDealOrderManager.open_limit_up(code, f"最新成交价：{fdatas[-1][0][1]}")
			radical_buy_strategy.clear_data(code)
			except:
			pass
			if not fdatas[-1][2]:
			if code not in cls.__not_limit_up_time_dict:
			cls.__not_limit_up_time_dict[code] = fdatas[-1][5]
			last_time = cls.__not_limit_up_time_dict[code]
			# 炸板时间持续500ms以上算炸板
			if tool.trade_time_sub_with_ms(fdatas[-1][5], last_time) > 500:
			# 没有涨停
			EveryLimitupBigDealOrderManager.open_limit_up(code, f"最新成交价：{fdatas[-1][0][1]}")
			radical_buy_strategy.clear_data(code, msg=f"没有涨停：{fdatas[-1][0]}")
			else:
			if code in cls.__not_limit_up_time_dict:
			cls.__not_limit_up_time_dict.pop(code)
			except Exception as e:
			async_log_util.error(logger_debug, f"L2成交开板计算错误：{str(e)}")

			total_datas = l2_data_util.local_today_datas.get(code)
			use_time_list = []
			@@ -174,21 +234,12 @@
			_start_time = time.time()

			try:
			last_data = fdatas[-1]
			# 统计上板时间
			try:
			for d in fdatas:
			if d[1]:
			# 主动买
			if d[2]:
			# 涨停
			current_price_process_manager.set_latest_not_limit_up_time(code, d[5])
			else:
			# 主动卖（板上）
			if d[2]:
			L2LimitUpSellDataManager.clear_data(code)
			break
			except:
			pass
			if last_data[1] and last_data[2]:
			current_price_process_manager.set_latest_not_limit_up_time(code, last_data[5])
			if not last_data[1] and last_data[2]:
			L2LimitUpSellDataManager.clear_data(code)
			big_sell_order_info = None
			# 统计卖单
			big_sell_order_info = HuaXinSellOrderStatisticManager.statistic_continue_limit_up_sell_transaction_datas(
			@@ -237,12 +288,13 @@
			# if big_money_count > 0:
			# LCancelRateManager.compute_big_num_deal_rate(code)

			buy_progress_index = cls.__compute_latest_trade_progress(code, fdatas)
			buy_progress_index, is_similar = cls.__compute_latest_trade_progress(code, fdatas,
			order_begin_pos.buy_exec_index)

			if buy_progress_index is not None:
			buy_progress_index_changed = cls.__TradeBuyQueue.set_traded_index(code, buy_progress_index,
			total_datas)
			l2_log.info(code, logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{}", code, buy_progress_index)
			l2_log.info(code, logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{} is_similar-{}", code, buy_progress_index, is_similar)
			if is_placed_order:
			# NewGCancelBigNumComputer().set_trade_progress(code, order_begin_pos.buy_single_index,
			# buy_progress_index)
			@@ -325,18 +377,19 @@
			# 统计涨停主动卖成交，为了F撤准备数据
			HuaXinSellOrderStatisticManager.statistic_active_sell_deal_volume(code, fdatas, limit_up_price)
			# 计算成交进度
			buy_progress_index = cls.__compute_latest_trade_progress(code, fdatas)
			if buy_progress_index is not None:
			_buy_progress_index, _is_similar = cls.__compute_latest_trade_progress(code, fdatas,
			order_begin_pos.buy_exec_index)
			if _buy_progress_index is not None:
			total_datas = l2_data_util.local_today_datas.get(code)
			buy_progress_index_changed = cls.__TradeBuyQueue.set_traded_index(code, buy_progress_index,
			buy_progress_index_changed = cls.__TradeBuyQueue.set_traded_index(code, _buy_progress_index,
			total_datas)
			async_log_util.info(logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{}", code,
			buy_progress_index)
			async_log_util.info(logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{} similar-{}", code,
			_buy_progress_index, _is_similar)
			if is_placed_order:
			LCancelBigNumComputer().set_trade_progress(code, order_begin_pos.buy_single_index,
			buy_progress_index,
			_buy_progress_index,
			total_datas)
			cancel_result = FCancelBigNumComputer().need_cancel_for_deal_fast(code, buy_progress_index)
			cancel_result = FCancelBigNumComputer().need_cancel_for_deal_fast(code, _buy_progress_index)
			if cancel_result[0]:
			L2TradeDataProcessor.cancel_buy(code, f"F撤:{cancel_result[1]}",
			cancel_type=trade_constant.CANCEL_TYPE_F)
			@@ -404,7 +457,7 @@

			# 如果是被动买就更新成交进度
			if not fdatas[-1][1]:
			buy_progress_index = cls.__compute_latest_trade_progress(code, fdatas)
			buy_progress_index, is_similar = cls.__compute_latest_trade_progress(code, fdatas)
			if buy_progress_index is not None:
			total_datas = l2_data_util.local_today_datas.get(code)
			cls.__TradeBuyQueue.set_traded_index(code, buy_progress_index,