"""
|
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 l2.l2_data_util import local_today_sellno_map, local_today_datas
|
from l2.place_order_single_data_manager import L2TradeSingleManager
|
|
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 = {}
|
# 最近所有的卖单
|
__latest_all_sell_orders_dict = {}
|
|
# 保存最近成交的价格
|
__latest_trade_price_dict = {}
|
|
# 返回最近1s的大单卖：(总卖金额,[(卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）),...])
|
@classmethod
|
def add_transaction_datas(cls, code, datas, limit_up_price=None):
|
# 是否为主动卖
|
def is_active_sell(sell_no, buy_no):
|
return sell_no > buy_no
|
|
# 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] = []
|
|
sell_no_map = local_today_sellno_map.get(code)
|
total_datas = local_today_datas.get(code)
|
if not sell_no_map:
|
sell_no_map = {}
|
|
# 保存最近的成交价格:(价格,成交时间)
|
cls.__latest_trade_price_dict[code] = (datas[-1][1], datas[-1][3])
|
|
for d in datas:
|
# 获取当前是否为主动买
|
_is_active_sell = is_active_sell(d[7], d[6])
|
if not _is_active_sell and d[1] == limit_up_price:
|
# 被动涨停卖,这个卖的订单是否在最近的涨停卖列表中
|
L2TradeSingleManager.process_passive_limit_up_sell_data(d)
|
|
if not _is_active_sell:
|
continue
|
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加入卖单
|
money = info[1] * info[2]
|
if money >= 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)
|
# 只保留10w以上的单
|
if money > 100000:
|
cls.__latest_all_sell_orders_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)
|
|
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
|
return total_sell_info
|
|
# 获取最近成交数据
|
@classmethod
|
def get_latest_transaction_datas(cls, code, min_sell_order_no=None, min_deal_time=None):
|
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
|
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)
|
