[Mod] type declaration

Author: noranhe

vnpy_excelrtd/engine.py

@@ -1,6 +1,4 @@
-""""""
-
-from typing import Set
+from typing import Set, Optional
 
 from vnpy.event import Event, EventEngine
 from vnpy.rpc import RpcServer
@@ -9,12 +7,12 @@
 from vnpy.trader.event import EVENT_TICK
 
 
-APP_NAME: str = "ExcelRtd"
+APP_NAME = "ExcelRtd"
 
-EVENT_RTD_LOG: str = "eRtdLog"
+EVENT_RTD_LOG = "eRtdLog"
 
-REP_ADDRESS: str = "tcp://*:9001"
-PUB_ADDRESS: str = "tcp://*:9002"
+REP_ADDRESS = "tcp://*:9001"
+PUB_ADDRESS = "tcp://*:9002"
 
 
 class RtdEngine(BaseEngine):
@@ -60,7 +58,7 @@ def subscribe(self, vt_symbol: str) -> None:
         """
         Subscribe tick data update.
         """
-        contract: ContractData = self.main_engine.get_contract(vt_symbol)
+        contract: Optional[ContractData] = self.main_engine.get_contract(vt_symbol)
         if not contract:
             return
 

vnpy_excelrtd/ui/widget.py

@@ -9,7 +9,7 @@
 
 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) -> None:
         """"""

vnpy_excelrtd/vnpy_rtd.py

@@ -1,5 +1,3 @@
-""""""
-
 from typing import Dict, Set, Any
 from collections import defaultdict
 
@@ -9,8 +7,8 @@
 from vnpy.trader.object import TickData
 
 
-REQ_ADDRESS: str = "tcp://localhost:9001"
-SUB_ADDRESS: str = "tcp://localhost:9002"
+REQ_ADDRESS = "tcp://localhost:9001"
+SUB_ADDRESS = "tcp://localhost:9002"
 
 
 rtd_client: "RtdClient" = None
@@ -68,7 +66,7 @@ def __init__(self) -> None:
     def callback(self, topic: str, data: Any) -> None:
         """"""
         tick: TickData = data
-        buf: dict = 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: dict = 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: dict = 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}")