wehub-resource-sync
1711 lines
59 KiB
Python
1711 lines
59 KiB
Python
"""
|
||
Basic widgets for UI.
|
||
"""
|
||
|
||
import csv
|
||
import platform
|
||
import time
|
||
from enum import Enum
|
||
from io import BytesIO
|
||
from typing import cast, Any
|
||
from copy import copy
|
||
from tzlocal import get_localzone_name
|
||
from datetime import datetime
|
||
from importlib import metadata
|
||
|
||
import qrcode
|
||
|
||
from .qt import QtCore, QtGui, QtWidgets, Qt
|
||
from ..constant import Direction, Exchange, Offset, OrderType
|
||
from ..engine import MainEngine, Event, EventEngine, WechatEngine
|
||
from ..event import (
|
||
EVENT_QUOTE,
|
||
EVENT_TICK,
|
||
EVENT_TRADE,
|
||
EVENT_ORDER,
|
||
EVENT_POSITION,
|
||
EVENT_ACCOUNT,
|
||
EVENT_LOG
|
||
)
|
||
from ..object import (
|
||
OrderRequest,
|
||
SubscribeRequest,
|
||
CancelRequest,
|
||
ContractData,
|
||
PositionData,
|
||
OrderData,
|
||
QuoteData,
|
||
TickData
|
||
)
|
||
from ..utility import load_json, save_json, get_digits, ZoneInfo
|
||
from ..setting import SETTING_FILENAME, SETTINGS
|
||
from ..locale import _
|
||
from ..wechat import (
|
||
Credentials,
|
||
WeixinError,
|
||
WeixinTimeout,
|
||
poll,
|
||
request_qrcode,
|
||
wait_for_login,
|
||
)
|
||
|
||
|
||
COLOR_LONG = QtGui.QColor("red")
|
||
COLOR_SHORT = QtGui.QColor("green")
|
||
COLOR_BID = QtGui.QColor(255, 174, 201)
|
||
COLOR_ASK = QtGui.QColor(160, 255, 160)
|
||
COLOR_BLACK = QtGui.QColor("black")
|
||
|
||
|
||
class BaseCell(QtWidgets.QTableWidgetItem):
|
||
"""
|
||
General cell used in tablewidgets.
|
||
"""
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self._text: str = ""
|
||
self._data: Any = None
|
||
|
||
self.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
|
||
self.set_content(content, data)
|
||
|
||
def set_content(self, content: Any, data: Any) -> None:
|
||
"""
|
||
Set text content.
|
||
"""
|
||
self._text = str(content)
|
||
self._data = data
|
||
|
||
self.setText(self._text)
|
||
|
||
def get_data(self) -> Any:
|
||
"""
|
||
Get data object.
|
||
"""
|
||
return self._data
|
||
|
||
def __lt__(self, other: "BaseCell") -> bool: # type: ignore
|
||
"""
|
||
Sort by text content.
|
||
"""
|
||
result: bool = self._text < other._text
|
||
return result
|
||
|
||
|
||
class EnumCell(BaseCell):
|
||
"""
|
||
Cell used for showing enum data.
|
||
"""
|
||
|
||
def __init__(self, content: Enum, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
def set_content(self, content: Any, data: Any) -> None:
|
||
"""
|
||
Set text using enum.constant.value.
|
||
"""
|
||
if content:
|
||
super().set_content(content.value, data)
|
||
|
||
|
||
class DirectionCell(EnumCell):
|
||
"""
|
||
Cell used for showing direction data.
|
||
"""
|
||
|
||
def __init__(self, content: Enum, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
def set_content(self, content: Any, data: Any) -> None:
|
||
"""
|
||
Cell color is set according to direction.
|
||
"""
|
||
super().set_content(content, data)
|
||
|
||
if content is Direction.SHORT:
|
||
self.setForeground(COLOR_SHORT)
|
||
else:
|
||
self.setForeground(COLOR_LONG)
|
||
|
||
|
||
class BidCell(BaseCell):
|
||
"""
|
||
Cell used for showing bid price and volume.
|
||
"""
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
self.setForeground(COLOR_BID)
|
||
|
||
|
||
class AskCell(BaseCell):
|
||
"""
|
||
Cell used for showing ask price and volume.
|
||
"""
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
self.setForeground(COLOR_ASK)
|
||
|
||
|
||
class PnlCell(BaseCell):
|
||
"""
|
||
Cell used for showing pnl data.
|
||
"""
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
def set_content(self, content: Any, data: Any) -> None:
|
||
"""
|
||
Cell color is set based on whether pnl is
|
||
positive or negative.
|
||
"""
|
||
super().set_content(content, data)
|
||
|
||
if str(content).startswith("-"):
|
||
self.setForeground(COLOR_SHORT)
|
||
else:
|
||
self.setForeground(COLOR_LONG)
|
||
|
||
|
||
class TimeCell(BaseCell):
|
||
"""
|
||
Cell used for showing time string from datetime object.
|
||
"""
|
||
|
||
local_tz = ZoneInfo(get_localzone_name())
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
def set_content(self, content: datetime | None, data: Any) -> None:
|
||
""""""
|
||
if content is None:
|
||
return
|
||
|
||
content = content.astimezone(self.local_tz)
|
||
timestamp: str = content.strftime("%H:%M:%S")
|
||
|
||
millisecond: int = int(content.microsecond / 1000)
|
||
if millisecond:
|
||
timestamp = f"{timestamp}.{millisecond}"
|
||
else:
|
||
timestamp = f"{timestamp}.000"
|
||
|
||
self.setText(timestamp)
|
||
self._data = data
|
||
|
||
|
||
class DateCell(BaseCell):
|
||
"""
|
||
Cell used for showing date string from datetime object.
|
||
"""
|
||
|
||
def __init__(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
|
||
def set_content(self, content: Any, data: Any) -> None:
|
||
""""""
|
||
if content is None:
|
||
return
|
||
|
||
self.setText(content.strftime("%Y-%m-%d"))
|
||
self._data = data
|
||
|
||
|
||
class MsgCell(BaseCell):
|
||
"""
|
||
Cell used for showing msg data.
|
||
"""
|
||
|
||
def __init__(self, content: str, data: Any) -> None:
|
||
""""""
|
||
super().__init__(content, data)
|
||
self.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignLeft | QtCore.Qt.AlignmentFlag.AlignVCenter)
|
||
|
||
|
||
class BaseMonitor(QtWidgets.QTableWidget):
|
||
"""
|
||
Monitor data update.
|
||
"""
|
||
|
||
event_type: str = ""
|
||
data_key: str = ""
|
||
sorting: bool = False
|
||
headers: dict = {}
|
||
|
||
signal: QtCore.Signal = QtCore.Signal(Event)
|
||
|
||
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.event_engine: EventEngine = event_engine
|
||
self.cells: dict[str, dict] = {}
|
||
|
||
self.init_ui()
|
||
self.load_setting()
|
||
self.register_event()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.init_table()
|
||
self.init_menu()
|
||
|
||
def init_table(self) -> None:
|
||
"""
|
||
Initialize table.
|
||
"""
|
||
self.setColumnCount(len(self.headers))
|
||
|
||
labels: list = [d["display"] for d in self.headers.values()]
|
||
self.setHorizontalHeaderLabels(labels)
|
||
|
||
self.verticalHeader().setVisible(False)
|
||
self.setEditTriggers(self.EditTrigger.NoEditTriggers)
|
||
self.setAlternatingRowColors(True)
|
||
self.setSortingEnabled(self.sorting)
|
||
|
||
def init_menu(self) -> None:
|
||
"""
|
||
Create right click menu.
|
||
"""
|
||
self.menu: QtWidgets.QMenu = QtWidgets.QMenu(self)
|
||
|
||
resize_action: QtGui.QAction = QtGui.QAction(_("调整列宽"), self)
|
||
resize_action.triggered.connect(self.resize_columns)
|
||
self.menu.addAction(resize_action)
|
||
|
||
save_action: QtGui.QAction = QtGui.QAction(_("保存数据"), self)
|
||
save_action.triggered.connect(self.save_csv)
|
||
self.menu.addAction(save_action)
|
||
|
||
def register_event(self) -> None:
|
||
"""
|
||
Register event handler into event engine.
|
||
"""
|
||
if self.event_type:
|
||
self.signal.connect(self.process_event)
|
||
self.event_engine.register(self.event_type, self.signal.emit)
|
||
|
||
def process_event(self, event: Event) -> None:
|
||
"""
|
||
Process new data from event and update into table.
|
||
"""
|
||
# Disable sorting to prevent unwanted error.
|
||
if self.sorting:
|
||
self.setSortingEnabled(False)
|
||
|
||
# Update data into table.
|
||
data = event.data
|
||
|
||
if not self.data_key:
|
||
self.insert_new_row(data)
|
||
else:
|
||
key: str = data.__getattribute__(self.data_key)
|
||
|
||
if key in self.cells:
|
||
self.update_old_row(data)
|
||
else:
|
||
self.insert_new_row(data)
|
||
|
||
# Enable sorting
|
||
if self.sorting:
|
||
self.setSortingEnabled(True)
|
||
|
||
def insert_new_row(self, data: Any) -> None:
|
||
"""
|
||
Insert a new row at the top of table.
|
||
"""
|
||
self.insertRow(0)
|
||
|
||
row_cells: dict = {}
|
||
for column, header in enumerate(self.headers.keys()):
|
||
setting: dict = self.headers[header]
|
||
|
||
content = data.__getattribute__(header)
|
||
cell: QtWidgets.QTableWidgetItem = setting["cell"](content, data)
|
||
self.setItem(0, column, cell)
|
||
|
||
if setting["update"]:
|
||
row_cells[header] = cell
|
||
|
||
if self.data_key:
|
||
key: str = data.__getattribute__(self.data_key)
|
||
self.cells[key] = row_cells
|
||
|
||
def update_old_row(self, data: Any) -> None:
|
||
"""
|
||
Update an old row in table.
|
||
"""
|
||
key: str = data.__getattribute__(self.data_key)
|
||
row_cells = self.cells[key]
|
||
|
||
for header, cell in row_cells.items():
|
||
content = data.__getattribute__(header)
|
||
cell.set_content(content, data)
|
||
|
||
def resize_columns(self) -> None:
|
||
"""
|
||
Resize all columns according to contents.
|
||
"""
|
||
self.horizontalHeader().resizeSections(QtWidgets.QHeaderView.ResizeMode.ResizeToContents)
|
||
|
||
def save_csv(self) -> None:
|
||
"""
|
||
Save table data into a csv file
|
||
"""
|
||
path, __ = QtWidgets.QFileDialog.getSaveFileName(
|
||
self, _("保存数据"), "", "CSV(*.csv)")
|
||
|
||
if not path:
|
||
return
|
||
|
||
with open(path, "w") as f:
|
||
writer = csv.writer(f, lineterminator="\n")
|
||
|
||
headers: list = [d["display"] for d in self.headers.values()]
|
||
writer.writerow(headers)
|
||
|
||
for row in range(self.rowCount()):
|
||
if self.isRowHidden(row):
|
||
continue
|
||
|
||
row_data: list = []
|
||
for column in range(self.columnCount()):
|
||
item: QtWidgets.QTableWidgetItem | None = self.item(row, column)
|
||
if item:
|
||
row_data.append(str(item.text()))
|
||
else:
|
||
row_data.append("")
|
||
writer.writerow(row_data)
|
||
|
||
def contextMenuEvent(self, event: QtGui.QContextMenuEvent) -> None:
|
||
"""
|
||
Show menu with right click.
|
||
"""
|
||
self.menu.popup(QtGui.QCursor.pos())
|
||
|
||
def save_setting(self) -> None:
|
||
""""""
|
||
settings: QtCore.QSettings = QtCore.QSettings(self.__class__.__name__, "custom")
|
||
settings.setValue("column_state", self.horizontalHeader().saveState())
|
||
|
||
def load_setting(self) -> None:
|
||
""""""
|
||
settings: QtCore.QSettings = QtCore.QSettings(self.__class__.__name__, "custom")
|
||
column_state = settings.value("column_state")
|
||
|
||
if isinstance(column_state, QtCore.QByteArray):
|
||
self.horizontalHeader().restoreState(column_state)
|
||
self.horizontalHeader().setSortIndicator(-1, QtCore.Qt.SortOrder.AscendingOrder)
|
||
|
||
|
||
class TickMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for tick data.
|
||
"""
|
||
|
||
event_type: str = EVENT_TICK
|
||
data_key: str = "vt_symbol"
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"symbol": {"display": _("代码"), "cell": BaseCell, "update": False},
|
||
"exchange": {"display": _("交易所"), "cell": EnumCell, "update": False},
|
||
"name": {"display": _("名称"), "cell": BaseCell, "update": True},
|
||
"last_price": {"display": _("最新价"), "cell": BaseCell, "update": True},
|
||
"volume": {"display": _("成交量"), "cell": BaseCell, "update": True},
|
||
"open_price": {"display": _("开盘价"), "cell": BaseCell, "update": True},
|
||
"high_price": {"display": _("最高价"), "cell": BaseCell, "update": True},
|
||
"low_price": {"display": _("最低价"), "cell": BaseCell, "update": True},
|
||
"bid_price_1": {"display": _("买1价"), "cell": BidCell, "update": True},
|
||
"bid_volume_1": {"display": _("买1量"), "cell": BidCell, "update": True},
|
||
"ask_price_1": {"display": _("卖1价"), "cell": AskCell, "update": True},
|
||
"ask_volume_1": {"display": _("卖1量"), "cell": AskCell, "update": True},
|
||
"datetime": {"display": _("时间"), "cell": TimeCell, "update": True},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
|
||
class LogMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for log data.
|
||
"""
|
||
|
||
event_type: str = EVENT_LOG
|
||
data_key: str = ""
|
||
sorting: bool = False
|
||
|
||
headers: dict = {
|
||
"time": {"display": _("时间"), "cell": TimeCell, "update": False},
|
||
"msg": {"display": _("信息"), "cell": MsgCell, "update": False},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
|
||
class TradeMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for trade data.
|
||
"""
|
||
|
||
event_type: str = EVENT_TRADE
|
||
data_key: str = ""
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"tradeid": {"display": _("成交号"), "cell": BaseCell, "update": False},
|
||
"orderid": {"display": _("委托号"), "cell": BaseCell, "update": False},
|
||
"symbol": {"display": _("代码"), "cell": BaseCell, "update": False},
|
||
"exchange": {"display": _("交易所"), "cell": EnumCell, "update": False},
|
||
"direction": {"display": _("方向"), "cell": DirectionCell, "update": False},
|
||
"offset": {"display": _("开平"), "cell": EnumCell, "update": False},
|
||
"price": {"display": _("价格"), "cell": BaseCell, "update": False},
|
||
"volume": {"display": _("数量"), "cell": BaseCell, "update": False},
|
||
"datetime": {"display": _("时间"), "cell": TimeCell, "update": False},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
|
||
class OrderMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for order data.
|
||
"""
|
||
|
||
event_type: str = EVENT_ORDER
|
||
data_key: str = "vt_orderid"
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"orderid": {"display": _("委托号"), "cell": BaseCell, "update": False},
|
||
"reference": {"display": _("来源"), "cell": BaseCell, "update": False},
|
||
"symbol": {"display": _("代码"), "cell": BaseCell, "update": False},
|
||
"exchange": {"display": _("交易所"), "cell": EnumCell, "update": False},
|
||
"type": {"display": _("类型"), "cell": EnumCell, "update": False},
|
||
"direction": {"display": _("方向"), "cell": DirectionCell, "update": False},
|
||
"offset": {"display": _("开平"), "cell": EnumCell, "update": False},
|
||
"price": {"display": _("价格"), "cell": BaseCell, "update": False},
|
||
"volume": {"display": _("总数量"), "cell": BaseCell, "update": True},
|
||
"traded": {"display": _("已成交"), "cell": BaseCell, "update": True},
|
||
"status": {"display": _("状态"), "cell": EnumCell, "update": True},
|
||
"datetime": {"display": _("时间"), "cell": TimeCell, "update": True},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
def init_ui(self) -> None:
|
||
"""
|
||
Connect signal.
|
||
"""
|
||
super().init_ui()
|
||
|
||
self.setToolTip(_("双击单元格撤单"))
|
||
self.itemDoubleClicked.connect(self.cancel_order)
|
||
|
||
def cancel_order(self, cell: BaseCell) -> None:
|
||
"""
|
||
Cancel order if cell double clicked.
|
||
"""
|
||
order: OrderData = cell.get_data()
|
||
req: CancelRequest = order.create_cancel_request()
|
||
self.main_engine.cancel_order(req, order.gateway_name)
|
||
|
||
|
||
class PositionMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for position data.
|
||
"""
|
||
|
||
event_type: str = EVENT_POSITION
|
||
data_key: str = "vt_positionid"
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"symbol": {"display": _("代码"), "cell": BaseCell, "update": False},
|
||
"exchange": {"display": _("交易所"), "cell": EnumCell, "update": False},
|
||
"direction": {"display": _("方向"), "cell": DirectionCell, "update": False},
|
||
"volume": {"display": _("数量"), "cell": BaseCell, "update": True},
|
||
"yd_volume": {"display": _("昨仓"), "cell": BaseCell, "update": True},
|
||
"frozen": {"display": _("冻结"), "cell": BaseCell, "update": True},
|
||
"price": {"display": _("均价"), "cell": BaseCell, "update": True},
|
||
"pnl": {"display": _("盈亏"), "cell": PnlCell, "update": True},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
|
||
class AccountMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for account data.
|
||
"""
|
||
|
||
event_type: str = EVENT_ACCOUNT
|
||
data_key: str = "vt_accountid"
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"accountid": {"display": _("账号"), "cell": BaseCell, "update": False},
|
||
"balance": {"display": _("余额"), "cell": BaseCell, "update": True},
|
||
"frozen": {"display": _("冻结"), "cell": BaseCell, "update": True},
|
||
"available": {"display": _("可用"), "cell": BaseCell, "update": True},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
|
||
class QuoteMonitor(BaseMonitor):
|
||
"""
|
||
Monitor for quote data.
|
||
"""
|
||
|
||
event_type: str = EVENT_QUOTE
|
||
data_key: str = "vt_quoteid"
|
||
sorting: bool = True
|
||
|
||
headers: dict = {
|
||
"quoteid": {"display": _("报价号"), "cell": BaseCell, "update": False},
|
||
"reference": {"display": _("来源"), "cell": BaseCell, "update": False},
|
||
"symbol": {"display": _("代码"), "cell": BaseCell, "update": False},
|
||
"exchange": {"display": _("交易所"), "cell": EnumCell, "update": False},
|
||
"bid_offset": {"display": _("买开平"), "cell": EnumCell, "update": False},
|
||
"bid_volume": {"display": _("买量"), "cell": BidCell, "update": False},
|
||
"bid_price": {"display": _("买价"), "cell": BidCell, "update": False},
|
||
"ask_price": {"display": _("卖价"), "cell": AskCell, "update": False},
|
||
"ask_volume": {"display": _("卖量"), "cell": AskCell, "update": False},
|
||
"ask_offset": {"display": _("卖开平"), "cell": EnumCell, "update": False},
|
||
"status": {"display": _("状态"), "cell": EnumCell, "update": True},
|
||
"datetime": {"display": _("时间"), "cell": TimeCell, "update": True},
|
||
"gateway_name": {"display": _("接口"), "cell": BaseCell, "update": False},
|
||
}
|
||
|
||
def init_ui(self) -> None:
|
||
"""
|
||
Connect signal.
|
||
"""
|
||
super().init_ui()
|
||
|
||
self.setToolTip(_("双击单元格撤销报价"))
|
||
self.itemDoubleClicked.connect(self.cancel_quote)
|
||
|
||
def cancel_quote(self, cell: BaseCell) -> None:
|
||
"""
|
||
Cancel quote if cell double clicked.
|
||
"""
|
||
quote: QuoteData = cell.get_data()
|
||
req: CancelRequest = quote.create_cancel_request()
|
||
self.main_engine.cancel_quote(req, quote.gateway_name)
|
||
|
||
|
||
class ConnectDialog(QtWidgets.QDialog):
|
||
"""
|
||
Start connection of a certain gateway.
|
||
"""
|
||
|
||
def __init__(self, main_engine: MainEngine, gateway_name: str) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.gateway_name: str = gateway_name
|
||
self.filename: str = f"connect_{gateway_name.lower()}.json"
|
||
|
||
self.widgets: dict[str, tuple[QtWidgets.QWidget, type]] = {}
|
||
|
||
self.init_ui()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setWindowTitle(_("连接{}").format(self.gateway_name))
|
||
|
||
# Default setting provides field name, field data type and field default value.
|
||
default_setting: dict | None = self.main_engine.get_default_setting(self.gateway_name)
|
||
|
||
# Saved setting provides field data used last time.
|
||
loaded_setting: dict = load_json(self.filename)
|
||
|
||
# Initialize line edits and form layout based on setting.
|
||
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
|
||
|
||
if default_setting:
|
||
for field_name, field_value in default_setting.items():
|
||
field_type: type = type(field_value)
|
||
|
||
if field_type is list:
|
||
combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
combo.addItems(field_value)
|
||
|
||
if field_name in loaded_setting:
|
||
saved_value = loaded_setting[field_name]
|
||
ix: int = combo.findText(saved_value)
|
||
combo.setCurrentIndex(ix)
|
||
|
||
form.addRow(f"{field_name} <{field_type.__name__}>", combo)
|
||
self.widgets[field_name] = (combo, field_type)
|
||
else:
|
||
line: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(field_value))
|
||
|
||
if field_name in loaded_setting:
|
||
saved_value = loaded_setting[field_name]
|
||
line.setText(str(saved_value))
|
||
|
||
if _("密码") in field_name:
|
||
line.setEchoMode(QtWidgets.QLineEdit.EchoMode.Password)
|
||
|
||
if field_type is int:
|
||
validator: QtGui.QIntValidator = QtGui.QIntValidator()
|
||
line.setValidator(validator)
|
||
|
||
form.addRow(f"{field_name} <{field_type.__name__}>", line)
|
||
self.widgets[field_name] = (line, field_type)
|
||
|
||
button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("连接"))
|
||
button.clicked.connect(self.connect_gateway)
|
||
form.addRow(button)
|
||
|
||
self.setLayout(form)
|
||
|
||
def connect_gateway(self) -> None:
|
||
"""
|
||
Get setting value from line edits and connect the gateway.
|
||
"""
|
||
setting: dict = {}
|
||
|
||
for field_name, tp in self.widgets.items():
|
||
widget, field_type = tp
|
||
if field_type is list:
|
||
combo: QtWidgets.QComboBox = cast(QtWidgets.QComboBox, widget)
|
||
field_value = str(combo.currentText())
|
||
else:
|
||
line: QtWidgets.QLineEdit = cast(QtWidgets.QLineEdit, widget)
|
||
try:
|
||
field_value = field_type(line.text())
|
||
except ValueError:
|
||
field_value = field_type()
|
||
setting[field_name] = field_value
|
||
|
||
save_json(self.filename, setting)
|
||
|
||
self.main_engine.connect(setting, self.gateway_name)
|
||
self.accept()
|
||
|
||
|
||
class TradingWidget(QtWidgets.QWidget):
|
||
"""
|
||
General manual trading widget.
|
||
"""
|
||
|
||
signal_tick: QtCore.Signal = QtCore.Signal(Event)
|
||
|
||
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.event_engine: EventEngine = event_engine
|
||
|
||
self.vt_symbol: str = ""
|
||
self.price_digits: int = 0
|
||
|
||
self.init_ui()
|
||
self.register_event()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setFixedWidth(300)
|
||
|
||
# Trading function area
|
||
exchanges: list[Exchange] = self.main_engine.get_all_exchanges()
|
||
self.exchange_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
self.exchange_combo.addItems([exchange.value for exchange in exchanges])
|
||
|
||
self.symbol_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit()
|
||
self.symbol_line.returnPressed.connect(self.set_vt_symbol)
|
||
|
||
self.name_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit()
|
||
self.name_line.setReadOnly(True)
|
||
|
||
self.direction_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
self.direction_combo.addItems(
|
||
[Direction.LONG.value, Direction.SHORT.value])
|
||
|
||
self.offset_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
self.offset_combo.addItems([offset.value for offset in Offset])
|
||
|
||
self.order_type_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
self.order_type_combo.addItems(
|
||
[order_type.value for order_type in OrderType])
|
||
|
||
double_validator: QtGui.QDoubleValidator = QtGui.QDoubleValidator()
|
||
double_validator.setBottom(0)
|
||
|
||
self.price_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit()
|
||
self.price_line.setValidator(double_validator)
|
||
|
||
self.volume_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit()
|
||
self.volume_line.setValidator(double_validator)
|
||
|
||
self.gateway_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
|
||
self.gateway_combo.addItems(self.main_engine.get_all_gateway_names())
|
||
|
||
self.price_check: QtWidgets.QCheckBox = QtWidgets.QCheckBox()
|
||
self.price_check.setToolTip(_("设置价格随行情更新"))
|
||
|
||
send_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("委托"))
|
||
send_button.clicked.connect(self.send_order)
|
||
|
||
cancel_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("全撤"))
|
||
cancel_button.clicked.connect(self.cancel_all)
|
||
|
||
grid: QtWidgets.QGridLayout = QtWidgets.QGridLayout()
|
||
grid.addWidget(QtWidgets.QLabel(_("交易所")), 0, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("代码")), 1, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("名称")), 2, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("方向")), 3, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("开平")), 4, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("类型")), 5, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("价格")), 6, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("数量")), 7, 0)
|
||
grid.addWidget(QtWidgets.QLabel(_("接口")), 8, 0)
|
||
grid.addWidget(self.exchange_combo, 0, 1, 1, 2)
|
||
grid.addWidget(self.symbol_line, 1, 1, 1, 2)
|
||
grid.addWidget(self.name_line, 2, 1, 1, 2)
|
||
grid.addWidget(self.direction_combo, 3, 1, 1, 2)
|
||
grid.addWidget(self.offset_combo, 4, 1, 1, 2)
|
||
grid.addWidget(self.order_type_combo, 5, 1, 1, 2)
|
||
grid.addWidget(self.price_line, 6, 1, 1, 1)
|
||
grid.addWidget(self.price_check, 6, 2, 1, 1)
|
||
grid.addWidget(self.volume_line, 7, 1, 1, 2)
|
||
grid.addWidget(self.gateway_combo, 8, 1, 1, 2)
|
||
grid.addWidget(send_button, 9, 0, 1, 3)
|
||
grid.addWidget(cancel_button, 10, 0, 1, 3)
|
||
|
||
# Market depth display area
|
||
bid_color: str = "rgb(255,174,201)"
|
||
ask_color: str = "rgb(160,255,160)"
|
||
|
||
self.bp1_label: QtWidgets.QLabel = self.create_label(bid_color)
|
||
self.bp2_label: QtWidgets.QLabel = self.create_label(bid_color)
|
||
self.bp3_label: QtWidgets.QLabel = self.create_label(bid_color)
|
||
self.bp4_label: QtWidgets.QLabel = self.create_label(bid_color)
|
||
self.bp5_label: QtWidgets.QLabel = self.create_label(bid_color)
|
||
|
||
self.bv1_label: QtWidgets.QLabel = self.create_label(
|
||
bid_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.bv2_label: QtWidgets.QLabel = self.create_label(
|
||
bid_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.bv3_label: QtWidgets.QLabel = self.create_label(
|
||
bid_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.bv4_label: QtWidgets.QLabel = self.create_label(
|
||
bid_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.bv5_label: QtWidgets.QLabel = self.create_label(
|
||
bid_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
|
||
self.ap1_label: QtWidgets.QLabel = self.create_label(ask_color)
|
||
self.ap2_label: QtWidgets.QLabel = self.create_label(ask_color)
|
||
self.ap3_label: QtWidgets.QLabel = self.create_label(ask_color)
|
||
self.ap4_label: QtWidgets.QLabel = self.create_label(ask_color)
|
||
self.ap5_label: QtWidgets.QLabel = self.create_label(ask_color)
|
||
|
||
self.av1_label: QtWidgets.QLabel = self.create_label(
|
||
ask_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.av2_label: QtWidgets.QLabel = self.create_label(
|
||
ask_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.av3_label: QtWidgets.QLabel = self.create_label(
|
||
ask_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.av4_label: QtWidgets.QLabel = self.create_label(
|
||
ask_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
self.av5_label: QtWidgets.QLabel = self.create_label(
|
||
ask_color, alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
|
||
self.lp_label: QtWidgets.QLabel = self.create_label()
|
||
self.return_label: QtWidgets.QLabel = self.create_label(alignment=QtCore.Qt.AlignmentFlag.AlignRight)
|
||
|
||
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
|
||
form.addRow(self.ap5_label, self.av5_label)
|
||
form.addRow(self.ap4_label, self.av4_label)
|
||
form.addRow(self.ap3_label, self.av3_label)
|
||
form.addRow(self.ap2_label, self.av2_label)
|
||
form.addRow(self.ap1_label, self.av1_label)
|
||
form.addRow(self.lp_label, self.return_label)
|
||
form.addRow(self.bp1_label, self.bv1_label)
|
||
form.addRow(self.bp2_label, self.bv2_label)
|
||
form.addRow(self.bp3_label, self.bv3_label)
|
||
form.addRow(self.bp4_label, self.bv4_label)
|
||
form.addRow(self.bp5_label, self.bv5_label)
|
||
|
||
# Overall layout
|
||
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
vbox.addLayout(grid)
|
||
vbox.addLayout(form)
|
||
self.setLayout(vbox)
|
||
|
||
def create_label(
|
||
self,
|
||
color: str = "",
|
||
alignment: int = QtCore.Qt.AlignmentFlag.AlignLeft
|
||
) -> QtWidgets.QLabel:
|
||
"""
|
||
Create label with certain font color.
|
||
"""
|
||
label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
if color:
|
||
label.setStyleSheet(f"color:{color}")
|
||
label.setAlignment(Qt.AlignmentFlag(alignment))
|
||
return label
|
||
|
||
def register_event(self) -> None:
|
||
""""""
|
||
self.signal_tick.connect(self.process_tick_event)
|
||
self.event_engine.register(EVENT_TICK, self.signal_tick.emit)
|
||
|
||
def process_tick_event(self, event: Event) -> None:
|
||
""""""
|
||
tick: TickData = event.data
|
||
if tick.vt_symbol != self.vt_symbol:
|
||
return
|
||
|
||
price_digits: int = self.price_digits
|
||
|
||
self.lp_label.setText(f"{tick.last_price:.{price_digits}f}")
|
||
self.bp1_label.setText(f"{tick.bid_price_1:.{price_digits}f}")
|
||
self.bv1_label.setText(str(tick.bid_volume_1))
|
||
self.ap1_label.setText(f"{tick.ask_price_1:.{price_digits}f}")
|
||
self.av1_label.setText(str(tick.ask_volume_1))
|
||
|
||
if tick.pre_close:
|
||
r: float = (tick.last_price / tick.pre_close - 1) * 100
|
||
self.return_label.setText(f"{r:.2f}%")
|
||
|
||
if tick.bid_price_2:
|
||
self.bp2_label.setText(f"{tick.bid_price_2:.{price_digits}f}")
|
||
self.bv2_label.setText(str(tick.bid_volume_2))
|
||
self.ap2_label.setText(f"{tick.ask_price_2:.{price_digits}f}")
|
||
self.av2_label.setText(str(tick.ask_volume_2))
|
||
|
||
self.bp3_label.setText(f"{tick.bid_price_3:.{price_digits}f}")
|
||
self.bv3_label.setText(str(tick.bid_volume_3))
|
||
self.ap3_label.setText(f"{tick.ask_price_3:.{price_digits}f}")
|
||
self.av3_label.setText(str(tick.ask_volume_3))
|
||
|
||
self.bp4_label.setText(f"{tick.bid_price_4:.{price_digits}f}")
|
||
self.bv4_label.setText(str(tick.bid_volume_4))
|
||
self.ap4_label.setText(f"{tick.ask_price_4:.{price_digits}f}")
|
||
self.av4_label.setText(str(tick.ask_volume_4))
|
||
|
||
self.bp5_label.setText(f"{tick.bid_price_5:.{price_digits}f}")
|
||
self.bv5_label.setText(str(tick.bid_volume_5))
|
||
self.ap5_label.setText(f"{tick.ask_price_5:.{price_digits}f}")
|
||
self.av5_label.setText(str(tick.ask_volume_5))
|
||
|
||
if self.price_check.isChecked():
|
||
self.price_line.setText(f"{tick.last_price:.{price_digits}f}")
|
||
|
||
def set_vt_symbol(self) -> None:
|
||
"""
|
||
Set the tick depth data to monitor by vt_symbol.
|
||
"""
|
||
symbol: str = str(self.symbol_line.text())
|
||
if not symbol:
|
||
return
|
||
|
||
# Generate vt_symbol from symbol and exchange
|
||
exchange_value: str = str(self.exchange_combo.currentText())
|
||
vt_symbol: str = f"{symbol}.{exchange_value}"
|
||
|
||
if vt_symbol == self.vt_symbol:
|
||
return
|
||
self.vt_symbol = vt_symbol
|
||
|
||
# Update name line widget and clear all labels
|
||
contract: ContractData | None = self.main_engine.get_contract(vt_symbol)
|
||
if not contract:
|
||
self.name_line.setText("")
|
||
gateway_name: str = self.gateway_combo.currentText()
|
||
else:
|
||
self.name_line.setText(contract.name)
|
||
gateway_name = contract.gateway_name
|
||
|
||
# Update gateway combo box.
|
||
ix: int = self.gateway_combo.findText(gateway_name)
|
||
self.gateway_combo.setCurrentIndex(ix)
|
||
|
||
# Update price digits
|
||
self.price_digits = get_digits(contract.pricetick)
|
||
|
||
self.clear_label_text()
|
||
self.volume_line.setText("")
|
||
self.price_line.setText("")
|
||
|
||
# Subscribe tick data
|
||
req: SubscribeRequest = SubscribeRequest(
|
||
symbol=symbol, exchange=Exchange(exchange_value)
|
||
)
|
||
|
||
self.main_engine.subscribe(req, gateway_name)
|
||
|
||
def clear_label_text(self) -> None:
|
||
"""
|
||
Clear text on all labels.
|
||
"""
|
||
self.lp_label.setText("")
|
||
self.return_label.setText("")
|
||
|
||
self.bv1_label.setText("")
|
||
self.bv2_label.setText("")
|
||
self.bv3_label.setText("")
|
||
self.bv4_label.setText("")
|
||
self.bv5_label.setText("")
|
||
|
||
self.av1_label.setText("")
|
||
self.av2_label.setText("")
|
||
self.av3_label.setText("")
|
||
self.av4_label.setText("")
|
||
self.av5_label.setText("")
|
||
|
||
self.bp1_label.setText("")
|
||
self.bp2_label.setText("")
|
||
self.bp3_label.setText("")
|
||
self.bp4_label.setText("")
|
||
self.bp5_label.setText("")
|
||
|
||
self.ap1_label.setText("")
|
||
self.ap2_label.setText("")
|
||
self.ap3_label.setText("")
|
||
self.ap4_label.setText("")
|
||
self.ap5_label.setText("")
|
||
|
||
def send_order(self) -> None:
|
||
"""
|
||
Send new order manually.
|
||
"""
|
||
symbol: str = str(self.symbol_line.text())
|
||
if not symbol:
|
||
QtWidgets.QMessageBox.critical(self, _("委托失败"), _("请输入合约代码"))
|
||
return
|
||
|
||
volume_text: str = str(self.volume_line.text())
|
||
if not volume_text:
|
||
QtWidgets.QMessageBox.critical(self, _("委托失败"), _("请输入委托数量"))
|
||
return
|
||
volume: float = float(volume_text)
|
||
|
||
price_text: str = str(self.price_line.text())
|
||
if not price_text:
|
||
price: float = 0
|
||
else:
|
||
price = float(price_text)
|
||
|
||
req: OrderRequest = OrderRequest(
|
||
symbol=symbol,
|
||
exchange=Exchange(str(self.exchange_combo.currentText())),
|
||
direction=Direction(str(self.direction_combo.currentText())),
|
||
type=OrderType(str(self.order_type_combo.currentText())),
|
||
volume=volume,
|
||
price=price,
|
||
offset=Offset(str(self.offset_combo.currentText())),
|
||
reference="ManualTrading"
|
||
)
|
||
|
||
gateway_name: str = str(self.gateway_combo.currentText())
|
||
|
||
self.main_engine.send_order(req, gateway_name)
|
||
|
||
def cancel_all(self) -> None:
|
||
"""
|
||
Cancel all active orders.
|
||
"""
|
||
order_list: list[OrderData] = self.main_engine.get_all_active_orders()
|
||
for order in order_list:
|
||
req: CancelRequest = order.create_cancel_request()
|
||
self.main_engine.cancel_order(req, order.gateway_name)
|
||
|
||
def update_with_cell(self, cell: BaseCell) -> None:
|
||
""""""
|
||
data = cell.get_data()
|
||
|
||
self.symbol_line.setText(data.symbol)
|
||
self.exchange_combo.setCurrentIndex(
|
||
self.exchange_combo.findText(data.exchange.value)
|
||
)
|
||
|
||
self.set_vt_symbol()
|
||
|
||
if isinstance(data, PositionData):
|
||
if data.direction == Direction.SHORT:
|
||
direction: Direction = Direction.LONG
|
||
elif data.direction == Direction.LONG:
|
||
direction = Direction.SHORT
|
||
else: # Net position mode
|
||
if data.volume > 0:
|
||
direction = Direction.SHORT
|
||
else:
|
||
direction = Direction.LONG
|
||
|
||
self.direction_combo.setCurrentIndex(
|
||
self.direction_combo.findText(direction.value)
|
||
)
|
||
self.offset_combo.setCurrentIndex(
|
||
self.offset_combo.findText(Offset.CLOSE.value)
|
||
)
|
||
self.volume_line.setText(str(abs(data.volume)))
|
||
|
||
|
||
class ActiveOrderMonitor(OrderMonitor):
|
||
"""
|
||
Monitor which shows active order only.
|
||
"""
|
||
|
||
def process_event(self, event: Event) -> None:
|
||
"""
|
||
Hides the row if order is not active.
|
||
"""
|
||
super().process_event(event)
|
||
|
||
order: OrderData = event.data
|
||
row_cells: dict = self.cells[order.vt_orderid]
|
||
row: int = self.row(row_cells["volume"])
|
||
|
||
if order.is_active():
|
||
self.showRow(row)
|
||
else:
|
||
self.hideRow(row)
|
||
|
||
|
||
class ContractManager(QtWidgets.QWidget):
|
||
"""
|
||
Query contract data available to trade in system.
|
||
"""
|
||
|
||
headers: dict[str, str] = {
|
||
"vt_symbol": _("本地代码"),
|
||
"symbol": _("代码"),
|
||
"exchange": _("交易所"),
|
||
"name": _("名称"),
|
||
"product": _("合约分类"),
|
||
"size": _("合约乘数"),
|
||
"pricetick": _("价格跳动"),
|
||
"min_volume": _("最小委托量"),
|
||
"option_portfolio": _("期权产品"),
|
||
"option_expiry": _("期权到期日"),
|
||
"option_strike": _("期权行权价"),
|
||
"option_type": _("期权类型"),
|
||
"gateway_name": _("交易接口"),
|
||
}
|
||
|
||
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.event_engine: EventEngine = event_engine
|
||
|
||
self.init_ui()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setWindowTitle(_("合约查询"))
|
||
self.resize(1000, 600)
|
||
|
||
self.filter_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit()
|
||
self.filter_line.setPlaceholderText(_("输入合约代码或者交易所，留空则查询所有合约"))
|
||
|
||
self.button_show: QtWidgets.QPushButton = QtWidgets.QPushButton(_("查询"))
|
||
self.button_show.clicked.connect(self.show_contracts)
|
||
|
||
labels: list = []
|
||
for name, display in self.headers.items():
|
||
label: str = f"{display}\n{name}"
|
||
labels.append(label)
|
||
|
||
self.contract_table: QtWidgets.QTableWidget = QtWidgets.QTableWidget()
|
||
self.contract_table.setColumnCount(len(self.headers))
|
||
self.contract_table.setHorizontalHeaderLabels(labels)
|
||
self.contract_table.verticalHeader().setVisible(False)
|
||
self.contract_table.setEditTriggers(self.contract_table.EditTrigger.NoEditTriggers)
|
||
self.contract_table.setAlternatingRowColors(True)
|
||
|
||
hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
|
||
hbox.addWidget(self.filter_line)
|
||
hbox.addWidget(self.button_show)
|
||
|
||
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
vbox.addLayout(hbox)
|
||
vbox.addWidget(self.contract_table)
|
||
|
||
self.setLayout(vbox)
|
||
|
||
def show_contracts(self) -> None:
|
||
"""
|
||
Show contracts by symbol
|
||
"""
|
||
flt: str = str(self.filter_line.text())
|
||
|
||
all_contracts: list[ContractData] = self.main_engine.get_all_contracts()
|
||
if flt:
|
||
contracts: list[ContractData] = [
|
||
contract for contract in all_contracts if flt in contract.vt_symbol
|
||
]
|
||
else:
|
||
contracts = all_contracts
|
||
|
||
self.contract_table.clearContents()
|
||
self.contract_table.setRowCount(len(contracts))
|
||
|
||
for row, contract in enumerate(contracts):
|
||
for column, name in enumerate(self.headers.keys()):
|
||
value: Any = getattr(contract, name)
|
||
|
||
if value in {None, 0}:
|
||
value = ""
|
||
|
||
cell: BaseCell
|
||
if isinstance(value, Enum):
|
||
cell = EnumCell(value, contract)
|
||
elif isinstance(value, datetime):
|
||
cell = DateCell(value, contract)
|
||
else:
|
||
cell = BaseCell(value, contract)
|
||
self.contract_table.setItem(row, column, cell)
|
||
|
||
self.contract_table.resizeColumnsToContents()
|
||
|
||
|
||
class AboutDialog(QtWidgets.QDialog):
|
||
"""
|
||
Information about the trading platform.
|
||
"""
|
||
|
||
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.event_engine: EventEngine = event_engine
|
||
|
||
self.init_ui()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setWindowTitle(_("关于VeighNa Trader"))
|
||
|
||
from ... import __version__ as vnpy_version
|
||
|
||
text: str = f"""
|
||
By Traders, For Traders.
|
||
|
||
Created by VeighNa Technology
|
||
|
||
|
||
License：MIT
|
||
Website：www.vnpy.com
|
||
Github：www.github.com/vnpy/vnpy
|
||
|
||
|
||
VeighNa - {vnpy_version}
|
||
Python - {platform.python_version()}
|
||
PySide6 - {metadata.version("pyside6")}
|
||
NumPy - {metadata.version("numpy")}
|
||
pandas - {metadata.version("pandas")}
|
||
"""
|
||
|
||
label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
label.setText(text)
|
||
label.setMinimumWidth(500)
|
||
|
||
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
vbox.addWidget(label)
|
||
self.setLayout(vbox)
|
||
|
||
|
||
class GlobalDialog(QtWidgets.QDialog):
|
||
"""
|
||
Start connection of a certain gateway.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.widgets: dict[str, tuple[QtWidgets.QLineEdit, type]] = {}
|
||
|
||
self.init_ui()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setWindowTitle(_("全局配置"))
|
||
self.setMinimumWidth(800)
|
||
|
||
settings: dict = copy(SETTINGS)
|
||
settings.update(load_json(SETTING_FILENAME))
|
||
|
||
# Initialize line edits and form layout based on setting.
|
||
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
|
||
|
||
for field_name, field_value in settings.items():
|
||
field_type: type = type(field_value)
|
||
widget: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(field_value))
|
||
|
||
form.addRow(f"{field_name} <{field_type.__name__}>", widget)
|
||
self.widgets[field_name] = (widget, field_type)
|
||
|
||
button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("确定"))
|
||
button.clicked.connect(self.update_setting)
|
||
form.addRow(button)
|
||
|
||
scroll_widget: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
scroll_widget.setLayout(form)
|
||
|
||
scroll_area: QtWidgets.QScrollArea = QtWidgets.QScrollArea()
|
||
scroll_area.setWidgetResizable(True)
|
||
scroll_area.setWidget(scroll_widget)
|
||
|
||
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
vbox.addWidget(scroll_area)
|
||
self.setLayout(vbox)
|
||
|
||
def update_setting(self) -> None:
|
||
"""
|
||
Get setting value from line edits and update global setting file.
|
||
"""
|
||
settings: dict = {}
|
||
for field_name, tp in self.widgets.items():
|
||
widget, field_type = tp
|
||
value_text: str = widget.text()
|
||
|
||
if field_type is bool:
|
||
if value_text == "True":
|
||
field_value: bool = True
|
||
else:
|
||
field_value = False
|
||
else:
|
||
field_value = field_type(value_text)
|
||
|
||
settings[field_name] = field_value
|
||
|
||
QtWidgets.QMessageBox.information(
|
||
self,
|
||
_("注意"),
|
||
_("全局配置的修改需要重启后才会生效！"),
|
||
QtWidgets.QMessageBox.StandardButton.Ok
|
||
)
|
||
|
||
save_json(SETTING_FILENAME, settings)
|
||
self.accept()
|
||
|
||
|
||
class WechatWorker(QtCore.QThread):
|
||
"""
|
||
Background worker for two-phase WeChat binding: QR-code login first,
|
||
then wait for the user's first inbound message to capture user info.
|
||
"""
|
||
|
||
signal_qr_ready: QtCore.Signal = QtCore.Signal(str)
|
||
signal_waiting_message: QtCore.Signal = QtCore.Signal()
|
||
signal_bound: QtCore.Signal = QtCore.Signal(object, object)
|
||
signal_failed: QtCore.Signal = QtCore.Signal(str)
|
||
|
||
def __init__(self) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self._stop: bool = False
|
||
|
||
def stop(self) -> None:
|
||
"""Request the worker loop to terminate at the next checkpoint."""
|
||
self._stop = True
|
||
|
||
def run(self) -> None:
|
||
""""""
|
||
try:
|
||
deadline: float = time.monotonic() + 600.0 # wait for 10 minutes
|
||
|
||
# Phase 1: request QR-code and wait for user confirmation
|
||
creds: Credentials | None = None
|
||
while not self._stop and time.monotonic() < deadline:
|
||
try:
|
||
qrcode_value, scan_url = request_qrcode()
|
||
self.signal_qr_ready.emit(scan_url)
|
||
|
||
creds = wait_for_login(
|
||
qrcode_value,
|
||
deadline,
|
||
)
|
||
if creds is not None:
|
||
break
|
||
except WeixinTimeout:
|
||
continue
|
||
|
||
if self._stop or creds is None:
|
||
self.signal_failed.emit(_("扫码超时或被取消"))
|
||
return
|
||
|
||
# Phase 2: long-poll inbound messages until the first one arrives
|
||
self.signal_waiting_message.emit()
|
||
sync_buf: str = ""
|
||
while not self._stop and time.monotonic() < deadline:
|
||
try:
|
||
user_ids, sync_buf = poll(
|
||
creds,
|
||
sync_buf,
|
||
)
|
||
if user_ids:
|
||
self.signal_bound.emit(creds, user_ids[0])
|
||
return
|
||
except WeixinTimeout:
|
||
continue
|
||
|
||
self.signal_failed.emit(_("等待首条消息超时"))
|
||
except WeixinError as exc:
|
||
self.signal_failed.emit(str(exc))
|
||
except Exception as exc:
|
||
self.signal_failed.emit(str(exc))
|
||
|
||
|
||
class WechatDialog(QtWidgets.QDialog):
|
||
"""
|
||
WeChat binding dialog: shows status when bound, drives the scan and
|
||
first-message flow when not bound.
|
||
"""
|
||
|
||
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
|
||
""""""
|
||
super().__init__()
|
||
|
||
self.main_engine: MainEngine = main_engine
|
||
self.event_engine: EventEngine = event_engine
|
||
self.wechat_engine: WechatEngine = cast(WechatEngine, main_engine.get_engine("wechat"))
|
||
|
||
self.worker: WechatWorker | None = None
|
||
|
||
self.init_ui()
|
||
self.refresh_status()
|
||
|
||
def init_ui(self) -> None:
|
||
""""""
|
||
self.setWindowTitle(_("微信通知"))
|
||
self.setMinimumWidth(380)
|
||
|
||
# Create Stacked Widget
|
||
self.stack: QtWidgets.QStackedWidget = QtWidgets.QStackedWidget()
|
||
|
||
# Create pages
|
||
self.page_status: QtWidgets.QWidget = self._create_status_page()
|
||
self.page_loading: QtWidgets.QWidget = self._create_loading_page()
|
||
self.page_qr: QtWidgets.QWidget = self._create_qr_page()
|
||
self.page_waiting: QtWidgets.QWidget = self._create_waiting_page()
|
||
self.page_result: QtWidgets.QWidget = self._create_result_page()
|
||
|
||
self.stack.addWidget(self.page_status)
|
||
self.stack.addWidget(self.page_loading)
|
||
self.stack.addWidget(self.page_qr)
|
||
self.stack.addWidget(self.page_waiting)
|
||
self.stack.addWidget(self.page_result)
|
||
|
||
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
vbox.addWidget(self.stack)
|
||
self.setLayout(vbox)
|
||
|
||
def _create_status_page(self) -> QtWidgets.QWidget:
|
||
"""Create the status page."""
|
||
page: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
|
||
# Status area
|
||
group_status: QtWidgets.QGroupBox = QtWidgets.QGroupBox(_("账号状态"))
|
||
self.status_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.status_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
|
||
# Info form layout
|
||
self.info_widget: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
self.info_form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
|
||
self.info_form.setLabelAlignment(QtCore.Qt.AlignmentFlag.AlignRight | QtCore.Qt.AlignmentFlag.AlignVCenter)
|
||
self.info_widget.setLayout(self.info_form)
|
||
|
||
self.bot_id_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.bot_id_label.setTextInteractionFlags(QtCore.Qt.TextInteractionFlag.TextSelectableByMouse)
|
||
self.bot_id_label.setWordWrap(True)
|
||
|
||
self.user_id_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.user_id_label.setTextInteractionFlags(QtCore.Qt.TextInteractionFlag.TextSelectableByMouse)
|
||
self.user_id_label.setWordWrap(True)
|
||
|
||
self.gateway_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.gateway_label.setTextInteractionFlags(QtCore.Qt.TextInteractionFlag.TextSelectableByMouse)
|
||
self.gateway_label.setWordWrap(True)
|
||
|
||
self.interval_spin: QtWidgets.QSpinBox = QtWidgets.QSpinBox()
|
||
self.interval_spin.setRange(1, 8640)
|
||
interval_tip: str = _(
|
||
"控制两次微信推送之间的间隔时间。\n"
|
||
"间隔内的新消息会暂存，并在下次推送时合并发送。\n"
|
||
"用户每发送 1 条消息，机器人可在 24 小时内推送 10 条；超限后需用户再次发送消息才能恢复。"
|
||
)
|
||
self.interval_spin.setSuffix(_(" 秒"))
|
||
self.interval_spin.setToolTip(interval_tip)
|
||
self.interval_spin.valueChanged.connect(self.change_interval)
|
||
|
||
self.interval_spin.setValue(self.wechat_engine.send_interval)
|
||
self.info_form.addRow("Bot ID：", self.bot_id_label)
|
||
self.info_form.addRow(_("用户 ID："), self.user_id_label)
|
||
self.info_form.addRow(_("网关："), self.gateway_label)
|
||
|
||
group_setting: QtWidgets.QGroupBox = QtWidgets.QGroupBox(_("推送设置"))
|
||
setting_form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
|
||
setting_form.setLabelAlignment(QtCore.Qt.AlignmentFlag.AlignRight | QtCore.Qt.AlignmentFlag.AlignVCenter)
|
||
setting_form.addRow(_("推送间隔："), self.interval_spin)
|
||
group_setting.setLayout(setting_form)
|
||
|
||
self.bind_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("开始绑定"))
|
||
self.bind_button.clicked.connect(self.start_bind)
|
||
|
||
self.test_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("测试消息"))
|
||
self.test_button.clicked.connect(self.send_test_message)
|
||
|
||
self.unbind_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("解除绑定"))
|
||
self.unbind_button.clicked.connect(self.unbind)
|
||
|
||
status_layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
status_layout.addStretch()
|
||
status_layout.addWidget(self.status_label)
|
||
status_layout.addWidget(self.info_widget)
|
||
status_layout.addStretch()
|
||
status_layout.addWidget(self.bind_button)
|
||
status_layout.addWidget(self.test_button)
|
||
status_layout.addWidget(self.unbind_button)
|
||
group_status.setLayout(status_layout)
|
||
|
||
layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
layout.addWidget(group_status)
|
||
layout.addWidget(group_setting)
|
||
page.setLayout(layout)
|
||
|
||
return page
|
||
|
||
def _create_loading_page(self) -> QtWidgets.QWidget:
|
||
"""Create the loading page."""
|
||
page: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
|
||
label: QtWidgets.QLabel = QtWidgets.QLabel(_("正在向服务器请求二维码，请稍候..."))
|
||
label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
|
||
cancel_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("取消绑定"))
|
||
cancel_button.clicked.connect(self.cancel_bind)
|
||
|
||
layout.addStretch()
|
||
layout.addWidget(label)
|
||
layout.addStretch()
|
||
layout.addWidget(cancel_button)
|
||
page.setLayout(layout)
|
||
return page
|
||
|
||
def _create_qr_page(self) -> QtWidgets.QWidget:
|
||
"""Create the QR page."""
|
||
page: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
|
||
label: QtWidgets.QLabel = QtWidgets.QLabel(_("第一步：请使用微信扫描下方二维码"))
|
||
label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
|
||
self.qr_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.qr_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
self.qr_label.setMinimumSize(260, 260)
|
||
|
||
cancel_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("取消绑定"))
|
||
cancel_button.clicked.connect(self.cancel_bind)
|
||
|
||
layout.addStretch()
|
||
layout.addWidget(label)
|
||
layout.addWidget(self.qr_label)
|
||
layout.addStretch()
|
||
layout.addWidget(cancel_button)
|
||
page.setLayout(layout)
|
||
return page
|
||
|
||
def _create_waiting_page(self) -> QtWidgets.QWidget:
|
||
"""Create the waiting page."""
|
||
page: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
|
||
label: QtWidgets.QLabel = QtWidgets.QLabel(_("第二步：扫码成功！\n\n请在微信中向机器人发送任意文字消息以完成最终激活。"))
|
||
label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
|
||
cancel_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("取消绑定"))
|
||
cancel_button.clicked.connect(self.cancel_bind)
|
||
|
||
layout.addStretch()
|
||
layout.addWidget(label)
|
||
layout.addStretch()
|
||
layout.addWidget(cancel_button)
|
||
page.setLayout(layout)
|
||
return page
|
||
|
||
def _create_result_page(self) -> QtWidgets.QWidget:
|
||
"""Create the result page."""
|
||
page: QtWidgets.QWidget = QtWidgets.QWidget()
|
||
layout: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
|
||
|
||
self.result_label: QtWidgets.QLabel = QtWidgets.QLabel()
|
||
self.result_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||
self.result_label.setWordWrap(True)
|
||
|
||
self.result_ok_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("完成"))
|
||
self.result_ok_button.clicked.connect(self.refresh_status)
|
||
|
||
self.result_retry_button: QtWidgets.QPushButton = QtWidgets.QPushButton(_("重试"))
|
||
self.result_retry_button.clicked.connect(self.start_bind)
|
||
|
||
layout.addStretch()
|
||
layout.addWidget(self.result_label)
|
||
layout.addStretch()
|
||
layout.addWidget(self.result_retry_button)
|
||
layout.addWidget(self.result_ok_button)
|
||
page.setLayout(layout)
|
||
return page
|
||
|
||
def refresh_status(self) -> None:
|
||
"""Refresh dialog widgets according to current binding state."""
|
||
creds: Credentials | None = self.wechat_engine.creds
|
||
bound: bool = bool(creds and self.wechat_engine.user_id)
|
||
|
||
if bound and creds:
|
||
self.status_label.setText(_("状态：已绑定"))
|
||
|
||
self.bot_id_label.setText(creds.bot_id)
|
||
self.user_id_label.setText(self.wechat_engine.user_id)
|
||
self.gateway_label.setText(creds.base_url)
|
||
|
||
self.info_widget.show()
|
||
self.bind_button.setText(_("重新绑定"))
|
||
self.test_button.show()
|
||
self.unbind_button.setEnabled(True)
|
||
else:
|
||
self.status_label.setText(_("状态：未绑定"))
|
||
self.bot_id_label.clear()
|
||
self.user_id_label.clear()
|
||
self.gateway_label.clear()
|
||
self.info_widget.hide()
|
||
self.bind_button.setText(_("开始绑定"))
|
||
self.test_button.hide()
|
||
self.unbind_button.setEnabled(False)
|
||
|
||
with QtCore.QSignalBlocker(self.interval_spin):
|
||
self.interval_spin.setValue(self.wechat_engine.send_interval)
|
||
|
||
self.stack.setCurrentWidget(self.page_status)
|
||
|
||
def send_test_message(self) -> None:
|
||
"""Send a test message via WeChat engine."""
|
||
self.wechat_engine.send_wechat(_("VeighNa Trader 消息推送测试"))
|
||
QtWidgets.QMessageBox.information(
|
||
self,
|
||
_("测试消息"),
|
||
_("测试消息已加入推送队列！"),
|
||
QtWidgets.QMessageBox.StandardButton.Ok
|
||
)
|
||
|
||
def start_bind(self) -> None:
|
||
"""Start the background binding worker."""
|
||
if self.worker and self.worker.isRunning():
|
||
return
|
||
|
||
self.stack.setCurrentWidget(self.page_loading)
|
||
|
||
self.worker = WechatWorker()
|
||
self.worker.signal_qr_ready.connect(self.show_qrcode)
|
||
self.worker.signal_waiting_message.connect(self.show_waiting_message)
|
||
self.worker.signal_bound.connect(self.finish_bind)
|
||
self.worker.signal_failed.connect(self.fail_bind)
|
||
self.worker.start()
|
||
|
||
def show_qrcode(self, url: str) -> None:
|
||
"""Render the scan URL as a QR-code pixmap in the dialog."""
|
||
qr: Any = qrcode.QRCode(border=1)
|
||
qr.add_data(url)
|
||
qr.make(fit=True)
|
||
img: Any = qr.make_image(fill_color="black", back_color="white")
|
||
|
||
buf: BytesIO = BytesIO()
|
||
img.save(buf, format="PNG")
|
||
|
||
pixmap: QtGui.QPixmap = QtGui.QPixmap()
|
||
if not pixmap.loadFromData(buf.getvalue()):
|
||
self.fail_bind(_("二维码图片加载失败"))
|
||
return
|
||
|
||
scaled: QtGui.QPixmap = pixmap.scaled(
|
||
260,
|
||
260,
|
||
QtCore.Qt.AspectRatioMode.KeepAspectRatio,
|
||
QtCore.Qt.TransformationMode.SmoothTransformation
|
||
)
|
||
self.qr_label.setPixmap(scaled)
|
||
|
||
self.stack.setCurrentWidget(self.page_qr)
|
||
|
||
def show_waiting_message(self) -> None:
|
||
"""Switch UI to the second binding phase."""
|
||
self.stack.setCurrentWidget(self.page_waiting)
|
||
|
||
def finish_bind(self, creds: Credentials, user_id: str) -> None:
|
||
"""Persist binding result via WechatEngine and refresh the UI."""
|
||
self.wechat_engine.bind(creds, user_id)
|
||
|
||
self.result_label.setText(_("微信通知绑定成功！"))
|
||
self.result_retry_button.hide()
|
||
self.result_ok_button.show()
|
||
|
||
self.stack.setCurrentWidget(self.page_result)
|
||
|
||
def fail_bind(self, error: str) -> None:
|
||
"""Show the failure reason and reset the UI for retry."""
|
||
self.result_label.setText(_("绑定失败：\n{}").format(error))
|
||
self.result_retry_button.show()
|
||
self.result_ok_button.show()
|
||
|
||
self.stack.setCurrentWidget(self.page_result)
|
||
|
||
def cancel_bind(self) -> None:
|
||
"""Cancel binding process and return to status page."""
|
||
if self.worker and self.worker.isRunning():
|
||
self.worker.stop()
|
||
self.worker.wait(2000)
|
||
self.refresh_status()
|
||
|
||
def unbind(self) -> None:
|
||
"""Confirm and clear current binding."""
|
||
reply: QtWidgets.QMessageBox.StandardButton = QtWidgets.QMessageBox.question(
|
||
self,
|
||
_("解除绑定"),
|
||
_("确认解除当前微信绑定？"),
|
||
QtWidgets.QMessageBox.StandardButton.Yes | QtWidgets.QMessageBox.StandardButton.No,
|
||
QtWidgets.QMessageBox.StandardButton.No,
|
||
)
|
||
|
||
if reply == QtWidgets.QMessageBox.StandardButton.Yes:
|
||
self.wechat_engine.unbind()
|
||
self.refresh_status()
|
||
|
||
def change_interval(self, value: int) -> None:
|
||
"""Persist send interval updates from the status page."""
|
||
self.wechat_engine.send_interval = value
|
||
self.wechat_engine.save_setting()
|
||
|
||
def closeEvent(self, event: QtGui.QCloseEvent) -> None:
|
||
"""Stop the binding worker before closing the dialog."""
|
||
if self.worker and self.worker.isRunning():
|
||
self.worker.stop()
|
||
self.worker.wait(2000)
|
||
|
||
super().closeEvent(event)
|