typo

Author: Edanflame

vnpy_algotrading/engine.py

@@ -1,49 +1,63 @@
+from typing import Any, Optional
 
 from vnpy.event import EventEngine, Event
 from vnpy.trader.engine import BaseEngine, MainEngine
 from vnpy.trader.event import (
-    EVENT_TICK, EVENT_TIMER, EVENT_ORDER, EVENT_TRADE)
-from vnpy.trader.constant import (Direction, Offset, OrderType)
-from vnpy.trader.object import (SubscribeRequest, OrderRequest, LogData)
+    EVENT_TICK,
+    EVENT_TIMER,
+    EVENT_ORDER,
+    EVENT_TRADE
+)
+from vnpy.trader.constant import Direction, Offset, OrderType
+from vnpy.trader.object import (
+    SubscribeRequest,
+    OrderRequest,
+    LogData,
+    ContractData,
+    OrderData,
+    TickData,
+    CancelRequest
+)
 from vnpy.trader.utility import load_json, save_json, round_to
-from vnpy.trader.setting import SETTINGS
 
 from .template import AlgoTemplate
 from .base import (
-    EVENT_ALGO_LOG, EVENT_ALGO_PARAMETERS,
-    EVENT_ALGO_SETTING, EVENT_ALGO_VARIABLES,
+    EVENT_ALGO_LOG,
+    EVENT_ALGO_PARAMETERS,
+    EVENT_ALGO_SETTING,
+    EVENT_ALGO_VARIABLES,
     APP_NAME
 )
 
 
 class AlgoEngine(BaseEngine):
     """"""
-    setting_filename = "algo_trading_setting.json"
+    setting_filename: str = "algo_trading_setting.json"
 
     def __init__(self, main_engine: MainEngine, event_engine: EventEngine):
         """构造函数"""
         super().__init__(main_engine, event_engine, APP_NAME)
 
-        self.algos = {}
-        self.symbol_algo_map = {}
-        self.orderid_algo_map = {}
+        self.algos: dict = {}
+        self.symbol_algo_map: dict = {}
+        self.orderid_algo_map: dict = {}
 
-        self.algo_templates = {}
-        self.algo_settings = {}
+        self.algo_templates: dict = {}
+        self.algo_settings: dict = {}
 
         self.load_algo_template()
         self.register_event()
 
-    def init_engine(self):
+    def init_engine(self) -> None:
         """"""
         self.write_log("算法交易引擎启动")
         self.load_algo_setting()
 
-    def close(self):
+    def close(self) -> None:
         """"""
         pass
 
-    def load_algo_template(self):
+    def load_algo_template(self) -> None:
         """"""
         from .algos.twap_algo import TwapAlgo
         from .algos.iceberg_algo import IcebergAlgo
@@ -63,94 +77,94 @@ def load_algo_template(self):
         self.add_algo_template(DmaAlgo)
         self.add_algo_template(ArbitrageAlgo)
 
-    def add_algo_template(self, template: AlgoTemplate):
+    def add_algo_template(self, template: AlgoTemplate) -> None:
         """"""
         self.algo_templates[template.__name__] = template
 
-    def load_algo_setting(self):
+    def load_algo_setting(self) -> None:
         """"""
-        self.algo_settings = load_json(self.setting_filename)
+        self.algo_settings: dict = load_json(self.setting_filename)
 
         for setting_name, setting in self.algo_settings.items():
             self.put_setting_event(setting_name, setting)
 
         self.write_log("算法配置载入成功")
 
-    def save_algo_setting(self):
+    def save_algo_setting(self) -> None:
         """"""
         save_json(self.setting_filename, self.algo_settings)
 
-    def register_event(self):
+    def register_event(self) -> None:
         """"""
         self.event_engine.register(EVENT_TICK, self.process_tick_event)
         self.event_engine.register(EVENT_TIMER, self.process_timer_event)
         self.event_engine.register(EVENT_ORDER, self.process_order_event)
         self.event_engine.register(EVENT_TRADE, self.process_trade_event)
 
-    def process_tick_event(self, event: Event):
+    def process_tick_event(self, event: Event) -> None:
         """"""
-        tick = event.data
+        tick: Any = event.data
 
