数据记录模块，也是比较容易单拎出来的
我的逻辑是：账户里面找标识为行情的号，再加入标记为收藏的品种
增加了关闭窗口隐藏，托盘图标后台记录
import sys
from datetime import datetime
from typing import Any, List
from PySide6 import QtCore, QtWidgets
from PySide6.QtCore import Qt
from PySide6.QtGui import QIcon, QAction
from PySide6.QtWidgets import QSpacerItem, QApplication, QSystemTrayIcon, QMenu
from qfluentwidgets import LineEdit, SpinBox, TextEdit, BodyLabel, PrimaryPushButton, setTheme, Theme, Action, \
FluentIcon, RoundMenu
from vnpy_ctp import CtpGateway
from vnpy.trader.engine import MainEngine
from guanlan_futures.common.modules import accounts
from vnpy.event import Event, EventEngine
from vnpy_datarecoder.engine import RecorderEngine, APP_NAME, EVENT_RECORDER_LOG, EVENT_RECORDER_UPDATE, \
EVENT_RECORDER_EXCEPTION
from vnpy.trader.event import EVENT_CONTRACT
from vnpy.trader.object import ContractData
from guanlan_futures.common import resource
from guanlan_widgets import MicaWindow
# from guanlan_futures.common import resource
class DataRecoderWindow(MicaWindow):
signal_log: QtCore.Signal = QtCore.Signal(Event)
signal_update: QtCore.Signal = QtCore.Signal(Event)
signal_contract: QtCore.Signal = QtCore.Signal(Event)
signal_exception: QtCore.Signal = QtCore.Signal(Event)
def __init__(self):
super().__init__()
self.setWindowIcon(QIcon(':/futures/icon/recorder.png'))
self.setWindowTitle("行情记录")
self.titleBar.minBtn.hide()
self.titleBar.maxBtn.hide()
self.setMicaEffectEnabled(True)
self.event_engine = EventEngine()
self.main_engine = MainEngine(self.event_engine)
self.main_engine.add_engine(RecorderEngine)
self.recorder_engine: RecorderEngine = self.main_engine.get_engine(APP_NAME)
self.init_ui()
self.init_tray_icon()
self.register_event()
self.recorder_engine.put_event()
# 连接服务器
self.connect_ctp()
def closeEvent(self, event):
# 忽略退出事件，而是隐藏到托盘
event.ignore()
self.hide()
def init_tray_icon(self):
# 配置系统托盘
self.tray_icon = QSystemTrayIcon(self)
self.tray_icon.setIcon(QIcon(':/futures/icon/recorder.png'))
self.tray_icon.setToolTip("观澜量化 - 行情记录")
self.tray_icon.show()
# 创建托盘的右键菜单
self.tray_menu = RoundMenu(parent=self)
self.tray_menu.addAction(Action(FluentIcon.ZOOM, "显示", triggered=self.show))
self.tray_menu.addAction(Action(FluentIcon.CLOSE, "退出", triggered=self.quit))
# 配置菜单并显示托盘
self.tray_icon.setContextMenu(self.tray_menu) # 把tpMenu设定为托盘的右键菜单
def quit(self):
# 真正的退出
self.recorder_engine.close()
self.main_engine.close()
sys.exit()
def connect_ctp(self):
# 找账户配置为行情的
self.gateway_name = ""
setting = {}
for key, value in accounts.account_data.items():
if value['md_data'] == "1":
self.gateway_name = key
setting = value
break
gateway = self.main_engine.add_gateway(CtpGateway, self.gateway_name)
self.main_engine.connect(setting, self.gateway_name)
def init_ui(self) -> None:
""""""
self.symbol_line: LineEdit = LineEdit()
self.interval_spin: SpinBox = SpinBox()
self.interval_spin.setMinimum(1)
self.interval_spin.setMaximum(60)
self.interval_spin.setValue(self.recorder_engine.timer_interval)
self.interval_spin.setSuffix(" 秒")
self.interval_spin.valueChanged.connect(self.set_interval)
contracts: List[ContractData] = self.main_engine.get_all_contracts()
self.vt_symbols: list = [contract.vt_symbol for contract in contracts]
self.symbol_completer: QtWidgets.QCompleter = QtWidgets.QCompleter(self.vt_symbols)
self.symbol_completer.setFilterMode(Qt.MatchContains)
self.symbol_line.setCompleter(self.symbol_completer)
add_bar_button: PrimaryPushButton = PrimaryPushButton(text="添加", parent=self)
add_bar_button.clicked.connect(self.add_bar_recording)
remove_bar_button: PrimaryPushButton = PrimaryPushButton(text="移除", parent=self)
remove_bar_button.clicked.connect(self.remove_bar_recording)
add_tick_button: PrimaryPushButton = PrimaryPushButton(text="添加", parent=self)
add_tick_button.clicked.connect(self.add_tick_recording)
remove_tick_button: PrimaryPushButton = PrimaryPushButton(text="移除", parent=self)
remove_tick_button.clicked.connect(self.remove_tick_recording)
self.bar_recording_edit: TextEdit = TextEdit()
self.bar_recording_edit.setReadOnly(True)
self.tick_recording_edit: TextEdit = TextEdit()
self.tick_recording_edit.setReadOnly(True)
self.log_edit: TextEdit = TextEdit()
self.log_edit.setReadOnly(True)
# Set layout
grid: QtWidgets.QGridLayout = QtWidgets.QGridLayout()
grid.addWidget(BodyLabel(text="K线记录", parent=self), 0, 0)
grid.addWidget(add_bar_button, 0, 1)
grid.addWidget(remove_bar_button, 0, 2)
grid.addWidget(BodyLabel(text="Tick记录", parent=self), 1, 0)
grid.addWidget(add_tick_button, 1, 1)
grid.addWidget(remove_tick_button, 1, 2)
self.symbol_line.setMinimumWidth(300)
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
form.addRow(BodyLabel(text="本地代码", parent=self), self.symbol_line)
form.addRow(BodyLabel(text="写入间隔", parent=self), self.interval_spin)
hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox.addLayout(form)
# hbox.addWidget(BodyLabel(text=" ", parent=self))
hbox.addItem(QSpacerItem(50, 0))
hbox.addLayout(grid)
hbox.addStretch(1)
hbox.setContentsMargins(0, 10, 0, 0)
grid2: QtWidgets.QGridLayout = QtWidgets.QGridLayout()
grid2.addWidget(BodyLabel(text="K线记录列表", parent=self), 0, 0)
grid2.addWidget(BodyLabel(text="Tick记录列表", parent=self), 0, 1)
grid2.addWidget(self.bar_recording_edit, 1, 0)
grid2.addWidget(self.tick_recording_edit, 1, 1)
grid2.addWidget(self.log_edit, 2, 0, 1, 2)
grid2.setContentsMargins(0, 10, 0, 0)
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addLayout(hbox)
vbox.addLayout(grid2)
vbox.setContentsMargins(20, 40, 12, 12)
self.setLayout(vbox)
def register_event(self) -> None:
""""""
self.signal_log.connect(self.process_log_event)
self.signal_contract.connect(self.process_contract_event)
self.signal_update.connect(self.process_update_event)
self.signal_exception.connect(self.process_exception_event)
self.event_engine.register(EVENT_CONTRACT, self.signal_contract.emit)
self.event_engine.register(EVENT_RECORDER_LOG, self.signal_log.emit)
self.event_engine.register(EVENT_RECORDER_UPDATE, self.signal_update.emit)
self.event_engine.register(EVENT_RECORDER_EXCEPTION, self.signal_exception.emit)
def process_log_event(self, event: Event) -> None:
""""""
timestamp: str = datetime.now().strftime("%H:%M:%S")
msg: str = f"{timestamp}\t{event.data}"
self.log_edit.append(msg)
def process_update_event(self, event: Event) -> None:
""""""
data: Any = event.data
self.bar_recording_edit.clear()
bar_text: str = "\n".join(data["bar"])
self.bar_recording_edit.setText(bar_text)
self.tick_recording_edit.clear()
tick_text: str = "\n".join(data["tick"])
self.tick_recording_edit.setText(tick_text)
def process_contract_event(self, event: Event) -> None:
""""""
contract: ContractData = event.data
self.vt_symbols.append(contract.vt_symbol)
model: QtCore.QAbstractItemModel = self.symbol_completer.model()
model.setStringList(self.vt_symbols)
def process_exception_event(self, event: Event) -> None:
""""""
exc_info = event.data
raise exc_info[1].with_traceback(exc_info[2])
def add_bar_recording(self) -> None:
""""""
vt_symbol: str = self.symbol_line.text()
self.recorder_engine.add_bar_recording(vt_symbol, self.gateway_name)
def add_tick_recording(self) -> None:
""""""
vt_symbol: str = self.symbol_line.text()
self.recorder_engine.add_tick_recording(vt_symbol)
def remove_bar_recording(self) -> None:
""""""
vt_symbol: str = self.symbol_line.text()
self.recorder_engine.remove_bar_recording(vt_symbol)
def remove_tick_recording(self) -> None:
""""""
vt_symbol: str = self.symbol_line.text()
self.recorder_engine.remove_tick_recording(vt_symbol)
def set_interval(self, interval) -> None:
""""""
self.recorder_engine.timer_interval = interval
if __name__ == '__main__':
setTheme(Theme.DARK)
# 初始化应用和窗口
app = QApplication(sys.argv)
win = DataRecoderWindow()
win.show()
# 运行应用
sys.exit(app.exec())
