diff --git a/pyproject.toml b/pyproject.toml index e987ab1..8830219 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "aiomql" -version = "4.0.14b" +version = "4.0.15b" readme = "README.md" requires-python = ">=3.11" classifiers = [ @@ -16,7 +16,7 @@ classifiers = [ keywords = ["MetaTrader5", "Asynchronous", "Algorithmic Trading", "Trading Bot", "Backtesting", "Technical Analysis", "Forex", "Stocks", "Cryptocurrency", "Futures", "Options", "Crypto", "Algo Trading"] -dependencies = ["MetaTrader5>=5.0.37", "pandas>=1.5.0", "pandas-ta>=0.3.14b0", "mplfinance>=0.10.1"] +dependencies = ["MetaTrader5>=5.0.5200", "pandas>=1.5.0", "pandas-ta>=0.3.14b0", "mplfinance>=0.10.1"] authors = [{name = "Ichinga Samuel", email = "ichingasamuel@gmail.com"}] diff --git a/requirements.txt b/requirements.txt index 59d63b5..b430bc3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ --e git+ssh://git@github.com/Chimezirim-Bassey/aiomql.git@8e0a5516774894505d8d915bbd510bf88c6e3e48#egg=aiomql +-e git+ssh://git@github.com/Chimezirim-Bassey/aiomql.git@bae2983bded5b3912f7234fd0acaa89ab84fa1c4#egg=aiomql anyio==4.3.0 argon2-cffi==23.1.0 argon2-cffi-bindings==21.2.0 @@ -71,7 +71,7 @@ MarkupSafe==2.1.3 matplotlib==3.8.4 matplotlib-inline==0.1.6 mdurl==0.1.2 -MetaTrader5==5.0.4424 +MetaTrader5==5.0.5200 mistune==3.0.2 more-itertools==10.1.0 mplfinance==0.12.10b0 diff --git a/src/aiomql/contrib/quants/change.py b/src/aiomql/contrib/quants/change.py index 2c9eaf5..6dbe3d6 100644 --- a/src/aiomql/contrib/quants/change.py +++ b/src/aiomql/contrib/quants/change.py @@ -1,10 +1,3 @@ -def relative_difference(first: float, second: float) -> float: - """Find the relative difference between two values""" - diff = abs(first - second) - div = (first + second) / 2 - return (diff / div) * 100 - - def percentage_difference(first: float, second: float) -> float: """Find the percentage difference between two values""" diff = abs(first - second) @@ -12,23 +5,11 @@ def percentage_difference(first: float, second: float) -> float: return (diff / div) * 100 -def relative_position(start: float, end: float, value: float): - """Find the relative position of a value between two values""" - return (value - start) / (end - start) - - def percentage_position(start: float, end: float, value: float): """Find the percentage position of a value between two values""" return ((value - start) / (end - start)) * 100 -def get_relative_position(start: float, end: float, factor: float): - """Find position within an interval based on a factor""" - span = end - start - position = start + (factor * span) - return position - - def get_percentage_position(start: float, end: float, rate: float): """Find position within an interval based on a percentage""" span = end - start @@ -36,12 +17,6 @@ def get_percentage_position(start: float, end: float, rate: float): return position -def extend_interval_by_factor(start: float, end: float, factor: float): - """Extend the end of an interval by a factor""" - span = end - start - span *= factor - return end + span - def extend_interval_by_percentage(start: float, end: float, rate: float): """Extend the end of an interval by a factor""" span = end - start @@ -49,28 +24,16 @@ def extend_interval_by_percentage(start: float, end: float, rate: float): return end + span -def relative_change(start_value: float, end_value: float) -> float: - """Find the relative change from one value to another""" - return (end_value - start_value) / start_value - - def percentage_change(start_value: float, end_value: float) -> float: """Find the percentage change from one value to another""" return ((end_value - start_value) / start_value) * 100 -def relative_increase(value: float, factor: float) -> float: - """Increase the value by a factor""" - return value * (1 + factor) def percentage_increase(value: float, rate: float) -> float: """Increase the value by a factor""" return value * ((100 + rate) / 100) -def relative_decrease(value: float, factor: float) -> float: - """Increase the value by a factor""" - return value * (1 - factor) - def percentage_decrease(value: float, rate: float) -> float: """Increase the value by a factor""" return value * ((100 - rate) / 100) diff --git a/src/aiomql/contrib/strategies/finger_trap.py b/src/aiomql/contrib/strategies/finger_trap.py index daf37ad..c45f76b 100644 --- a/src/aiomql/contrib/strategies/finger_trap.py +++ b/src/aiomql/contrib/strategies/finger_trap.py @@ -1,6 +1,6 @@ import logging -from ...lib.symbol import Symbol +from ..symbols import ForexSymbol from ...lib.trader import Trader from ...lib.candle import Candles from ...lib.strategy import Strategy @@ -29,7 +29,7 @@ class FingerTrap(Strategy): parameters = {"fast_ema": 8, "slow_ema": 20, "etf": TimeFrame.M5, "ttf": TimeFrame.H1, "entry_ema": 5, "tcc": 720, "ecc": 8640} - def __init__(self, *,symbol: Symbol, params: dict | None = None, trader: Trader = None, sessions: Sessions = None, + def __init__(self, *, symbol: ForexSymbol, params: dict | None = None, trader: Trader = None, sessions: Sessions = None, name: str = "FingerTrap"): super().__init__(symbol=symbol, params=params, sessions=sessions, name=name) self.trader = trader or SimpleTrader(symbol=self.symbol) diff --git a/src/aiomql/contrib/trackers/open_position.py b/src/aiomql/contrib/trackers/open_position.py index 16cb487..3224b99 100644 --- a/src/aiomql/contrib/trackers/open_position.py +++ b/src/aiomql/contrib/trackers/open_position.py @@ -1,11 +1,12 @@ from dataclasses import dataclass, field -from typing import ClassVar +from typing import ClassVar, Self from logging import getLogger from ...lib import Symbol, Positions, Order from ...core.models import TradePosition, TradeAction, OrderSendResult from ...core.constants import OrderType from ...core.config import Config +from ..quants import percentage_increase, percentage_decrease from .position_tracker import PositionTracker logger = getLogger() @@ -17,9 +18,13 @@ class OpenPosition: ticket: int position: TradePosition is_open: bool = True - trackers: list[PositionTracker] = field(default_factory=list) use_checkpoint: bool = False - checkpoint: float = 0 + checkpoint: float = None + is_hedged: bool = False + is_a_hedge: bool = False + hedge: Self | None = None + pending_hedge: OrderSendResult | None = None + _trackers: dict[str, PositionTracker] = field(default_factory=dict) positions: ClassVar[Positions] state_key: ClassVar[str] = "tracked_positions" config: ClassVar[Config] @@ -34,6 +39,16 @@ class OpenPosition: def __post_init__(self): self.config.state.setdefault(self.state_key, {}).setdefault(self.ticket, self) + def add_tracker(self, *, tracker: PositionTracker, number: int = None, name: str = ""): + number = number or len(self._trackers) + 1 + tracker.set_position(self, number) + self._trackers[name or tracker.function.__name__] = tracker + + @property + def trackers(self): + for tracker in sorted(self._trackers, key=lambda key: self._trackers[key].number): + yield self._trackers[tracker] + async def remove_closed(self): try: await self.update_position() @@ -42,13 +57,6 @@ class OpenPosition: except Exception as exe: logger.error("%s: Unable to remove closed position from state", exe) - - def add_tracker(self, *, tracker: PositionTracker, number: int = None): - number = number or len(self.trackers) - tracker.set_position(self, number) - self.trackers.append(tracker) - self.trackers.sort(key=lambda x: x.number) - async def update_position(self) -> bool: pos = await self.positions.get_position_by_ticket(ticket=self.ticket) if pos is not None: @@ -121,6 +129,16 @@ class OpenPosition: self.symbol.name, self.ticket) return False, None + @staticmethod + async def check_pending_order(self): + await self.update_position() + if self.is_open and self.is_hedged and self.pending_hedge is not None: + pos = await self.positions.get_position_by_ticket(ticket=self.pending_hedge.order) + if pos is not None: + self.pending_hedge = None + self.hedge = OpenPosition(symbol=self.symbol, position=pos, ticket=pos.ticket, + is_a_hedge=True, hedge=self) + async def track(self): try: for tracker in self.trackers: @@ -129,35 +147,52 @@ class OpenPosition: logger.error("%s: Error occurred in track method of Open Position for %d:%s", exe, self.symbol.name, self.ticket) + async def get_price_from_profit(self, profit): + action = OrderType.BUY if self.position.type == 0 else OrderType.SELL + volume = self.position.volume + price_open = self.position.price_open + price_close = percentage_increase(price_open, 50) if action == 0 else percentage_decrease(price_open, 50) + half_profit = await Order.mt5.order_calc_profit(symbol=self.symbol.name, action=action, volume=volume, + price_open=price_open, price_close=price_close) + rate = profit / half_profit * 50 + rate = percentage_increase(price_open, rate) if action == 0 else percentage_decrease(price_open, rate) + return rate - async def hedge_position(self, *, hedge_params: dict = None) -> tuple[bool, OrderSendResult | None]: + async def hedge_order(self, price, **order_params) -> tuple[bool, OrderSendResult | None]: + try: + order_type = OrderType.SELL_STOP if self.position.type == OrderType.BUY else OrderType.BUY_STOP + volume = order_params.get("volume", self.position.volume) + order = Order(action=TradeAction.PENDING, symbol=self.symbol.name, type=order_type, price=price, volume=volume) + res = await order.send() + if res.retcode != 10009: + return False, res + self.pending_hedge = res + self.is_hedged = True + self.add_tracker(tracker=PositionTracker(self.check_pending_order), number=0, name="check_pending_order") + return True, res + except Exception as exe: + logger.error("%s: Error occurred in hedge_order method of Open Position for %d:%s", exe, self.symbol.name, self.ticket) + return False, None + + async def hedge_position(self, *, hedge_params: dict = None) -> tuple[bool, Self]: try: # hedge_params for customizing the hedge hedge_params = hedge_params or {} volume = hedge_params.get("volume", self.position.volume) order_type = hedge_params.get("type") or OrderType(self.position.type).opposite - tick = await self.symbol.info_tick() - price = hedge_params.get("price") or (tick.ask if order_type == OrderType.BUY else tick.bid) + if (price := hedge_params.get("price", None)) is None: + tick = await self.symbol.info_tick() + price = (tick.ask if order_type == OrderType.BUY else tick.bid) comment = hedge_params.get("comment", f"Hedge_{self.ticket}") hedge_order = Order(type=order_type, symbol=self.symbol.name, volume=volume, price=price, comment=comment) - - # if position has stops get them from the sl and tp of the position - if self.position.sl and self.position.tp: - osl = abs(self.position.sl - self.position.price_open) - otp = abs(self.position.tp - self.position.price_open) - if order_type == OrderType.BUY: - sl = hedge_params.get("sl") or tick.ask - osl - tp = hedge_params.get("tp") or tick.ask + otp - hedge_order.set_attributes(tp=tp, sl=sl) - if order_type == OrderType.SELL: - sl = hedge_params.get("sl") or tick.bid + osl - tp = hedge_params.get("sl") or tick.bid - otp - hedge_order.set_attributes(tp=tp, sl=sl) - res = await hedge_order.send() - if res.retcode == 10009: - return True, res - return False, res + if res.retcode != 10009: + return False, None + self.is_hedged = True + hedge_pos = await self.positions.get_position_by_ticket(ticket=res.order) + self.hedge = OpenPosition(symbol=self.symbol, ticket=res.order, position=hedge_pos, + hedge=self, is_a_hedge=True) + return True, self.hedge except Exception as exe: logger.error("%s: Error occurred in %s.hedge_position for %s:%d", exe, self.__class__.__name__, self.symbol.name, self.ticket) diff --git a/src/aiomql/contrib/trackers/position_tracking_functions.py b/src/aiomql/contrib/trackers/position_tracking_functions.py index d9c42af..16032ca 100644 --- a/src/aiomql/contrib/trackers/position_tracking_functions.py +++ b/src/aiomql/contrib/trackers/position_tracking_functions.py @@ -1,6 +1,5 @@ from logging import getLogger -from aiomql import OrderType from .open_position import OpenPosition from ..quants import extend_interval_by_percentage, get_percentage_position, percentage_position @@ -8,7 +7,7 @@ from ..quants import extend_interval_by_percentage, get_percentage_position, per logger = getLogger(__name__) -async def exit_at_price(*, pos: OpenPosition, tp: float = None, sl: float = None): +async def exit_at_price(pos: OpenPosition, /, tp: float = None, sl: float = None): """Exit a trade at a particular price""" if not await pos.update_position(): return @@ -19,7 +18,7 @@ async def exit_at_price(*, pos: OpenPosition, tp: float = None, sl: float = None logger.warning("Unable to close %s:%d due to %s in exit_at_price", pos.symbol, pos.ticket, res.comment) -async def extend_take_profit(*, pos: OpenPosition, increase: float = 20, start: float = 80, +async def extend_take_profit(pos: OpenPosition, /, increase: float = 20, start: float = 80, use_stop_levels: bool = True): is_open = await pos.update_position() if is_open is False or pos.position.profit < 0: @@ -35,13 +34,13 @@ async def extend_take_profit(*, pos: OpenPosition, increase: float = 20, start: position.symbol, position.ticket, res.comment) -async def extend_stop_loss(*, pos: OpenPosition, increase: float = 20, start: float = 80, +async def extend_stop_loss(pos: OpenPosition, /, increase: float = 20, start: float = 80, use_stop_levels: bool = True): is_open = await pos.update_position() if is_open is False or pos.position.profit > 0: return position = pos.position - if percentage_position(position.price_open, position.tp, position.price_current) >= start: + if percentage_position(position.price_open, position.sl, position.price_current) >= start: new_sl = extend_interval_by_percentage(position.price_open, position.sl, increase) ok, res = await pos.modify_stops(sl=new_sl, use_stop_levels=use_stop_levels) if ok: @@ -52,17 +51,17 @@ async def extend_stop_loss(*, pos: OpenPosition, increase: float = 20, start: fl position.symbol, position.ticket, res.comment) -async def exit_at_checkpoint(*, pos: OpenPosition, start: float = 80, trail: float = 15): +async def exit_at_checkpoint(pos: OpenPosition, /, start: float = 80, trail: float = 15): is_open = await pos.update_position() - if is_open is False or pos.position.profit < 0: + if is_open is False: return position = pos.position - if percentage_position(position.price_open, position.tp, position.price_current) >= start: + if position.profit > 0 and percentage_position(position.price_open, position.tp, position.price_current) >= start: new_checkpoint = get_percentage_position(position.price_open, position.price_current, 100-trail) change_checkpoint = False - if position.type == OrderType.BUY and new_checkpoint > pos.checkpoint: + if position.type.long and new_checkpoint > (pos.checkpoint or position.price_open): change_checkpoint = True - elif position.type == OrderType.SELL and new_checkpoint < pos.checkpoint: + elif position.type.short and new_checkpoint < (pos.checkpoint or position.price_open): change_checkpoint = True if change_checkpoint: pos.checkpoint = new_checkpoint @@ -70,10 +69,10 @@ async def exit_at_checkpoint(*, pos: OpenPosition, start: float = 80, trail: flo logger.info("New checkpoint created for %s:%d at %f:%f", position.symbol, position.ticket, new_checkpoint, position.profit) close = False - if position.type == OrderType.BUY and position.price_current <= pos.checkpoint and pos.use_checkpoint: + if position.type.long and position.price_current <= pos.checkpoint and pos.use_checkpoint: close = True - elif position.type == OrderType.SELL and position.price_current >= pos.checkpoint and pos.use_checkpoint: + elif position.type.short and position.price_current >= pos.checkpoint and pos.use_checkpoint: close = True if close: diff --git a/src/aiomql/contrib/trackers/positions_tracker.py b/src/aiomql/contrib/trackers/positions_tracker.py index 9c88d04..6ce1ba3 100644 --- a/src/aiomql/contrib/trackers/positions_tracker.py +++ b/src/aiomql/contrib/trackers/positions_tracker.py @@ -2,7 +2,7 @@ import asyncio from logging import getLogger from typing import ClassVar -from ...core import Config, State, Store, sleep +from ...core import Config, State, sleep from ...lib import Positions @@ -10,19 +10,20 @@ logger = getLogger(__name__) class OpenPositionsTracker: - state: ClassVar[State] - store: ClassVar[Store] config: ClassVar[Config] + positions: ClassVar[Positions] + state: ClassVar[State] def __new__(cls, *args, **kwargs): if not hasattr(cls, "config"): cls.config = Config() - if not hasattr(cls, "state"): - cls.state = cls.config.state + if not hasattr(cls, "positions"): cls.positions = Positions() - if not hasattr(cls, "store"): - cls.store = cls.config.store + + if not hasattr(cls, "state"): + cls.state = State() + return super().__new__(cls) def __init__(self, interval: int = 10, state_key: str = "tracked_positions", autocommit=False): diff --git a/src/aiomql/core/base.py b/src/aiomql/core/base.py index 4daba57..561f74d 100644 --- a/src/aiomql/core/base.py +++ b/src/aiomql/core/base.py @@ -19,8 +19,8 @@ class Base: include (set [str]): A set of attributes to be included when retrieving attributes using the get_dict and dict method. """ - exclude: set[str] - include: set[str] + exclude: set[str] = {"mt5", "config", "exclude", "include", "annotations", "class_vars", "dict", "_instance"} + include: set[str] = {} def __init__(self, **kwargs): """ @@ -29,8 +29,6 @@ class Base: Args: **kwargs: Set instance attributes with keyword arguments. Only if they are annotated on the class body. """ - self.exclude = {"mt5", "config", "exclude", "include", "annotations", "class_vars", "dict", "_instance"} - self.include = set() self.set_attributes(**kwargs) def __repr__(self): @@ -138,7 +136,18 @@ class _Base(Base): """ mt5: MetaTrader | MetaBackTester config: Config + + def __new__(cls, *args, **kwargs): + if not hasattr(cls, 'config'): + cls.config = Config() + if not hasattr(cls, 'mt5'): + cls.mt5 = MetaTrader() if cls.config.mode != "backtest" else MetaBackTester() + return super().__new__(cls) + + def __getstate__(self): + state = self.__dict__.copy() + state.pop("mt5", None) + return state + def __init__(self, **kwargs): - self.__class__.config = Config() - self.__class__.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester() super().__init__(**kwargs) diff --git a/src/aiomql/core/config.py b/src/aiomql/core/config.py index 4e3a3c7..0067b1d 100644 --- a/src/aiomql/core/config.py +++ b/src/aiomql/core/config.py @@ -45,7 +45,9 @@ class Config: shutdown: bool force_shutdown: bool db_commit_interval: float - auto_commit: bool = False + auto_commit: bool + flush_state: bool + stop_trading: bool lock: Lock _defaults = { "timeout": 60000, @@ -67,7 +69,9 @@ class Config: "root": None, "plots_dir_name": "plots", "db_commit_interval": 30, - "auto_commit": False + "auto_commit": False, + "flush_state": False, + "stop_trading": False } def __new__(cls, *args, **kwargs): @@ -224,10 +228,10 @@ class Config: self._store = value def init_state(self): - self.state = State(db_name=self.db_name) + self.state = State(db_name=self.root / self.db_name, flush=self.flush_state) def init_store(self): - self.store = Store(db_name=self.db_name) + self.store = Store(db_name=self.root / self.db_name, flush=self.flush_state) @property def records_dir(self): diff --git a/src/aiomql/core/constants.py b/src/aiomql/core/constants.py index 47afb58..1927f84 100644 --- a/src/aiomql/core/constants.py +++ b/src/aiomql/core/constants.py @@ -25,15 +25,13 @@ class TradeAction(Repr, IntEnum): """TRADE_REQUEST_ACTION Enum. Attributes: - DEAL (int): Delete the pending order placed previously Place a trade order for an immediate execution with the - specified parameters (market order). - PENDING (int): Delete the pending order placed previously + DEAL (int): Place a trade order for an immediate execution with the specified parameters (market order) + PENDING (int): Place a trade order for the execution under specified conditions (pending order) SLTP (int): Modify Stop Loss and Take Profit values of an opened position MODIFY (int): Modify the parameters of the order placed previously REMOVE (int): Delete the pending order placed previously CLOSE_BY (int): Close a position by an opposite one """ - __enum_name__ = "TRADE_ACTION" DEAL = mt5.TRADE_ACTION_DEAL PENDING = mt5.TRADE_ACTION_PENDING @@ -93,12 +91,12 @@ class OrderType(Repr, IntEnum): Attributes: BUY (int): Market buy order SELL (int): Market sell order - BUY_LIMIT (int): Buy Limit pending order - SELL_LIMIT (int): Sell Limit pending order - BUY_STOP (int): Buy Stop pending order - SELL_STOP (int): Sell Stop pending order - BUY_STOP_LIMIT (int): Upon reaching the order price, Buy Limit pending order is placed at StopLimit price - SELL_STOP_LIMIT (int): Upon reaching the order price, Sell Limit pending order is placed at StopLimit price + BUY_LIMIT (int): Buy Limit pending order, buy when price drops to level below the current price + SELL_LIMIT (int): Sell Limit pending order, sell when price rises to a level above the current price + BUY_STOP (int): Buy Stop pending order, buy when price rises to a level above the current price + SELL_STOP (int): Sell Stop pending order, sell when price drops to a level below the current price + BUY_STOP_LIMIT (int): Buy stop limit order, buy when price rises to a level above the current price, with a limit to how far above the current price + SELL_STOP_LIMIT (int): Sell stop limit order, sell when price drops to a level below the current price, with a limit to how far below the current price CLOSE_BY (int): Order for closing a position by an opposite one Properties: @@ -126,6 +124,13 @@ class OrderType(Repr, IntEnum): _type = {0: 1, 1: 0, 2: 3, 3: 2, 4: 5, 5: 4, 6: 7, 7: 6, 8: 8}[self] return OrderType(_type) + @property + def long(self): + return self in [0, 2, 4, 6] + + @property + def short(self): + return self in [1, 3, 5, 7] class BookType(Repr, IntEnum): """BOOK_TYPE Enum. diff --git a/src/aiomql/core/meta_trader.py b/src/aiomql/core/meta_trader.py index 260a8bd..def14ae 100644 --- a/src/aiomql/core/meta_trader.py +++ b/src/aiomql/core/meta_trader.py @@ -28,8 +28,12 @@ logger = getLogger() class MetaTrader(MetaCore): + def __new__(cls, *args, **kwargs): + if not hasattr(cls, "config"): + cls.config = Config() + return super().__new__(cls) + def __init__(self): - self.config = Config() self.error: Error = Error(code=1) self._lock = asyncio.Lock() diff --git a/src/aiomql/core/models.py b/src/aiomql/core/models.py index 6e34d46..2f35875 100644 --- a/src/aiomql/core/models.py +++ b/src/aiomql/core/models.py @@ -542,6 +542,15 @@ class OrderSendResult(Base): profit: float = None loss: float = None + def __getstate__(self): + state = self.__dict__.copy() + state["request"] = state.pop('request')._asdict() + return state + + def __setstate__(self, state): + state['request'] = mt5.TradeRequest(state['request']) + self.__dict__.update(state) + class TradePosition(Base): """Trade Position diff --git a/src/aiomql/core/state.py b/src/aiomql/core/state.py index ab6f69a..b004b39 100644 --- a/src/aiomql/core/state.py +++ b/src/aiomql/core/state.py @@ -1,6 +1,7 @@ import os import pickle import sqlite3 +from pathlib import Path from threading import Lock from typing import Self from typing import MutableMapping, Iterable, Any, ClassVar @@ -27,7 +28,7 @@ class State(MutableMapping): cls._instance = super().__new__(cls) return cls._instance - def __init__(self, db_name: str = "", data: dict = None, flush: bool = False, autocommit: bool = False): + def __init__(self, db_name: str | Path = "", data: dict = None, flush: bool = False, autocommit: bool = False): with self._lock: self.autocommit = autocommit self.init(data=data, flush=flush, db_name=db_name) @@ -99,14 +100,14 @@ class State(MutableMapping): def items(self): return self.data.items() - def load(self, *, conn = None, db_name: str = "", data: dict = None): + def load(self, *, conn = None, data: dict = None): try: - db_name = db_name or self.db_name conn = conn or self.conn res = conn.execute("SELECT value FROM state where key = 'data'").fetchone() db_data = pickle.loads(res[0]) if res else {} db_data |= (data or {}) self.update(db_data) + conn.close() except Exception as err: logger.error("%s: Failed to load data from database", err) @@ -121,7 +122,7 @@ class State(MutableMapping): self.commit(conn=conn) else: self.load(conn=conn, data=data) - conn.close() + return if flush: self.data = data or {} diff --git a/src/aiomql/core/store.py b/src/aiomql/core/store.py index ec26d87..3d8aa1b 100644 --- a/src/aiomql/core/store.py +++ b/src/aiomql/core/store.py @@ -2,6 +2,7 @@ import os import sqlite3 from typing import MutableMapping, Iterable, Any from logging import getLogger +from pathlib import Path SENTINEL = object() @@ -14,7 +15,7 @@ class Store(MutableMapping): cursor: sqlite3.Cursor conn: sqlite3.Connection - def __init__(self, db_name: str = "", data: dict = None, flush: bool = False, autocommit: bool = True): + def __init__(self, db_name: str | Path = "", data: dict = None, flush: bool = False, autocommit: bool = True): self.autocommit = autocommit self.db_name = db_name or os.environ.get("DB_NAME", "db.sqlite3") self.conn = sqlite3.connect(self.db_name, check_same_thread=False) diff --git a/src/aiomql/core/sync/__init__.py b/src/aiomql/core/sync/__init__.py new file mode 100644 index 0000000..3b57075 --- /dev/null +++ b/src/aiomql/core/sync/__init__.py @@ -0,0 +1 @@ +from .meta_trader import MetaTrader diff --git a/src/aiomql/core/sync/meta_trader.py b/src/aiomql/core/sync/meta_trader.py new file mode 100644 index 0000000..b0abde5 --- /dev/null +++ b/src/aiomql/core/sync/meta_trader.py @@ -0,0 +1,344 @@ +from datetime import datetime +from logging import getLogger +from typing import Literal, Self +from pathlib import Path + +import numpy as np +from MetaTrader5 import ( + BookInfo, + SymbolInfo, + AccountInfo, + Tick, + TerminalInfo, + TradeOrder, + TradeDeal, + TradePosition, + OrderSendResult, + OrderCheckResult, +) + +from ..constants import OrderType, CopyTicks + +from .._core import MetaCore +from ..errors import Error +from ..config import Config + +logger = getLogger() + + +class MetaTrader(MetaCore): + def __init__(self): + self.config = Config() + self.error: Error = Error(code=1) + + def __enter__(self) -> Self: + """ + Async context manager entry point. + Initializes the connection to the MetaTrader terminal. + + Returns: + MetaTrader: An instance of the MetaTrader class. + """ + self.initialize() + self.login() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """ + Async context manager exit point. Closes the connection to the MetaTrader terminal. + """ + self.shutdown() + + def _handler(self, api: dict, retries=3): + func = api["func"] + args = api.get("args", ()) + kwargs = api.get("kwargs", {}) + error_msg = api.get("error_msg", "An error occurred") + res = func(*args, **kwargs) + + if res is not None: + return res + + if res is None and self.error.is_connection_error() and retries > 0: + self.initialize() + self.login() + return self._handler(api, retries=retries - 1) + else: + err = self.last_error() + self.error = Error(*err) + logger.warning(f"{error_msg}:{self.error.description}") + return res + + def login(self, *, login: int = 0, password: str = "", server: str = "", timeout: int = 60000) -> bool: + """ + Connects to the MetaTrader terminal using the specified login, password and server. + + Args: + login (int): The trading account number. + password (str): The trading account password. + server (str): The trading server name. + timeout (int): The timeout for the connection in seconds. + + Returns: + bool: True if successful, False otherwise. + """ + acc_details = self.config.account_info + login = login or acc_details.get("login", 0) + password = password or acc_details.get("password", "") + server = server or acc_details.get("server", "") + return self._login(login, password=password, server=server, timeout=timeout) + + def initialize(self, path: str = None, login: int = 0, password: str = "", server: str = "", + timeout: int | None = None, portable=False) -> bool: + """ + Initializes the connection to the MetaTrader terminal. All parameters are optional. + + Keyword Args: + path (str): The path to the MetaTrader terminal executable. + login (int): The trading account number. + password (str): The trading account password. + server (str): The trading server name. + timeout (int): The timeout for the connection in milliseconds. + portable (bool): If True, the terminal will be launched in portable mode. + + Returns: + bool: True if successful, False otherwise. + """ + path = self.config.path if path is None else path + path = "" if Path(path).exists() is False else path + args = (str(path),) if path else () + acc = self.config.account_info + kwargs = {key: value for key, value in (("login", login or acc.get("login")), + ("password", password or acc.get("password")), + ("server", server or acc.get("server")), ("timeout", timeout or 45000), + ("portable", portable)) if key is not None} + res = self._initialize(*args, **kwargs) + if not res: + err = self._last_error() + self.error = Error(*err) + return res + + def shutdown(self) -> None: + """Closes the connection to the MetaTrader terminal.""" + self._shutdown() + + def last_error(self) -> tuple[int, str]: + try: + return self.last_error() + except Exception as err: + logger.warning("%s: Error in obtaining last error.", err) + return -1, str(err) + + def version(self) -> tuple[int, int, str] | None: + """""" + api = {"func": self._version, "error_msg": "Error in obtaining version."} + return self._handler(api) + + def account_info(self) -> AccountInfo | None: + """""" + api = {"func": self._account_info, "error_msg": "Error in obtaining account information"} + return self._handler(api) + + def terminal_info(self) -> TerminalInfo | None: + api = {"func": self._terminal_info, "error_msg": "Error in obtaining terminal information"} + return self._handler(api) + + def symbols_total(self) -> int: + api = {"func": self._symbols_total, "error_msg": "Error in obtaining total symbols."} + return self._handler(api) + + def symbols_get(self, group: str = "") -> tuple[SymbolInfo] | None: + kwargs = {"group": group} if group else {} + api = {"func": self._symbols_get, "kwargs": kwargs, "error_msg": "Error in obtaining symbols."} + return self._handler(api) + + def symbol_info(self, symbol: str) -> SymbolInfo | None: + api = { + "func": self._symbol_info, + "args": (symbol,), + "error_msg": f"Error in obtaining information for {symbol}", + } + return self._handler(api) + + def symbol_info_tick(self, symbol: str) -> Tick | None: + api = {"func": self._symbol_info_tick, "args": (symbol,), "error_msg": f"Error in obtaining tick for {symbol}"} + return self._handler(api) + + def symbol_select(self, symbol: str, enable: bool) -> bool: + api = {"func": self._symbol_select, "args": (symbol, enable), "error_msg": f"Error in selecting {symbol}"} + return self._handler(api) + + def market_book_add(self, symbol: str) -> bool: + api = { + "func": self._market_book_add, + "args": (symbol,), + "error_msg": f"Error in adding {symbol} to market book", + } + return self._handler(api) + + def market_book_get(self, symbol: str) -> tuple[BookInfo] | None: + api = { + "func": self._market_book_get, + "args": (symbol,), + "error_msg": f"Error in obtaining market depth for {symbol}", + } + return self._handler(api) + + def market_book_release(self, symbol: str) -> bool: + api = { + "func": self._market_book_release, + "args": (symbol,), + "error_msg": f"Error in releasing market depth for {symbol}", + } + return self._handler(api) + + def copy_rates_from( + self, symbol: str, timeframe: int, date_from: datetime | float, count: int + ) -> np.ndarray | None: + api = { + "func": self._copy_rates_from, + "args": (symbol, timeframe, date_from, count), + "error_msg": f"Error in obtaining rates for {symbol}", + } + return self._handler(api) + + def copy_rates_from_pos(self, symbol: str, timeframe: int, start_pos: int, count: int) -> np.ndarray | None: + api = { + "func": self._copy_rates_from_pos, + "args": (symbol, timeframe, start_pos, count), + "error_msg": f"Error in obtaining rates for {symbol}", + } + return self._handler(api) + + def copy_rates_range( + self, symbol: str, timeframe: int, date_from: datetime | float, date_to: datetime | float + ) -> np.ndarray | None: + api = { + "func": self._copy_rates_range, + "args": (symbol, timeframe, date_from, date_to), + "error_msg": f"Error in obtaining rates for {symbol}", + } + return self._handler(api) + + def copy_ticks_from( + self, symbol: str, date_from: datetime | float, count: int, flags: CopyTicks + ) -> np.ndarray | None: + api = { + "func": self._copy_ticks_from, + "args": (symbol, date_from, count, flags), + "error_msg": f"Error in obtaining ticks for {symbol}", + } + return self._handler(api) + + def copy_ticks_range( + self, symbol: str, date_from: datetime | float, date_to: datetime | float, flags: CopyTicks + ) -> np.ndarray | None: + api = { + "func": self._copy_ticks_range, + "args": (symbol, date_from, date_to, flags), + "error_msg": f"Error in obtaining ticks for {symbol}", + } + return self._handler(api) + + def orders_total(self) -> int: + api = {"func": self._orders_total, "error_msg": "Error in obtaining total orders."} + return self._handler(api) + + def orders_get(self, group: str = "", ticket: int = 0, symbol: str = "") -> tuple[TradeOrder] | None: + kwargs = {key: value for key, value in (("group", group), ("ticket", ticket), ("symbol", symbol)) if value} + api = {"func": self._orders_get, "kwargs": kwargs, "error_msg": "Error in obtaining orders."} + return self._handler(api) + + def order_calc_margin( + self, action: Literal[OrderType.BUY, OrderType.SELL], symbol: str, volume: float, price: float + ) -> float | None: + api = { + "func": self._order_calc_margin, + "args": (action, symbol, volume, price), + "error_msg": "Error in calculating margin.", + } + return self._handler(api) + + def order_calc_profit( + self, + action: Literal[OrderType.BUY, OrderType.SELL], + symbol: str, + volume: float, + price_open: float, + price_close: float, + ) -> float | None: + api = { + "func": self._order_calc_profit, + "args": (action, symbol, volume, price_open, price_close), + "error_msg": "Error in calculating profit.", + } + return self._handler(api) + + def order_check(self, request: dict) -> OrderCheckResult: + api = {"func": self._order_check, "args": (request,), "error_msg": "Error in checking order."} + return self._handler(api) + + def order_send(self, request: dict) -> OrderSendResult: + api = {"func": self._order_send, "args": (request,), "error_msg": "Error in sending order."} + return self._handler(api) + + def positions_total(self) -> int: + api = {"func": self._positions_total, "error_msg": "Error in obtaining total positions."} + return self._handler(api) + + def positions_get(self, group: str = "", ticket: int = None, symbol: str = "") -> tuple[TradePosition] | None: + kwargs = {key: value for key, value in (("group", group), ("ticket", ticket), ("symbol", symbol)) if value} + api = {"func": self._positions_get, "kwargs": kwargs, "error_msg": "Error in obtaining open positions."} + return self._handler(api) + + def history_orders_total(self, date_from: datetime | float, date_to: datetime | float) -> int: + api = { + "func": self._history_orders_total, + "args": (date_from, date_to), + "error_msg": "Error in obtaining total history orders.", + } + return self._handler(api) + + def history_orders_get( + self, + date_from: datetime | float = None, + date_to: datetime | float = None, + group: str = "", + ticket: int = None, + position: int = None, + ) -> tuple[TradeOrder] | None: + kwargs = {key: value for key, value in (("group", group), ("ticket", ticket), ("position", position)) if value} + args = tuple(arg for arg in (date_from, date_to) if arg) + api = { + "func": self._history_orders_get, + "args": args, + "kwargs": kwargs, + "error_msg": "Error in obtaining history orders", + } + return self._handler(api) + + def history_deals_total(self, date_from: datetime | float, date_to: datetime | float) -> int: + api = { + "func": self._history_deals_total, + "args": (date_from, date_to), + "error_msg": "Error in obtaining total history deals", + } + return self._handler(api) + + def history_deals_get( + self, + date_from: datetime | float = None, + date_to: datetime | float = None, + group: str = "", + ticket: int = None, + position: int = None, + ) -> tuple[TradeDeal] | None: + kwargs = {key: value for key, value in (("group", group), ("ticket", ticket), ("position", position)) if value} + args = tuple(arg for arg in (date_from, date_to) if arg) + api = { + "func": self._history_deals_get, + "args": args, + "kwargs": kwargs, + "error_msg": "Error in obtaining history deals", + } + return self._handler(api) diff --git a/src/aiomql/lib/account.py b/src/aiomql/lib/account.py index fc27fba..a4c06e2 100644 --- a/src/aiomql/lib/account.py +++ b/src/aiomql/lib/account.py @@ -1,6 +1,6 @@ from threading import Lock from logging import getLogger -from typing import Self +from typing import Self, ClassVar from ..core.base import _Base from ..core.models import AccountInfo @@ -17,7 +17,7 @@ class Account(_Base, AccountInfo): connected (bool): Status of connection to MetaTrader 5 Terminal """ _instance: Self - _lock: Lock + _lock: ClassVar[Lock] connected: bool def __new__(cls, *args, **kwargs): @@ -26,6 +26,7 @@ class Account(_Base, AccountInfo): cls._lock = lock cls._instance = super().__new__(cls) cls._instance.connected = False + cls._instance.exclude.add("_lock") return cls._instance async def __aenter__(self) -> Self: @@ -51,7 +52,13 @@ class Account(_Base, AccountInfo): async def refresh(self): """Refreshes the account instance with the latest account details from the MetaTrader 5 terminal""" - account_info = await self.mt5.account_info + account_info = await self.mt5.account_info() + acc = account_info._asdict() + self.connected = True + self.set_attributes(**acc) + + def refresh_sync(self): + account_info = self.mt5._account_info() acc = account_info._asdict() self.connected = True self.set_attributes(**acc) diff --git a/src/aiomql/lib/bot.py b/src/aiomql/lib/bot.py index bf67bb7..48b230f 100644 --- a/src/aiomql/lib/bot.py +++ b/src/aiomql/lib/bot.py @@ -7,6 +7,7 @@ import logging from .executor import Executor from ..core.config import Config from ..core.meta_trader import MetaTrader +from ..core.meta_backtester import MetaBackTester from .symbol import Symbol as Symbol from .strategy import Strategy as Strategy @@ -25,12 +26,16 @@ class Bot: executor: Executor mt: MetaTrader strategies: list[Strategy] + initialized: bool + login: bool def __init__(self): self.config = Config(bot=self) self.executor = Executor() - self.mt5 = MetaTrader() + self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester() self.strategies = [] + self.initialized = False + self.login = False @classmethod def process_pool(cls, processes: dict[Callable:dict] = None, num_workers: int = None): @@ -46,17 +51,36 @@ class Bot: for bot, kwargs in processes.items(): executor.submit(bot, **kwargs) + async def start_terminal(self): + """Start terminal and login asynchronously""" + res = await self.mt5.initialize() + if res: + self.initialized = True + res = await self.mt5.login() + if res: + self.login = True + return res + + def start_terminal_sync(self): + """Start terminal and login synchronously""" + res = self.mt5.initialize_sync() + if res: + self.initialized = True + res = self.mt5.login_sync() + if res: + self.login = True + return res + async def initialize(self): """Prepares the bot by signing in to the trading account and initializing the symbols for each strategy. Only strategies with successfully initialized symbols will be added to the executor. Starts the global task queue. Raises: - SystemExit if sign in was not successful + SystemExit if sign_in was not successful """ try: - await self.mt5.initialize() - login = await self.mt5.login() - if not login: + await self.start_terminal() + if not self.login: logger.critical("Unable to sign in to MetaTrder 5 Terminal") raise Exception("Unable to sign in to MetaTrader 5 Terminal") logger.info("Login Successful") @@ -78,12 +102,11 @@ class Bot: Starts the global task queue. Raises: - SystemExit if sign in was not successful + SystemExit if sign_in was not successful """ try: - self.mt5.initialize_sync() - login = self.mt5.login_sync() - if not login: + self.start_terminal_sync() + if not self.login: logger.critical("Unable to sign in to MetaTrder 5 Terminal") raise Exception("Unable to sign in to MetaTrader 5 Terminal") logger.info("Login Successful") @@ -163,7 +186,7 @@ class Bot: async def init_strategy(self, *, strategy: Strategy) -> bool: """Initialize a single strategy. This method is called internally by the bot.""" - res = await strategy.symbol.initialize() + res = await strategy.initialize() if res: self.executor.add_strategy(strategy=strategy) return res @@ -171,11 +194,11 @@ class Bot: async def init_strategies(self): """Initialize the symbols for the current trading session. This method is called internally by the bot.""" tasks = [self.init_strategy(strategy=strategy) for strategy in self.strategies] - await asyncio.gather(*tasks) + await asyncio.gather(*tasks, return_exceptions=True) def init_strategy_sync(self, *, strategy: Strategy) -> bool: """Initialize a single strategy. This method is called internally by the bot.""" - res = strategy.symbol.initialize_sync() + res = strategy.initialize_sync() if res: self.executor.add_strategy(strategy=strategy) return res diff --git a/src/aiomql/lib/history.py b/src/aiomql/lib/history.py index 8d7cc6b..682cc5e 100644 --- a/src/aiomql/lib/history.py +++ b/src/aiomql/lib/history.py @@ -1,9 +1,8 @@ import asyncio -from datetime import datetime +from typing import ClassVar +from datetime import datetime, UTC from logging import getLogger -import pytz - from ..core.config import Config from ..core.meta_trader import MetaTrader from ..core.models import TradeDeal, TradeOrder @@ -24,34 +23,41 @@ class History: mt5 (MetaTrader): MetaTrader instance config (Config): Config instance """ - - mt5: MetaTrader | MetaBackTester - config: Config + mt5: ClassVar[MetaTrader | MetaBackTester] + config: ClassVar[Config] deals: tuple[TradeDeal, ...] orders: tuple[TradeOrder, ...] total_deals: int total_orders: int group: str + def __new__(cls, *args, **kwargs): + instance = super().__new__(cls) + if not hasattr(instance.__class__, 'config'): + instance.__class__.config = Config() + if not hasattr(instance.__class__, 'mt5'): + instance.__class__.mt5 = MetaTrader() if instance.config.mode != "backtest" else MetaBackTester() + return instance + def __init__( - self, *, date_from: datetime | float, date_to: datetime | float, group: str = "", use_utc: bool = True + self, *, date_from: datetime | float, date_to: datetime | float, group: str = "", use_utc: bool = float ): """ Args: date_from (datetime, float): Date the orders are requested from. Set by the 'datetime' object or as a - number of seconds elapsed since 1970.01.01. Defaults to twenty-four hours from the current time in 'utc' + number of seconds elapsed since 1970.01.01. date_to (datetime, float): Date up to which the orders are requested. Set by the 'datetime' object or as a - number of seconds elapsed since 1970.01.01. Defaults to the current time in "utc" + number of seconds elapsed since 1970.01.01. + + use_utc (bool): Convert date_from and date_to to UTC. Default is False. group (str): Filter for selecting history by symbols. This defaults to an empty string """ - self.config = Config() - self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester() date_from = date_from if isinstance(date_from, datetime) else datetime.fromtimestamp(date_from) date_to = date_to if isinstance(date_to, datetime) else datetime.fromtimestamp(date_to) - self.date_from = date_from.astimezone(pytz.UTC) if use_utc else date_from - self.date_to = date_to.astimezone(pytz.UTC) if use_utc else date_to + self.date_from = date_from.astimezone(UTC) if use_utc else date_from + self.date_to = date_to.astimezone(UTC) if use_utc else date_to self.group = group self.deals: tuple[TradeDeal, ...] = () self.orders: tuple[TradeOrder, ...] = () diff --git a/src/aiomql/lib/order.py b/src/aiomql/lib/order.py index 6c7e262..fabc13d 100644 --- a/src/aiomql/lib/order.py +++ b/src/aiomql/lib/order.py @@ -11,7 +11,6 @@ logger = getLogger(__name__) class Order(_Base, TradeRequest): """Trade order related functions and properties. Subclass of TradeRequest.""" - def __init__(self, **kwargs): """Initialize the order object with keyword arguments, symbol must be provided. Provide default values for action, type_time and type_filling if not provided. diff --git a/src/aiomql/lib/positions.py b/src/aiomql/lib/positions.py index f25091e..1bcdf8e 100644 --- a/src/aiomql/lib/positions.py +++ b/src/aiomql/lib/positions.py @@ -20,22 +20,24 @@ class Positions: Attributes: mt5 (MetaTrader): MetaTrader instance. """ - mt5: MetaTrader | MetaBackTester positions: tuple[TradePosition, ...] - total_positions: int + config: Config + + def __new__(cls, *args, **kwargs): + if not hasattr(cls, "config"): + cls.config = Config() + if not hasattr(cls, "mt5"): + cls.mt5 = MetaTrader() + return super().__new__(cls) def __init__(self): - """Get Open Positions""" - self.config = Config() - self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester() self.positions = () - self.total_positions = 0 async def get_positions(self, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]: """Get open positions with the ability to filter by symbol, ticket or group of symbols. Args: - symbol (Optional[str]): Financial instrument name. If symbol is provided, ticket is ignored. + symbol (Optional[str]): Financial instrument name. If a symbol is provided, the ticket is ignored. ticket (Optional[int]): Position ticket. group (Optional[str]): Group of symbols. @@ -53,12 +55,28 @@ class Positions: positions = await self.mt5.positions_get(**kwargs) if positions is not None: self.positions = tuple(TradePosition(**pos._asdict()) for pos in positions) - self.total_positions = len(self.positions) return self.positions logger.warning("Failed to get open positions") return () - async def get_position_by_ticket(self, *, ticket: int) -> TradePosition | None: + @classmethod + async def get_all_positions(cls, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]: + kwargs = {} + if symbol is not None: + kwargs["symbol"] = symbol + ticket = None + if ticket is not None: + kwargs["ticket"] = ticket + if group is not None: + kwargs["group"] = group + positions = await cls.mt5.positions_get(**kwargs) + if positions is not None: + return cls.positions + logger.warning("Failed to get open positions") + return () + + @classmethod + async def get_position_by_ticket(cls, *, ticket: int) -> TradePosition | None: """Get an open position by ticket. Args: ticket (int): Position ticket. @@ -66,13 +84,14 @@ class Positions: Returns: TradePosition: Return an open position """ - positions = await self.mt5.positions_get(ticket=ticket) + positions = await cls.mt5.positions_get(ticket=ticket) position = positions[0] if positions else None if position is None or position.ticket != ticket: return None return TradePosition(**position._asdict()) - async def get_positions_by_symbol(self, *, symbol: str) -> tuple[TradePosition, ...]: + @classmethod + async def get_positions_by_symbol(cls, *, symbol: str) -> tuple[TradePosition, ...]: """Get open positions by symbol. Args: symbol (str): Financial instrument name. @@ -80,7 +99,7 @@ class Positions: Returns: tuple[TradePosition, ...]: A tuple of open trade positions """ - positions = await self.mt5.positions_get(symbol=symbol) + positions = await cls.mt5.positions_get(symbol=symbol) return tuple(TradePosition(**pos._asdict()) for pos in (positions or ())) @staticmethod @@ -104,9 +123,10 @@ class Positions: ) return await order.send() - async def close_position_by_ticket(self, *, ticket: int) -> OrderSendResult | None: + @classmethod + async def close_position_by_ticket(cls, *, ticket: int) -> OrderSendResult | None: """Close an open position using the ticket.""" - position = await self.get_position_by_ticket(ticket=ticket) + position = await cls.get_position_by_ticket(ticket=ticket) if position is None: return None order = Order( @@ -144,8 +164,15 @@ class Positions: ) return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)]) - async def get_total_positions(self) -> int: + @classmethod + async def close_all_positions(cls): + positions = await cls.mt5.positions_get() + results = await asyncio.gather( + *(cls.close_position(position=position) for position in positions), return_exceptions=True + ) + return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)]) + + @classmethod + async def get_total_positions(cls) -> int: """Get the total number of open positions.""" - total = await self.mt5.positions_total() - self.total_positions = total or self.total_positions - return self.total_positions + return await cls.mt5.positions_total() diff --git a/src/aiomql/lib/sessions.py b/src/aiomql/lib/sessions.py index a615b46..180fb81 100644 --- a/src/aiomql/lib/sessions.py +++ b/src/aiomql/lib/sessions.py @@ -2,6 +2,7 @@ import asyncio from datetime import time, timedelta, datetime, UTC from typing import Literal, Callable, Iterable, NamedTuple from logging import getLogger +from time import sleep from ..core.models import OrderSendResult, TradePosition from ..core.config import Config @@ -26,11 +27,11 @@ def delta(obj: time) -> timedelta: async def backtest_sleep(secs): - """An async sleep function for use during backtesting.""" + """A sleep function for use during backtesting.""" config = Config() btc = config.backtest_controller - sleep = config.backtest_engine.cursor.time + secs - while sleep > config.backtest_engine.cursor.time: + sleep_secs = config.backtest_engine.cursor.time + secs + while sleep_secs > config.backtest_engine.cursor.time: btc.wait() @@ -59,7 +60,6 @@ class Session: name: str = "", ): """Create a session. - Keyword Args: start (int | datetime.time): The start time of the session in UTC. end (int | datetime.time): The end time of the session in UTC. @@ -119,7 +119,7 @@ class Session: return Duration(hours=hours, minutes=minutes, seconds=seconds) async def close_positions(self, *, positions: tuple[TradePosition, ...]): - results = asyncio.gather( + results = await asyncio.gather( *(self.positions_manager.close_position(position=position) for position in positions), return_exceptions=True, ) diff --git a/src/aiomql/lib/strategy.py b/src/aiomql/lib/strategy.py index 5e44f00..72d3f48 100644 --- a/src/aiomql/lib/strategy.py +++ b/src/aiomql/lib/strategy.py @@ -1,21 +1,19 @@ """The base class for creating strategies.""" import asyncio -from time import time -from typing import TypeVar -from abc import ABC, abstractmethod +import time +from abc import ABC from datetime import time as dtime from logging import getLogger -from ..core.meta_trader import MetaTrader +from .sessions import Sessions, Session +from .symbol import Symbol from ..core import Config +from ..core.backtesting.backtest_controller import BackTestController from ..core.exceptions import StopTrading from ..core.meta_backtester import MetaBackTester -from ..core.backtesting.backtest_controller import BackTestController -from .sessions import Sessions, Session -from .symbol import Symbol as _Symbol +from ..core.meta_trader import MetaTrader -Symbol = TypeVar("Symbol", bound=_Symbol) logger = getLogger(__name__) @@ -93,7 +91,10 @@ class Strategy(ABC): async def initialize(self): """Perform any initialization tasks here.""" - ... + return await self.symbol.initialize() + + def initialize_sync(self): + return self.symbol.initialize_sync() @staticmethod async def live_sleep(*, secs: float): @@ -104,7 +105,7 @@ class Strategy(ABC): Args: secs (float): The time in seconds. Usually the timeframe you are trading on. """ - mod = time() % secs + mod = time.time() % secs secs = secs - mod if mod != 0 else mod await asyncio.sleep(secs + 0.1) @@ -117,26 +118,19 @@ class Strategy(ABC): secs (float): The time in seconds. Usually the timeframe you are trading on. """ if self.config.mode == "backtest": - await self.backtest_sleep(secs=secs) + self.backtest_sleep(secs=secs) else: await self.live_sleep(secs=secs) async def delay(self, *, secs: float): """Sleep for the input amount of seconds""" if self.config.mode == "backtest": - await self._backtest_sleep(secs=secs) + self._backtest_sleep(secs=secs) else: await asyncio.sleep(secs) - async def _backtest_sleep(self, *, secs: float): + def _backtest_sleep(self, *, secs: float): try: - # if self.backtest_controller.parties == 2: - # this is not needed, as the backtest_engine iterator will handle this - # steps = int(secs) // self.config.backtest_engine.speed - # steps = max(steps, 1) - # self.config.backtest_engine.fast_forward(steps=steps) - # self.backtest_controller.wait() - if self.backtest_controller.parties >= 2: _time = self.config.backtest_engine.cursor.time + secs while _time > self.config.backtest_engine.cursor.time: @@ -147,7 +141,7 @@ class Strategy(ABC): self.backtest_controller.wait() logger.error("Error: %s in backtest_sleep", err) - async def backtest_sleep(self, *, secs: float): + def backtest_sleep(self, *, secs: float): """Sleep for the needed amount of seconds in between requests to the terminal. Args: @@ -157,7 +151,7 @@ class Strategy(ABC): _time = self.config.backtest_engine.cursor.time mod = _time % secs secs = secs - mod if mod != 0 else mod - await self._backtest_sleep(secs=secs) + self._backtest_sleep(secs=secs) except Exception as err: logger.error("Error: %s in backtest_sleep", err) @@ -172,7 +166,6 @@ class Strategy(ABC): """Run the strategy.""" async with self as _: logger.info("Running %s strategy on %s", self.name, self.symbol.name) - await self.initialize() while self.running: try: await self.sessions.check() @@ -208,7 +201,6 @@ class Strategy(ABC): logger.error(f"Error: {err} in backtest_strategy") return - @abstractmethod async def trade(self): """Place trades using this method. This is the main method of the strategy. It will be called by the strategy runner. diff --git a/src/aiomql/lib/symbol.py b/src/aiomql/lib/symbol.py index 99e1179..ab68c0d 100644 --- a/src/aiomql/lib/symbol.py +++ b/src/aiomql/lib/symbol.py @@ -25,7 +25,7 @@ class Symbol(_Base, SymbolInfo): Notes: Full properties are on the SymbolInfo Object. - Make sure Symbol is always initialized with a name argument + Make sure a Symbol is always initialized with a name argument """ initialized: bool tick: Tick @@ -46,7 +46,7 @@ class Symbol(_Base, SymbolInfo): """Get the current price tick of a financial instrument. Args: - name: if name is supplied get price tick of that financial instrument. Optional unnamed parameter. + name: if name is supplied, get price tick of that financial instrument. Optional unnamed parameter. Returns: Tick: Return a Tick Object @@ -112,12 +112,14 @@ class Symbol(_Base, SymbolInfo): self.set_attributes(**info._asdict()) info_tick = await self.mt5.symbol_info_tick(self.name) + if info_tick: self.tick = Tick(**info_tick._asdict()) if info is not None and info_tick is not None: self.initialized = True return True + logger.warning("Unable to initialize %s", self.name) return False except Exception as err: diff --git a/src/aiomql/lib/sync/__init__.py b/src/aiomql/lib/sync/__init__.py new file mode 100644 index 0000000..1224715 --- /dev/null +++ b/src/aiomql/lib/sync/__init__.py @@ -0,0 +1 @@ +from .lib import * diff --git a/src/aiomql/lib/sync/order.py b/src/aiomql/lib/sync/order.py new file mode 100644 index 0000000..961e0f9 --- /dev/null +++ b/src/aiomql/lib/sync/order.py @@ -0,0 +1,148 @@ +from logging import getLogger + +from ...core.models import TradeRequest, TradeOrder, OrderCheckResult, OrderSendResult +from ...core.constants import TradeAction, OrderTime, OrderFilling +from ...core.exceptions import OrderError +from ...core.base import _Base +from ...core.sync.meta_trader import MetaTrader +from ...utils import error_handler_sync + +logger = getLogger(__name__) + + +class Order(_Base, TradeRequest): + """Trade order related functions and properties. Subclass of TradeRequest.""" + mt5: MetaTrader + + def __init__(self, **kwargs): + """Initialize the order object with keyword arguments, symbol must be provided. + Provide default values for action, type_time and type_filling if not provided. + + Args: + **kwargs: Keyword arguments must match the attributes of TradeRequest as well as the attributes of + Order class as specified in the annotations in the class definition. + + Default Values: + action (TradeAction.DEAL): Trade action + type_time (OrderTime.DAY): Order time + type_filling (OrderFilling.FOK): Order filling + """ + kwargs = {"action": TradeAction.DEAL, "type_time": OrderTime.DAY, "type_filling": OrderFilling.FOK, **kwargs} + super().__init__(**kwargs) + + def modify(self, **kwargs): + """Modify the order object with keyword arguments. + + Args: + **kwargs: Keyword arguments must match the attributes of TradeRequest as well as the attributes of + Order class as specified in the annotations in the class definition. + """ + self.set_attributes(**kwargs) + + def orders_total(self): + """Get the number of active pending orders. + + Returns: + (int): total number of active pending orders + """ + return self.mt5.orders_total() + + def get_pending_order(self, *, ticket: int) -> TradeOrder | None: + """ + Get a pending order by ticket number. + + Args: + ticket (int): Order ticket number + + Returns: + """ + orders = self.mt5.orders_get(ticket=ticket) + order = None + for order_ in orders: + if order_.ticket == ticket: + return TradeOrder(**order_._asdict()) + return order + + def get_pending_orders(self, *, ticket: int = 0, symbol: str = "", group: str = "") -> tuple[TradeOrder, ...]: + """Get the list of active pending orders for the current symbol. + + Args: + ticket (int): Order ticket number + symbol (str): Symbol name + group (str): Group name + + Returns: + tuple[TradeOrder, ...]: A Tuple of active pending trade orders as TradeOrder objects + """ + orders = self.mt5.orders_get(symbol=symbol, ticket=ticket, group=group) + if orders is not None: + return tuple(TradeOrder(**order._asdict()) for order in orders) + return tuple() + + def check(self, **kwargs) -> OrderCheckResult: + """Check funds sufficiency for performing a required trading operation and the possibility of executing it. + + Returns: + OrderCheckResult: An OrderCheckResult object + + Raises: + OrderError: If not successful + """ + req = self.request | kwargs + res = self.mt5.order_check(req) + if res is None: + raise OrderError(f"Order check failed for {self.symbol}") + return OrderCheckResult(**res._asdict()) + + def send(self) -> OrderSendResult: + """Send a request to perform a trading operation from the terminal to the trade server. + + Returns: + OrderSendResult: An OrderSendResult object + + Raises: + OrderError: If not successful + """ + res = self.mt5.order_send(self.request) + if res is None: + raise OrderError(f"Failed to send order {self.symbol}") + return OrderSendResult(**res._asdict()) + + @error_handler_sync(log_error_msg=False) + def calc_margin(self) -> float | None: + """Return the required margin in the account currency to perform a specified trading operation. + + Returns: + float: Returns float value if successful + """ + res = self.mt5.order_calc_margin(self.type, self.symbol, self.volume, self.price) + return res + + @error_handler_sync(response=0, log_error_msg=False) + def calc_profit(self) -> float: + """Return profit in the account currency for a specified trading operation. + + Returns: + float: Returns float value if successful + None: If not successful + """ + action, symbol, volume, price_open, price_close = (self.type, self.symbol, self.volume, self.price, self.tp) + res = self.mt5.order_calc_profit(action, symbol, volume, price_open, price_close) + return res + + @error_handler_sync(response=0, log_error_msg=False) + def calc_loss(self) -> float: + """Return profit in the account currency for a specified trading operation. + + Returns: + float: Returns float value if successful + None: If not successful + """ + action, symbol, volume, price_open, price_close = (self.type, self.symbol, self.volume, self.price, self.sl) + res = self.mt5.order_calc_profit(action, symbol, volume, price_open, price_close) + return res + + @property + def request(self) -> dict: + """Return the order request as a dictionary.""" + return {key: value for key, value in self.dict.items() if key in self.mt5.TradeRequest.__match_args__} diff --git a/src/aiomql/lib/sync/positions.py b/src/aiomql/lib/sync/positions.py new file mode 100644 index 0000000..d9e4ef7 --- /dev/null +++ b/src/aiomql/lib/sync/positions.py @@ -0,0 +1,146 @@ +"""Handle Open positions.""" +from logging import getLogger + +from ...core.meta_trader import MetaTrader +from ...core.models import TradePosition, OrderSendResult +from ...core.constants import OrderType, TradeAction +from ...core.config import Config +from ...core.sync.meta_trader import MetaTrader +from ..order import Order + + +logger = getLogger(__name__) + + +class Positions: + """Get Open Positions. + + Attributes: + mt5 (MetaTrader): MetaTrader instance. + """ + mt5: MetaTrader + config: Config + positions: tuple[TradePosition, ...] + + def __new__(cls, *args, **kwargs): + if not hasattr(cls, "config"): + cls.config = Config() + if not hasattr(cls, "mt5"): + cls.mt5 = MetaTrader() + return super().__new__(cls) + + def __init__(self): + self.positions = () + + def get_positions(self, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]: + """Get open positions with the ability to filter by symbol, ticket or group of symbols. + Args: + symbol (Optional[str]): Financial instrument name. If a symbol is provided, the ticket is ignored. + ticket (Optional[int]): Position ticket. + group (Optional[str]): Group of symbols. + + Returns: + tuple[TradePosition, ...]: A tuple of open trade positions + """ + kwargs = {} + if symbol is not None: + kwargs["symbol"] = symbol + ticket = None + if ticket is not None: + kwargs["ticket"] = ticket + if group is not None: + kwargs["group"] = group + positions = self.mt5.positions_get(**kwargs) + if positions is not None: + return tuple(TradePosition(**pos._asdict()) for pos in positions) + logger.warning("Failed to get open positions") + return () + + def get_position_by_ticket(self, *, ticket: int) -> TradePosition | None: + """Get an open position by ticket. + Args: + ticket (int): Position ticket. + + Returns: + TradePosition: Return an open position + """ + positions = self.mt5.positions_get(ticket=ticket) + position = positions[0] if positions else None + if position is None or position.ticket != ticket: + return None + return TradePosition(**position._asdict()) + + def get_positions_by_symbol(self, *, symbol: str) -> tuple[TradePosition, ...]: + """Get open positions by symbol. + Args: + symbol (str): Financial instrument name. + + Returns: + tuple[TradePosition, ...]: A tuple of open trade positions + """ + positions = self.mt5.positions_get(symbol=symbol) + return tuple(TradePosition(**pos._asdict()) for pos in (positions or ())) + + @staticmethod + def close(*, ticket: int, symbol: str, price: float, volume: float, order_type: OrderType) -> OrderSendResult: + """Close an open position for the trading account using the ticket and other parameters. + + Args: + ticket (int): Position ticket. + symbol (str): Financial instrument name. + price (float): Closing price. + volume (float): Volume to close. + order_type (OrderType): Order type. + """ + order = Order( + action=TradeAction.DEAL, + price=price, + position=ticket, + symbol=symbol, + volume=volume, + type=order_type.opposite, + ) + return order.send() + + def close_position_by_ticket(self, *, ticket: int) -> OrderSendResult | None: + """Close an open position using the ticket.""" + position = self.get_position_by_ticket(ticket=ticket) + if position is None: + return None + order = Order( + position=position.ticket, + symbol=position.symbol, + volume=position.volume, + type=position.type.opposite, + price=position.price_current, + action=TradeAction.DEAL, + ) + return order.send() + + @staticmethod + def close_position(*, position: TradePosition): + """Close an open position for the trading account. Using a position object.""" + order = Order( + position=position.ticket, + symbol=position.symbol, + volume=position.volume, + type=position.type.opposite, + price=position.price_current, + action=TradeAction.DEAL, + ) + return order.send() + + def close_all(self) -> int: + """Close all open positions for the trading account. Specify a symbol or group to filter positions. + + Returns: + int: Return number of positions closed. + """ + positions = self.get_positions() + results = [self.close_position(position=position) for position in positions] + + return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)]) + + def get_total_positions(self) -> int: + """Get the total number of open positions.""" + return self.mt5.positions_total() diff --git a/src/aiomql/lib/sync/sessions.py b/src/aiomql/lib/sync/sessions.py new file mode 100644 index 0000000..e1d64b6 --- /dev/null +++ b/src/aiomql/lib/sync/sessions.py @@ -0,0 +1,254 @@ +from datetime import time, timedelta, datetime, UTC +from typing import Literal, Callable, Iterable, NamedTuple +from logging import getLogger +from time import sleep + +from ...core.config import Config +from .positions import Positions + +logger = getLogger(__name__) + + +class Duration(NamedTuple): + hours: int + minutes: int + seconds: int + + +def delta(obj: time) -> timedelta: + """Get the timedelta of a datetime.time object. + + Args: + obj (datetime.time): A datetime.time object. + """ + return timedelta(hours=obj.hour, minutes=obj.minute, seconds=obj.second, microseconds=obj.microsecond) + + +def backtest_sleep(secs): + """A sleep function for use during backtesting.""" + config = Config() + btc = config.backtest_controller + sleep_secs = config.backtest_engine.cursor.time + secs + while sleep_secs > config.backtest_engine.cursor.time: + btc.wait() + +class Session: + """A session is a time period between two datetime.time objects specified in utc. + + Attributes: + start (datetime.time): The start time of the session. + end (datetime.time): The end time of the session. + on_start (str): The action to take when the session starts. Default is None. + on_end (str): The action to take when the session ends. Default is None. + custom_start (Callable): A custom function to call when the session starts. Default is None. + custom_end (Callable): A custom function to call when the session ends. Default is None. + name (str): A name for the session. Default is a combination of start and end. + """ + + def __init__( + self, + *, + start: int | time, + end: int | time, + on_start: Literal["close_all", "close_win", "close_loss", "custom_start"] = None, + on_end: Literal["close_all", "close_win", "close_loss", "custom_end"] = None, + custom_start: Callable = None, + custom_end: Callable = None, + name: str = "", + ): + """Create a session. + Keyword Args: + start (int | datetime.time): The start time of the session in UTC. + end (int | datetime.time): The end time of the session in UTC. + on_start (Literal['close_all', 'close_win', 'close_loss', 'custom_start']): The action to take when the + session starts. Default is None. + on_end (Literal['close_all', 'close_win', 'close_loss', 'custom_end']): The action to take when the session + ends. Default is None. + custom_start (Callable): A custom function to call when the session starts. Default is None. + custom_end (Callable): A custom function to call when the session ends. Default is None. + name (str): A name for the session. Default is a combination of start and end. + """ + self.start = start.replace(tzinfo=UTC) if isinstance(start, time) else time(hour=start, tzinfo=UTC) + self.end = end if isinstance(end, time) else time(hour=end, tzinfo=UTC) + self.on_start = on_start + self.on_end = on_end + self.custom_start = custom_start + self.custom_end = custom_end + self.name = name or f"{self.start}<-->{self.end}" + self.positions_manager = Positions() + self.config = Config() + + def __contains__(self, item: time): + span = (delta(self.end) - delta(self.start)).seconds + item_span = (delta(self.end) - delta(item)).seconds + return item_span <= span + + def __str__(self): + return f"{self.start}<-->{self.end}" + + def __repr__(self): + return f"{self.start}<-->{self.end}" + + def __len__(self): + return int((delta(self.end) - delta(self.start)).seconds) + + def in_session(self) -> bool: + """Check if the current time is within the session.""" + now = ( + datetime.now(tz=UTC).time() + if self.config.mode == "live" + else datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time() + ) + return now in self + + def begin(self): + """Call the action specified in on_start or custom_start.""" + self.action(action=self.on_start) + + def close(self): + """Call the action specified in on_end or custom_end.""" + self.action(action=self.on_end) + + def duration(self) -> Duration: + """Get the duration of the session in seconds.""" + hours, seconds = divmod(len(self), 3600) + minutes, seconds = divmod(seconds, 60) + return Duration(hours=hours, minutes=minutes, seconds=seconds) + + def action(self, *, action): + """Used by begin and close to call the action specified. + + Args: + action (Literal['close_all', 'close_win', 'close_loss', 'custom_start', 'custom_end']): The action to take. + """ + try: + match action: + case "close_all": + raise NotImplementedError("To be implemented") + + case "close_win": + raise NotImplementedError("To be implemented") + + case "close_loss": + raise NotImplementedError("To be implemented") + + case "custom_end": + raise NotImplementedError("To be implemented") + + case "custom_start": + raise NotImplementedError("To be implemented") + + case _: + pass + except Exception as exe: + logger.warning(f"Failed to call action {action} due to {exe}") + + def until(self): + """Get the seconds until the session starts from the current time in seconds.""" + if self.config.mode == "backtest": + now = datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time() + secs = (delta(self.start) - delta(now)).seconds + else: + secs = (delta(self.start) - delta(datetime.now(tz=UTC).time())).seconds + return secs + + +class Sessions: + """Sessions allow you to run code at specific times of the day. It is a collection of Session objects. + Sessions are sorted by start time. The sessions object is an asynchronous context manager. + + Attributes: + sessions (list[Session]): A list of Session objects. + current_session (Session): The current session. + """ + + sessions: list[Session] + current_session: Session | None + + def __init__(self, *, sessions: Iterable[Session]): + self.sessions = list(sessions) + self.sessions.sort(key=lambda x: (x.start.hour, x.end.hour)) + self.current_session = None + self.config = Config() + + def find(self, *, moment: time = None) -> Session | None: + """Find a session that contains a datetime.time object, if not found return None. + + Keyword Args: + moment (datetime.time | None): A datetime.time object. if not provided, the current time is used. + + Returns: + Session | None: A Session object or None if not found. + """ + moment = ( + moment or datetime.now(tz=UTC).time() + if self.config.mode == "live" + else (datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time()) + ) + for session in self.sessions: + if moment in session: + return session + return None + + def find_next(self, *, moment: time = None) -> Session: + """Find the next session that contains a datetime.time object. + + Args: + moment (datetime.time | None): A datetime.time object, if not provided, the current time is used. + + Returns: + Session: A Session object. + """ + moment = ( + moment or datetime.now(tz=UTC).time() + if self.config.mode == "live" + else (datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time()) + ) + for session in self.sessions: + if delta(moment) < delta(session.start): + return session + return self.sessions[0] + + def __contains__(self, moment: time): + return True if self.find(moment=moment) is not None else False + + def __enter__(self): + self.check() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.current_session.close() if self.current_session is not None else ... + + def check(self): + """Check if the current session has started and if not, wait until it starts.""" + if self.current_session is not None and self.current_session.in_session(): + return + + if self.config.mode == "backtest": + now = datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time() + else: + now = datetime.now(tz=UTC).time() + + next_session = self.find(moment=now) + + if next_session and self.current_session is None: + self.current_session = next_session + self.current_session.begin() + return + + if next_session and self.current_session is not None: + self.current_session.close() + self.current_session = next_session + self.current_session.begin() + return + + if next_session is None and self.current_session is not None: + self.current_session.close() + + next_session = self.find_next(moment=now) + secs = next_session.until() + 10 + logger.info(f"sleeping for {secs} seconds until next {next_session} session") + sleep_func = sleep if self.config.mode == "live" else backtest_sleep + sleep_func(secs) + self.current_session = next_session + self.current_session.begin() diff --git a/src/aiomql/lib/sync/strategy.py b/src/aiomql/lib/sync/strategy.py new file mode 100644 index 0000000..3df20d4 --- /dev/null +++ b/src/aiomql/lib/sync/strategy.py @@ -0,0 +1,200 @@ +"""The base class for creating strategies.""" +import time +from abc import ABC +from datetime import time as dtime +from logging import getLogger + +from .sessions import Sessions, Session +from .symbol import Symbol +from ...core import Config +from ...core.backtesting.backtest_controller import BackTestController +from ...core.exceptions import StopTrading +from ...core.meta_backtester import MetaBackTester +from ...core.meta_trader import MetaTrader + + +logger = getLogger(__name__) + + +class Strategy(ABC): + """The base class for creating strategies. + + Attributes: + name (str): The name of the strategy. + symbol (Symbol): The Financial Instrument as a Symbol Object + parameters (Dict): A dictionary of parameters for the strategy. + sessions (Sessions): The sessions to use for the strategy. + running (bool): A flag to indicate if the strategy is running. + backtest_controller (BackTestController): A controller for running the backtester. + current_session (Session): The current session. + mt5 (MetaTrader|MetaBackTester): The MetaTrader object. + config (Config): The config object. + + Notes: + Define the name of a strategy as a class attribute. If not provided, the class name will be used as the name. + """ + name: str + symbol: Symbol + sessions: Sessions + mt5: MetaTrader | MetaBackTester + config: Config + running: bool + parameters = {} + backtest_controller = BackTestController + current_session = Session + + def __init__(self, *, symbol: Symbol, params: dict = None, sessions: Sessions = None, name=""): + """Initiate the parameters dict and add name and symbol fields. + Use class name as strategy name if name is not provided + + Args: + symbol (Symbol): The Financial instrument + params (Dict): Trading strategy parameters + """ + self.parameters = {**self.parameters} | (params or {}) + self.symbol = symbol + self.name = name or self.__class__.__name__ + self.parameters["symbol"] = symbol.name + self.parameters["name"] = self.name + self.running = True + self.sessions = sessions or Sessions(sessions=[Session(start=0, end=dtime(hour=23, minute=59, second=59))]) + self.config = Config() + self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester() + self.backtest_controller = BackTestController() + + def __repr__(self): + return f"{self.name}({self.symbol!r})" + + def __getattr__(self, item): + if item in self.parameters: + return self.parameters[item] + raise AttributeError(f"{item} not an attribute of {self.name}") + + def __setattr__(self, key, value): + if key in self.parameters: + self.parameters[key] = value + super().__setattr__(key, value) + + def __enter__(self): + self.sessions.check() + self.running = True + self.current_session = self.sessions.current_session + + def __exit__(self, exc_type, exc_val, exc_tb): + try: + self.current_session.close() if self.current_session else ... + self.running = False + except Exception as err: + logger.error(f"Error: {err}") + + def initialize(self): + return self.symbol.initialize_sync() + + @staticmethod + def live_sleep(*, secs: float): + """Sleep for the needed amount of seconds in between requests to the terminal. + computes the accurate amount of time needed to sleep ensuring that the next request is made at the start of + a new bar and making cooperative multitasking possible. + + Args: + secs (float): The time in seconds. Usually the timeframe you are trading on. + """ + mod = time.time() % secs + secs = secs - mod if mod != 0 else mod + time.sleep(secs + 0.1) + + def sleep(self, *, secs: float): + """Sleep for the needed amount of seconds in between requests to the terminal. + computes the accurate amount of time needed to sleep ensuring that the next request is made at the start of + a new bar and making cooperative multitasking possible. + + Args: + secs (float): The time in seconds. Usually the timeframe you are trading on. + """ + if self.config.mode == "backtest": + self.backtest_sleep(secs=secs) + else: + self.live_sleep(secs=secs) + + def delay(self, *, secs: float): + """Sleep for the input amount of seconds""" + if self.config.mode == "backtest": + self._backtest_sleep(secs=secs) + else: + time.sleep(secs) + + def _backtest_sleep(self, *, secs: float): + try: + if self.backtest_controller.parties >= 2: + _time = self.config.backtest_engine.cursor.time + secs + while _time > self.config.backtest_engine.cursor.time: + self.backtest_controller.wait() + else: + self.backtest_controller.wait() + except Exception as err: + self.backtest_controller.wait() + logger.error("Error: %s in backtest_sleep", err) + + def backtest_sleep(self, *, secs: float): + """Sleep for the needed amount of seconds in between requests to the terminal. + + Args: + secs (float): The time in seconds. Usually the timeframe you are trading on. + """ + try: + _time = self.config.backtest_engine.cursor.time + mod = _time % secs + secs = secs - mod if mod != 0 else mod + self._backtest_sleep(secs=secs) + except Exception as err: + logger.error("Error: %s in backtest_sleep", err) + + def run_strategy(self): + """Run the strategy.""" + if self.config.mode == "live": + self.live_strategy() + elif self.config.mode == "backtest": + self.backtest_strategy() + + def live_strategy(self): + """Run the strategy.""" + with self as _: + logger.info("Running %s strategy on %s", self.name, self.symbol.name) + while self.running: + try: + self.sessions.check() + self.trade() + + except StopTrading: + self.running = False + break + + except Exception as err: + logger.error("Error: %s in live_strategy", err) + self.running = False + break + + def backtest_strategy(self): + """Backtest the strategy.""" + with self as _: + logger.info("Testing %s strategy on %s with Backtester", self.name, self.symbol.name) + while self.running: + try: + self.sessions.check() + self.backtest_controller.wait() + self.test() + except StopTrading: + self.running = False + break + except Exception as err: + logger.error(f"Error: {err} in backtest_strategy") + return + + def trade(self): + """Place trades using this method. This is the main method of the strategy. + It will be called by the strategy runner. + """ + raise NotImplementedError("Implement this method in your subclass") + + def test(self): + self.trade() diff --git a/src/aiomql/lib/sync/symbol.py b/src/aiomql/lib/sync/symbol.py new file mode 100644 index 0000000..b5fe545 --- /dev/null +++ b/src/aiomql/lib/sync/symbol.py @@ -0,0 +1,383 @@ +"""Symbol class for handling a financial instrument.""" + +from datetime import datetime +from logging import getLogger + +from ...core.constants import TimeFrame, CopyTicks +from ...core.base import _Base +from ...core.config import Config +from ...core.sync.meta_trader import MetaTrader +from ...core.models import SymbolInfo, BookInfo +from ...utils import round_off +from ..ticks import Tick +from ..account import Account +from ..candle import Candles +from ..ticks import Ticks + +logger = getLogger(__name__) + + +class Symbol(_Base, SymbolInfo): + """Main class for handling a financial instrument. A subclass of SymbolInfo it has attributes and methods + for working with a financial instrument. + + Attributes: + tick (Tick): Price tick object for instrument + account: An instance of the current trading account + + Notes: + Full properties are on the SymbolInfo Object. + Make sure a Symbol is always initialized with a name argument + """ + initialized: bool + tick: Tick + account: Account + + def __new__(cls, *args, **kwargs): + instance = super().__new__(cls) + instance.__class__.config = Config() + instance.__class__.mt5 = MetaTrader() + return instance + + def __init__(self, **kwargs): + """Initialize the Symbol object with the name of the financial instrument. + + Args: + name (str): Name of the financial instrument + """ + assert "name" in kwargs, "Symbol Object Must be initialized with a name" + super().__init__(**kwargs) + self.account = Account() + self.initialized = False + + def info_tick(self, *, name: str = "") -> Tick | None: + """Get the current price tick of a financial instrument. + + Args: + name: if name is supplied, get price tick of that financial instrument. Optional unnamed parameter. + + Returns: + Tick: Return a Tick Object + None: If request was unsuccessful + """ + try: + if not name: + tick = self.mt5.symbol_info_tick(self.name) + else: + self.mt5.symbol_select(name, True) + self.mt5.market_book_add(name) + tick = self.mt5.symbol_info_tick(name) + if tick is not None: + tick = Tick(**tick._asdict()) + setattr(self, "tick", tick) if not name else ... + return tick + except Exception as err: + logger.warning("%s: Unable to get tick for %s", err, self.name) + return None + + def symbol_select(self, *, enable: bool = True) -> bool: + """Select a symbol in the MarketWatch window or remove a symbol from the window. + Update the select property + + Args: + enable (bool): Switch. Optional unnamed parameter. If 'false', a symbol should be removed from + the MarketWatch window. + + Returns: + bool: True if successful, otherwise – False. + """ + self.select = self.mt5.symbol_select(self.name, enable) + return self.select + + def info(self) -> SymbolInfo | None: + """Get data on the specified financial instrument and update the symbol object properties + + Returns: + (SymbolInfo): SymbolInfo if successful + (None): If request was unsuccessful + """ + info = self.mt5.symbol_info(self.name) + if info is not None: + info = info._asdict() + self.set_attributes(**info) + return SymbolInfo(**info) + return None + + def initialize(self) -> bool: + """Initialize the symbol by pulling properties from the terminal + + Returns: + bool: Returns True if symbol info was successfully initialized + """ + try: + select = self.mt5.symbol_select(self.name, True) + self.select = select + + self.mt5.market_book_add(self.name) + + info = self.mt5.symbol_info(self.name) + if info is not None: + self.set_attributes(**info._asdict()) + + info_tick = self.mt5.symbol_info_tick(self.name) + + if info_tick: + self.tick = Tick(**info_tick._asdict()) + + if info is not None and info_tick is not None: + self.initialized = True + return True + + logger.warning("Unable to initialize %s", self.name) + return False + except Exception as err: + logger.warning("%s: Unable to initialize %s", err, self.name) + return False + + def initialize_sync(self) -> bool: + """Synchronous version of the initialize method""" + try: + select = self.mt5._symbol_select(self.name, True) + self.select = select + + self.mt5._market_book_add(self.name) + + info = self.mt5._symbol_info(self.name) + if info is not None: + self.set_attributes(**info._asdict()) + + info_tick = self.mt5._symbol_info_tick(self.name) + if info_tick: + self.tick = Tick(**info_tick._asdict()) + + if info is not None and info_tick is not None: + self.initialized = True + return True + logger.warning("Unable to initialize %s", self.name) + return False + except Exception as err: + logger.warning("%s: Unable to initialize %s", err, self.name) + return False + + def book_add(self) -> bool: + """Subscribes the MetaTrader 5 terminal to the Market Depth change events for a specified symbol. + If the symbol is not in the list of instruments for the market, This method will return False + + Returns: + bool: True if successful, otherwise – False. + """ + res = self.mt5.market_book_add(self.name) + if res is False: + logger.debug("Could not add %s to market book", self.name) + return res + + def book_get(self) -> tuple[BookInfo, ...]: + """Returns a tuple of BookInfo featuring Market Depth entries for the specified symbol. + + Returns: + tuple[BookInfo]: Returns the Market Depth contents as a tuples of BookInfo Objects + """ + book_info = self.mt5.market_book_get(self.name) + + if book_info is not None: + book_infos = (BookInfo(**info._asdict()) for info in book_info) + return tuple(book_infos) + + raise ValueError(f"Could not get book info for {self.name}") + + def book_release(self) -> bool: + """Cancels subscription of the MetaTrader 5 terminal to the Market Depth change events for a specified symbol. + + Returns: + bool: True if successful, otherwise – False. + """ + return self.mt5.market_book_release(self.name) + + def check_volume(self, *, volume: float) -> tuple[bool, float]: + """Check if the volume is within the limits of the symbol. If not, return the nearest limit. + + Args: + volume (float): Volume to check + + Returns: tuple[bool, float]: Returns a tuple of a boolean and a float. The boolean indicates if the volume is + within the limits of the symbol. The float is the volume to use if the volume is not within the limits of the + symbol. + """ + if check := self.volume_min <= volume <= self.volume_max: + return check, volume + else: + return check, self.volume_min if volume <= self.volume_min else self.volume_max + + def round_off_volume(self, *, volume: float, round_down: bool = False) -> float: + """Round off the volume to the nearest volume step. + + Args: + volume (float): Volume to round off + round_down (bool): If True, round down. If False, round up. Optional unnamed parameter. Defaults to True. + + Returns: + float: Rounded off volume + """ + return round_off(value=volume, step=self.volume_step, round_down=round_down) + + def amount_in_quote_currency(self, *, amount: float) -> float: + """Convert the amount to the quote currency of the symbol.""" + if self.currency_profit != self.account.currency: + amount = self.convert_currency( + amount=amount, from_currency=self.account.currency, to_currency=self.currency_profit + ) + return amount + + def compute_volume(self) -> float: + """Computes the volume required for a trade usually based on the amount and any other keyword arguments. + This is a dummy method that returns the minimum volume of the symbol. It is meant to be overridden by a subclass + that implements the computation of volume. + + Returns: + float: Returns the volume of the trade + """ + return self.volume_min + + def convert_currency(self, *, amount: float, from_currency: str, to_currency: str) -> float | None: + """Convert a given amount from one currency to the other. + Args: + amount: Amount to convert + from_currency: Currency to convert from + to_currency: Currency to convert to + """ + base, quote = to_currency, from_currency + try: + pair = f"{quote}{base}" + tick = self.info_tick(name=pair) + if tick is not None: + return round(amount * tick.bid, 2) + + pair = f"{base}{quote}" + tick = self.info_tick(name=pair) + return round(amount / tick.ask, 2) + except Exception as err: + logger.warning(f"{err}: Currency conversion failed: Unable to convert {amount} in {quote} to {base}") + return None + + def copy_rates_from(self, *, timeframe: TimeFrame, date_from: datetime | int, count: int = 500) -> Candles: + """ + Get bars from the MetaTrader 5 terminal starting from the specified date. + + Args: timeframe (TimeFrame): Timeframe the bars are requested for. Set by a value from the TimeFrame + enumeration. Required unnamed parameter. + + date_from (datetime | int): Date of opening of the first bar from the requested sample. Set by the + 'datetime' object or as a number of seconds elapsed since 1970.01.01. Required unnamed parameter. + + count (int): Number of bars to receive. Required unnamed parameter. + + Returns: + Candles: Returns a Candles object as a collection of rates ordered chronologically + + Raises: + ValueError: If request was unsuccessful and None was returned + """ + rates = self.mt5.copy_rates_from(self.name, timeframe, date_from, count) + if rates is not None: + return Candles(data=rates) + raise ValueError(f"Could not get rates for {self.name}.") + + def copy_rates_from_pos(self, *, timeframe: TimeFrame, count: int = 500, start_position: int = 0) -> Candles: + """Get bars from the MetaTrader 5 terminal starting from the specified index. + + Args: + timeframe (TimeFrame): TimeFrame value from TimeFrame Enum. Required keyword only parameter + + count (int): Number of bars to return. Keyword argument defaults to 500 + + start_position (int): Initial index of the bar the data are requested from. The numbering of bars goes from + present to past. Thus, the zero bar means the current one. Keyword argument defaults to 0. + + Returns: + Candles: Returns a Candles object as a collection of rates ordered chronologically. + + Raises: + ValueError: If request was unsuccessful and None was returned + """ + rates = self.mt5.copy_rates_from_pos(self.name, timeframe, start_position, count) + if rates is not None: + return Candles(data=rates) + raise ValueError(f"Could not get rates for {self.name}.") + + def copy_rates_range( + self, *, timeframe: TimeFrame, date_from: datetime | int, date_to: datetime | int + ) -> Candles: + """Get bars in the specified date range from the MetaTrader 5 terminal. + + Args: + timeframe (TimeFrame): Timeframe for the bars using the TimeFrame enumeration. Required unnamed parameter. + + date_from (datetime | int): Date the bars are requested from. Set by the 'datetime' object or as a number + of seconds elapsed since 1970.01.01. Bars with the open time >= date_from are returned. Required unnamed + parameter. + + date_to (datetime | int): Date, up to which the bars are requested. Set by the 'datetime' object or as a + number of seconds elapsed since 1970.01.01. Bars with the open time <= date_to are returned. + Required unnamed parameter. + + Returns: + Candles: Returns a Candles object as a collection of rates ordered chronologically. + + Raises: + ValueError: If request was unsuccessful and None was returned + """ + rates = self.mt5.copy_rates_range( + symbol=self.name, timeframe=timeframe, date_from=date_from, date_to=date_to + ) + if rates is not None: + return Candles(data=rates) + raise ValueError(f"Could not get rates for {self.name}.") + + def copy_ticks_from(self, *, date_from: datetime | int, count: int = 100, + flags: CopyTicks = CopyTicks.ALL) -> Ticks: + """ + Get ticks from the MetaTrader 5 terminal starting from the specified date. + + Args: date_from (datetime | int): Date the ticks are requested from. Set by the 'datetime' object or as a + number of seconds elapsed since 1970.01.01. + + count (int): Number of requested ticks. Defaults to 100 + + flags (CopyTicks): A flag to define the type of the requested ticks from CopyTicks enum. INFO is the default + + Returns: + Candles: Returns a Candles object as a collection of ticks ordered chronologically. + + Raises: + ValueError: If request was unsuccessful and None was returned + """ + ticks = self.mt5.copy_ticks_from(self.name, date_from, count, flags) + if ticks is not None: + return Ticks(data=ticks) + raise ValueError(f"Could not get ticks for {self.name}.") + + def copy_ticks_range( + self, *, date_from: datetime | int, date_to: datetime | int, flags: CopyTicks = CopyTicks.ALL + ) -> Ticks: + """Get ticks for the specified date range from the MetaTrader 5 terminal. + + Args: + date_from: Date the bars are requested from. Set by the 'datetime' object or as a number of seconds elapsed + since 1970.01.01. Bars with the open time >= date_from are returned. Required unnamed parameter. + + date_to: Date, up to which the bars are requested. Set by the 'datetime' object or as a number of + seconds elapsed since 1970.01.01. Bars with the open time <= date_to are returned. + Required unnamed parameter. + + flags (CopyTicks): + + Returns: + Candles: Returns a Candles object as a collection of ticks ordered chronologically. + + Raises: + ValueError: If request was unsuccessful and None was returned. + """ + ticks = self.mt5.copy_ticks_range(self.name, date_from, date_to, flags) + if ticks is not None: + return Ticks(data=ticks) + raise ValueError(f"Could not get ticks for {self.name}.") diff --git a/tests/conftest.py b/tests/conftest.py index 51cb628..355b4ad 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,6 +7,11 @@ def btc_usd(): return Symbol(name="BTCUSD") +@pytest.fixture(scope="function") +def eth_usd(): + return Symbol(name="ETHUSD") + + @pytest.fixture(scope="function") async def buy_order(btc_usd): sym = btc_usd @@ -26,8 +31,8 @@ async def buy_order(btc_usd): @pytest.fixture(scope="function") -async def sell_order(btc_usd): - sym = btc_usd +async def sell_order(eth_usd): + sym = eth_usd sym_info = await sym.mt5.symbol_info(sym.name) return { "action": sym.mt5.TRADE_ACTION_DEAL, diff --git a/tests/live/unit/test_account.py b/tests/live/unit/test_account.py index 982941b..a58a43f 100644 --- a/tests/live/unit/test_account.py +++ b/tests/live/unit/test_account.py @@ -15,6 +15,6 @@ class TestAccount: assert self.account.connected is True async def test_account_info(self): - acc_info = await self.account.mt5.account_info + acc_info = await self.account.mt5.account_info() assert acc_info.login == self.account.login assert acc_info.server == self.account.server diff --git a/todo/thread_task_queue.py b/todo/thread_task_queue.py new file mode 100644 index 0000000..e69de29