Improved typing

flumine/markets/blotter.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Iterable, Optional, List
 from collections import defaultdict
+from typing import TYPE_CHECKING
 
 from ..order.ordertype import OrderTypes
 from ..utils import (
@@ -10,6 +11,12 @@
 )
 from ..order.order import BaseOrder, OrderStatus
 
+if TYPE_CHECKING:
+    from betfairlightweight.resources.bettingresources import MarketBook
+    from ..clients.baseclient import BaseClient
+    from ..markets.market import Market
+    from ..strategy.strategy import BaseStrategy
+
 logger = logging.getLogger(__name__)
 
 # https://www.betfair.com/aboutUs/Betfair.Charges/#charges6
@@ -58,10 +65,10 @@ def get_order_bet_id(self, bet_id: str) -> Optional[BaseOrder]:
 
     def strategy_orders(
         self,
-        strategy,
+        strategy: "BaseStrategy",
         order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
-    ) -> list:
+    ) -> list[BaseOrder]:
         """Returns all orders related to a strategy."""
         orders = self._strategy_orders[strategy]
         if order_status:
@@ -72,12 +79,12 @@ def strategy_orders(
 
     def strategy_selection_orders(
         self,
-        strategy,
+        strategy: "BaseStrategy",
         selection_id: int,
         handicap: float = 0,
         order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
-    ) -> list:
+    ) -> list[BaseOrder]:
         """Returns all orders related to a strategy selection."""
         orders = self._strategy_selection_orders[(strategy, selection_id, handicap)]
         if order_status:
@@ -88,10 +95,10 @@ def strategy_selection_orders(
 
     def client_orders(
         self,
-        client,
+        client: "BaseClient",
         order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
-    ) -> list:
+    ) -> list[BaseOrder]:
         orders = self._client_orders[client]
         if order_status:
             orders = [o for o in orders if o.status in order_status]
@@ -101,11 +108,11 @@ def client_orders(
 
     def client_strategy_orders(
         self,
-        client,
-        strategy,
+        client: "BaseClient",
+        strategy: "BaseStrategy",
         order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
-    ) -> list:
+    ) -> list[BaseOrder]:
         orders = self._client_strategy_orders[(client, strategy)]
         if order_status:
             orders = [o for o in orders if o.status in order_status]
@@ -114,14 +121,16 @@ def client_strategy_orders(
         return orders
 
     @property
-    def live_orders(self) -> Iterable:
+    def live_orders(self) -> Iterable[BaseOrder]:
         return iter(list(self._live_orders))
 
     @property
     def has_live_orders(self) -> bool:
         return bool(self._live_orders)
 
-    def process_closed_market(self, market, market_book) -> None:
+    def process_closed_market(
+        self, market: "Market", market_book: "MarketBook"
+    ) -> None:
         number_of_winners = len(
             [runner for runner in market_book.runners if runner.status == "WINNER"]
         )
@@ -152,7 +161,7 @@ def process_closed_market(self, market, market_book) -> None:
                                 "line_range_result unavailable, for simulation results update the market.context['line_range_result']"
                             )
 
-    def process_cleared_orders(self, cleared_orders) -> list:
+    def process_cleared_orders(self, cleared_orders) -> list[BaseOrder]:
         for cleared_order in cleared_orders.orders:
             order_id = cleared_order.customer_order_ref[STRATEGY_NAME_HASH_LENGTH + 1 :]
             if order_id in self:
@@ -162,7 +171,9 @@ def process_cleared_orders(self, cleared_orders) -> list:
 
     """ position """
 
-    def market_exposure(self, strategy, market_book) -> float:
+    def market_exposure(
+        self, strategy: "BaseStrategy", market_book: "MarketBook"
+    ) -> float:
         """Returns worst-case exposure for market, which is the maximum potential loss (negative),
         arising from the worst race outcome, or the minimum potential profit (positive).
         """
@@ -181,7 +192,7 @@ def market_exposure(self, strategy, market_book) -> float:
         worst_differences = sorted(differences)[: market_book.number_of_winners]
         return sum(worst_possible_profits_on_loses) + sum(worst_differences)
 