-        algos = self.symbol_algo_map.get(tick.vt_symbol, None)
+        algos: Any = self.symbol_algo_map.get(tick.vt_symbol, None)
         if algos:
             for algo in algos:
                 algo.update_tick(tick)
 
-    def process_timer_event(self, event: Event):
+    def process_timer_event(self, event: Event) -> None:
         """"""
         # 生成列表避免字典改变
-        algos = list(self.algos.values())
+        algos: list = list(self.algos.values())
 
         for algo in algos:
             algo.update_timer()
 
-    def process_trade_event(self, event: Event):
+    def process_trade_event(self, event: Event) -> None:
         """"""
-        trade = event.data
+        trade: Any = event.data
 
-        algo = self.orderid_algo_map.get(trade.vt_orderid, None)
+        algo: Optional[AlgoTemplate] = self.orderid_algo_map.get(trade.vt_orderid, None)
         if algo:
             algo.update_trade(trade)
 
-    def process_order_event(self, event: Event):
+    def process_order_event(self, event: Event) -> None:
         """"""
-        order = event.data
+        order: Any = event.data
 
-        algo = self.orderid_algo_map.get(order.vt_orderid, None)
+        algo: Optional[AlgoTemplate] = self.orderid_algo_map.get(order.vt_orderid, None)
         if algo:
             algo.update_order(order)
 
-    def start_algo(self, setting: dict):
+    def start_algo(self, setting: dict) -> str:
         """"""
         template_name: str = setting["template_name"]
-        algo_template = self.algo_templates[template_name]
+        algo_template: AlgoTemplate = self.algo_templates[template_name]
 
-        algo = algo_template.new(self, setting)
+        algo: AlgoTemplate = algo_template.new(self, setting)
         algo.start()
 
         self.algos[algo.algo_name] = algo
         return algo.algo_name
 
-    def stop_algo(self, algo_name: str):
+    def stop_algo(self, algo_name: str) -> None:
         """"""
-        algo = self.algos.get(algo_name, None)
+        algo: Optional[AlgoTemplate] = self.algos.get(algo_name, None)
         if algo:
             algo.stop()
             self.algos.pop(algo_name)
 
-    def stop_all(self):
+    def stop_all(self) -> None:
         """"""
         for algo_name in list(self.algos.keys()):
             self.stop_algo(algo_name)
 
-    def subscribe(self, algo: AlgoTemplate, vt_symbol: str):
+    def subscribe(self, algo: AlgoTemplate, vt_symbol: str) -> None:
         """"""
-        contract = self.main_engine.get_contract(vt_symbol)
+        contract: Optional[ContractData] = self.main_engine.get_contract(vt_symbol)
         if not contract:
             self.write_log(f'订阅行情失败，找不到合约：{vt_symbol}', algo)
             return
 
-        algos = self.symbol_algo_map.setdefault(vt_symbol, set())
+        algos: Any = self.symbol_algo_map.setdefault(vt_symbol, set())
 
         if not algos:
             req = SubscribeRequest(
@@ -170,18 +184,18 @@ def send_order(
         volume: float,
         order_type: OrderType,
         offset: Offset
-    ):
+    ) -> str:
         """"""
-        contract = self.main_engine.get_contract(vt_symbol)
+        contract: Optional[ContractData] = self.main_engine.get_contract(vt_symbol)
         if not contract:
             self.write_log(f'委托下单失败，找不到合约：{vt_symbol}', algo)
             return
 
-        volume = round_to(volume, contract.min_volume)
+        volume: float = round_to(volume, contract.min_volume)
         if not volume:
             return ""
 
