""" L2成交数据处理器 """ import json import time import l2_data_util from db import redis_manager_delegate as redis_manager from db.redis_manager_delegate import RedisUtils from l2 import l2_log from l2.huaxin import l2_huaxin_util from l2.place_order_single_data_manager import L2TradeSingleDataProcessor, L2TradeSingleDataManager from log_module import log_export from log_module.log import hx_logger_l2_transaction_desc, hx_logger_l2_transaction_sell_order, hx_logger_l2_active_sell, \ hx_logger_l2_transaction_big_buy_order, hx_logger_l2_transaction_big_sell_order, hx_logger_l2_upload from utils import tool class BigOrderDealManager: """ 成交大单管理 """ __total_buy_datas_dict = {} __total_sell_datas_dict = {} __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(BigOrderDealManager, cls).__new__(cls, *args, **kwargs) cls.__load_datas() return cls.__instance @classmethod def __load_datas(cls): # 从日志加载数据 buy_order_map = log_export.load_huaxin_big_buy_order() if buy_order_map: for code in buy_order_map: cls.__total_buy_datas_dict[code] = buy_order_map[code] sell_order_map = log_export.load_huaxin_big_sell_order() if sell_order_map: for code in sell_order_map: cls.__total_sell_datas_dict[code] = sell_order_map[code] def add_buy_datas(self, code, datas): """ 添加大买单数据 @param code: 代码 @param datas:[(买单号,总股数,总成交额,成交开始时间,成交结束时间)] @return: """ if not datas: return if code not in self.__total_buy_datas_dict: self.__total_buy_datas_dict[code] = [] self.__total_buy_datas_dict[code].extend(datas) # 将数据保存到日志 l2_log.info(code, hx_logger_l2_transaction_big_buy_order, f"{datas}") def add_sell_datas(self, code, datas): """ 添加大卖单数据:只计算板上成交 @param code: 代码 @param datas:[(卖单号,总股数,总成交额)] @return: """ if not datas: return if code not in self.__total_sell_datas_dict: self.__total_sell_datas_dict[code] = [] self.__total_sell_datas_dict[code].extend(datas) l2_log.info(code, hx_logger_l2_transaction_big_sell_order, f"{datas}") def get_total_buy_money(self, code): """ 获取总共的大单买金额 @param code: @return: """ if code not in self.__total_buy_datas_dict: return 0 return int(sum([x[2] for x in self.__total_buy_datas_dict[code]])) def get_total_buy_order_ids(self, code): """ 获取成交大单的订单号 @param code: @return: """ if code not in self.__total_buy_datas_dict: return set() return set([x[0] for x in self.__total_buy_datas_dict[code]]) def get_total_buy_count(self, code): """ 获取大单成交的笔数 @param code: @return: """ if code not in self.__total_buy_datas_dict: return 0 return len(self.__total_buy_datas_dict[code]) def get_total_buy_money_list(self, code): """ 获取大单列表 @param code: @return: """ if code not in self.__total_buy_datas_dict: return [] return [x[2] for x in self.__total_buy_datas_dict[code]] def get_total_buy_data_list(self, code): """ 获取所有的买单数据列表 @param code: @return:[(买单号,总股数,总成交额,成交开始时间,成交结束时间)] """ return self.__total_buy_datas_dict.get(code) def get_total_sell_money(self, code, threshold_money=None): """ 获取总共的大单卖金额 @param threshold_money: 大单阈值 @param code: @return: """ if code not in self.__total_sell_datas_dict: return 0 if threshold_money: return int(sum([x[2] if x[2] >= threshold_money else 0 for x in self.__total_sell_datas_dict[code]])) else: return int(sum([x[2] for x in self.__total_sell_datas_dict[code]])) # 成交数据统计 class HuaXinBuyOrderManager: __db = 0 __instance = None __redis_manager = redis_manager.RedisManager(0) # 正在成交的订单 __dealing_order_info_dict = {} # 正在成交的主动买的订单 __dealing_active_buy_order_info_dict = {} # 最近成交的订单{"code":(订单号，是否成交完成)} __latest_deal_order_info_dict = {} # 成交的买单号集合：{"代码":set()} __deal_buy_order_nos_dict = {} def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(HuaXinBuyOrderManager, cls).__new__(cls, *args, **kwargs) cls.__load_datas() return cls.__instance @classmethod def __get_redis(cls): return cls.__redis_manager.getRedis() @classmethod def __load_datas(cls): __redis = cls.__get_redis() keys = RedisUtils.keys(__redis, "dealing_order_info-*") for k in keys: code = k.split("-")[-1] val = RedisUtils.get(__redis, k) val = json.loads(val) tool.CodeDataCacheUtil.set_cache(cls.__dealing_order_info_dict, code, val) # 读取已经成交的大单 deal_data_list_dict = log_export.load_huaxin_deal_record_all() for code in deal_data_list_dict: cls.__deal_buy_order_nos_dict[code] = set([x[0] for x in deal_data_list_dict[code]]) # 将数据持久化到数据库 def sync_dealing_data_to_db(self): for code in self.__dealing_order_info_dict: RedisUtils.setex(self.__get_redis(), f"dealing_order_info-{code}", tool.get_expire(), json.dumps(self.__dealing_order_info_dict[code])) @classmethod def get_dealing_order_info(cls, code): """ 获取当前正在成交的数据 @param code: @return: [订单号，总股数，成交金额,成交开始时间,成交结束时间] """ return cls.__dealing_order_info_dict.get(code) @classmethod def get_dealing_active_order_info(cls, code): """ 获取正在主动成交的数据 @param code: @return:[订单号，总股数，成交金额,成交开始时间,成交结束时间] """ return cls.__dealing_active_buy_order_info_dict.get(code) @classmethod def statistic_big_buy_data(cls, code, fdatas, limit_up_price): """ 统计大单买 @param code: @param fdatas: [(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)] @return: 返回数据里面(成交的大单,50w以上的单） """ big_buy_datas = [] normal_buy_datas = [] # 大单阈值 threshold_big_money = l2_data_util.get_big_money_val(limit_up_price, tool.is_ge_code(code)) dealing_order_info = cls.__dealing_order_info_dict.get(code) if code not in cls.__deal_buy_order_nos_dict: cls.__deal_buy_order_nos_dict[code] = set() deal_buy_order_nos = cls.__deal_buy_order_nos_dict[code] for data in fdatas: # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType'])) is_limit_up = data[2] money = data[3] current_order_info = data[0] if dealing_order_info is None: # 数据格式[订单号，总股数，成交金额,成交开始时间,成交结束时间, 最近的成交价格, 最近的卖单号, 涨停价成交金额] dealing_order_info = [current_order_info[6], current_order_info[2], money, current_order_info[3], current_order_info[3], current_order_info[1], current_order_info[7], 0] if is_limit_up: dealing_order_info[7] += money cls.__dealing_order_info_dict[code] = dealing_order_info else: if dealing_order_info[0] == current_order_info[6]: # 成交同一个订单号 dealing_order_info[1] += current_order_info[2] dealing_order_info[2] += money dealing_order_info[4] = current_order_info[3] dealing_order_info[5] = current_order_info[1] dealing_order_info[6] = current_order_info[7] if is_limit_up: dealing_order_info[7] += money else: # 保存上一条数据 l2_log.info(code, hx_logger_l2_transaction_desc, f"{code}#{dealing_order_info}") # 设置最近成交完成的一条数据 deal_info = dealing_order_info cls.__latest_deal_order_info_dict[code] = deal_info # 是否为大买单 if deal_info[7] >= threshold_big_money: # # 如果最后一笔成交价格不是涨停价就不算 # if abs(deal_info[5] - limit_up_price) < 0.0001: big_buy_datas.append(deal_info) if deal_info[2] >= 500000: deal_buy_order_nos.add(deal_info[0]) normal_buy_datas.append(deal_info) # 初始化本条数据 dealing_order_info = [current_order_info[6], current_order_info[2], money, current_order_info[3], current_order_info[3], current_order_info[1], current_order_info[7], 0] if is_limit_up: dealing_order_info[7] += money cls.__dealing_order_info_dict[code] = dealing_order_info # 统计主动买(买单号大于卖单号) if data[1]: if code not in cls.__dealing_active_buy_order_info_dict: # 数据格式[订单号，总股数，成交金额,成交开始时间,成交结束时间] cls.__dealing_active_buy_order_info_dict[code] = [current_order_info[6], current_order_info[2], money, current_order_info[3], current_order_info[3]] else: dealing_active_buy_order_info = cls.__dealing_active_buy_order_info_dict[code] if dealing_active_buy_order_info[0] == current_order_info[6]: # 成交同一个订单号 dealing_active_buy_order_info[1] += current_order_info[2] dealing_active_buy_order_info[2] += money dealing_active_buy_order_info[4] = current_order_info[3] else: # 初始化本条数据 cls.__dealing_active_buy_order_info_dict[code] = [current_order_info[6], current_order_info[2], money, current_order_info[3], current_order_info[3]] return big_buy_datas, normal_buy_datas def get_deal_buy_order_nos(self, code): """ 获取成交大单集合 @param code: @return: """ return self.__deal_buy_order_nos_dict.get(code) # 卖单统计数据 class HuaXinSellOrderStatisticManager: # 最近的大卖单成交，格式:{code:[卖单信息,...]} # __latest_sell_order_info_list_dict = {} # 大单卖单号的集合，格式：{code:{卖单号}} __big_sell_order_ids_dict = {} # 大卖单的卖单号->卖单信息映射 __big_sell_order_info_dict = {} # 大单列表 __big_sell_order_info_list_dict = {} # 最近的卖单, 格式{code:[卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）]} __latest_sell_order_dict = {} # 最近所有的卖单 __latest_all_sell_orders_dict = {} # 保存最近成交的价格 __latest_trade_price_dict = {} __last_trade_data_dict = {} # 用来暂时存放统计结果 __dealing_order_info_dict = {} @classmethod def statistic_big_sell_data(cls, code, fdatas): """ 统计大卖单(299w以上) @param code: @param fdatas: [(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)] @return: 返回数据里面成交的大单 """ big_sell_datas = [] # 正在成交的卖单 dealing_order_info = cls.__dealing_order_info_dict.get(code) for data in fdatas: if not data[2]: # 只计算板上成交 continue money = data[3] # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType'])) if dealing_order_info is None: # 数据格式[订单号，总股数，成交金额] dealing_order_info = [data[0][7], data[0][2], money] else: if dealing_order_info[0] == data[0][7]: # 成交同一个订单号 dealing_order_info[1] += data[0][2] dealing_order_info[2] += money else: # 保存上一条数据 # 设置最近成交完成的一条数据 deal_info = dealing_order_info # 保存上一条数据 if deal_info[2] >= 2990000: # 是否为大买单 l2_log.info(code, hx_logger_l2_transaction_desc, f"{code}#{deal_info}") big_sell_datas.append(deal_info) # 初始化本条数据 dealing_order_info = [data[0][7], data[0][2], money] # 保存最后一条数据的状态 cls.__dealing_order_info_dict[code] = dealing_order_info return big_sell_datas # 统计所有的成交量 __deal_volume_list_dict = {} @classmethod def statistic_active_sell_deal_volume(cls, code, fdatas, limit_up_price): """ 统计主动卖成交 @param code: @param fdatas: [(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)] @param limit_up_price: @return: """ # 只统计被动买 if code not in cls.__deal_volume_list_dict: cls.__deal_volume_list_dict[code] = [] # 为了加速处理，如果第一条数据和最后一条数据都是主动买就返回 if fdatas[0][1] and fdatas[-1][1]: return for d in fdatas: # 只统计被动买 if d[1]: continue time_str = d[4] if cls.__deal_volume_list_dict[code]: if cls.__deal_volume_list_dict[code][-1][0] == time_str: # 如果是同一秒 cls.__deal_volume_list_dict[code][-1][1] += d[0][2] else: # 不是同一秒 cls.__deal_volume_list_dict[code].append([time_str, d[0][2]]) else: cls.__deal_volume_list_dict[code].append([time_str, d[0][2]]) # 删除超过5条数据 if len(cls.__deal_volume_list_dict[code]) > 5: cls.__deal_volume_list_dict[code] = cls.__deal_volume_list_dict[code][-5:] @classmethod def get_latest_3s_continue_deal_volumes(cls, code): """ 获取最近3s的成交量分布 @param code: @return: [(时间,量)] """ deal_list = cls.__deal_volume_list_dict.get(code) if not deal_list: return [] fdatas = [deal_list[-1]] # 从倒数第二个数据计算 for i in range(len(deal_list) - 2, -1, -1): if tool.trade_time_sub(fdatas[0][0], deal_list[i][0]) < 3: fdatas.append(deal_list[i]) return fdatas @classmethod def get_latest_2s_continue_deal_volumes(cls, code): """ 获取最近2s的成交量分布 @param code: @return: [(时间,量)] """ deal_list = cls.__deal_volume_list_dict.get(code) if not deal_list: return [] fdatas = [deal_list[-1]] # 从倒数第二个数据计算 for i in range(len(deal_list) - 1, -1, -1): if tool.trade_time_sub(fdatas[0][0], deal_list[i][0]) < 2: fdatas.append(deal_list[i]) return fdatas @classmethod def clear_latest_deal_volume(cls, code): if code in cls.__deal_volume_list_dict: cls.__deal_volume_list_dict.pop(code) # 返回最近1s的大单卖：(总卖金额,[(卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）),...]) @classmethod def statistic_continue_limit_up_sell_transaction_datas(cls, code, fdatas, limit_up_price=None): """ 统计连续涨停卖成交的数据 @param code: @param fdatas: [(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)] @param limit_up_price: @return: """ f_start_time = time.time() use_time_list = [] # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'], # data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'], # data['SellNo'], data['ExecType'])) # if code not in cls.__latest_sell_order_info_list_dict: # cls.__latest_sell_order_info_list_dict[code] = [] if code not in cls.__big_sell_order_ids_dict: cls.__big_sell_order_ids_dict[code] = set() if code not in cls.__big_sell_order_info_dict: cls.__big_sell_order_info_dict[code] = {} if code not in cls.__big_sell_order_info_list_dict: cls.__big_sell_order_info_list_dict[code] = [] if code not in cls.__latest_all_sell_orders_dict: cls.__latest_all_sell_orders_dict[code] = [] # 保存最近的成交价格:(价格,成交时间) cls.__latest_trade_price_dict[code] = (fdatas[-1][0][1], fdatas[-1][0][3]) __start_time = time.time() # 是否还有涨停卖剩下 no_left_limit_up_sell = L2TradeSingleDataProcessor.process_passive_limit_up_sell_data(code, fdatas) use_time = time.time() - __start_time __start_time = time.time() use_time_list.append(("处理涨停卖", use_time)) latest_sell_order = cls.__latest_sell_order_dict.get(code) big_sell_order_ids = cls.__big_sell_order_ids_dict.get(code) if big_sell_order_ids is None: big_sell_order_ids = set() big_sell_order_info = cls.__big_sell_order_info_dict.get(code) if big_sell_order_info is None: big_sell_order_info = {} big_sell_order_info_list = cls.__big_sell_order_info_list_dict.get(code) if big_sell_order_info_list is None: big_sell_order_info_list = [] latest_all_sell_orders = cls.__latest_all_sell_orders_dict.get(code) if latest_all_sell_orders is None: latest_all_sell_orders = [] last_trade_data = cls.__last_trade_data_dict.get(code) for d in fdatas: # 获取当前是否为主动买 if d[1]: # 主动买 continue try: if d[2]: # 涨停主动卖 L2TradeSingleDataProcessor.add_active_limit_up_sell_data(d[0]) # 判断是否是涨停被动变主动 if last_trade_data and last_trade_data[1] and last_trade_data[2]: if d[2]: # 涨停被动变主动 L2TradeSingleDataManager.set_sell_passive_to_active_datas(code, last_trade_data[0], d[0]) # cls.__latest_sell_order_info_list_dict[code].append(d) if latest_sell_order is None: latest_sell_order = [d[0][7], d[0][2], d[0][1], (d[0][3], d[0][6]), (d[0][3], d[0][6])] else: if latest_sell_order[0] == d[0][7]: latest_sell_order[1] += d[0][2] latest_sell_order[2] = d[0][1] latest_sell_order[4] = (d[0][3], d[0][6]) else: info = latest_sell_order # 上个卖单成交完成 # 封存数据，计算新起点 # 大于50w的卖单才会保存 # 大于50w加入卖单 money = info[1] * info[2] if money >= 500000: big_sell_order_ids.add(info[0]) big_sell_order_info[info[0]] = info big_sell_order_info_list.append(info) # 只保留10w以上的单 if money > 100000: latest_all_sell_orders.append(info) l2_log.info(code, hx_logger_l2_transaction_sell_order, f"{info}") if limit_up_price == info[2]: # 将涨停主动卖记入日志 l2_log.info(code, hx_logger_l2_active_sell, f"{info}") latest_sell_order = [d[0][7], d[0][2], d[0][1], (d[0][3], d[0][6]), (d[0][3], d[0][6])] finally: last_trade_data = d cls.__last_trade_data_dict[code] = last_trade_data if latest_sell_order: cls.__latest_sell_order_dict[code] = latest_sell_order if big_sell_order_ids: cls.__big_sell_order_ids_dict[code] = big_sell_order_ids if big_sell_order_info: cls.__big_sell_order_info_dict[code] = big_sell_order_info if big_sell_order_info_list: cls.__big_sell_order_info_list_dict[code] = big_sell_order_info_list if latest_all_sell_orders: cls.__latest_all_sell_orders_dict[code] = latest_all_sell_orders use_time = time.time() - __start_time __start_time = time.time() use_time_list.append(("大单统计", use_time)) latest_time = fdatas[-1][5] min_time = tool.trade_time_add_millionsecond(latest_time, -1000) min_time_int = int(min_time.replace(":", "").replace(".", "")) # 计算最近1s的大单成交 total_big_sell_datas = cls.__big_sell_order_info_list_dict.get(code) total_sell_info = [0, None] # 总资金,开始成交信息,结束成交信息 latest_sell_order_info = cls.__latest_sell_order_dict.get(code) if latest_sell_order_info: # 不是第一次非主动卖上传 big_sell_order_ids = cls.__big_sell_order_ids_dict[code] # print("大卖单", big_sell_order_ids) big_sell_orders = [] temp_sell_order_ids = set() # 统计已经结算出的大单 for i in range(len(total_big_sell_datas) - 1, -1, -1): bd = total_big_sell_datas[i] if min_time_int > int( l2_huaxin_util.convert_time(bd[3][0], with_ms=True).replace(":", "").replace(".", "")): break if bd[0] != latest_sell_order_info[0]: # 不是最近的成交且不是大单直接过滤 if bd[0] not in big_sell_order_ids: continue else: if bd[0] not in temp_sell_order_ids: big_sell_orders.append(cls.__big_sell_order_info_dict[code].get(bd[0])) temp_sell_order_ids.add(bd[0]) else: # 是最近的但不是大单需要过滤 if latest_sell_order_info[1] * latest_sell_order_info[2] < 500000: continue else: if latest_sell_order_info[0] not in temp_sell_order_ids: big_sell_orders.append(latest_sell_order_info) temp_sell_order_ids.add(latest_sell_order_info[0]) # 统计最近1s的大卖单数据 total_sell_info[0] += int(bd[1] * bd[2]) # 统计最近的大单 if latest_sell_order_info[1] * latest_sell_order_info[2] >= 500000: if latest_sell_order_info[0] not in temp_sell_order_ids: # if is_active_sell(latest_sell_order_info[0], latest_sell_order_info[3][1]): big_sell_orders.append(latest_sell_order_info) temp_sell_order_ids.add(latest_sell_order_info[0]) total_sell_info[0] += int(latest_sell_order_info[1] * latest_sell_order_info[2]) big_sell_orders.reverse() total_sell_info[1] = big_sell_orders use_time = time.time() - __start_time __start_time = time.time() use_time_list.append(("最近大单统计", use_time)) # ----------------统计涨停主动买----------------- try: limit_up_active_buy_datas = [] for d in fdatas: if not d[1]: # 被动买 continue # 是否是涨停 if d[2]: # 有涨停主动买 limit_up_active_buy_datas.append(d[0]) L2TradeSingleDataManager.set_limit_up_active_buy(code, limit_up_active_buy_datas, no_left_limit_up_sell) use_time = time.time() - __start_time __start_time = time.time() use_time_list.append(("涨停主动买成交", use_time)) except: pass use_time = time.time() - f_start_time if use_time > 0.01: l2_log.info(code, hx_logger_l2_upload, f"{code}处理成交详细用时：{use_time} 数据数量：{len(fdatas)} 详情:{use_time_list}") return total_sell_info # 获取最近成交数据 @classmethod def get_latest_transaction_datas(cls, code, min_sell_order_no=None, min_deal_time=None, min_sell_money=None): """ 获取最近的主动卖成交信息 @param code: @param min_sell_order_no: @param min_deal_time: @param min_sell_money: @return: """ total_orders = [] sell_orders = cls.__latest_all_sell_orders_dict.get(code) if sell_orders: for i in range(len(sell_orders) - 1, -1, -1): if min_deal_time and tool.trade_time_sub(min_deal_time, l2_huaxin_util.convert_time(sell_orders[i][3][0])) > 0: break if min_sell_order_no and min_sell_order_no > sell_orders[i][0]: continue if min_sell_money and sell_orders[i][1] * sell_orders[i][2] < min_sell_money: # 过滤小金额 continue total_orders.append(sell_orders[i]) if code in cls.__latest_sell_order_dict: if min_sell_order_no: if cls.__latest_sell_order_dict[code][0] >= min_sell_order_no: total_orders.append(cls.__latest_sell_order_dict[code]) else: total_orders.append(cls.__latest_sell_order_dict[code]) return total_orders # 获取最近成交价格信息, 返回格式：（价格,时间） @classmethod def get_latest_trade_price_info(cls, code): return cls.__latest_trade_price_dict.get(code) if __name__ == "__main__": code = "603300" order_nos = HuaXinBuyOrderManager().get_deal_buy_order_nos(code) print(order_nos)