import builtins
|
import hashlib
|
import json
|
import logging
|
import os
|
import queue
|
import socket
|
import socketserver
|
import time
|
|
import constant
|
from log_module import log
|
from utils import socket_util
|
|
trade_data_request_queue = queue.Queue()
|
|
|
class MyTCPServer(socketserver.TCPServer):
|
def __init__(self, server_address, RequestHandlerClass):
|
socketserver.TCPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate=True)
|
|
|
# 如果使用异步的形式则需要再重写ThreadingTCPServer
|
class MyThreadingTCPServer(socketserver.ThreadingMixIn, MyTCPServer): pass
|
|
|
class MyBaseRequestHandle(socketserver.BaseRequestHandler):
|
__inited = False
|
|
def setup(self):
|
self.__init()
|
|
@classmethod
|
def __init(cls):
|
if cls.__inited:
|
return True
|
cls.__inited = True
|
cls.__req_socket_dict = {}
|
|
def __is_sign_right(self, data_json):
|
list_str = []
|
sign = data_json["sign"]
|
data_json.pop("sign")
|
for k in data_json:
|
list_str.append(f"{k}={data_json[k]}")
|
list_str.sort()
|
__str = "&".join(list_str) + "JiaBei@!*."
|
md5 = hashlib.md5(__str.encode(encoding='utf-8')).hexdigest()
|
if md5 != sign:
|
raise Exception("签名出错")
|
|
@classmethod
|
def getRecvData(cls, skk):
|
data = ""
|
header_size = 10
|
buf = skk.recv(header_size)
|
header_str = buf
|
if buf:
|
start_time = time.time()
|
buf = buf.decode('utf-8')
|
if buf.startswith("##"):
|
content_length = int(buf[2:10])
|
received_size = 0
|
while not received_size == content_length:
|
r_data = skk.recv(10240)
|
received_size += len(r_data)
|
data += r_data.decode('utf-8')
|
else:
|
data = skk.recv(1024 * 1024)
|
data = buf + data.decode('utf-8')
|
return data, header_str
|
|
def handle(self):
|
host = self.client_address[0]
|
super().handle()
|
sk: socket.socket = self.request
|
while True:
|
try:
|
data, header = self.getRecvData(sk)
|
if data:
|
data_str = data
|
# print("收到数据------", f"{data_str[:20]}......{data_str[-20:]}")
|
data_json = None
|
try:
|
data_json = json.loads(data_str)
|
except json.decoder.JSONDecodeError as e:
|
# JSON解析失败
|
sk.sendall(socket_util.load_header(json.dumps(
|
{"code": 100, "msg": f"JSON解析失败"}).encode(
|
encoding='utf-8')))
|
continue
|
type_ = data_json["type"]
|
__start_time = time.time()
|
try:
|
if data_json["type"] == 'l1_data':
|
datas = data_json["data"]
|
L1DataManager().add_datas(datas)
|
break
|
elif data_json["type"] == 'get_l1_target_codes':
|
# 获取目标代码
|
codes = L1DataManager().get_target_codes()
|
sk.sendall(socket_util.load_header(json.dumps(
|
{"code": 0, "data": list(codes)}).encode(
|
encoding='utf-8')))
|
break
|
except Exception as e:
|
log.logger_tuoguan_request_debug.exception(e)
|
finally:
|
if time.time() - __start_time > 2:
|
log.logger_tuoguan_request_debug.info(
|
f"耗时：{int(time.time() - __start_time)}s 数据：{data_json}")
|
else:
|
# 断开连接
|
break
|
# sk.close()
|
except Exception as e:
|
# log.logger_tuoguan_request_debug.exception(e)
|
logging.exception(e)
|
break
|
|
def finish(self):
|
super().finish()
|
|
|
# L1数据管理
|
class L1DataManager:
|
__l1_datas_dict = {}
|
__target_codes = set()
|
__instance = None
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(L1DataManager, cls).__new__(cls, *args, **kwargs)
|
cls.__load_datas()
|
return cls.__instance
|
|
@classmethod
|
def __load_datas(cls):
|
try:
|
with open(f"{constant.LOG_DIR}/l1_codes.txt", 'r', encoding='utf-8') as f:
|
lines = f.readlines()
|
for line in lines:
|
if line:
|
line = line.strip()
|
if line:
|
cls.__target_codes.add(line)
|
except:
|
pass
|
|
def add_datas(cls, datas):
|
for data in datas:
|
"""
|
data数据结构：(代码,昨日收盘价,最新价,总成交量,总成交额,更新时间)
|
"""
|
cls.__l1_datas_dict[data[0]] = data
|
|
def get_current_l1_data(self):
|
return [self.__l1_datas_dict[x] for x in self.__l1_datas_dict]
|
|
def save_target_codes(self, codes):
|
# 保存目标代码
|
self.__target_codes = codes
|
# 将代码保存到文件
|
path = f"{constant.LOG_DIR}/l1_codes.txt"
|
if not os.path.exists(constant.LOG_DIR):
|
os.mkdir(constant.LOG_DIR)
|
with open(path, 'w', encoding='utf-8') as f:
|
for code in codes:
|
f.write(f"{code}\n")
|
|
def get_target_codes(self):
|
return self.__target_codes
|
|
|
def run(port):
|
print("create MiddleL1DataServer")
|
laddr = "0.0.0.0", port
|
print("MiddleServer is at: http://%s:%d/" % (laddr))
|
tcpserver = MyThreadingTCPServer(laddr, MyBaseRequestHandle) # 注意：参数是MyBaseRequestHandle
|
tcpserver.serve_forever()
|
|
|
if __name__ == "__main__":
|
print(L1DataManager().get_target_codes())
|
