From 28811df0cef402ffef6b1758ba8030bdfec2e2aa Mon Sep 17 00:00:00 2001 From: Ichinga Samuel Date: Thu, 23 Jan 2025 18:18:18 +0100 Subject: [PATCH] v4.0.9 --- .gitignore | 2 + CHANGELOG.md | 17 ++++- check.py | 36 ++++++++++ docs/lib/strategy.md | 8 +++ docs/lib/symbol.md | 14 +++- .../backtest_data_01_01_24_06_05_24.json | 24 +++---- src/aiomql/contrib/strategies/chaos.py | 12 ++-- .../core/backtesting/backtest_engine.py | 48 +++++++++++++ src/aiomql/core/meta_backtester.py | 15 +++++ src/aiomql/core/meta_trader.py | 5 +- src/aiomql/core/task_queue.py | 13 +++- src/aiomql/lib/backtester.py | 64 +++++------------- src/aiomql/lib/bot.py | 39 ++++------- src/aiomql/lib/candle.py | 4 +- src/aiomql/lib/executor.py | 67 +++++++------------ src/aiomql/lib/strategy.py | 10 ++- src/aiomql/lib/symbol.py | 47 +++++++++++-- tests/live/integration/test_bot.py | 7 +- tests/live/integration/test_bot_sync.py | 3 +- tests/live/unit/test_task_queue.py | 6 +- 20 files changed, 284 insertions(+), 157 deletions(-) create mode 100644 check.py diff --git a/.gitignore b/.gitignore index 2674615..10839b0 100644 --- a/.gitignore +++ b/.gitignore @@ -79,4 +79,6 @@ aiomql.json # development terminals/ *.pkl +backtesting/ +trade_records/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 558c6cb..f48cf47 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,21 @@ # Changelog -## [4.0.7](https://github.com/Ichinga-Samuel/aiomql/releases/tag/v4.0.8) - 2025-01-21 +## [4.0.9](https://github.com/Ichinga-Samuel/aiomql/releases/tag/v4.0.8) - 2025-01-23 + +### Fixed + +- Fixed `__add__` to return a new Candles object + +### Changed + +- Removed tasks attribute from executor class + +### Added + +- Added `initialize_sync` method for synchronous initialization of a symbol + + +## [4.0.8](https://github.com/Ichinga-Samuel/aiomql/releases/tag/v4.0.8) - 2025-01-21 ### Fixed diff --git a/check.py b/check.py new file mode 100644 index 0000000..b9c5cec --- /dev/null +++ b/check.py @@ -0,0 +1,36 @@ +import asyncio + + +async def rt(): + await asyncio.sleep(1) + print("Hello") + +async def rrt(): + while True: + await asyncio.sleep(15) + print("rtt") + + +async def stop(task, tm=10): + await asyncio.sleep(tm) + print("Stopping", task) + res = task.cancel() + print(res, task) + +def main(): + task1 = rt() + task2 = rrt() + # task3 = stop(task2) + tasks = [task1, task2, task3] + # task = asyncio.gather(*tasks, return_exceptions=True) + asyncio.run(asyncio.gather(*tasks, return_exceptions=True)) + # await asyncio.gather(task, stop(task, tm=3), return_exceptions=True) + +main = main() +# asyncio.run(main) + +# def main(a, b=6, **kwargs): + # print(a, b, kwargs) + + +# main(4, c=9, g=4, b=99) diff --git a/docs/lib/strategy.md b/docs/lib/strategy.md index 1aedc8a..3f43242 100644 --- a/docs/lib/strategy.md +++ b/docs/lib/strategy.md @@ -13,6 +13,7 @@ The base class for creating strategies. - [backtest_strategy](#strategy.backtest_strategy) - [trade](#strategy.trade) - [test](#strategy.test) +- [initialize](#strategy.initialize) @@ -134,3 +135,10 @@ Runs the strategy in live mode. async def live_strategy() ``` Runs the strategy in backtest mode. + + +### initialize +```python +async def initialize() +``` +Initialize a strategy diff --git a/docs/lib/symbol.md b/docs/lib/symbol.md index fc43e38..d981b89 100644 --- a/docs/lib/symbol.md +++ b/docs/lib/symbol.md @@ -7,6 +7,7 @@ Symbol class for handling a financial instrument. - [symbol_select](#symbol.symbol_select) - [info](#symbol.info) - [initialize](#symbol.initialize) +- [initialize_sync](#symbol.initialize_sync) - [book_add](#symbol.book_add) - [book_get](#symbol.book_get) - [book_release](#symbol.book_release) @@ -94,7 +95,7 @@ Get data on the specified financial instrument and update the symbol object prop -### init +### initialize ```python async def initialize() -> bool ``` @@ -105,6 +106,17 @@ Initialized the symbol by pulling properties from the terminal |--------|--------------------------------------------------------| | `bool` | Returns True if symbol info was successful initialized | + +### initialize_sync +```python +def initialize_sync() -> bool +``` +Initialized the symbol by pulling properties from the terminal +#### Returns: +| Type | Description | +|--------|--------------------------------------------------------| +| `bool` | Returns True if symbol info was successful initialized | + ### book_add diff --git a/examples/backtesting/backtest_data_01_01_24_06_05_24.json b/examples/backtesting/backtest_data_01_01_24_06_05_24.json index 234edaf..c45e86d 100644 --- a/examples/backtesting/backtest_data_01_01_24_06_05_24.json +++ b/examples/backtesting/backtest_data_01_01_24_06_05_24.json @@ -1,17 +1,17 @@ { - "balance": 297007.45, + "balance": 367056.24, "profit": 0, - "equity": 297007.45, + "equity": 367056.24, "margin": 0.0, - "margin_free": 297007.45, + "margin_free": 367056.24, "margin_level": 0, - "wins": 307, - "losses": 336, - "total": 643, - "win_percentage": 47.74, - "win": 619421.92, - "loss": -323164.47, - "net_profit": 296257.45, - "profit_factor": 1.92, - "profitability": 39500.99 + "wins": 374, + "losses": 394, + "total": 768, + "win_percentage": 48.7, + "win": 946684.28, + "loss": -580378.04, + "net_profit": 366306.24, + "profit_factor": 1.63, + "profitability": 48840.83 } \ No newline at end of file diff --git a/src/aiomql/contrib/strategies/chaos.py b/src/aiomql/contrib/strategies/chaos.py index d67529a..3eb6a57 100644 --- a/src/aiomql/contrib/strategies/chaos.py +++ b/src/aiomql/contrib/strategies/chaos.py @@ -20,11 +20,13 @@ class Chaos(Strategy): fast_ema: int slow_ema: int tracker: Tracker - parameters = {"fast_ema": 8, "slow_ema": 20, "ltf": TimeFrame.M1, "htf": TimeFrame.M2, "lcc": 100, "hcc": 100} + interval: int + parameters = {"fast_ema": 8, "slow_ema": 20, "ltf": TimeFrame.M1, "htf": TimeFrame.M2, "lcc": 100, "hcc": 100, + "interval": 0} def __init__(self, *, symbol: ForexSymbol, params: dict = None, sessions=None, name="Chaos"): super().__init__(symbol=symbol, params=params, sessions=sessions, name=name) - self.tracker = Tracker(snooze=self.ltf.seconds) + self.tracker = Tracker(snooze=self.interval or self.ltf.seconds) self.trader = ScalpTrader(symbol=self.symbol) async def check_trend(self): @@ -43,12 +45,12 @@ class Chaos(Strategy): candles.rename(inplace=True, **{f"EMA_{self.fast_ema}": "fast", f"EMA_{self.slow_ema}": "slow"}) order_type = random.choice([OrderType.BUY, OrderType.SELL]) if order_type == OrderType.BUY: - self.tracker.update(trend="bullish", snooze=self.htf.seconds, order_type=OrderType.BUY) + self.tracker.update(trend="bullish", snooze=self.interval or self.htf.seconds, order_type=OrderType.BUY) else: - self.tracker.update(trend="bearish", snooze=self.htf.seconds, order_type=OrderType.SELL) + self.tracker.update(trend="bearish", snooze=self.interval or self.htf.seconds, order_type=OrderType.SELL) except Exception as err: logger.error(f"{err}. Failed to check trend") - self.tracker.update(trend="ranging", snooze=self.ltf.seconds, order_type=None) + self.tracker.update(trend="ranging", snooze=self.interval or self.ltf.seconds, order_type=None) async def trade(self): try: diff --git a/src/aiomql/core/backtesting/backtest_engine.py b/src/aiomql/core/backtesting/backtest_engine.py index 06bb563..b083747 100644 --- a/src/aiomql/core/backtesting/backtest_engine.py +++ b/src/aiomql/core/backtesting/backtest_engine.py @@ -1067,6 +1067,54 @@ class BackTestEngine: tick = await self.get_price_tick(symbol=symbol, time=self.cursor.time) return tick + async def symbol_select(self, *, symbol: str, enable: bool) -> bool: + if self.use_terminal: + info = await self.mt5.symbol_select(symbol, enable) + return info + else: + return symbol in self._data.symbols.keys() + + def symbol_select_sync(self, *, symbol: str, enable: bool = True) -> bool: + if self.use_terminal: + info = self.mt5._symbol_select(symbol, enable) + return info + else: + return symbol in self._data.symbols.keys() + + def symbol_info_tick_sync(self, *, symbol) -> Tick | None: + if self.use_terminal: + time = datetime.fromtimestamp(self.cursor.time, tz=UTC) + tick = self.mt5._copy_ticks_from(symbol, time, 1, CopyTicks.ALL) + tick = Tick(tick[-1]) if tick is not None else None + else: + tick = self.prices[symbol].loc[self.cursor.time] + tick = Tick(tick) if tick is not None else None + return tick + + def symbol_info_sync(self, *, symbol) -> SymbolInfo | None: + if self.use_terminal: + info = self.mt5._symbol_info(symbol) + time = datetime.fromtimestamp(self.cursor.time, tz=UTC) + tick = self.mt5._copy_ticks_from(symbol, time, 1, CopyTicks.ALL) + tick = Tick(tick[-1]) if tick is not None else None + else: + info = self.symbols[symbol] + tick = self.prices[symbol].loc[self.cursor.time] + tick = Tick(tick) if tick is not None else None + + if info and tick: + info = info._asdict() | { + "bid": tick.bid, + "bidhigh": tick.bid, + "bidlow": tick.bid, + "ask": tick.ask, + "askhigh": tick.ask, + "asklow": tick.bid, + "last": tick.last, + "volume_real": tick.volume_real, + } + return SymbolInfo((info.get(key) for key in SymbolInfo.__match_args__)) + @async_cache async def _symbol_info(self, *, symbol: str) -> SymbolInfo: if self.use_terminal: diff --git a/src/aiomql/core/meta_backtester.py b/src/aiomql/core/meta_backtester.py index a749d7a..c93a9a0 100644 --- a/src/aiomql/core/meta_backtester.py +++ b/src/aiomql/core/meta_backtester.py @@ -87,6 +87,15 @@ class MetaBackTester(MetaTrader): return super().login_sync(login=login, password=password, server=server, timeout=timeout) return True + def _symbol_select(self, symbol: str, enable: bool) -> bool: + return self.backtest_engine.symbol_select_sync(symbol=symbol, enable=enable) + + def _market_book_add(self, symbol: str) -> bool: + return True + + async def market_book_add(self, symbol: str) -> bool: + return True + async def login(self, *, login: int = 0, password: str = "", server: str = "", timeout: int = 60000) -> bool: if self.config.use_terminal_for_backtesting: return await super().login(login=login, password=password, server=server, timeout=timeout) @@ -121,6 +130,12 @@ class MetaBackTester(MetaTrader): sym = await self.backtest_engine.get_symbol_info(symbol=symbol) return sym + def _symbol_info(self, symbol) -> SymbolInfo | None: + return self.backtest_engine.symbol_info_sync(symbol=symbol) + + def _symbol_info_tick(self, symbol) -> Tick | None: + return self.backtest_engine.symbol_info_tick_sync(symbol=symbol) + @error_handler(msg="test data not available") async def symbol_info_tick(self, symbol: str) -> Tick | None: tick = await self.backtest_engine.get_symbol_info_tick(symbol=symbol) diff --git a/src/aiomql/core/meta_trader.py b/src/aiomql/core/meta_trader.py index 4f10745..df677b6 100644 --- a/src/aiomql/core/meta_trader.py +++ b/src/aiomql/core/meta_trader.py @@ -17,7 +17,6 @@ from MetaTrader5 import ( OrderSendResult, OrderCheckResult, ) -import MetaTrader5 as mt5 from .constants import OrderType, CopyTicks @@ -111,7 +110,7 @@ class MetaTrader(MetaCore): login = login or acc_details.get("login", 0) password = password or acc_details.get("password", "") server = server or acc_details.get("server", "") - res = mt5.login(login, password=password, server=server, timeout=timeout) + res = self._login(login, password=password, server=server, timeout=timeout) return res async def initialize( @@ -200,7 +199,7 @@ class MetaTrader(MetaCore): ) if key is not None } - res = mt5.initialize(*args, **kwargs) + res = self._initialize(*args, **kwargs) if res is False: self._shutdown() if not res: diff --git a/src/aiomql/core/task_queue.py b/src/aiomql/core/task_queue.py index f9ce06a..25e1be3 100644 --- a/src/aiomql/core/task_queue.py +++ b/src/aiomql/core/task_queue.py @@ -64,7 +64,7 @@ class TaskQueue: mode: Literal['finite', 'infinite'] = 'finite', worker_timeout: int = 1): self.queue = queue or asyncio.PriorityQueue(maxsize=size) self.workers = workers - self.worker_tasks = {} + self.worker_tasks: dict[int|float, asyncio.Task] = {} self.queue_timeout = queue_timeout self.absolute_timeout = absolute_timeout self.stop = False @@ -72,6 +72,7 @@ class TaskQueue: self.mode = mode self.worker_timeout = worker_timeout self.queue_task_cancelled = False + self.start_time = time.perf_counter() signal(SIGINT, self.sigint_handle) def add(self, *, item: QueueItem, priority=3, must_complete=False): @@ -187,6 +188,14 @@ class TaskQueue: wr = range(no_of_workers) [self.worker_tasks.setdefault(wi:=ri(), ct(wi)) for _ in wr] + async def watch(self): + while True: + await asyncio.sleep(1) + if (time.perf_counter() - self.start_time) > self.absolute_timeout: + self.stop = True + self.cancel() + break + async def run(self, queue_timeout: int = None, absolute_timeout: int = None): """Run the queue until all tasks are completed or the timeout is reached. @@ -202,6 +211,8 @@ class TaskQueue: self.start_timer(queue_timeout=queue_timeout, absolute_timeout=absolute_timeout, start=True) await self.add_workers(no_of_workers=self.workers) self.queue_task = asyncio.create_task(self.queue.join()) + if self.absolute_timeout: + asyncio.create_task(self.watch()) await self.queue_task except asyncio.TimeoutError: diff --git a/src/aiomql/lib/backtester.py b/src/aiomql/lib/backtester.py index 8b89cf6..323279a 100644 --- a/src/aiomql/lib/backtester.py +++ b/src/aiomql/lib/backtester.py @@ -1,16 +1,13 @@ import asyncio import logging +import time from typing import Type, Iterable, Callable, Coroutine -from datetime import datetime, UTC - -from MetaTrader5 import Tick from .executor import Executor from ..core.config import Config from ..core.backtesting.backtest_controller import BackTestController from ..core.meta_backtester import MetaBackTester from ..core.backtesting.backtest_engine import BackTestEngine -from ..core.constants import CopyTicks from .symbol import Symbol as Symbol from .strategy import Strategy as Strategy @@ -58,8 +55,10 @@ class BackTester: self.backtest_engine.setup_account_sync() self.init_strategies_sync() if (strategies := len(self.executor.strategy_runners)) == 0: - logger.warning("No strategies were added to the backtester. Exiting ...") - raise Exception("No strategies added to the backtester") + self.config.shutdown = True + logger.warning("No strategies were added to the backtester. Exiting in one second") + time.sleep(1) + return self.config.task_queue.worker_timeout = 5 self.add_coroutine(coroutine=self.config.task_queue.run, on_separate_thread=True) self.add_coroutine(coroutine=self.executor.exit) @@ -87,8 +86,10 @@ class BackTester: await self.backtest_engine.setup_account() await self.init_strategies() if (strategies := len(self.executor.strategy_runners)) == 0: - logger.warning("No strategies were added to the backtester. Exiting ...") - raise Exception("No strategies added to the backtester") + self.config.shutdown = True + logger.warning("No strategies were added to the backtester. Exiting in one second") + await asyncio.sleep(1) + return self.config.task_queue.worker_timeout = 5 self.add_coroutine(coroutine=self.config.task_queue.run, on_separate_thread=True) self.add_coroutine(coroutine=self.executor.exit) @@ -113,16 +114,19 @@ class BackTester: def execute(self): """Execute the bot.""" self.initialize_sync() - self.executor.execute() + if self.config.shutdown is False: + self.executor.execute() async def start(self): """Initialize the bot and execute it. Similar to calling `execute` method but is a coroutine.""" await self.initialize() + if self.config.shutdown is False: + self.executor.execute() self.executor.execute() def add_strategy(self, *, strategy: Strategy): """Add a strategy to the list of strategies. - An added strategy will only run if it's symbol was successfully initialized and it is added to the executor. + An added strategy will only run if it's symbol was successfully initialized, and it is added to the executor. Args: strategy (Strategy): A Strategy instance to run on bot @@ -162,44 +166,10 @@ class BackTester: def init_strategy_sync(self, *, strategy: Strategy) -> bool: """Initialize a single strategy. This method is called internally by the bot.""" try: - if self.backtest_engine.use_terminal is False: - info = self.backtest_engine.symbols[strategy.symbol.name] - tick = self.backtest_engine.prices[strategy.symbol.name].loc[self.backtest_engine.cursor.time] - tick = Tick(tick) - info = info._asdict() | { - "bid": tick.bid, - "bidhigh": tick.bid, - "bidlow": tick.bid, - "ask": tick.ask, - "askhigh": tick.ask, - "asklow": tick.bid, - "last": tick.last, - "volume_real": tick.volume_real, - } - strategy.symbol.set_attributes(**info) + res = strategy.symbol.initialize_sync() + if res: self.executor.add_strategy(strategy=strategy) - return True - else: - self.mt._symbol_select(strategy.symbol.name, True) - self.mt._market_book_add(strategy.symbol.name) - info = self.mt._symbol_info(strategy.symbol.name) - time = datetime.fromtimestamp(self.backtest_engine.cursor.time, tz=UTC) - tick = self.mt._copy_ticks_from(strategy.symbol.name, time, 1, CopyTicks.ALL) - tick = Tick(tick[-1]) if tick is not None else None - info = info._asdict() | { - "bid": tick.bid, - "bidhigh": tick.bid, - "bidlow": tick.bid, - "ask": tick.ask, - "askhigh": tick.ask, - "asklow": tick.bid, - "last": tick.last, - "volume_real": tick.volume_real, - } - strategy.symbol.set_attributes(**info) - strategy.symbol.initialized = True - self.executor.add_strategy(strategy=strategy) - return True + return res except Exception as err: logger.error("%s: Unable to initialize strategy", err) return False diff --git a/src/aiomql/lib/bot.py b/src/aiomql/lib/bot.py index 52f5153..f1bab00 100644 --- a/src/aiomql/lib/bot.py +++ b/src/aiomql/lib/bot.py @@ -8,7 +8,6 @@ from .executor import Executor from ..core.config import Config from ..core.meta_trader import MetaTrader from .symbol import Symbol as Symbol -from .ticks import Tick from .strategy import Strategy as Strategy logger = logging.getLogger(__name__) @@ -67,7 +66,7 @@ class Bot: self.add_coroutine(coroutine=self.executor.exit) if len(self.executor.strategy_runners) == 0: - logger.warning("No strategies were added to the bot. Exiting in five seconds") + logger.warning("No strategies were added to the bot. Exiting in one second") await asyncio.sleep(1) self.config.shutdown = True except Exception as err: @@ -76,7 +75,8 @@ class Bot: def initialize_sync(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. + 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 @@ -93,7 +93,7 @@ class Bot: self.add_coroutine(coroutine=self.executor.exit) if len(self.executor.strategy_runners) == 0: - logger.warning("No strategies were added to the bot. Exiting in 5 seconds") + logger.warning("No strategies were added to the bot. Exiting in one second") time.sleep(1) self.config.shutdown = True except Exception as err: @@ -120,14 +120,16 @@ class Bot: self.executor.add_coroutine(coroutine=coroutine, kwargs=kwargs, on_separate_thread=on_separate_thread) def execute(self): - """Execute the bot using asyncio.run""" + """Start the bot in sync mode""" self.initialize_sync() - self.executor.execute() + if self.config.shutdown is False: + self.executor.execute() async def start(self): """Initialize the bot and call the executor it.""" await self.initialize() - self.executor.execute() + if self.config.shutdown is False: + self.executor.execute() def add_strategy(self, *, strategy: Strategy): """Add a strategy to the list of strategies. @@ -174,25 +176,10 @@ class Bot: def init_strategy_sync(self, *, strategy: Strategy) -> bool: """Initialize a single strategy. This method is called internally by the bot.""" - try: - select = self.mt5._symbol_select(strategy.symbol.name, True) - info = self.mt5._symbol_info(strategy.symbol.name) - tick = self.mt5._symbol_info_tick(strategy.symbol.name) - self.mt5._market_book_add(strategy.symbol.name) - if info is not None and tick is not None: - info = info._asdict() - info["swap_rollover3days"] = info.get("swap_rollover3days", 0) % 7 - info["select"] = select - tick = Tick(**tick._asdict()) - strategy.symbol.tick = tick - strategy.symbol.initialized = True - strategy.symbol.set_attributes(**info) - self.executor.add_strategy(strategy=strategy) - return True - return False - except Exception as err: - logger.warning("%s: Unable to initialize strategy", err) - return False + res = strategy.symbol.initialize_sync() + if res: + self.executor.add_strategy(strategy=strategy) + return res def init_strategies_sync(self): """Initialize the symbols for the current trading session. This method is called internally by the bot.""" diff --git a/src/aiomql/lib/candle.py b/src/aiomql/lib/candle.py index ed01c13..da3ab59 100644 --- a/src/aiomql/lib/candle.py +++ b/src/aiomql/lib/candle.py @@ -298,11 +298,11 @@ class Candles: """Add a new row to the candles class.""" if isinstance(row, Series): data = self. pd.concat([self._data, pd.DataFrame(row).T]) - return self.Candle(data=data) + return self.__class__(data=data) elif isinstance(row, DataFrame): data = pd.concat([self._data, row]) - return self.Candle(data=data) + return self.__class__(data=data) def add(self, row: DataFrame | Series) -> bool: """Add a new row to the candles class.""" diff --git a/src/aiomql/lib/executor.py b/src/aiomql/lib/executor.py index 6a0fac4..98b5f7b 100644 --- a/src/aiomql/lib/executor.py +++ b/src/aiomql/lib/executor.py @@ -17,20 +17,18 @@ class Executor: Attributes: executor (ThreadPoolExecutor): The executor object. strategy_runners (list): List of strategies. - coroutines (list[Coroutine]): A list of coroutines to run in the executor + coroutines (dict[Coroutine, dict]): A list of coroutines to run in the executor + coroutine_threads (dict[Coroutine, dict]): A list of coroutines to run in the executor functions (dict[Callable, dict]): A dictionary of functions to run in the executor """ - executor: ThreadPoolExecutor - tasks: list[asyncio.Task | asyncio.Future] config: Config def __init__(self): self.strategy_runners: list[Strategy] = [] - self.coroutines: list[Coroutine] = [] - self.coroutine_threads: list[Coroutine] = [] + self.coroutines: dict[Coroutine: dict] = {} + self.coroutine_threads: dict[Coroutine: dict] = {} self.functions: dict[Callable:dict] = {} - self.tasks = [] self.config = Config() self.timeout = None # Timeout for the executor. For testing purposes only signal(SIGINT, self.sigint_handle) @@ -41,8 +39,10 @@ class Executor: def add_coroutine(self, *, coroutine: Callable | Coroutine, kwargs: dict = None, on_separate_thread=False): kwargs = kwargs or {} - coroutine = coroutine(**kwargs) - self.coroutines.append(coroutine) if on_separate_thread is False else self.coroutine_threads.append(coroutine) + if on_separate_thread: + self.coroutine_threads[coroutine] = kwargs + else: + self.coroutines[coroutine] = kwargs def add_strategies(self, *, strategies: tuple[Strategy]): """Add multiple strategies at once @@ -60,12 +60,8 @@ class Executor: """ self.strategy_runners.append(strategy) - # async def create_strategy_task(self, strategy: Strategy): - # task = asyncio.create_task(strategy.run_strategy()) - # self.tasks.append(task) - # await task - - def run_strategy(self, strategy: Strategy): + @staticmethod + def run_strategy(strategy: Strategy): """Wraps the coroutine trade method of each strategy with 'asyncio.run'. Args: @@ -73,38 +69,23 @@ class Executor: """ asyncio.run(strategy.run_strategy()) - async def create_coroutine_task(self, coroutine: Coroutine): - task = asyncio.create_task(coroutine) - self.tasks.append(task) - await task - - async def create_coroutines_task(self): - """""" - tasks = [asyncio.create_task(coroutine) for coroutine in self.coroutines] - self.tasks.extend(tasks) - task = asyncio.gather(*tasks, return_exceptions=True) - self.tasks.append(task) - await task - - def run_coroutine_tasks(self): + async def run_coroutine_tasks(self): """Run all coroutines in the executor""" - asyncio.run(self.create_coroutines_task()) + await asyncio.gather(*[coroutine(**kwargs) for coroutine, kwargs in self.coroutines.items()], + return_exceptions=True) - def run_coroutine_task(self, coroutine): - asyncio.run(self.create_coroutine_task(coroutine)) + @staticmethod + def run_coroutine_task(coroutine, kwargs): + asyncio.run(coroutine(**kwargs)) @staticmethod def run_function(function: Callable, kwargs: dict): """Run a function - Args: function: The function to run kwargs: A dictionary of keyword arguments for the function """ - try: - function(**kwargs) - except Exception as err: - logger.error(f"Error: {err}. Unable to run function: {function.__name__}") + function(**kwargs) def sigint_handle(self, signum, frame): self.config.shutdown = True @@ -117,19 +98,20 @@ class Executor: if self.timeout is not None and self.timeout < (asyncio.get_event_loop().time() - start): self.config.shutdown = True break - timeout = self.timeout or 30 + + timeout = 1 if self.timeout else 30 await asyncio.sleep(timeout) + for strategy in self.strategy_runners: strategy.running = False + self.config.task_queue.cancel() + if self.config.backtest_engine is not None: self.config.backtest_engine.stop_testing = True self.executor.shutdown(wait=False, cancel_futures=False) - for task in self.tasks: - task.cancel() - if self.config.force_shutdown: os._exit(1) except Exception as err: @@ -151,5 +133,6 @@ class Executor: self.executor = executor [self.executor.submit(self.run_strategy, strategy) for strategy in self.strategy_runners] [self.executor.submit(function, **kwargs) for function, kwargs in self.functions.items()] - [self.executor.submit(self.run_coroutine_task, coroutine) for coroutine in self.coroutine_threads] - self.executor.submit(self.run_coroutine_tasks) + [self.executor.submit(self.run_coroutine_task, coroutine, kwargs) for coroutine, kwargs in + self.coroutine_threads.items()] + self.executor.submit(asyncio.run, self.run_coroutine_tasks()) diff --git a/src/aiomql/lib/strategy.py b/src/aiomql/lib/strategy.py index b57e4b0..5e44f00 100644 --- a/src/aiomql/lib/strategy.py +++ b/src/aiomql/lib/strategy.py @@ -177,17 +177,25 @@ class Strategy(ABC): try: await self.sessions.check() await self.trade() + except StopTrading: self.running = False break + + except asyncio.CancelledError: + self.running = False + break + except Exception as err: logger.error("Error: %s in live_strategy", err) - return + self.running = False + break async def backtest_strategy(self): """Backtest the strategy.""" async with self as _: logger.info("Testing %s strategy on %s with Backtester", self.name, self.symbol.name) + await self.initialize() while self.running: try: await self.sessions.check() diff --git a/src/aiomql/lib/symbol.py b/src/aiomql/lib/symbol.py index 97642d7..2f11ef1 100644 --- a/src/aiomql/lib/symbol.py +++ b/src/aiomql/lib/symbol.py @@ -40,6 +40,7 @@ class Symbol(_Base, SymbolInfo): assert "name" in kwargs, "Symbol Object Must be initialized with a name" super().__init__(**kwargs) self.account = Account() + self.initialized = False @backoff_decorator async def info_tick(self, *, name: str = "") -> Tick | None: @@ -63,7 +64,6 @@ class Symbol(_Base, SymbolInfo): tick = Tick(**tick._asdict()) setattr(self, "tick", tick) if not name else ... return tick - return None except Exception as err: logger.warning("%s: Unable to get tick for %s", err, self.name) return None @@ -92,8 +92,7 @@ class Symbol(_Base, SymbolInfo): info = await self.mt5.symbol_info(self.name) if info is not None: info = info._asdict() - info["swap_rollover3days"] = info.get("swap_rollover3days", 0) % 7 - self.set_attributes(**info) + # self.set_attributes(**info) return SymbolInfo(**info) return None @@ -104,10 +103,44 @@ class Symbol(_Base, SymbolInfo): bool: Returns True if symbol info was successful initialized """ try: - await self.symbol_select() - info = await self.info() - info_tick = await self.info_tick() - await self.book_add() + select = await self.mt5.symbol_select(self.name, True) + self.select = select + + await self.mt5.market_book_add(self.name) + + info = await self.mt5.symbol_info(self.name) + if info is not None: + 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: + 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 diff --git a/tests/live/integration/test_bot.py b/tests/live/integration/test_bot.py index 50435a6..755c0e8 100644 --- a/tests/live/integration/test_bot.py +++ b/tests/live/integration/test_bot.py @@ -4,17 +4,20 @@ from aiomql.lib.bot import Bot from aiomql.contrib.strategies import Chaos from aiomql.contrib.symbols import ForexSymbol +logger = logging.getLogger(__name__) async def test_bot(): logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") syms = ["BTCUSD", "SOLUSD", "ETHUSD"] symbols = [ForexSymbol(name=sym) for sym in syms] - strategies = [Chaos(symbol=symbol, name="test_chaos") for symbol in symbols] + strategies = [Chaos(symbol=symbol, name="test_chaos", params={"interval": 3}) for symbol in symbols] bot = Bot() - bot.executor.timeout = 5 + bot.executor.timeout = 10 bot.add_strategies(strategies=strategies) await bot.initialize() bot.executor.execute() assert len(bot.executor.coroutines) == 1 assert len(bot.executor.coroutine_threads) == 1 + assert len(bot.executor.strategy_runners) == 3 + assert bot.config.shutdown is True diff --git a/tests/live/integration/test_bot_sync.py b/tests/live/integration/test_bot_sync.py index b955be2..c3639ff 100644 --- a/tests/live/integration/test_bot_sync.py +++ b/tests/live/integration/test_bot_sync.py @@ -9,9 +9,8 @@ def test_bot_sync(): logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") syms = ["BTCUSD", "SOLUSD", "ETHUSD"] symbols = [ForexSymbol(name=sym) for sym in syms] - strategies = [Chaos(symbol=symbol, name="test_chaos") for symbol in symbols] + strategies = [Chaos(symbol=symbol, name="test_chaos", params={"interval": 3}) for symbol in symbols] bot = Bot() - bot.config.task_queue.worker_timeout = 2 bot.executor.timeout = 10 bot.add_strategies(strategies=strategies) bot.initialize_sync() diff --git a/tests/live/unit/test_task_queue.py b/tests/live/unit/test_task_queue.py index 31d0cf4..411b32f 100644 --- a/tests/live/unit/test_task_queue.py +++ b/tests/live/unit/test_task_queue.py @@ -6,7 +6,7 @@ from aiomql.core.task_queue import TaskQueue, QueueItem class TestTaskQueue: @classmethod def setup_class(cls): - cls.task_queue = TaskQueue(timeout=5, worker_timeout=1) + cls.task_queue = TaskQueue(absolute_timeout=6, worker_timeout=1) cls.data = {} async def task_one(self): @@ -27,14 +27,10 @@ class TestTaskQueue: async def test_queue(self): item_one = QueueItem(self.task_one) self.task_queue.add(item=item_one, must_complete=False) - assert len(self.task_queue.priority_tasks) == 0 assert self.task_queue.queue.qsize() == 1 self.task_queue.add(item=QueueItem(self.task_two), must_complete=True) - assert len(self.task_queue.priority_tasks) == 1 assert self.task_queue.queue.qsize() == 2 self.task_queue.add(item=QueueItem(self.task_three), must_complete=False) await self.task_queue.run() assert len(self.data["task_one"]) >= 2 assert len(self.data["task_two"]) == 10 - assert len(self.data["task_three"]) == 1 - assert len(self.task_queue.priority_tasks) == 0