# -*- coding: utf-8 -*-
|
import json
|
import logging
|
import multiprocessing
|
import queue
|
import threading
|
import time
|
from huaxin_client import socket_util
|
|
from huaxin_client.client_network import SendResponseSkManager
|
|
# 活动时间
|
from huaxin_client.code_queue_distribute_manager import CodeQueueDistributeManager
|
from log_module import async_log_util
|
from log_module.log import logger_local_huaxin_l2_error, logger_system
|
from utils import tool
|
|
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, order_queue_distribute_manager: CodeQueueDistributeManager,
|
transaction_queue_distribute_manager: CodeQueueDistributeManager,
|
market_data_queue: multiprocessing.Queue):
|
self.order_queue_distribute_manager = order_queue_distribute_manager
|
self.transaction_queue_distribute_manager = transaction_queue_distribute_manager
|
self.market_data_queue = market_data_queue
|
|
# 添加委托详情
|
def add_l2_order_detail(self, data, start_time, istransaction=False):
|
code = data["SecurityID"]
|
queue_info = self.order_queue_distribute_manager.get_distributed_queue(code)
|
if not queue_info:
|
return
|
queue_info[1].put_nowait(
|
(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"]
|
queue_info = self.transaction_queue_distribute_manager.get_distributed_queue(code)
|
if not queue_info:
|
return
|
# 判断是否为大单成交
|
queue_info[1].put_nowait((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
|
data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
|
data['SellNo'], data['ExecType']))
|
|
def add_market_data(self, data):
|
# 加入上传队列
|
self.market_data_queue.put_nowait(data)
|
|
# 分配上传队列
|
def distribute_upload_queue(self, code):
|
self.order_queue_distribute_manager.distribute_queue(code)
|
self.transaction_queue_distribute_manager.distribute_queue(code)
|
|
# 释放已经分配的队列
|
def release_distributed_upload_queue(self, code):
|
self.order_queue_distribute_manager.release_distribute_queue(code)
|
self.transaction_queue_distribute_manager.release_distribute_queue(code)
|
|
|
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):
|
print("add_subscript_codes", codes)
|
# 加入上传队列
|
common_queue.put(('', "l2_subscript_codes", list(codes)))
|
|
|
def __send_response(sk, msg):
|
msg = socket_util.load_header(msg)
|
sk.sendall(msg)
|
result, header_str = socket_util.recv_data(sk)
|
if result:
|
result_json = json.loads(result)
|
if result_json.get("code") == 0:
|
return True
|
return False
|
|
|
# 发送消息
|
def send_response(type, msg):
|
try:
|
sk = SendResponseSkManager.get_send_response_sk(type)
|
if __send_response(sk, msg):
|
return True
|
else:
|
# 再次发送
|
print("再次发送")
|
return __send_response(sk, msg)
|
except ConnectionResetError as e:
|
SendResponseSkManager.del_send_response_sk(type)
|
sk = SendResponseSkManager.get_send_response_sk(type)
|
return __send_response(sk, msg)
|
except BrokenPipeError as e:
|
SendResponseSkManager.del_send_response_sk(type)
|
sk = SendResponseSkManager.get_send_response_sk(type)
|
return __send_response(sk, msg)
|
|
|
# 上传数据
|
def upload_data(code, _type, datas, new_sk=False):
|
key = f"{_type}_{code}"
|
fdata = json.dumps(
|
{"type": _type, "data": {"code": code, "data": datas, "time": round(time.time() * 1000)}})
|
result = None
|
try:
|
if new_sk:
|
sk = SendResponseSkManager.create_send_response_sk()
|
result = __send_response(sk, fdata.encode('utf-8'))
|
else:
|
result = send_response(key, fdata.encode('utf-8'))
|
except Exception as e:
|
logging.exception(e)
|
finally:
|
pass
|
|
|
def __run_upload_common():
|
print("__run_upload_common")
|
logger_system.info(f"l2_client __run_upload_common 线程ID:{tool.get_thread_id()}")
|
while True:
|
try:
|
while not common_queue.empty():
|
temp = common_queue.get()
|
upload_data(temp[0], temp[1], temp[2])
|
|
except Exception as e:
|
logger_local_huaxin_l2_error.exception(e)
|
logger_local_huaxin_l2_error.error(f"上传普通数据出错：{str(e)}")
|
finally:
|
time.sleep(0.01)
|
|
|
def __run_log():
|
print("__run_log")
|
logger_system.info(f"l2_client __run_log 线程ID:{tool.get_thread_id()}")
|
async_log_util.huaxin_l2_log.run_sync()
|
|
|
# 采用socket传输数据
|
def run_upload_common():
|
t = threading.Thread(target=lambda: __run_upload_common(), daemon=True)
|
t.start()
|
|
|
def run_log():
|
t = threading.Thread(target=lambda: __run_log(), daemon=True)
|
t.start()
|
|
|
def __test():
|
code = "002073"
|
pass
|
|
|
def run_test():
|
t = threading.Thread(target=lambda: __test(), daemon=True)
|
t.start()
|
|
|
def test():
|
pass
|
