Merge pull request vnpy#3 from noranhe/main

[Add] 类型声明

Author: vnpy

vnpy_excelrtd/__init__.py

@@ -21,16 +21,18 @@
 # SOFTWARE.
 
 from pathlib import Path
+
 from vnpy.trader.app import BaseApp
+
 from .engine import RtdEngine, APP_NAME
 
 
 class ExcelRtdApp(BaseApp):
     """"""
-    app_name = APP_NAME
-    app_module = __module__
-    app_path = Path(__file__).parent
-    display_name = "Excel RTD"
-    engine_class = RtdEngine
-    widget_name = "RtdManager"
-    icon_name = str(app_path.joinpath("ui", "rtd.ico"))
+    app_name: str = APP_NAME
+    app_module: str = __module__
+    app_path: Path = Path(__file__).parent
+    display_name: str = "Excel RTD"
+    engine_class: RtdEngine = RtdEngine
+    widget_name: str = "RtdManager"
+    icon_name: str = str(app_path.joinpath("ui", "rtd.ico"))

vnpy_excelrtd/engine.py

@@ -1,11 +1,9 @@
-""""""
-
-from typing import Set
+from typing import Set, Optional
 
 from vnpy.event import Event, EventEngine
 from vnpy.rpc import RpcServer
 from vnpy.trader.engine import BaseEngine, MainEngine
-from vnpy.trader.object import TickData, LogData, SubscribeRequest
+from vnpy.trader.object import TickData, ContractData, LogData, SubscribeRequest
 from vnpy.trader.event import EVENT_TICK
 
 
@@ -22,7 +20,7 @@ class RtdEngine(BaseEngine):
     The engine for managing RTD objects and data update.
     """
 
-    def __init__(self, main_engine: MainEngine, event_engine: EventEngine):
+    def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
         """"""
         super().__init__(main_engine, event_engine, APP_NAME)
 
@@ -52,29 +50,29 @@ def write_log(self, msg: str) -> None:
         """
         Output RTD related log message.
         """
-        log = LogData(msg=msg, gateway_name=APP_NAME)
-        event = Event(EVENT_RTD_LOG, log)
+        log: LogData = LogData(msg=msg, gateway_name=APP_NAME)
+        event: Event = Event(EVENT_RTD_LOG, log)
         self.event_engine.put(event)
 
     def subscribe(self, vt_symbol: str) -> None:
         """
         Subscribe tick data update.
         """
-        contract = self.main_engine.get_contract(vt_symbol)
+        contract: Optional[ContractData] = self.main_engine.get_contract(vt_symbol)
         if not contract:
             return
 
         if vt_symbol in self.subscribed:
             return
         self.subscribed.add(vt_symbol)
 
-        req = SubscribeRequest(
+        req: SubscribeRequest = SubscribeRequest(
             contract.symbol,
             contract.exchange
         )
         self.main_engine.subscribe(req, contract.gateway_name)
 
-    def close(self):
+    def close(self) -> None:
         """"""
         self.server.stop()
         self.server.join()

vnpy_excelrtd/ui/widget.py

@@ -4,20 +4,20 @@
 from vnpy.trader.engine import MainEngine
 from vnpy.trader.ui import QtWidgets, QtCore
 from vnpy.trader.object import LogData
-from ..engine import APP_NAME, EVENT_RTD_LOG
+from ..engine import APP_NAME, EVENT_RTD_LOG, BaseEngine
 
 
 class RtdManager(QtWidgets.QWidget):
     """"""
-    signal_log = QtCore.pyqtSignal(Event)
+    signal_log: QtCore.pyqtSignal = QtCore.pyqtSignal(Event)
 
-    def __init__(self, main_engine: MainEngine, event_engine: EventEngine):
+    def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
         """"""
         super().__init__()
 
-        self.main_engine = main_engine
-        self.event_engine = event_engine
-        self.rm_engine = main_engine.get_engine(APP_NAME)
+        self.main_engine: MainEngine = main_engine
+        self.event_engine: EventEngine = event_engine
+        self.rm_engine: BaseEngine = main_engine.get_engine(APP_NAME)
 
         self.init_ui()
         self.register_event()
@@ -29,26 +29,26 @@ def init_ui(self) -> None:
         self.setWindowTitle("Excel RTD")
         self.resize(600, 600)
 
-        module_path = Path(__file__).parent.parent
-        client_path = module_path.joinpath("vnpy_rtd.py")
-        self.client_line = QtWidgets.QLineEdit(str(client_path))
+        module_path: Path = Path(__file__).parent.parent
+        client_path: Path = module_path.joinpath("vnpy_rtd.py")
+        self.client_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(client_path))
         self.client_line.setReadOnly(True)
 
-        copy_button = QtWidgets.QPushButton("复制")
+        copy_button: QtWidgets.QPushButton = QtWidgets.QPushButton("复制")
         copy_button.clicked.connect(self.copy_client_path)
 
-        self.log_monitor = QtWidgets.QTextEdit()
+        self.log_monitor: QtWidgets.QTextEdit = QtWidgets.QTextEdit()
         self.log_monitor.setReadOnly(True)
 
-        self.port_label = QtWidgets.QLabel(
+        self.port_label: QtWidgets.QLabel = QtWidgets.QLabel(
             "使用Socket端口：请求回应9001、广播推送9002"
         )
 
-        hbox = QtWidgets.QHBoxLayout()
+        hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
         hbox.addWidget(self.client_line)
         hbox.addWidget(copy_button)
 
-        vbox = QtWidgets.QVBoxLayout()
+        vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
         vbox.addLayout(hbox)
         vbox.addWidget(self.log_monitor)
         vbox.addWidget(self.port_label)
@@ -68,7 +68,7 @@ def process_log_event(self, event: Event) -> None:
         """
         log: LogData = event.data
 
