# -*- coding: utf-8 -*-
|
import logging
|
import marshal
|
import queue
|
import threading
|
import time
|
|
from huaxin_client.code_queue_distribute_manager import CodeDataCallbackDistributeManager
|
from log_module.async_log_util import huaxin_l2_log
|
import collections
|
|
from log_module.log import logger_local_huaxin_contact_debug
|
|
order_detail_upload_active_time_dict = {}
|
transaction_upload_active_time_dict = {}
|
# 临时数据
|
tmep_order_detail_queue_dict = {}
|
tmep_transaction_queue_dict = {}
|
target_codes = set()
|
target_codes_add_time = {}
|
common_queue = queue.Queue()
|
|
|
# L2上传数据管理器
|
class L2DataUploadManager:
|
def __init__(self, data_callback_distribute_manager: CodeDataCallbackDistributeManager):
|
self.data_callback_distribute_manager = data_callback_distribute_manager
|
# 代码分配的对象
|
self.temp_order_queue_dict = {}
|
self.temp_transaction_queue_dict = {}
|
self.temp_log_queue_dict = {}
|
|
self.filter_order_condition_dict = {}
|
self.upload_l2_data_task_dict = {}
|
self.l2_order_codes = set()
|
self.l2_transaction_codes = set()
|
self.__real_time_buy1_data = {}
|
|
# 过滤订单
|
def __filter_order(self, item):
|
if item[1] * item[2] < 500000:
|
return None
|
return item
|
# 过滤订单
|
|
def __filter_transaction(self, item):
|
return item
|
|
# 添加委托详情
|
def add_l2_order_detail(self, data, start_time=0, istransaction=False):
|
code = data["SecurityID"]
|
if code in self.__real_time_buy1_data:
|
if self.__real_time_buy1_data[code][1] == data["Price"]:
|
# 与买的价格一致
|
if data["Side"] == '1':
|
if data["OrderStatus"] == 'D':
|
# 买撤
|
self.__real_time_buy1_data[code][3] -= data["Volume"]
|
else:
|
# 买
|
self.__real_time_buy1_data[code][3] += data["Volume"]
|
q: collections.deque = self.temp_order_queue_dict.get(code)
|
q.append((data['SecurityID'], data['Price'], data['Volume'], data['Side'], data['OrderType'], data['OrderTime'],
|
data['MainSeq'], data['SubSeq'], data['OrderNO'], data['OrderStatus'], time.time(), start_time))
|
|
# 添加逐笔成交
|
def add_transaction_detail(self, data):
|
code = data["SecurityID"]
|
if code in self.__real_time_buy1_data:
|
if self.__real_time_buy1_data[code][1] == data["TradePrice"]:
|
# 与买的价格一致
|
self.__real_time_buy1_data[code][3] -= data["TradeVolume"]
|
q: collections.deque = self.temp_transaction_queue_dict.get(code)
|
q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
|
data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
|
data['SellNo'], data['ExecType']))
|
|
def add_market_data(self, data):
|
code = data["securityID"]
|
# [时间,买1价格,原始买1量,计算后的买1量]}
|
self.__real_time_buy1_data[code] = [data["dataTimeStamp"], data["buy"][0][0], data["buy"][0][1],
|
data["buy"][0][1]]
|
|
self.data_callback_distribute_manager.get_distributed_callback(code).OnMarketData(code, [data])
|
|
# 分配上传队列
|
def distribute_upload_queue(self, code):
|
if not self.data_callback_distribute_manager.get_distributed_callback(code):
|
self.data_callback_distribute_manager.distribute_callback(code)
|
|
if code not in self.temp_order_queue_dict:
|
self.temp_order_queue_dict[code] = collections.deque()
|
if code not in self.temp_transaction_queue_dict:
|
self.temp_transaction_queue_dict[code] = collections.deque()
|
if code not in self.temp_log_queue_dict:
|
self.temp_log_queue_dict[code] = queue.Queue()
|
if code not in self.upload_l2_data_task_dict:
|
t1 = threading.Thread(target=lambda: self.__run_upload_order_task(code), daemon=True)
|
t1.start()
|
t2 = threading.Thread(target=lambda: self.__run_upload_transaction_task(code), daemon=True)
|
t2.start()
|
t3 = threading.Thread(target=lambda: self.__run_upload_real_time_buy1_task(code), daemon=True)
|
t3.start()
|
self.upload_l2_data_task_dict[code] = (t1, t2, t3)
|
# 释放已经分配的队列
|
|
def release_distributed_upload_queue(self, code):
|
self.data_callback_distribute_manager.release_distribute_callback(code)
|
if code in self.temp_order_queue_dict:
|
self.temp_order_queue_dict[code].clear()
|
self.temp_order_queue_dict.pop(code)
|
if code in self.temp_transaction_queue_dict:
|
self.temp_transaction_queue_dict[code].clear()
|
self.temp_transaction_queue_dict.pop(code)
|
if code in self.temp_log_queue_dict:
|
self.temp_log_queue_dict.pop(code)
|
|
if code in self.upload_l2_data_task_dict:
|
self.upload_l2_data_task_dict.pop(code)
|
|
def __upload_l2_data(self, code, _queue, datas):
|
_queue.put_nowait(marshal.dumps([code, datas, time.time()]))
|
|
# 处理订单数据并上传
|
def __run_upload_order_task(self, code):
|
q: collections.deque = self.temp_order_queue_dict.get(code)
|
temp_list = []
|
while True:
|
try:
|
while len(q) > 0:
|
data = q.popleft()
|
# 前置数据处理，过滤掉无用的数据
|
data = self.__filter_order(data)
|
if data:
|
temp_list.append(data)
|
|
if temp_list:
|
# 上传数据
|
# self.__upload_l2_data(code, upload_queue, temp_list)
|
# self.__upload_l2_order_data(code, temp_list)
|
__start_time = time.time()
|
last_data = temp_list[-1]
|
self.data_callback_distribute_manager.get_distributed_callback(code).OnL2Order(code, temp_list,
|
time.time())
|
use_time = time.time() - __start_time
|
if use_time > 0.01:
|
# 记录10ms以上的数据
|
huaxin_l2_log.info(logger_local_huaxin_contact_debug, f"耗时:{use_time}s 结束数据：{last_data}")
|
temp_list = []
|
else:
|
if code not in self.temp_order_queue_dict:
|
self.l2_order_codes.discard(code)
|
break
|
self.l2_order_codes.add(code)
|
time.sleep(0.001)
|
|
except Exception as e:
|
logging.exception(e)
|
finally:
|
pass
|
|
# 处理成交数据并上传
|
def __run_upload_transaction_task(self, code):
|
q: collections.deque = self.temp_transaction_queue_dict.get(code)
|
temp_list = []
|
while True:
|
try:
|
while len(q) > 0:
|
data = q.popleft()
|
data = self.__filter_transaction(data)
|
if data:
|
temp_list.append(data)
|
if temp_list:
|
# 上传数据
|
# self.__upload_l2_data(code, upload_queue, temp_list)
|
self.data_callback_distribute_manager.get_distributed_callback(code).OnL2Transaction(code,
|
temp_list)
|
temp_list = []
|
else:
|
if code not in self.temp_transaction_queue_dict:
|
self.l2_transaction_codes.discard(code)
|
break
|
self.l2_transaction_codes.add(code)
|
time.sleep(0.001)
|
except:
|
pass
|
finally:
|
pass
|
|
# 处理实时买1数据
|
def __run_upload_real_time_buy1_task(self, code):
|
while True:
|
try:
|
if code in self.__real_time_buy1_data:
|
data = self.__real_time_buy1_data[code]
|
# 如果最新的买1是原来买1的1/2时开始上传
|
if data[2] > 0 and data[3] / data[2] <= 0.5:
|
self.data_callback_distribute_manager.get_distributed_callback(code).OnRealTimeBuy1Info(code, data)
|
except:
|
pass
|
finally:
|
time.sleep(0.1)
|
|
|
def add_target_code(code):
|
target_codes.add(code)
|
# 记录代码加入时间
|
target_codes_add_time[code] = time.time()
|
|
|
def del_target_code(code):
|
target_codes.discard(code)
|
if code in target_codes_add_time:
|
target_codes_add_time.pop(code)
|
|
|
def add_subscript_codes(codes):
|
pass
|
|
|
if __name__ == "__main__":
|
add_subscript_codes(["000333"])
|
