""" L2成交数据处理器 """ import json from db import redis_manager from db.redis_manager_delegate import RedisUtils from l2 import l2_log from l2.huaxin import l2_huaxin_util from log_module import async_log_util from log_module.log import hx_logger_l2_transaction_desc, hx_logger_l2_transaction_sell_order from utils import tool # 成交数据统计 class HuaXinBuyOrderManager: __db = 0 __instance = None __redis_manager = redis_manager.RedisManager(0) # 正在成交的订单 __dealing_order_info_dict = {} # 最近成交的订单{"code":(订单号，是否成交完成)} __latest_deal_order_info_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() try: 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) finally: RedisUtils.realse(__redis) # 将数据持久化到数据库 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): return cls.__dealing_order_info_dict.get(code) # 统计成交的情况 @classmethod def statistic_deal_desc(cls, code, data, total_buy_num): if code not in cls.__dealing_order_info_dict: # 数据格式[订单号，总股数，开始成交时间，结束成交时间, 总买] cls.__dealing_order_info_dict[code] = [data[6], 0, data[3], data[3], total_buy_num] if cls.__dealing_order_info_dict[code][0] == data[6]: # 成交同一个订单号 cls.__dealing_order_info_dict[code][1] += data[2] cls.__dealing_order_info_dict[code][3] = data[3] else: # 保存上一条数据 async_log_util.info(hx_logger_l2_transaction_desc, f"{code}#{cls.__dealing_order_info_dict[code]}") # 设置最近成交完成的一条数据 deal_info = ( cls.__dealing_order_info_dict[code][0], cls.__dealing_order_info_dict[code][4] == cls.__dealing_order_info_dict[code][1]) cls.__latest_deal_order_info_dict[code] = deal_info # 初始化本条数据 cls.__dealing_order_info_dict[code] = [data[6], data[2], data[3], data[3], total_buy_num] return deal_info return None # 卖单统计数据 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 = {} # 返回最近1s的大单卖：(总卖金额,[(卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）),...]) @classmethod def add_transaction_datas(cls, code, datas): # 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] = [] for d in datas: cls.__latest_sell_order_info_list_dict[code].append(d) if code not in cls.__latest_sell_order_dict: cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])] else: if cls.__latest_sell_order_dict[code][0] == d[7]: cls.__latest_sell_order_dict[code][1] += d[2] cls.__latest_sell_order_dict[code][2] = d[1] cls.__latest_sell_order_dict[code][4] = (d[3], d[6]) else: info = cls.__latest_sell_order_dict[code] # 上个卖单成交完成 # 封存数据，计算新起点 # 大于50w的卖单才会保存 # 大于50w加入卖单 if info[1] * info[2] >= 500000: l2_log.info(code, hx_logger_l2_transaction_sell_order, f"{cls.__latest_sell_order_dict[code]}") cls.__big_sell_order_ids_dict[code].add(info[0]) cls.__big_sell_order_info_dict[code][info[0]] = info cls.__big_sell_order_info_list_dict[code].append(info) cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])] latest_time = l2_huaxin_util.convert_time(datas[-1][3], with_ms=True) 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) start_index = 0 total_sell_info = [0, None] # 总资金,开始成交信息,结束成交信息 latest_sell_order_info = cls.__latest_sell_order_dict[code] 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 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]) if min_time_int > int( l2_huaxin_util.convert_time(bd[3][0], with_ms=True).replace(":", "").replace(".", "")): start_index = i break # 统计最近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: 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 cls.__latest_sell_order_info_list_dict[code] = cls.__latest_sell_order_info_list_dict[code][start_index:] return total_sell_info