diff --git a/src/aiomql/contrib/backtester/types.py b/src/aiomql/contrib/backtester/types.py index c89e8f4..f500646 100644 --- a/src/aiomql/contrib/backtester/types.py +++ b/src/aiomql/contrib/backtester/types.py @@ -9,40 +9,26 @@ TradeData = TypeVar('TradeData', bound=TradePosition | TradeOrder | TradeDeal) class TradingData(Generic[TradeData]): _data: dict[int, TradeData] - _open_items: set[int] def __init__(self, open_items: set[int] = None, data: dict = None): self._data = data or {} - self._open_items = open_items or {trade.ticket for trade in self._data.values()} + + def __len__(self): + return len(self._data) def __getitem__(self, item): return self._data[item] def __setitem__(self, key, value: TradeData): - self._open_items.add(value.ticket) self._data[key] = value - def __delitem__(self, key): - try: - self._open_items.discard(key) - - except KeyError: - logger.warning(f'{key} not found') - def __contains__(self, item: int): return item in self._open_items def __iter__(self): return iter(self._data) - def __len__(self): - return len(self._data) - def get(self, key, default=None) -> TradeData | None: - return self._data.get(key, default) if key in self._open_items else default - - def pop(self, key, default=None) -> TradeData | None: - self._open_items.discard(key) return self._data.get(key, default) def update(self, *, ticket: int, **kwargs): @@ -57,14 +43,54 @@ class TradingData(Generic[TradeData]): except KeyError: logger.error(f"Update Operation Failed: Could Not Find Ticket") - @property - def open_items(self) -> set[int]: - return self._open_items + def get_by_symbol(self, *, symbol: str) -> tuple[TradeData, ...]: + return tuple(position for position in self._data.values() if position.symbol == symbol) + + def get_by_ticket(self, *, ticket: int) -> tuple[TradeData, ...]: + return tuple(position for position in self._data.values() if position.ticket == ticket) class PositionsManager(TradingData): _data: dict[int, TradePosition] + def __init__(self, open_items: set[int] = None, data: dict = None): + super().__init__(data=data) + self._open_items = open_items or {trade.ticket for trade in self._data.values()} + + def __len__(self): + return len(self._open_items) + + def __setitem__(self, key, value: TradeData): + self._open_items.add(value.ticket) + self._data[key] = value + + def __delitem__(self, key): + try: + self._open_items.discard(key) + except KeyError: + logger.warning(f'{key} not found') + + def get(self, key, default=None) -> TradeData | None: + return self._data.get(key, default) if key in self._open_items else default + + def pop(self, key, default=None) -> TradeData | None: + self._open_items.discard(key) + return self._data.get(key, default) + + def positions_get(self, *, ticket: int = None, symbol: str = None, group: None) -> tuple[TradePosition, ...]: + if ticket and (symbol == group == None): + return self.get_by_ticket(ticket=ticket) + if symbol and (ticket == group == None): + return self.get_by_symbol(symbol=symbol) + if group and (ticket == symbol == None): + return tuple(position for position in self._data.values()) + if ticket == group == symbol == None: + return tuple(position for position in self._data.values()) + return tuple() + + def positions_total(self) -> int: + return len(self) + @property def open_positions(self) -> tuple[TradePosition, ...]: return tuple(position for position in self._data.values() if position.ticket in self.open_items) @@ -73,10 +99,33 @@ class PositionsManager(TradingData): class OrdersManager(TradingData): _data = dict[int, TradeOrder] - @property - def active_orders(self) -> tuple[TradeOrder, ...]: - return tuple(order for order in self._data.values() if order.ticket in self.open_items) - + def get_orders_range(self, *, date_from: float, date_to: float) -> tuple[TradeData, ...]: + start = date_from.timestamp() if isinstance(date_from, datetime) else date_from + end = date_to.timestamp() if isinstance(date_to, datetime) else date_to + return tuple(v for v in self._data.values() if start <= v.time_setup <= end) + + def history_orders_get(self, *, date_from: float | datetime, date_to: float | datetime, + group: str = '', ticket: int = None, position: int = None) -> tuple[TradeOrder, ...]: + orders = self.get_orders_range(date_from=date_from, date_to=date_to) + if ticket and (position == None): + return tuple(order for order in orders if order.ticket == ticket) + if position and (ticket == None): + return tuple(order for order in orders if order.position == position) + return orders class DealsManager(TradingData): - ... + _data = dict[int, TradeDeal] + + def get_deals_range(self, *, date_from: float, date_to: float) -> tuple[TradeData, ...]: + start = date_from.timestamp() if isinstance(date_from, datetime) else date_from + end = date_to.timestamp() if isinstance(date_to, datetime) else date_to + return tuple(v for v in self._data.values() if start <= v.time <= end) + + def history_deals_get(self, date_from: float | datetime, date_to: float | datetime, + group: str = '', ticket: int = None, position: int = None) -> tuple[TradeDeal, ...]: + deals = self.get_deals_range(date_from, date_to) + if ticket and (position == None): + return tuple(deal for deal in deals if deal.ticket == ticket) + if position and (ticket == None): + return tuple(deal for deal in deals if deal.position == position) + return deals