diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py index bed6f158..9f389728 100644 --- a/flumine/markets/blotter.py +++ b/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 diff --git a/flumine/markets/markets.py b/flumine/markets/markets.py index 652d8ea9..51270186 100644 --- a/flumine/markets/markets.py +++ b/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 diff --git a/flumine/order/order.py b/flumine/order/order.py index f53e7855..e11d303e 100644 --- a/flumine/order/order.py +++ b/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