from PySide6 import QtWidgets, QtCore from vnpy_tts.gateway.tts_gateway import TtsGateway, TtsMdApi class SimpleWidget(QtWidgets.QWidget): """简单图形控件""" singal = QtCore.Signal(str) # QtCore.Signal def __init__(self) -> None: """构造函数""" super().__init__() # 调用Qt对象c++中的构造函数 self.api = None self.log_monitor = QtWidgets.QTextEdit() self.log_monitor.setReadOnly(True) self.subscribe_button = QtWidgets.QPushButton("订阅") self.symbol_line = QtWidgets.QLineEdit() self.subscribe_button.clicked.connect(self.subscribe_symbol) self.singal.connect(self.log_monitor.append) # 设置布局组合 vbox = QtWidgets.QVBoxLayout() vbox.addWidget(self.log_monitor) vbox.addWidget(self.symbol_line) vbox.addWidget(self.subscribe_button) self.setLayout(vbox) def subscribe_symbol(self): symbol =self.symbol_line.text() self.api.subscribeMarketData(symbol) class CtpMdApi(TtsMdApi): def __init__(self, widget) -> None: super().__init__() self.widget = widget def onFrontConnected(self): self.widget.signal.emit("服务器连接成功") ctp_req = { "UserID":"1234", "Password":"123456", "BrokerID":"9999" } self.reqUserLogin(ctp_req, 1) def onFrontDisconnected(self, reason): self.widget.signal.emit("服务器连接断开", reason) def onRspUserLogin(self, data, error, reqid, last): if not error["ErrorID"]: self.widget.signal.emit("行情服务器登录成功") self.subscribeMarketData("rb2410") else: self.widget.signal.emit("行情服务器登录失败 {error}") def onRtnDepthMarketData(self, data): self.widget.signal.emit(str(data)) def main(): app = QtWidgets.QApplication() widget = SimpleWidget() widget.show() api = CtpMdApi(widget.log_monitor) api.createFtdcMdApi(".") api.registerFront("tcp://122.51.136.165:20004") api.init() app.exec() if __name__ == "__main__": main()
api = CtpMdApi(xxx)
这行的xxx里应该有2个参数，第一个是gateway类
`from PySide6 import QtWidgets, QtCore
from vnpy_tts.gateway.tts_gateway import TtsGateway, TtsMdApi
class SimpleWidget(QtWidgets.QWidget):
"""简单图形控件"""
singal = QtCore.Signal(str) # QtCore.Signal
def __init__(self) -> None:
"""构造函数"""
super().__init__() # 调用Qt对象c++中的构造函数
self.api = None
# self.gateway_name: str = "CTP"
self.log_monitor = QtWidgets.QTextEdit()
self.log_monitor.setReadOnly(True)
self.subscribe_button = QtWidgets.QPushButton("订阅")
self.symbol_line = QtWidgets.QLineEdit()
self.subscribe_button.clicked.connect(self.subscribe_symbol)
self.singal.connect(self.log_monitor.append)
# 设置布局组合
vbox = QtWidgets.QVBoxLayout()
vbox.addWidget(self.log_monitor)
vbox.addWidget(self.symbol_line)
vbox.addWidget(self.subscribe_button)
self.setLayout(vbox)
def subscribe_symbol(self):
symbol =self.symbol_line.text()
self.api.subscribeMarketData(symbol)class CtpMdApi(TtsMdApi):
def __init__(self, widget) -> None:
super().__init__(widget)
self.widget = widget
# self.gateway_name: str = widget.gateway_name
# self.gateway_name: str = "CTP"
def onFrontConnected(self):
self.widget.signal.emit("服务器连接成功")
ctp_req = {
"UserID":"9643",
"Password":"123456",
"BrokerID":"9999"
}
self.reqUserLogin(ctp_req, 1)
def onFrontDisconnected(self, reason):
self.widget.signal.emit("服务器连接断开", reason)
def onRspUserLogin(self, data, error, reqid, last):
if not error["ErrorID"]:
self.widget.signal.emit("行情服务器登录成功")
self.subscribeMarketData("rb2410")
else:
self.widget.signal.emit("行情服务器登录失败 {error}")
def onRtnDepthMarketData(self, data):
self.widget.signal.emit(str(data))
def main():
app = QtWidgets.QApplication()
widget = SimpleWidget()
widget.show()
api = CtpMdApi(TtsGateway, widget)
# api = CtpMdApi(widget.log_monitor)
widget.api = api
api.createFtdcMdApi(".")
api.registerFront("tcp://122.51.136.165:20004")
# api.registerFront("tcp://114.80.54.218:41213")
api.init()
app.exec()if name == "main":
main()
`
super().__init__的时候应该传gateway而不是widget