-        req = OrderRequest(
+        req: OrderRequest = OrderRequest(
             symbol=contract.symbol,
             exchange=contract.exchange,
             direction=direction,
@@ -191,73 +205,73 @@ def send_order(
             offset=offset,
             reference=f"{APP_NAME}_{algo.algo_name}"
         )
-        vt_orderid = self.main_engine.send_order(req, contract.gateway_name)
+        vt_orderid: str = self.main_engine.send_order(req, contract.gateway_name)
 
         self.orderid_algo_map[vt_orderid] = algo
         return vt_orderid
 
-    def cancel_order(self, algo: AlgoTemplate, vt_orderid: str):
+    def cancel_order(self, algo: AlgoTemplate, vt_orderid: str) -> None:
         """"""
-        order = self.main_engine.get_order(vt_orderid)
+        order: Optional[OrderData] = self.main_engine.get_order(vt_orderid)
 
         if not order:
             self.write_log(f"委托撤单失败，找不到委托：{vt_orderid}", algo)
             return
 
-        req = order.create_cancel_request()
+        req: CancelRequest = order.create_cancel_request()
         self.main_engine.cancel_order(req, order.gateway_name)
 
-    def get_tick(self, algo: AlgoTemplate, vt_symbol: str):
+    def get_tick(self, algo: AlgoTemplate, vt_symbol: str) -> Optional[TickData]:
         """"""
-        tick = self.main_engine.get_tick(vt_symbol)
+        tick: Optional[TickData] = self.main_engine.get_tick(vt_symbol)
 
         if not tick:
             self.write_log(f"查询行情失败，找不到行情：{vt_symbol}", algo)
 
         return tick
 
-    def get_contract(self, algo: AlgoTemplate, vt_symbol: str):
+    def get_contract(self, algo: AlgoTemplate, vt_symbol: str) -> Optional[ContractData]:
         """"""
-        contract = self.main_engine.get_contract(vt_symbol)
+        contract: Optional[ContractData] = self.main_engine.get_contract(vt_symbol)
 
         if not contract:
             self.write_log(f"查询合约失败，找不到合约：{vt_symbol}", algo)
 
         return contract
 
-    def write_log(self, msg: str, algo: AlgoTemplate = None):
+    def write_log(self, msg: str, algo: AlgoTemplate = None) -> None:
         """"""
         if algo:
             msg = f"{algo.algo_name}：{msg}"
 
-        log = LogData(msg=msg, gateway_name=APP_NAME)
-        event = Event(EVENT_ALGO_LOG, data=log)
+        log: LogData = LogData(msg=msg, gateway_name=APP_NAME)
+        event: Event = Event(EVENT_ALGO_LOG, data=log)
         self.event_engine.put(event)
 
-    def put_setting_event(self, setting_name: str, setting: dict):
+    def put_setting_event(self, setting_name: str, setting: dict) -> None:
         """"""
-        event = Event(EVENT_ALGO_SETTING)
+        event: Event = Event(EVENT_ALGO_SETTING)
         event.data = {
             "setting_name": setting_name,
             "setting": setting
         }
         self.event_engine.put(event)
 
-    def update_algo_setting(self, setting_name: str, setting: dict):
+    def update_algo_setting(self, setting_name: str, setting: dict) -> None:
         """"""
         self.algo_settings[setting_name] = setting
 
         self.save_algo_setting()
 
         self.put_setting_event(setting_name, setting)
 
-    def remove_algo_setting(self, setting_name: str):
+    def remove_algo_setting(self, setting_name: str) -> None:
         """"""
         if setting_name not in self.algo_settings:
             return
         self.algo_settings.pop(setting_name)
 
-        event = Event(EVENT_ALGO_SETTING)
+        event: Event = Event(EVENT_ALGO_SETTING)
         event.data = {
             "setting_name": setting_name,
             "setting": None
@@ -266,18 +280,18 @@ def remove_algo_setting(self, setting_name: str):
 
         self.save_algo_setting()
 
-    def put_parameters_event(self, algo: AlgoTemplate, parameters: dict):
+    def put_parameters_event(self, algo: AlgoTemplate, parameters: dict) -> None:
         """"""
-        event = Event(EVENT_ALGO_PARAMETERS)
+        event: Event = Event(EVENT_ALGO_PARAMETERS)
         event.data = {
             "algo_name": algo.algo_name,
             "parameters": parameters
         }
         self.event_engine.put(event)
 
-    def put_variables_event(self, algo: AlgoTemplate, variables: dict):
+    def put_variables_event(self, algo: AlgoTemplate, variables: dict) -> None:
         """"""
-        event = Event(EVENT_ALGO_VARIABLES)
+        event: Event = Event(EVENT_ALGO_VARIABLES)
         event.data = {
             "algo_name": algo.algo_name,
             "variables": variables