""" 下单信号管理 """ import time from l2 import l2_log from l2.huaxin import l2_huaxin_util from log_module import async_log_util from log_module.log import logger_l2_trade_buy, logger_debug from utils import tool class L2TradeSingleDataProcessor: """ L2逐笔成交数据信号管理 """ __latest_sell_data = {} # 最近的涨停卖委托列表 __latest_limit_up_sell_list_dict = {} __latest_limit_up_sell_order_no_set_dict = {} # 主动卖订单号集合 __active_sell_order_no_set_dict = {} # 主动买单数据,格式：{"code":[买单号，当前成交手数, 当前成交金额, 开始时间, 结束时间]} __latest_active_buy_order_data_dict = {} # 所有的买单数据，格式：{code:[[买单数据],...]} __active_buy_order_datas_dict = {} @classmethod def clear_passive_sell_data(cls, code): """ 清除被动卖相关的数据 @param code: @return: """ if code in cls.__latest_active_buy_order_data_dict: cls.__latest_active_buy_order_data_dict.pop(code) if code in cls.__active_buy_order_datas_dict: cls.__active_buy_order_datas_dict.pop(code) if code in cls.__latest_limit_up_sell_list_dict: cls.__latest_limit_up_sell_list_dict.pop(code) if code in cls.__latest_limit_up_sell_order_no_set_dict: cls.__latest_limit_up_sell_order_no_set_dict.pop(code) @classmethod def add_l2_delegate_limit_up_sell(cls, code, data): """ 添加涨停卖单数据，当前不是涨停成交时才加 @param code: @param data: @return: """ if code not in cls.__latest_limit_up_sell_list_dict: cls.__latest_limit_up_sell_list_dict[code] = [] cls.__latest_limit_up_sell_list_dict[code].append(data) # async_log_util.info(logger_debug, f"添加涨停卖数据：{code}-{data}") if code not in cls.__latest_limit_up_sell_order_no_set_dict: cls.__latest_limit_up_sell_order_no_set_dict[code] = set() cls.__latest_limit_up_sell_order_no_set_dict[code].add(data['val']['orderNo']) # 只保留前20的数据 if len(cls.__latest_limit_up_sell_list_dict[code]) > 20: delete_datas = cls.__latest_limit_up_sell_list_dict[code][:-20] cls.__latest_limit_up_sell_list_dict[code] = cls.__latest_limit_up_sell_list_dict[code][-20:] # 删除之前的map for d in delete_datas: cls.__latest_limit_up_sell_order_no_set_dict[code].discard(d["val"]["orderNo"]) @classmethod def add_l2_delegate_limit_up_sell_cancel(cls, code, order_no): """ 涨停卖撤 @param code: @param order_no: @return: """ if code not in cls.__latest_limit_up_sell_order_no_set_dict: return order_no_set = cls.__latest_limit_up_sell_order_no_set_dict[code] if order_no not in order_no_set: return cls.__latest_limit_up_sell_order_no_set_dict[code].discard(order_no) for i in range(0, len(cls.__latest_limit_up_sell_list_dict[code])): if cls.__latest_limit_up_sell_list_dict[code][i]['val']['orderNo'] == order_no: cls.__latest_limit_up_sell_list_dict[code].pop(i) break @classmethod def get_latest_limit_up_sell_order_count(cls, code): sell_list = cls.__latest_limit_up_sell_list_dict.get(code) if not sell_list: return 0 return len(sell_list) @classmethod def process_passive_limit_up_sell_data(cls, code, fdatas): """ 添加涨停被动卖成交数据 @param fdata: 数据格式：(data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType']) [(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)] @return: 涨停卖是否已经吃完 """ try: start_time = time.time() sell_list = cls.__latest_limit_up_sell_list_dict.get(code) if not sell_list: return False last_sell_info = sell_list[-1] for data in fdatas: # 过滤被动买 if not data[1]: # 出现被动买需要将历史大单清空 if cls.__active_buy_order_datas_dict.get(code): cls.__active_buy_order_datas_dict[code].clear() continue money = data[3] # 统计买单数据 if code not in cls.__latest_active_buy_order_data_dict: # [买单号，当前成交股数, 当前成交金额, 开始时间, 结束时间] cls.__latest_active_buy_order_data_dict[code] = [data[0][6], data[0][2], money, data[0][3], data[0][3]] else: if data[0][6] == cls.__latest_active_buy_order_data_dict[code][0]: # 同一买单号 cls.__latest_active_buy_order_data_dict[code][1] += data[0][2] cls.__latest_active_buy_order_data_dict[code][2] += money cls.__latest_active_buy_order_data_dict[code][4] = data[0][3] else: # 不同买单号 if cls.__latest_active_buy_order_data_dict[code][2] >= 2990000: # 加入大单列表 if code not in cls.__active_buy_order_datas_dict: cls.__active_buy_order_datas_dict[code] = [] cls.__active_buy_order_datas_dict[code].append( cls.__latest_active_buy_order_data_dict[code]) cls.__latest_active_buy_order_data_dict[code] = [data[0][6], data[0][2], money, data[0][3], data[0][3]] if not data[2]: # 排除主动卖/非涨停卖 continue sell_no = data[0][7] if sell_no != last_sell_info['val']['orderNo']: continue # 需要判断当前单是否已经成交完成 if code not in cls.__latest_sell_data: cls.__latest_sell_data[code] = [sell_no, data[0][2]] else: if cls.__latest_sell_data[code][0] == sell_no: cls.__latest_sell_data[code][1] += data[0][2] else: cls.__latest_sell_data[code] = [sell_no, data[0][2]] sell_info_num = last_sell_info['val']['num'] deal_num = cls.__latest_sell_data[code][1] // 100 if sell_info_num == deal_num: use_time = round((time.time() - start_time) * 1000, 3) l2_log.info(code, logger_l2_trade_buy, f"找到最近的被动涨停卖单数据：{last_sell_info['val']['orderNo']}, 成交数据：{data} 计算耗时：{use_time}ms, 可以触发下单") # 将历史大单列表与最近的大单加入列表 big_buy_order_datas = [] if code in cls.__active_buy_order_datas_dict: big_buy_order_datas.extend(cls.__active_buy_order_datas_dict[code]) if cls.__latest_active_buy_order_data_dict[code][2] >= 2990000: big_buy_order_datas.append(cls.__latest_active_buy_order_data_dict[code]) # 成交完成 L2TradeSingleDataManager.set_latest_sell_data(code, data, big_buy_order_datas) l2_log.info(code, logger_l2_trade_buy, "被动卖数据处理完毕") if tool.is_sz_code(code): # 涨停主动卖已经被吃完，可以清除 return True break except Exception as e: logger_debug.exception(e) return False @classmethod def filter_last_limit_up_sell_data(cls, code, fdatas): """ 筛选出最后一条涨停卖成交数据 @param code: @param fdatas: @return: (成交数据, 卖单数据) """ def compute_last_sell(): # 最大买单号 max_buy_order_no = fdatas[-1][0][6] for i in range(len(sell_list) - 1, -1, -1): if sell_list[i]['val']['orderNo'] > max_buy_order_no: continue return sell_list[i] return None if not fdatas[-1][2]: # 最后一条数据不是涨停成交数据 return None sell_list = cls.__latest_limit_up_sell_list_dict.get(code) if not sell_list: return None last_sell_info = compute_last_sell() if not last_sell_info: return None for data in fdatas: if not data[2]: # 排除主动卖/非涨停卖 continue sell_no = data[0][7] if sell_no != last_sell_info['val']['orderNo']: continue # 需要判断当前单是否已经成交完成 if code not in cls.__latest_sell_data: cls.__latest_sell_data[code] = [sell_no, data[0][2]] else: if cls.__latest_sell_data[code][0] == sell_no: cls.__latest_sell_data[code][1] += data[0][2] else: cls.__latest_sell_data[code] = [sell_no, data[0][2]] sell_info_num = last_sell_info['val']['num'] deal_num = cls.__latest_sell_data[code][1] // 100 if sell_info_num == deal_num: # 最后一笔涨停卖已经成交完成 l2_log.info(code, logger_l2_trade_buy, f"找到最近的被动涨停卖单数据：{last_sell_info}, 成交数据：{data} 可以触发下单") return data, last_sell_info return None @classmethod def add_active_limit_up_sell_data(cls, data): """ 添加主动卖数据 @param data: @return: """ try: code = data[0] sell_no = data[7] if code not in cls.__active_sell_order_no_set_dict: cls.__active_sell_order_no_set_dict[code] = set() cls.__active_sell_order_no_set_dict[code].add(sell_no) except Exception as e: logger_debug.exception(e) class L2TradeSingleCallback: """ 交易信号回调 """ def OnTradeSingle(self, code, big_buy_order_count, _type, data): """ 交易数据信号回调 @param code: @param big_buy_order_count: 大买单数量 @param _type: 类型：0-TYPE_PASSIVE 1-TYPE_ACTIVE @param data: (逐笔成交数据,生效时间) @return: """ def OnLimitUpActiveBuy(self, code, transaction_data, no_left_limit_up_sell): """ 涨停主动买触发 @param code: @param transaction_data: 成交数据 @param no_left_limit_up_sell: 是否还剩余涨停卖 @return: """ def OnLastLimitUpSellDeal(self, code, data): """ 最后一笔涨停卖成交 @param code: 代码 @param data: 成交的数据 @return: """ class L2TradeSingleDataManager: __callback = None TYPE_PASSIVE = 0 TYPE_ACTIVE = 1 """ 买入信号管理 """ # 最近的涨停卖被动成交数据 __latest_sell_data_dict = {} # 格式为：{code:(逐笔成交数据,生效时间（带ms）)} # 由被动向主动卖成交转变的数据 __latest_sell_active_deal_data_dict = {} # 格式为：{code:(逐笔成交数据,生效时间（带ms）)} @classmethod def set_callback(cls, callback: L2TradeSingleCallback): cls.__callback = callback @classmethod def set_latest_sell_data(cls, code, fdata, big_active_buy_order_datas): """ 设置最近成交的涨停卖被动成交数据 @param code: 代码 @param data: L2逐笔成交数据 数据格式：(data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType']) @param big_active_buy_order_datas: 大主动买单数据：[[买单号，当前成交股数, 当前成交金额, 开始时间, 结束时间],....] @return: """ deal_time = fdata[5] # 生效时间在1s以内 cls.__latest_sell_data_dict[code] = (fdata[0], tool.trade_time_add_millionsecond(deal_time, 1000)) if cls.__callback: big_buy_order_count = 0 if big_active_buy_order_datas: for b in big_active_buy_order_datas: if b[0] > fdata[0][7]: # 买单在卖单之后 big_buy_order_count += 1 cls.__callback.OnTradeSingle(code, big_buy_order_count, cls.TYPE_PASSIVE, cls.__latest_sell_data_dict[code]) @classmethod def set_sell_passive_to_active_datas(cls, code, passive_data, active_data): """ 设置被动卖成交向主动卖成交的转变数据 @param code: 代码 @param passive_data: 被动卖成交逐笔 @param active_data: 主动卖成交逐笔 @return: """ # 暂时不需要生效 l2_log.info(code, logger_l2_trade_buy, f"被动卖变主动卖：{passive_data} => {active_data}") # deal_time = l2_huaxin_util.convert_time(passive_data[3], True) # # 生效时间在1s以内 # cls.__latest_sell_active_deal_data_dict[code] = ( # active_data, tool.trade_time_add_millionsecond(deal_time, 1000)) # if cls.__callback: # cls.__callback.OnTradeSingle(code, 0, cls.TYPE_ACTIVE, cls.__latest_sell_active_deal_data_dict[code]) @classmethod def set_limit_up_active_buy(cls, code, transaction_datas, no_left_limit_up_sell): if transaction_datas: cls.__callback.OnLimitUpActiveBuy(code, transaction_datas, no_left_limit_up_sell) @classmethod def get_valid_trade_single(cls, code, latest_time_with_ms): """ 获取有效的成交下单信号 @param code: @param latest_time_with_ms: @return: （逐笔成交数据, 类型） """ # 如果有最近卖涨停成交完就用最近的数据 data = cls.__latest_sell_data_dict.get(code) if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0: return data, cls.TYPE_PASSIVE data = cls.__latest_sell_active_deal_data_dict.get(code) if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0: return data, cls.TYPE_ACTIVE return None @classmethod def is_can_place_order(cls, code, buy_data): """ 是否可以下单 @param code: 代码 @param buy_data: L2逐笔委托大于50w的买入数据 @return: (是否可以下单, 原因) """ buy_data_time_with_ms = tool.to_time_with_ms(buy_data['val']['time'], buy_data['val']['tms']) single = cls.get_valid_trade_single(code, buy_data_time_with_ms) if not single: return False, "没有找到可用信号" if single[0][0][6] < buy_data['val']['orderNo']: # 信号位置的成交买单号要小于买入数据的买单号才行 return True, f"找到信号：{single}" else: return False, f"买单没有在信号位之后 信号：{single}" @classmethod def clear_data(cls, code): if code in cls.__latest_sell_data_dict: cls.__latest_sell_data_dict.pop(code) if code in cls.__latest_sell_active_deal_data_dict: cls.__latest_sell_active_deal_data_dict.pop(code) L2TradeSingleDataProcessor.clear_passive_sell_data(code)