-    def selection_exposure(self, strategy, lookup: tuple) -> float:
+    def selection_exposure(self, strategy: "BaseStrategy", lookup: tuple) -> float:
         """Returns strategy/selection exposure, which is the worse-case loss arising
         from the selection either winning or losing. Can be positive or zero.
             positive = potential loss
@@ -194,7 +205,9 @@ def selection_exposure(self, strategy, lookup: tuple) -> float:
         )
         return max(exposure, 0.0)
 
-    def get_exposures(self, strategy, lookup: tuple, exclusion=None) -> dict:
+    def get_exposures(
+        self, strategy: "BaseStrategy", lookup: tuple, exclusion=None
+    ) -> dict:
         """Returns strategy/selection exposures as a dict."""
         mb, ml = [], []  # matched bets, (price, size)
         ub, ul = [], []  # unmatched bets, (price, size)
@@ -258,7 +271,7 @@ def get_exposures(self, strategy, lookup: tuple, exclusion=None) -> dict:
 
     """ getters / setters """
 
-    def complete_order(self, order) -> None:
+    def complete_order(self, order: BaseOrder) -> None:
         self._live_orders.remove(order)
 
     def has_order(self, customer_order_ref: str) -> bool:
@@ -269,7 +282,7 @@ def has_trade(self, trade_id: str) -> bool:
 
     __contains__ = has_order
 
-    def __setitem__(self, customer_order_ref: str, order) -> None:
+    def __setitem__(self, customer_order_ref: str, order: BaseOrder) -> None:
         self.active = True
         self._orders[customer_order_ref] = order
         self._bet_id_lookup[order.bet_id] = order

flumine/markets/markets.py

@@ -52,11 +52,11 @@ def get_order_from_bet_id(
         return blotter.get_order_bet_id(bet_id)
 
     @property
-    def markets(self) -> dict:
+    def markets(self) -> dict[str, Market]:
         return self._markets
 
     @property
-    def open_market_ids(self) -> list:
+    def open_market_ids(self) -> list[str]:
         return [m.market_id for m in self if m.status == "OPEN"]
 
     @property

flumine/order/order.py

@@ -5,7 +5,7 @@
 import string
 import collections
 from enum import Enum
-from typing import Union, Optional
+from typing import Any, Optional, TYPE_CHECKING, Union
 
 import betdaq.filters
 from betfairlightweight.resources.bettingresources import CurrentOrder
@@ -17,6 +17,9 @@
 from ..simulation.simulatedorder import SimulatedOrder
 from .. import config
 
+if TYPE_CHECKING:
+    from ..order.trade import Trade
+
 logger = logging.getLogger(__name__)
 
 
@@ -59,7 +62,7 @@ class BaseOrder:
 
     def __init__(
         self,
-        trade,
+        trade: "Trade",
         side: str,
         order_type: Union[LimitOrder, LimitOnCloseOrder, MarketOnCloseOrder],
         handicap: float = 0,
@@ -284,7 +287,7 @@ def notes_str(self) -> str:
         return ",".join(str(x) for x in self.notes.values())
 
     @property
-    def info(self) -> dict:
+    def info(self) -> dict[str, Any]:
         return {
             "market_id": self.market_id,
             "selection_id": self.selection_id,
@@ -421,19 +424,19 @@ def create_place_instruction(self) -> dict:
                 "handicap": self.handicap,
             }
 
-    def create_cancel_instruction(self) -> dict:
+    def create_cancel_instruction(self) -> dict[str, Any]:
         return {
             "betId": self.bet_id,
             "sizeReduction": self.update_data.get("size_reduction"),
         }
 
-    def create_update_instruction(self) -> dict:
+    def create_update_instruction(self) -> dict[str, Any]:
         return {
             "betId": self.bet_id,
             "newPersistenceType": self.order_type.persistence_type,
         }
 
-    def create_replace_instruction(self) -> dict:
+    def create_replace_instruction(self) -> dict[str, Any]:
         return {"betId": self.bet_id, "newPrice": self.update_data["new_price"]}
 
     # currentOrder