import json
|
import logging
|
import multiprocessing
|
import threading
|
import time
|
from multiprocessing import freeze_support
|
import sys
|
import torch
|
|
from PyQt5.QtWebChannel import QWebChannel
|
from PyQt5.QtWebEngineWidgets import QWebEngineView, QWebEngineSettings
|
from PyQt5.QtWidgets import QMainWindow, QApplication, QAction, QMessageBox
|
|
from PyQt5.QtCore import pyqtSlot, QObject, pyqtSignal, QTimer, QUrl, QPoint
|
|
import gui_wx
|
import network_util
|
import setting
|
|
freeze_support()
|
|
|
class JSBridgeClass(QObject):
|
signal_request = pyqtSignal(str, str, str)
|
|
def __request_result_callback(self, method, key, result):
|
self.webview.page().runJavaScript(f"{method}('{key}','{result}')")
|
|
def __init__(self, window, webview):
|
super().__init__()
|
self.window = window
|
self.webview = webview
|
self.signal_request.connect(self.__request_result_callback)
|
|
def __http_request(self, url,callback_info):
|
try:
|
result = network_util.http_get(url)
|
print("请求结果:", result)
|
self.signal_request.emit(callback_info[0],callback_info[1],result)
|
return result
|
except Exception as e:
|
logging.exception(e)
|
|
@pyqtSlot(str, str, str)
|
def http_request(self, path, params, callback_info):
|
url = path + "?"
|
if params:
|
params = json.loads(params)
|
url += "&".join([f"{key}={params[key]}" for key in params])
|
print("http请求", url)
|
callback_info = json.loads(callback_info)
|
t1 = threading.Thread(target=lambda: self.__http_request(url, callback_info))
|
t1.setDaemon(True)
|
t1.start()
|
|
@pyqtSlot(str, result=str)
|
def socket_request(self, text):
|
print("socket_request", text)
|
return network_util.socket_request(json.loads(text))
|
|
@pyqtSlot(str)
|
def show_info(self, msg):
|
QMessageBox.information(self.window, "提示", msg, QMessageBox.Yes)
|
|
@pyqtSlot(str, str, str, str)
|
def set_trade_info(self, code, name, trade_data, trade_record):
|
self.window.set_trade_data(code, name, trade_data, trade_record)
|
|
@pyqtSlot(str, str)
|
def show_want_codes(self, plate, codes):
|
WantBuyWindow(plate, json.loads(codes)).show()
|
|
|
class WantBuyWindow(QMainWindow):
|
def __init__(self, plate, codes):
|
super(WantBuyWindow, self).__init__()
|
self.setWindowTitle(f'想买单({plate})')
|
self.resize(300, 200)
|
|
|
class SecondWindow(QMainWindow):
|
signal_update_kpl = pyqtSignal(str)
|
|
def update_kpl_func(self):
|
while True:
|
time.sleep(3)
|
try:
|
self.__update_kpl()
|
except:
|
pass
|
|
def __init__(self, parent=None):
|
super(SecondWindow, self).__init__(parent)
|
self.setWindowTitle('看盘副屏')
|
window_info = setting.get_kp_second_window_info()
|
if window_info:
|
self.move(window_info[0], window_info[1])
|
self.resize(window_info[2], window_info[3])
|
else:
|
self.resize(500, 800)
|
|
self.webview = QWebEngineView()
|
self.webview.settings().setAttribute(QWebEngineSettings.JavascriptEnabled, True)
|
self.webview.page().setZoomFactor(1)
|
self.setCentralWidget(self.webview)
|
|
self.signal_update_kpl.connect(self.set_kpl_data)
|
|
t1 = threading.Thread(target=self.update_kpl_func)
|
t1.setDaemon(True)
|
t1.start()
|
|
def loadUrl(self, url):
|
self.webview.load(QUrl(url))
|
|
# 设置交易数据
|
def set_trade_data(self, code, code_name, trade_data, trade_record):
|
self.webview.page().runJavaScript(f"app.set_trade_info('{code}','{code_name}','{trade_data}','{trade_record}')")
|
|
def set_kpl_data(self, data):
|
self.webview.page().runJavaScript(f"fill_kpl_data('{data}')")
|
|
# 更新开盘啦数据
|
def __update_kpl(self):
|
datas = network_util.http_get("/get_kpl_data")
|
self.signal_update_kpl.emit(datas)
|
|
def closeEvent(self, event):
|
try:
|
setting.set_kp_second_window_info(
|
(self.pos().x(), self.pos().y(), self.size().width(), self.size().height()))
|
except Exception as e:
|
print("")
|
|
|
class MainWindow(QMainWindow):
|
signal_update_code = pyqtSignal(str)
|
|
def show_info(self, msg):
|
QMessageBox.information(self, "提示", msg, QMessageBox.Yes)
|
|
def show_warning(self, msg):
|
QMessageBox.warning(self, "提示", msg, QMessageBox.Yes)
|
|
def set_trade_data(self, code, code_name, trade_data, trade_record):
|
self.secondWindow.set_trade_data(code, code_name, trade_data, trade_record)
|
|
# 设置目标代码
|
def set_target_code(self, code):
|
print("set_target_code")
|
self.webview.page().runJavaScript(f"app.set_target_code('{code}')")
|
|
# 菜单及菜单点击事件
|
def __menu(self):
|
def __gpu_is_avaiable():
|
if torch.cuda.is_available():
|
self.show_info("GPU可用")
|
else:
|
self.show_warning("GPU不可用")
|
|
def __set_juejin_params():
|
ps = (self.x(), self.y())
|
size = (self.width(), self.height())
|
self.wx_pipe.send(json.dumps({"type": "juejin_setting", "pos": (ps[0] + size[0] / 2, ps[1] + size[1] / 2)}))
|
|
def __download_codes():
|
try:
|
result = self.__request("", 72)
|
result = json.loads(result)
|
if result['code'] == 0:
|
codes_dict = result['data']
|
if codes_dict:
|
for key in codes_dict:
|
# 将代码保存到桌面
|
path = f"C:\\Users\\Administrator\\Desktop\\涨停代码{key}.txt"
|
with open(path, mode='w') as f:
|
for c in codes_dict[key]:
|
f.write(c)
|
f.write("\n")
|
self.show_info("下载成功")
|
else:
|
self.show_warning(result['msg'])
|
except Exception as e:
|
self.show_warning(str(e))
|
|
def __manage_code():
|
ps = (self.x(), self.y())
|
size = (self.width(), self.height())
|
self.wx_pipe.send(json.dumps({"type": "codes_setting", "pos": (ps[0] + size[0] / 2, ps[1] + size[1] / 2)}))
|
|
def __manage_ths_pos():
|
ps = (self.x(), self.y())
|
size = (self.width(), self.height())
|
self.wx_pipe.send(json.dumps({"type": "manage_ths_pos", "pos": (ps[0] + size[0] / 2, ps[1] + size[1] / 2)}))
|
|
def __show_float_callback():
|
self.wx_pipe.send(json.dumps({"type": "show_float_callback"}))
|
|
def __show_main_callback():
|
self.wx_pipe.send(json.dumps({"type": "show_main_callback"}))
|
|
def __show_second_window():
|
self.secondWindow.show()
|
|
menubar = self.menuBar()
|
|
setting_ = menubar.addMenu('&设置')
|
|
juejin_action = QAction("&掘金参数配置", self)
|
juejin_action.triggered.connect(__set_juejin_params)
|
setting_.addAction(juejin_action)
|
|
action = QAction("&代码管理", self)
|
action.triggered.connect(__manage_code)
|
setting_.addAction(action)
|
|
action = QAction("&下载涨停代码", self)
|
action.triggered.connect(__download_codes)
|
setting_.addAction(action)
|
|
action = QAction("&GPU检测", self)
|
action.triggered.connect(__gpu_is_avaiable)
|
setting_.addAction(action)
|
|
auto_ = menubar.addMenu('&自动化')
|
action = QAction("&同花顺设置", self)
|
action.triggered.connect(__manage_ths_pos)
|
auto_.addAction(action)
|
|
view_ = menubar.addMenu('&视图')
|
action = QAction("&打开副屏", self)
|
action.triggered.connect(__show_second_window)
|
view_.addAction(action)
|
|
action = QAction("&打开悬浮盯盘", self)
|
action.triggered.connect(__show_float_callback)
|
view_.addAction(action)
|
|
action = QAction("&打开分时看盘", self)
|
action.triggered.connect(__show_main_callback)
|
view_.addAction(action)
|
|
def __init__(self, wx_pipe):
|
super().__init__()
|
self.wx_pipe = wx_pipe
|
self.setWindowTitle('看盘页面')
|
window_info = setting.get_kp_window_info()
|
if window_info:
|
self.move(window_info[0], window_info[1])
|
self.resize(window_info[2], window_info[3])
|
else:
|
self.resize(1100, 1000)
|
self.center()
|
self.webview = QWebEngineView()
|
self.webview.settings().setAttribute(QWebEngineSettings.JavascriptEnabled, True)
|
self.__menu()
|
# JS桥设置
|
channel = QWebChannel(self.webview.page())
|
self.webview.page().setWebChannel(channel)
|
self.python_bridge = JSBridgeClass(self,self.webview)
|
channel.registerObject("Bridge", self.python_bridge)
|
# 设置副屏
|
self.secondWindow = SecondWindow(self)
|
|
self.setCentralWidget(self.webview)
|
self.show()
|
self.webview.load(QUrl("http://192.168.3.122:8848/kp/index23-05-04.html"))
|
|
self.secondWindow.show()
|
self.secondWindow.loadUrl("http://192.168.3.122:8848/kp/banshuping.html")
|
|
# 绑定槽函数
|
self.signal_update_code.connect(self.set_target_code)
|
|
def closeEvent(self, event):
|
event.accept()
|
try:
|
setting.set_kp_window_info((self.pos().x(), self.pos().y(), self.size().width(), self.size().height()))
|
except Exception as e:
|
print("")
|
self.wx_pipe.send(json.dumps({"type": "exit"}))
|
sys.exit(0)
|
|
|
# 打包命令
|
# cd D:\workspace\GP\trade_desk
|
# D:\workspace\GP\trade_desk\dist\env\pk_env\Scripts\pyinstaller.exe main.spec
|
|
|
def recieve_code(pipe, mainWindow):
|
latest_code = ''
|
while True:
|
try:
|
data = pipe.recv()
|
if data:
|
data = json.loads(data)
|
if data["type"] == "code":
|
if latest_code != data["code"]:
|
latest_code = data["code"]
|
mainWindow.signal_update_code.emit(latest_code)
|
except Exception as e:
|
logging.exception(e)
|
|
|
if __name__ == "__main__":
|
p1, p2 = multiprocessing.Pipe()
|
wxGuiProcess = multiprocessing.Process(target=gui_wx.run, args=(p1,))
|
wxGuiProcess.start()
|
|
app = QApplication(sys.argv)
|
browser = MainWindow(p2)
|
t1 = threading.Thread(target=lambda: recieve_code(p2, browser))
|
# 后台运行
|
t1.setDaemon(True)
|
t1.start()
|
sys.exit(app.exec_())
|