-        msg = f"{log.time}: {log.msg}"
+        msg: str = f"{log.time}: {log.msg}"
         self.log_monitor.append(msg)
 
     def copy_client_path(self) -> None:

vnpy_excelrtd/vnpy_rtd.py

@@ -1,5 +1,3 @@
-""""""
-
 from typing import Dict, Set, Any
 from collections import defaultdict
 
@@ -21,13 +19,13 @@ class ObjectRtd(RTD):
     RTD proxy for object in Python.
     """
 
-    def __init__(self, engine: "RtdClient", name: str, field: str):
+    def __init__(self, engine: "RtdClient", name: str, field: str) -> None:
         """Constructor"""
         super().__init__(value=0)
 
-        self.engine = engine
-        self.name = name
-        self.field = field
+        self.engine: RtdClient = engine
+        self.name: str = name
+        self.field: str = field
 
     def connect(self) -> None:
         """
@@ -56,19 +54,19 @@ class RtdClient(RpcClient):
     The engine for managing RTD objects and data update.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         """"""
         super().__init__()
 
         self.rtds: Dict[str, Set[ObjectRtd]] = defaultdict(set)
 
         global rtd_client
-        rtd_client = self
+        rtd_client: RtdClient = self
 
     def callback(self, topic: str, data: Any) -> None:
         """"""
         tick: TickData = data
-        buf = self.rtds[tick.vt_symbol]
+        buf: Set[ObjectRtd] = self.rtds[tick.vt_symbol]
 
         for rtd in buf:
             rtd.update(tick)
@@ -77,7 +75,7 @@ def add_rtd(self, rtd: ObjectRtd) -> None:
         """
         Add a new RTD into the engine..
         """
-        buf = self.rtds[rtd.name]
+        buf: Set[ObjectRtd] = self.rtds[rtd.name]
         buf.add(rtd)
         self.write_log(f"新增RTD连接：{rtd.name} {rtd.field}")
 
@@ -88,7 +86,7 @@ def remove_rtd(self, rtd: ObjectRtd) -> None:
         """
         Remove an existing RTD from the engine.
         """
-        buf = self.rtds[self.name]
+        buf: Set[ObjectRtd] = self.rtds[self.name]
         if self in buf:
             buf.remove(rtd)
             self.write_log(f"移除RTD连接：{rtd.name} {rtd.field}")
@@ -97,7 +95,7 @@ def remove_rtd(self, rtd: ObjectRtd) -> None:
 def init_client() -> None:
     """Initialize vnpy rtd client"""
     global rtd_client
-    rtd_client = RtdClient()
+    rtd_client: RtdClient = RtdClient()
     rtd_client.subscribe_topic("")
     rtd_client.start(REQ_ADDRESS, SUB_ADDRESS)
 
@@ -110,5 +108,5 @@ def rtd_tick_data(vt_symbol: str, field: str) -> ObjectRtd:
     if not rtd_client:
         init_client()
 
-    rtd = ObjectRtd(rtd_client, vt_symbol, field)
+    rtd: ObjectRtd = ObjectRtd(rtd_client, vt_symbol, field)
     return rtd