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