testdata

src/aiomql/contrib/backtester/__init__.py

@@ -3,5 +3,4 @@ from .test_data import TestData
 from .get_data import GetData
 from .test_strategy import TestStrategy
 from .event_manager import EventManager
-from .strategy_tester import StrategyTester
-# from .test_executor import FingerTrapTest
+from .strategy_tester import StrategyTester, SingleStrategyTester

src/aiomql/contrib/backtester/check1.py

@@ -1,4 +1,75 @@
-def fun(a, b=6, *c, **d):
-    print(f"{a=}, {b=}, {c=}, {d=}")
+import asyncio
+import inspect
+from functools import cache, lru_cache, cached_property, wraps, partial
 
-fun(1, 3, 4, 5, six=6, seven=7)
+# from
+
+def async_cache(fun):
+
+    @wraps(fun)
+    async def wrapper(*args, **kwargs):
+        print(wrapper.cache)
+        key = (args, frozenset(kwargs.items()))
+        async with wrapper.lock:
+            if key not in wrapper.cache:
+                print('not in cache')
+                wrapper.cache[key] = await fun(*args, **kwargs)
+        return wrapper.cache[key]
+
+    wrapper.lock = asyncio.Lock()
+    wrapper.cache = {}
+    return wrapper
+
+
+class Test:
+
+    def __init__(self):
+        self.rr = 0
+
+    def __repr__(self):
+        return f'{self.__class__.__name__}(...)'
+
+    @async_cache
+    async def check(self, a, b):
+        an = a + b - self.rr
+        return an
+
+# async def main(a, b):
+#     t = Test()
+#
+#     @async_cache
+#     def check(_a, _b):
+#         an = _a + _b
+#         print('check', _a, _b, t.rr)
+#         return an
+#
+#     return check(a, b)
+
+t = Test()
+y = asyncio.run(t.check(1, 2))
+t.rr = 6
+y1 = asyncio.run(t.check(1, 2))
+t.rr = 7
+y2 = asyncio.run(t.check(1, 2))
+print(y, y1, y2)
+
+@async_cache
+async def func(a, b):
+    an = a + b
+    print('func')
+    return an
+
+@async_cache
+async def func1(e, c=6, d=6):
+    an = e - c + d
+    print('func1')
+    return an
+
+
+# asyncio.run(func(1, 2))
+# asyncio.run(func1(1, d=8))
+# asyncio.run(func(1, 2))
+# asyncio.run(func(1, 3))
+# asyncio.run(func1(1, d=6))
+# asyncio.run(func1(1, d=6))
+# asyncio.run(func1(1, d=7))

src/aiomql/contrib/backtester/event_manager.py

@@ -29,11 +29,9 @@ class EventManager:
         self.tasks.extend(tasks)
 
     def sigint_handler(self, sig, frame):
-        print('KeyboardInterrupt')
-        print(self.config.test_data.orders)
         for task in self.tasks:
             task.cancel()
-        # self.mt.cancel()
+        self.config.test_data.save()
 
     async def acquire(self):
         await self.condition.acquire()
@@ -49,10 +47,9 @@ class EventManager:
                     await self.config.test_data.tracker()
                     self.config.test_data.next()
                     self.condition.notify_all()
-                    if (timestamp := self.config.test_data.cursor.time % int(60 * 60 * 24)) == 0:
-                        print(f"Time: {datetime.fromtimestamp(timestamp)}")
-                    else:
-                        print(f"Time: {datetime.fromtimestamp(timestamp)}")
+                    if ((timestamp := self.config.test_data.cursor.time) % int(60 * 60 * 24)) == 0:
+                        print(f"if Time: {datetime.fromtimestamp(timestamp)}")
+
                     await asyncio.sleep(0)
 
     async def wait(self):

src/aiomql/contrib/backtester/get_data.py

@@ -47,7 +47,23 @@ class Data:
 
     _fields: list[ClassVar[str]] = field(default_factory=list)
 
-    def setattrs(self, **kwargs):
+    def __str__(self):
+        return f"""
+        Data: {self.name}
+        Terminal: {str(list(self.terminal.keys())[0:2]) + '...' if len(self.terminal) > 3 else list(self.terminal.keys())}
+        Version: {self.version}
+        Account: {str(list(self.account.keys())[0:2]) + '...' if len(self.account) > 3 else list(self.account.keys())}
+        Symbols: {len(self.symbols)} symbols
+        Prices: Prices for {len(self.prices)} symbols
+        Ticks: Ticks for {len(self.ticks)} symbols 
+        Rates: Bars for {len(self.rates)} symbols
+        Span: {datetime.fromtimestamp(self.span.start)} to {datetime.fromtimestamp(self.span.stop)}
+        """
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}({self.name})"
+
+    def set_attrs(self, **kwargs):
         [setattr(self, k, v) for k, v in kwargs.items() if k in self.fields]
 
     @property
@@ -71,19 +87,20 @@ class GetData:
         diff = int((self.end - self.start).total_seconds())
         self.range = range(diff)
         self.span = range(start := int(self.start.timestamp()), diff + start)
-        self.data = Data(name=name)
+        self.data = Data(name=name, span=self.span, range=self.range)
         self.mt5 = MetaTrader()
-        self.task_queue = TaskQueue()
+        self.task_queue = TaskQueue(workers=250)
 
     @classmethod
     def dump_data(cls, data: Data, name: str | Path, compress: bool = False):
         """"""
         try:
             fo = open(name, 'wb')
+
             if compress:
-                data = lzma.compress(pickle.dumps(data))
+                data = lzma.compress(pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL))
             else:
-                data = pickle.dumps(data)
+                data = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)
 
             fo.write(data)
             fo.close()
@@ -110,28 +127,39 @@ class GetData:
             logger.error(f"Error: {err}")
             return None
 
-    async def get_data(self):
+    async def get_data(self, workers: int = None):
         """"""
-        q_items = [QueueItem(self.get_symbol_rates, must_complete=True),
-                   QueueItem(self.get_symbol_ticks, must_complete=True),
-                   QueueItem(self.get_symbol_prices, must_complete=True),
-                   QueueItem(self.get_symbol_info, must_complete=True),
-                   QueueItem(self.get_account_info, must_complete=True),
+        if workers:
+            self.task_queue.workers = workers
+
+        q_items = [QueueItem(self.get_symbols_rates, must_complete=True),
+                   QueueItem(self.get_symbols_ticks, must_complete=True),
+                   QueueItem(self.get_symbols_prices, must_complete=True),
                    QueueItem(self.get_symbols_info, must_complete=True),
-                   QueueItem(self.get_version, must_complete=True),
-                   QueueItem(self.get_terminal_info, must_complete=True)]
+                   ]
+
         [self.task_queue.add(item=item, priority=0) for item in q_items]
+
+        if not self.data.account:
+            self.task_queue.add(item=QueueItem(self.get_account_info, must_complete=True))
+
+        if not self.data.terminal:
+            self.task_queue.add(item=QueueItem(self.get_terminal_info, must_complete=True))
+
+        if not self.data.version:
+            self.task_queue.add(item=QueueItem(self.get_version, must_complete=True))
+
         await self.task_queue.run()
 
     def pickle_data(self):
         """"""
-        fh = open(f'{self.config.test_data_dir}/{self.name}', 'wb')
-        pickle.dump(self.data, fh)
+        fh = open(f'{self.config.test_data_dir}/{self.name}.pkl', 'wb')
+        pickle.dump(self.data, fh, protocol=pickle.HIGHEST_PROTOCOL)
         fh.close()
 
     async def compress_data(self):
         """"""
-        bdata = pickle.dumps(self.data)
+        bdata = pickle.dumps(self.data, protocol=pickle.HIGHEST_PROTOCOL)
         name = self.name + 'xz'
         with lzma.open(f'{self.config.test_data_dir}/{name}', 'w') as fh:
             fh.write(bdata)
@@ -140,44 +168,48 @@ class GetData:
         """"""
         terminal = await self.mt5.terminal_info()
         terminal = terminal._asdict()
-        self.data.setattrs(terminal=terminal)
+        self.data.set_attrs(terminal=terminal)
 
     async def get_version(self):
         """"""
         version = await self.mt5.version()
-        self.data.setattrs(version=version)
+        self.data.set_attrs(version=version)
 
-    async def get_symbols_info(self):
-        """"""
-        [self.task_queue.add(item=QueueItem(self.get_symbol_info, symbol)) for symbol in self.symbols]
-
-    async def get_symbols_ticks(self):
-        """"""
-        [self.task_queue.add(item=QueueItem(self.get_symbol_ticks, symbol)) for symbol in self.symbols]
-
-    async def get_symbols_prices(self):
-        """"""
-        [self.task_queue.add(item=QueueItem(self.get_symbol_prices, symbol)) for symbol in self.symbols]
-
-    async def get_symbols_rates(self):
-        """"""
-        [self.task_queue.add(item=QueueItem(self.get_symbol_rates, symbol, timeframe), priority=4)
-                  for symbol in self.symbols for timeframe in self.timeframes]
-        
-    @backoff_decorator(max_retries=5)
+    @backoff_decorator
     async def get_account_info(self):
         """"""
         res = await self.mt5.account_info()
         res = res._asdict()
-        self.data.setattrs(account=res)
+        self.data.set_attrs(account=res)
 
-    @backoff_decorator(max_retries=5)
+    async def get_symbols_info(self):
+        """"""
+        [self.task_queue.add(item=QueueItem(self.get_symbol_info, symbol))
+         for symbol in self.symbols if self.data.symbols.get(symbol) is None]
+
+    async def get_symbols_ticks(self):
+        """"""
+        [self.task_queue.add(item=QueueItem(self.get_symbol_ticks, symbol))
+         for symbol in self.symbols if self.data.ticks.get(symbol) is None]
+
+    async def get_symbols_prices(self):
+        """"""
+        [self.task_queue.add(item=QueueItem(self.get_symbol_prices, symbol))
+         for symbol in self.symbols if self.data.prices.get(symbol) is None]
+
+    async def get_symbols_rates(self):
+        """"""
+        [self.task_queue.add(item=QueueItem(self.get_symbol_rates, symbol, timeframe), priority=4)
+                  for symbol in self.symbols for timeframe in self.timeframes
+         if self.data.rates.get(symbol, {}).get(timeframe.name) is None]
+
+    @backoff_decorator
     async def get_symbol_info(self, symbol: str):
         """"""
         res = await self.mt5.symbol_info(symbol)
         self.data.symbols[symbol] = res._asdict()
 
-    @backoff_decorator(max_retries=5)
+    @backoff_decorator
     async def get_symbol_ticks(self, symbol: str):
         """"""
         res = await self.mt5.copy_ticks_range(symbol, self.start, self.end, CopyTicks.ALL)
@@ -186,17 +218,17 @@ class GetData:
         res.set_index('time', inplace=True, drop=False)
         self.data.ticks[symbol] = res
 
-    @backoff_decorator(max_retries=5)
+    @backoff_decorator
     async def get_symbol_prices(self, symbol: str):
         """"""
         res = await self.mt5.copy_ticks_range(symbol, self.start, self.end, CopyTicks.ALL)
         res = pd.DataFrame(res)
         res.drop_duplicates(subset=['time'], keep='last', inplace=True)
         res.set_index('time', inplace=True, drop=False)
-        res = res.reindex(self.span, method='nearest')
+        res = res.reindex(self.span) # change method back to 'nearest'
         self.data.prices[symbol] = res
 
-    @backoff_decorator(max_retries=5)
+    @backoff_decorator
     async def get_symbol_rates(self, symbol: str, timeframe: TimeFrame):
         """"""
         res = await self.mt5.copy_rates_range(symbol, timeframe, self.start, self.end)

src/aiomql/contrib/backtester/meta_tester.py

@@ -20,20 +20,19 @@ class MetaTester(MetaTrader):
 
     def __init__(self, test_data: TestData = None):
         super().__init__()
-        if self.test_data:
+        if test_data is not None:
             self.config.test_data = test_data
 
     @property
     def test_data(self) -> TestData | None:
-        test_data = self.config.test_data
-        if test_data is None:
-            ...
-            # logger.error('No Test Data Available')
-        return test_data
+        return self.config.test_data
 
     @test_data.setter
     def test_data(self, value: TestData):
         self.config.test_data = value
+
+    async def last_error(self) -> tuple[int, str]:
+        return -1, ''
     
     async def initialize(self, path: str = "", login: int = 0, password: str = "", server: str = "", 
                    timeout: int | None = None, portable=False, load_test_data: bool = False,
@@ -63,16 +62,10 @@ class MetaTester(MetaTrader):
     async def shutdown(self) -> None:
         await super().shutdown() if self.config.use_terminal_for_backtesting else ...
 
-        # self.test_data.save()
-        # name = self.test_data.data.name
-        # if self.config.compress_test_data:
-        #     name += '.xz'
-        # name = self.config.test_data_dir/name
-        # GetData.dump_data(data=self.test_data.data, name=name, compress=self.config.compress_test_data)
-
     @error_handler(msg='test data not available', exe=AttributeError)
     async def terminal_info(self) -> TerminalInfo:
-        return self.test_data.get_terminal_info()
+        res = await self.test_data.get_terminal_info()
+        return res
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def account_info(self) -> AccountInfo:
@@ -81,49 +74,62 @@ class MetaTester(MetaTrader):
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def symbol_select(self, symbol: str, enable: bool = True) -> bool:
-        return symbol in self.test_data.symbols and enable
+        if self.config.use_terminal_for_backtesting:
+            res = await super().symbol_select(symbol, enable)
+            return res
+        res = (symbol in self.test_data.symbols) and enable
+        return res
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def symbols_total(self) -> int:
-        return self.test_data.get_symbols_total()
+        tot = await self.test_data.get_symbols_total()
+        return tot
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def symbols_get(self, group: str = "") -> tuple[SymbolInfo, ...] | None:
         """"""
-        return self.test_data.get_symbols(group)
+        syms = await self.test_data.get_symbols(group)
+        return syms
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def symbol_info(self, symbol: str) -> SymbolInfo | None:
-        return self.test_data.symbols.get(symbol)
+        sym = await self.test_data.get_symbol_info(symbol)
+        return sym
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def symbol_info_tick(self, symbol: str) -> Tick | None:
-        return self.test_data.get_symbol_info_tick(symbol)
+        tick = await self.test_data.get_symbol_info_tick(symbol)
+        return tick
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def copy_rates_from(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float,
                               count: int) -> ndarray | None:
-        return self.test_data.get_rates_from(symbol, timeframe, date_from, count)
+        rates = await self.test_data.get_rates_from(symbol, timeframe, date_from, count)
+        return rates
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def copy_rates_from_pos(self, symbol: str, timeframe: TimeFrame, start_pos: int,
                                   count: int) -> ndarray | None:
-        return self.test_data.get_rates_from_pos(symbol, timeframe, start_pos, count)
+        rates = await self.test_data.get_rates_from_pos(symbol, timeframe, start_pos, count)
+        return rates
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def copy_rates_range(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float,
                                date_to: datetime | float) -> ndarray | None:
-        return self.test_data.get_rates_range(symbol, timeframe, date_from, date_to)
+        rates = await self.test_data.get_rates_range(symbol, timeframe, date_from, date_to)
+        return rates
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def copy_ticks_from(self, symbol: str, date_from: datetime | float, count: int,
                               flags: CopyTicks) -> ndarray | None:
-        return self.test_data.get_ticks_from(symbol, date_from, count, flags)
+        ticks = await self.test_data.get_ticks_from(symbol, date_from, count, flags)
+        return ticks
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def copy_ticks_range(self, symbol: str, date_from: datetime | float, date_to: datetime | float,
                                flags: CopyTicks) -> ndarray | None:
-        return self.test_data.get_ticks_range(symbol, date_from, date_to, flags)
+        ticks = await self.test_data.get_ticks_range(symbol, date_from, date_to, flags)
+        return ticks
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def orders_total(self) -> int:
@@ -135,23 +141,26 @@ class MetaTester(MetaTrader):
         return self.test_data.get_orders(**kwargs)
 
     @error_handler(msg='test data not available', exe=AttributeError)
-    async def order_calc_margin(self, action: OrderType, symbol: str, volume: float,
-                                price: float, use_terminal: bool = True) -> float | None:
-        res = await self.test_data.order_calc_margin(action, symbol, volume, price, use_terminal=use_terminal)
+    async def order_calc_margin(self, action: OrderType, symbol: str, volume: float, price: float) -> float | None:
+        res = await self.test_data.order_calc_margin(action, symbol, volume, price)
         return res
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def order_calc_profit(self, action: OrderType, symbol: str, volume: float, price_open: float,
-                                price_close: float, use_terminal: bool = True) -> float | None:
-        return await self.test_data.order_calc_profit(action, symbol, volume,
-                                                      price_open, price_close, use_terminal=use_terminal)
+                                price_close: float) -> float | None:
+
+        profit = await self.test_data.order_calc_profit(action, symbol, volume,
+                                                      price_open, price_close)
+        return profit
 
     @error_handler(msg='test data not available', exe=AttributeError)
-    async def order_check(self, request: dict, use_terminal: bool = True) -> OrderCheckResult:
-        return await self.test_data.order_check(request, use_terminal=use_terminal)
+    async def order_check(self, request: dict) -> OrderCheckResult:
+        ocr = await self.test_data.order_check(request)
+        return ocr
 
-    async def order_send(self, request: dict, use_terminal: bool = True) -> OrderSendResult:
-        return await self.test_data.order_send(request, use_terminal=use_terminal)
+    async def order_send(self, request: dict) -> OrderSendResult:
+        osr = await self.test_data.order_send(request)
+        return osr
 
     @error_handler(msg='test data not available', exe=AttributeError)
     async def positions_total(self) -> int:

src/aiomql/contrib/backtester/strategy_tester.py

@@ -1,42 +1,55 @@
 import asyncio
 import signal
+from logging import getLogger
 
 from .event_manager import EventManager
-from .get_data import GetData
 from .test_data import TestData
 from .meta_tester import MetaTester
-
+from .test_strategy import TestStrategy, TestSingleStrategy
 from ...core import Config
 
+logger = getLogger(__name__)
+
+
 class StrategyTester:
-    def __init__(self, *, strategies: list = None, test_data: TestData = None, test_data_file: str = ''):
-        self.config = Config()
-        self.mt5 = MetaTester()
+    def __init__(self, *, strategies: list[TestStrategy] = None):
         self.strategies = strategies or []
-        self.test_data = test_data or self.get_test_data(name=test_data_file)
-        self.config.test_data = self.test_data
         self.event_manager = EventManager(num_tasks=len(self.strategies))
         signal.signal(signal.SIGINT, self.event_manager.sigint_handler)
 
-    def get_test_data(self, name: str) -> TestData | None:
-        name = f"{self.config.test_data_dir_name}/{name or self.config.test_data_file}"
-        data = GetData.load_data(name=name, compressed=self.config.compress_test_data)
-        return TestData(data) if data is not None else None
-
-    async def start(self):
-        acc = self.config.account_info()
-        await self.mt5.initialize(**acc)
-        await self.mt5.login(**acc)
-
-    async def run(self):
+    async def run(self, test_data: TestData):
         try:
-            await self.start()
-            tasks = [*[asyncio.create_task(strategy.test()) for strategy in self.strategies],
+            config = Config()
+            config.test_data = test_data
+            mt5 = MetaTester()
+            acc = config.account_info()
+            await mt5.initialize(**acc)
+            await mt5.login(**acc)
+            strategies = [strategy for strategy in self.strategies if await strategy.symbol.init()]
+            print(f"Number of strategies: {len(strategies)}")
+            self.event_manager.num_main_tasks = len(strategies)
+            tasks = [*[asyncio.create_task(strategy.test()) for strategy in strategies],
                      asyncio.create_task(self.event_manager.event_monitor())]
             self.event_manager.add_task(*tasks)
-            await asyncio.gather(*tasks, return_exceptions=True)
-            # self.mt = asyncio.create_task(asyncio.gather(*tasks))
+            await asyncio.gather(*tasks, return_exceptions=True) if strategies else ...
         except Exception as err:
-            print(f"Error {err} occurred in StrategyTester")
-        finally:
-            await self.mt5.shutdown()
+            logger.error(f"Error {err} occurred in StrategyTester")
+
+
+class SingleStrategyTester:
+    def __init__(self, *, strategy: TestSingleStrategy):
+        self.strategy = strategy
+
+    async def run(self, test_data: TestData):
+        try:
+            config = Config()
+            config.test_data = test_data
+            mt5 = MetaTester()
+            acc = config.account_info()
+            await mt5.initialize(**acc)
+            await mt5.login(**acc)
+            sym = self.strategy.symbol
+            res = await sym.init()
+            await self.strategy.test() if res else ...
+        except Exception as err:
+            logger.error(f"Error {err} occurred in SingleStrategyTester")

src/aiomql/contrib/backtester/test_account.py

@@ -1,4 +1,6 @@
-from dataclasses import dataclass, asdict
+from dataclasses import dataclass, asdict, field, fields
+from typing import ClassVar
+
 from ...core.constants import AccountTradeMode, AccountMarginMode, AccountStopOutMode
 
 
@@ -33,5 +35,16 @@ class AccountInfo:
     name: str = ''
     company: str = ''
 
+    _fields: list[ClassVar[str]] = field(default_factory=list)
+
     def asdict(self):
-        return asdict(self)
+        res = asdict(self)
+        res.pop('_fields', None)
+        return res
+
+    def set_attrs(self, **kwargs):
+        [setattr(self, k, v) for k, v in kwargs.items() if k in self.fields]
+
+    @property
+    def fields(self):
+        return self._fields or [name for f in fields(self) if (name := f.name) != '_fields']

src/aiomql/contrib/backtester/test_data.py

@@ -4,6 +4,8 @@ from datetime import datetime
 from typing import Literal
 from itertools import zip_longest
 import random
+import json
+from functools import cached_property
 
 import pandas as pd
 import pytz
@@ -14,10 +16,10 @@ from MetaTrader5 import (Tick, SymbolInfo, AccountInfo, TradeOrder, TradePositio
 
 from ...core.meta_trader import MetaTrader
 from ...core.constants import TimeFrame, CopyTicks, OrderType, TradeAction, AccountStopOutMode
-from .get_data import Data
+from ...core.config import Config
+from .get_data import Data, GetData
 from .test_account import AccountInfo as Account
-from ...utils import round_down, round_up, error_handler, error_handler_sync
-# from .event_manager import EventManager
+from ...utils import round_down, round_up, error_handler, error_handler_sync, async_cache
 
 tz = pytz.timezone('Etc/UTC')
 Cursor = namedtuple('Cursor', ['index', 'time'])
@@ -27,69 +29,103 @@ class TestData:
     history_orders: DataFrame
     history_deals: DataFrame
     
-    def __init__(self, data: Data):
-        self._data = data
-        self.version: tuple[int, int, str] = data.version
-        self.terminal_info = TerminalInfo(data.terminal)
-        self.account: Account = Account(**data.account)
-        self.symbols: dict[str, SymbolInfo] = {symbol: SymbolInfo(info) for symbol, info in data.symbols.items()}
-        self.prices: dict[str, DataFrame] = data.prices
-        self.ticks: dict[str, DataFrame] = data.ticks
-        self.rates: dict[str, dict[str, DataFrame]] = data.rates
-        self.span: range = data.span
-        self.range: range = data.range
+    def __init__(self, data: Data = None, speed: int = 1, start: float | datetime = 0, end: float | datetime = 0):
+        self._data = data or Data()
+        self._account: Account = Account(**self._data.account)
+        self.prices: dict[str, DataFrame] = self._data.prices
+        self.ticks: dict[str, DataFrame] = self._data.ticks
+        self.rates: dict[str, dict[str, DataFrame]] = self._data.rates
+        span_start = (int(start.timestamp()) if isinstance(start, datetime) else int(start)) or self._data.span.start
+        span_end = (int(end.timestamp()) if isinstance(end, datetime) else int(end)) or self._data.span.stop
+        self.span: range = range(span_start, span_end, speed)
+        self.range: range = range(0, span_end - span_start, speed)
         self.orders: dict[str, dict[int, TradeOrder]] = {}
         self.deals: dict[str, dict[int, TradeDeal]] = {}
         self.open_orders: dict[int, TradeOrder] = {}
         self.positions: dict[str, dict[int, TradePosition]] = {}
         self.open_positions: dict[int, TradePosition] = {}
-        self.history_orders = data.history_orders
-        self.history_deals = data.history_deals
+        self.history_orders = self._data.history_orders
+        self.history_deals = self._data.history_deals
         self.margins: dict[int, float] = {}
         self.mt5 = MetaTrader()
         self.iter = zip_longest(self.range, self.span)
-        self.cursor = next(self)
-        # self.event_manager = EventManager()
+        self.cursor: Cursor = Cursor(index=self.range.start, time=self.span.start)
+        self.config = Config()
+        self._data.name = self._data.name or f"{datetime.fromtimestamp(span_start):%d-%m-%y}_{datetime.fromtimestamp(span_end):%d-%m-%y}"
+        self.fh = open(f'{self.config.test_data_dir}/data.json', 'a')
 
     def __next__(self) -> Cursor:
         index, time = next(self.iter)
         self.cursor = Cursor(index=index, time=time)
         return self.cursor
 
+    def __repr__(self):
+        return f"{self.__class__.__name__}()"
+
     def next(self) -> Cursor:
         return next(self)
 
     @property
     def data(self):
         return self._data
+
+    def to_json(self, data):
+        json.dump(data, self.fh)
     
     def reset(self):
         self.iter = zip_longest(self.range, self.span)
-        self.cursor = Cursor(index=self.range[0], time=self.span[0])
-        return self.cursor
+        self.cursor = Cursor(index=self.range.start, time=self.span.start)
 
-    def go_to(self, index: int, time: int):
-        range_ = range(time, self.range.stop, self.range.step)
-        span = range(index, self.span.stop, self.span.step)
+    def go_to(self, time: datetime | int):
+        time =  int(time.timestamp()) if isinstance(time, datetime) else int(time)
+        steps = time - self.cursor.time
+        if steps > 0:
+            self.fast_forward(steps)
+            return
+        span = range(time, self.span.stop, self.span.step)
+        start = span.start - self.span.start
+        range_ = range(start, self.range.stop, self.range.step)
         self.iter = zip_longest(range_, span)
-        self.cursor = next(self)
+        self.cursor = Cursor(index=range_.start, time=span.start)
+
+    def fast_forward(self, steps: int):
+        for _ in range(steps):
+            self.next()
 
     def get_dtype(self, df: DataFrame) -> list[tuple[str, str]]:
         return [(c, t) for c, t in zip(df.columns, df.dtypes)]
 
+    @async_cache
+    async def get_price_tick(self, symbol, time: int) -> Tick | None:
+        if self.config.use_terminal_for_backtesting:
+            tick = await self.mt5.copy_ticks_from(symbol, time, 1, CopyTicks.ALL)
+            return Tick(tick[-1]) if tick else None
+        return self.prices[symbol].loc[self.cursor.time]
+
     async def tracker(self):
         pos_tasks = [self.check_position(ticket) for ticket in self.open_positions]
         await asyncio.gather(*pos_tasks)
         order_tasks = [self.check_order(ticket) for ticket in self.open_orders]
         await asyncio.gather(*order_tasks)
+        profit = sum(pos.profit for pos in self.open_positions.values())
+        self.update_account(profit=profit)
 
     def save(self):
-        for symbol in self.orders:
-            self.history_orders = pd.concat([DataFrame(self.orders[symbol].values()), self.history_orders])
-        self._data.history_orders = self.history_orders
-        for symbol in self.deals:
-            self.history_deals = pd.concat([DataFrame(self.deals[symbol].values()), self.history_deals])
-        self._data.history_deals = self.history_deals
+        self.fh.close()
+        try:
+            if len(self.orders) or len(self.deals):
+                for symbol in self.orders:
+                    self.history_orders = pd.concat([DataFrame(self.orders[symbol].values()), self.history_orders])
+                self._data.history_orders = self.history_orders
+
+                for symbol in self.deals:
+                    self.history_deals = pd.concat([DataFrame(self.deals[symbol].values()), self.history_deals])
+                self._data.history_deals = self.history_deals
+
+                path = self.config.test_data_dir/f"{self._data.name}.pkl"
+                GetData.dump_data(data=self._data, name=path, compress=self.config.compress_test_data)
+        except Exception as err:
+            print(err)
 
     @error_handler
     async def check_order(self, ticket: int):
@@ -116,7 +152,6 @@ class TestData:
         tick = self.prices[symbol].loc[self.cursor.time]
         price_current = tick.bid if order_type == OrderType.BUY else tick.ask
         profit = await self.order_calc_profit(order_type, symbol, volume, price_open, price_current, use_terminal)
-        self.update_account(equity=profit - prev_profit)
         pos = pos._asdict()
         pos.update(profit=profit, price_current=price_current, time_update=self.cursor.time)
         pos = TradePosition(pos)
@@ -129,7 +164,7 @@ class TestData:
         order = self.open_orders.pop(ticket)
         order = order._asdict()
         order.update(time_done=self.cursor.time)
-        self.update_account(profit=position.profit, margin=-margin)
+        self.update_account(gain=position.profit, margin=-margin)  # ToDo: Create a deal object here? modify update account
 
     def modify_stops(self, ticket: int, sl: int = None, tp: int = None):
         pos = self.open_positions.pop(ticket)
@@ -149,18 +184,45 @@ class TestData:
         self.positions[pos.symbol][ticket] = pos
         self.orders[order.symbol][ticket] = order
 
-    def update_account(self, *, profit: float = 0, margin: float = 0, equity: float = 0):
-        self.account.balance += profit
-        self.account.equity += equity
-        self.account.margin += margin
-        self.account.margin_free = self.account.equity - self.account.margin
-        self.account.margin_level = (self.account.equity / (self.account.margin or 1)) * 100 \
-            if self.account.margin_mode == AccountStopOutMode.PERCENT else self.account.margin_free
+    def update_account(self, *, profit: float = None, margin: float = 0, gain: float = 0):
+        self._account.balance += gain
+        self._account.profit = profit if profit is not None else self._account.profit
+        self._account.equity = self._account.balance + self._account.profit
+        self._account.margin += margin
+        self._account.margin_free = self._account.equity - self._account.margin
+
+        if self._account.margin == 0:
+            self._account.margin_level = 0
+        else:
+            mode = self._account.margin_mode
+            level = self._account.equity / self._account.margin * 100
+            self._account.margin_level = level if mode == AccountStopOutMode.PERCENT else self._account.margin_free
+
+    def deposit(self, amount: float):
+        self.update_account(gain=amount)
+
+    def withdraw(self, amount: float):
+        self.update_account(gain=-amount)
+
+    @error_handler
+    async def setup_account(self):
+        if self.config.use_terminal_for_backtesting:
+            acc = self._account
+            default = {'profit': acc.profit, 'margin': acc.margin, 'equity': acc.equity, 'margin_free': acc.margin_free,
+                       'margin_level': acc.margin_level, 'balance': acc.balance}
+            acc = await self.mt5.account_info()
+            acc = acc._asdict() | default
+            self._account.set_attrs(**acc)
+
+    @cached_property
+    def symbols(self) -> dict[str, SymbolInfo]:
+        return {symbol: SymbolInfo(info.values()) for symbol, info in self._data.symbols.items()}
 
     @error_handler
     async def order_send(self, request: dict, use_terminal: bool = True) -> OrderSendResult:
+        print('sending orders')
+        ticket = random.randint(100_000_000, 999_999_999)
         osr = {'retcode': 10009, 'comment': 'Request completed', 'request': TradeRequest(request)}
-
         if (position := request.get('position')) in self.open_positions:
             pos = self.open_positions[position]
             order_type = OrderType(request['type'])
@@ -168,6 +230,7 @@ class TestData:
             if order_type.opposite == pos_type:  # ToDo: is there another way to check if the order is a close order?
                 # close position
                 self.close_position(pos.ticket)
+                self.to_json(osr) # ToDo: remove later
                 return OrderSendResult(osr)  # ToDo: Create a deal object here
             action = request['action']
             if action == TradeAction.SLTP:
@@ -178,9 +241,9 @@ class TestData:
             ocr = await self.order_check(request, use_terminal=use_terminal)
             if ocr.retcode != 0:
                 osr.update({'comment': ocr.comment, 'retcode': ocr.retcode})
+                self.to_json(osr) # ToDo: remove later
                 return OrderSendResult(osr)
 
-            ticket = random.randint(100_000_000, 999_999_999)
             deal_ticket = random.randint(100_000_000, 999_999_999)
             tick = self.get_symbol_info_tick(request['symbol'])
             order_type = request['type']
@@ -210,6 +273,7 @@ class TestData:
             margin = await self.order_calc_margin(action, symbol, volume, price, use_terminal=use_terminal)
             self.margins[ticket] = margin
             self.update_account(margin=margin)
+            self.to_json(osr) # ToDo: remove later
             return OrderSendResult(osr)
 
     @error_handler
@@ -264,104 +328,138 @@ class TestData:
 
         return OrderCheckResult(ocr)
 
-    @error_handler_sync
-    def get_terminal_info(self) -> TerminalInfo:
-        return self.terminal_info
+    @error_handler
+    async def get_terminal_info(self) -> TerminalInfo:
+        if self.config.use_terminal_for_backtesting:
+            res = await self.mt5.terminal_info()
+            return res
+        return TerminalInfo(self._data.terminal)
 
-    @error_handler_sync
-    def get_version(self) -> tuple[int, int, str]:
-        return self.version
+    @error_handler
+    async def get_version(self) -> tuple[int, int, str]:
+        if self.config.use_terminal_for_backtesting:
+            res = await self.mt5.version()
+            return res
+        return self._data.version
 
-    @error_handler_sync
-    def get_symbols_total(self) -> int:
+    @error_handler
+    async def get_symbols_total(self) -> int:
+        if self.config.use_terminal_for_backtesting:
+            syms = await self.mt5.symbols_total()
+            return syms
         return len(self.symbols)
 
-    @error_handler_sync
-    def get_symbols(self, group: str = '') -> tuple[SymbolInfo, ...]:
+    @error_handler
+    async def get_symbols(self, group: str = '') -> tuple[SymbolInfo, ...]:
+        if self.config.use_terminal_for_backtesting:
+            syms = await self.mt5.symbols_get(group=group)
+            return syms
         return tuple(list(self.symbols.values()))
 
     @error_handler_sync
     def get_account_info(self) -> AccountInfo:
-        return AccountInfo(self.account.asdict())
+        return AccountInfo(self._account.asdict().values())
 
-    @error_handler_sync
-    def get_symbol_info_tick(self, symbol: str) -> Tick:
-        tick = self.prices[symbol].iloc[self.cursor.index]
-        return Tick(tick)
+    @error_handler
+    async def get_symbol_info_tick(self, symbol: str) -> Tick | None:
+        tick = await self.get_price_tick(symbol, self.cursor.time)
+        return tick
+
+    @error_handler
+    async def get_symbol_info(self, symbol: str) -> SymbolInfo:
+        if self.config.use_terminal_for_backtesting:
+            info = await self.mt5.symbol_info(symbol)
+            return info
 
-    @error_handler_sync
-    def get_symbol_info(self, symbol: str) -> SymbolInfo:
         info = self.symbols[symbol]
-        tick = self.get_symbol_info_tick(symbol)
+        tick = await self.get_symbol_info_tick(symbol)
         info = info._asdict()
         info |= {'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)
 
-    @error_handler_sync
-    def get_rates_from(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float, count: int) -> np.ndarray:
+    @error_handler
+    async def get_rates_from(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float, count: int) -> np.ndarray:
+        if self.config.use_terminal_for_backtesting:
+            rates = await self.mt5.copy_rates_from(symbol, timeframe, date_from, count)
+            return rates
+
         rates = self.rates[symbol][timeframe.name]
         start = int(datetime.timestamp(date_from)) if isinstance(date_from, datetime) else int(date_from)
         start = round_down(start, timeframe.time)
-        start = rates[rates.index <= start].iloc[-1].name
-        start = rates.index.get_loc(start)
-        end = start + count
-        return np.fromiter((tuple(i) for i in rates.iloc[start:end].iloc), dtype=self.get_dtype(rates))
+        rates = rates[rates.time <= start].iloc[-count:]
+        return np.fromiter((tuple(i) for i in rates.iloc), dtype=self.get_dtype(rates))
+
+    @error_handler
+    async def get_rates_from_pos(self, symbol: str, timeframe: TimeFrame, start_pos: int, count: int) -> np.ndarray:
+        if self.config.use_terminal_for_backtesting:
+            now = datetime.now(tz=tz)
+            b_now = self.cursor.time
+            diff = (now.timestamp() - b_now) // timeframe.time
+            start_pos = int(diff + start_pos)
+            res = await self.mt5.copy_rates_from_pos(symbol, timeframe, start_pos, count)
+            return res
 
-    @error_handler_sync
-    def get_rates_from_pos(self, symbol: str, timeframe: TimeFrame, start_pos: int, count: int) -> np.ndarray:
         rates = self.rates[symbol][timeframe.name]
-        end = -start_pos + count
-        end = end or None
-        return np.fromiter((tuple(i) for i in rates.iloc[-start_pos:end].iloc), dtype=self.get_dtype(rates))
+        end = abs(self.cursor.index - start_pos)
+        start = abs(end - count)
+        rates = rates.iloc[start:end]
+        return np.fromiter((tuple(i) for i in rates.iloc), dtype=self.get_dtype(rates))
+
+    @error_handler
+    async def get_rates_range(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float, date_to: datetime | float) -> np.ndarray:
+        if self.config.use_terminal_for_backtesting:
+            rates = await self.mt5.copy_rates_range(symbol, timeframe, date_from, date_to)
+            return rates
 
-    @error_handler_sync
-    def get_rates_range(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float, date_to: datetime | float) -> np.ndarray:
         rates = self.rates[symbol][timeframe.name]
         start = int(datetime.timestamp(date_from)) if isinstance(date_from, datetime) else int(date_from)
         start = round_down(start, timeframe.time)
-        start = rates[rates.index <= start].iloc[-1].name
         end = int(datetime.timestamp(date_to)) if isinstance(date_to, datetime) else int(date_to)
         end = round_up(end, timeframe.time)
-        end = rates[rates.index >= end].iloc[-1].name
-        return np.fromiter((tuple(i) for i in rates.loc[start:end].iloc), dtype=self.get_dtype(rates))
+        rates = rates[(rates.time >= start) & (rates.time <= end)]
+        return np.fromiter((tuple(i) for i in rates.iloc), dtype=self.get_dtype(rates))
+
+    @error_handler
+    async def get_ticks_from(self, symbol: str, date_from: datetime | float, count: int, flags: CopyTicks) -> np.ndarray:
+        if self.config.use_terminal_for_backtesting:
+            ticks = await self.mt5.copy_ticks_from(symbol, date_from, count, flags)
+            return ticks
 
-    @error_handler_sync
-    def get_ticks_from(self, symbol: str, date_from: datetime | float, count: int, flags: CopyTicks) -> np.ndarray:
         ticks = self.ticks[symbol]
         start = int(datetime.timestamp(date_from)) if isinstance(date_from, datetime) else int(date_from)
-        start = ticks[ticks.index <= start].iloc[-1].name
-        start = ticks.index.get_loc(start)
-        end = start + count
-        return np.fromiter((tuple(i) for i in ticks.iloc[start:end].iloc), dtype=self.get_dtype(ticks))
+        ticks = ticks[ticks.time <= start].iloc[-count:]
+        return np.fromiter((tuple(i) for i in ticks.iloc), dtype=self.get_dtype(ticks))
 
-    @error_handler_sync
-    def get_ticks_range(self, symbol: str, date_from: datetime | float, date_to: datetime | float, flags) -> np.ndarray:
+    @error_handler
+    async def get_ticks_range(self, symbol: str, date_from: datetime | float, date_to: datetime | float, flags) -> np.ndarray:
         ticks = self.ticks[symbol]
         start = int(datetime.timestamp(date_from)) if isinstance(date_from, datetime) else int(date_from)
-        start = ticks[ticks.index <= start].iloc[-1].index
         end = int(datetime.timestamp(date_to)) if isinstance(date_to, datetime) else int(date_to)
-        end = ticks[ticks.index >= end].iloc[-1].index
-        return np.fromiter((tuple(i) for i in ticks.loc[start:end].iloc), dtype=self.get_dtype(ticks))
+        ticks = ticks[(ticks.time >= start) & (ticks.time <= end)]
+        return np.fromiter((tuple(i) for i in ticks.iloc), dtype=self.get_dtype(ticks))
 
     @error_handler
     async def order_calc_margin(self, action: Literal[OrderType.BUY, OrderType.SELL], symbol: str, volume: float,
-                                price: float, use_terminal=False):
-        if use_terminal and self.mt5.config.use_terminal_for_backtesting:
+                                price: float):
+        if self.mt5.config.use_terminal_for_backtesting:
             return await self.mt5.order_calc_margin(action, symbol, volume, price)
+
         sym = self.symbols[symbol]
-        margin = (volume * sym.trade_contract_size * price) / (self.account.leverage / (sym.margin_initial or 1))
-        return round(margin, self.account.currency_digits)
+        margin = (volume * sym.trade_contract_size * price) / (self._account.leverage / (sym.margin_initial or 1))
+        return round(margin, self._account.currency_digits)
 
     @error_handler
     async def order_calc_profit(self, action: Literal[OrderType.BUY, OrderType.SELL], symbol: str, volume: float,
-                                price_open: float, price_close: float, use_terminal=True):
-        if use_terminal and self.mt5.config.use_terminal_for_backtesting:
+                                price_open: float, price_close: float):
+
+        if self.mt5.config.use_terminal_for_backtesting:
             return await self.mt5.order_calc_profit(action, symbol, volume, price_open, price_close)
+
         sym = self.symbols[symbol]
         profit = (volume * sym.trade_contract_size *
                   ((price_close - price_open) if action == OrderType.BUY else (price_open - price_close)))
-        return round(profit, self.account.currency_digits)
+        return round(profit, self._account.currency_digits)
 
     @error_handler_sync
     def get_orders_total(self) -> int:
@@ -369,6 +467,7 @@ class TestData:
 
     @error_handler_sync
     def get_orders(self, symbol: str = '', group: str = '', ticket: int = None) -> tuple[TradeOrder, ...]:
+
         if ticket:
             order = self.open_orders.get(ticket)
             return (order,) if order else ()
@@ -405,18 +504,15 @@ class TestData:
     def get_history_orders_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
         start =  int(date_from.timestamp()) if isinstance(date_from, datetime) else int(date_from)
         end = int(date_to.timestamp()) if isinstance(date_to, datetime) else int(date_to)
-        start = self.history_orders[self.history_orders.index >= start].iloc[0].name
-        end = self.history_orders[self.history_orders.index <= end].iloc[-1].name
-        return self.history_orders.loc[start:end].shape[0]
+        orders = self.history_orders[self.history_orders.time >= start & self.history_orders.time <= end]
+        return orders.shape[0]
 
     @error_handler_sync
     def get_history_orders(self, date_from: datetime | float, date_to: datetime | float, group: str = '',
                            ticket: int = None, position: int = None) -> tuple[TradeOrder, ...]:
         start =  int(date_from.timestamp()) if isinstance(date_from, datetime) else int(date_from)
         end = int(date_to.timestamp()) if isinstance(date_to, datetime) else int(date_to)
-        start = self.history_orders[self.history_orders.index >= start].iloc[0].name
-        end = self.history_orders[self.history_orders.index <= end].iloc[-1].name
-        orders = self.history_orders.loc[start:end]
+        orders = self.history_orders[self.history_orders.time >= start & self.history_orders.time <= end]
 
         if ticket:
             orders = orders[orders.ticket == ticket]
@@ -426,26 +522,21 @@ class TestData:
 
         elif group:
             ...
-
-        orders.drop(columns=['symbol'], inplace=True)
         return tuple(TradeOrder(order) for order in orders.iloc)
 
     @error_handler_sync
     def get_history_deals_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
         start =  int(date_from.timestamp()) if isinstance(date_from, datetime) else int(date_from)
         end = int(date_to.timestamp()) if isinstance(date_to, datetime) else int(date_to)
-        start = self.history_deals[self.history_deals.index >= start].iloc[0].name
-        end = self.history_deals[self.history_deals.index <= end].iloc[-1].name
-        return self.history_deals.loc[start:end].shape[0]
+        deals = self.history_deals[self.history_deals.time >= start & self.history_deals.time <= end]
+        return deals.shape[0]
 
     @error_handler_sync
     def get_history_deals(self, date_from: datetime | float, date_to: datetime | float, group: str = '',
                           position: int = None, ticket: int = None) -> tuple[TradeDeal, ...]:
         start = int(date_from.timestamp()) if isinstance(date_from, datetime) else int(date_from)
         end = int(date_to.timestamp()) if isinstance(date_to, datetime) else int(date_to)
-        start = self.history_deals[self.history_deals.index >= start].iloc[0].name
-        end = self.history_deals[self.history_deals.index <= end].iloc[-1].name
-        deals = self.history_deals.loc[start:end]
+        deals = self.history_deals[self.history_deals.time >= start & self.history_deals.time <= end]
 
         if ticket:
             deals = deals[deals.ticket == ticket]
@@ -455,5 +546,5 @@ class TestData:
 
         elif group:
             ...
-
+        
         return tuple(TradeDeal(deal) for deal in deals.iloc)

src/aiomql/contrib/backtester/test_strategy.py

@@ -1,14 +1,17 @@
-from .event_manager import EventManager
+from typing import TypeVar
 
+from .event_manager import EventManager
 from ...core.config import Config
 
+Symbol = TypeVar("Symbol")
+
 
 class TestStrategy:
     event_manager: EventManager
     config: Config
+    symbol: Symbol
 
     def set_up(self):
-        self.config = Config()
         self.event_manager = EventManager()
 
     async def sleep(self, secs: float):
@@ -16,6 +19,22 @@ class TestStrategy:
         mod = time % secs
         secs = secs - mod if mod != 0 else mod
         time = self.config.test_data.cursor.time + secs
-        print(f"Sleeping for {secs} seconds")
         while time > self.config.test_data.cursor.time:
           await self.event_manager.wait()
+
+    def test(self):
+        raise NotImplementedError("Implement this method in your subclass")
+
+
+class TestSingleStrategy:
+    config: Config
+    symbol: Symbol
+
+    async def sleep(self, secs: float):
+        time = self.config.test_data.cursor.time
+        mod = time % secs
+        secs = secs - mod if mod != 0 else mod
+        self.config.test_data.fast_forward(secs)
+
+    def test(self):
+        raise NotImplementedError("Implement this method in your subclass")

src/aiomql/core/constants.py

@@ -216,7 +216,14 @@ class TimeFrame(Repr, IntEnum):
         times = {60: 1, 120: 2, 180: 3, 240: 4, 300: 5, 360: 6, 600: 10, 900: 15, 1200: 20, 1800: 30, 3600: 16385,
                  7200: 16386, 10800: 16387, 14400: 16388, 21600: 16390, 28800: 16392, 43200: 16396, 86400: 16408,
                  604800: 32769, 2592000: 49153}
-        return TimeFrame(times[int(time)])
+        return TimeFrame(times[time])
+
+    @classmethod
+    @property
+    def all(cls) -> tuple['TimeFrame', ...]:
+        return (TimeFrame.M1, TimeFrame.M2, TimeFrame.M3, TimeFrame.M4, TimeFrame.M5, TimeFrame.M6, TimeFrame.M10,
+                TimeFrame.M15, TimeFrame.M20, TimeFrame.M30, TimeFrame.H1, TimeFrame.H2, TimeFrame.H3, TimeFrame.H4,
+                TimeFrame.H6, TimeFrame.H8, TimeFrame.H12, TimeFrame.D1, TimeFrame.W1, TimeFrame.MN1)
 
 
 class CopyTicks(Repr, IntEnum):

src/aiomql/core/meta_trader.py

@@ -78,6 +78,30 @@ class MetaTrader(metaclass=BaseMeta):
         """
         await self.shutdown()
 
+    async def _handler(self, api: dict):
+        func = api['func']
+        args = api.get('args', ())
+        kwargs = api.get('kwargs', {})
+        error_msg = api.get('error_msg', 'An error occurred')
+
+        res = await asyncio.to_thread(func, *args, **kwargs)
+        if res is None:
+            err = await self.last_error()
+            self.error = Error(*err)
+
+            if self.error.is_connection_error():
+                await self.initialize(**self.config.account_info(), path=self.config.path)
+                await self.login(**self.config.account_info())
+                res = await asyncio.to_thread(func, *args, **kwargs)
+
+                if res is None:
+                    err = await self.last_error()
+                    self.error = Error(*err)
+                    logger.warning(f'{error_msg}:{self.error.description}')
+            else:
+                logger.warning(f'{error_msg}:{self.error.description}')
+        return res
+
     async def login(self, login: int, password: str, server: str, timeout: int = 60000) -> bool:
         """
         Connects to the MetaTrader terminal using the specified login, password and server.
@@ -112,7 +136,8 @@ class MetaTrader(metaclass=BaseMeta):
         args = (str(path),) if path else ()
         kwargs = {key: value for key, value in (('login', login), ('password', password), ('server', server),
                                                 ('timeout', timeout), ('portable', portable)) if value}
-        return await asyncio.to_thread(self._initialize, *args, **kwargs)
+        res = await asyncio.to_thread(self._initialize, *args, **kwargs)
+        return res
 
     async def shutdown(self) -> None:
         """
@@ -121,229 +146,179 @@ class MetaTrader(metaclass=BaseMeta):
         Returns:
             None: None
         """
-        return await asyncio.to_thread(self._shutdown)
+        res = await asyncio.to_thread(self._shutdown)
+        return res
 
     async def last_error(self) -> tuple[int, str]:
         try:
-            return await asyncio.to_thread(self._last_error)
+            res = await asyncio.to_thread(self._last_error)
+            return res
         except Exception as err:
             logger.warning(f'Error in obtaining last error.')
             return -1, str(err)
 
     async def version(self) -> tuple[int, int, str] | None:
         """"""
-        res = await asyncio.to_thread(self._version)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining version information.{self.error.description}')
+        api = {'func': self._version, 'error_msg': 'Error in obtaining version.'}
+        res = await self._handler(api)
         return res
 
     async def account_info(self) -> AccountInfo | None:
         """"""
-        res = await asyncio.to_thread(self._account_info)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining account information.{self.error.description}')
+        api = {'func': self._account_info, 'error_msg': 'Error in obtaining account information'}
+        res = await self._handler(api)
         return res
 
     async def terminal_info(self) -> TerminalInfo | None:
-        res = await asyncio.to_thread(self._terminal_info)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining terminal information.{self.error.description}')
-            return res
+        api = {'func': self._terminal_info, 'error_msg': 'Error in obtaining terminal information'}
+        res = await self._handler(api)
         return res
 
     async def symbols_total(self) -> int:
-        return await asyncio.to_thread(self._symbols_total)
+        api = {'func': self._symbols_total, 'error_msg': 'Error in obtaining total symbols.'}
+        res = await self._handler(api)
+        return res
 
     async def symbols_get(self, group: str = "") -> tuple[SymbolInfo] | None:
         kwargs = {'group': group} if group else {}
-        res = await asyncio.to_thread(self._symbols_get, **kwargs)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining symbols.{self.error.description}')
-            return res
+        api = {'func': self._symbols_get, 'kwargs': kwargs, 'error_msg': 'Error in obtaining symbols.'}
+        res = await self._handler(api)
         return res
 
     async def symbol_info(self, symbol: str) -> SymbolInfo | None:
-        res = await asyncio.to_thread(self._symbol_info, symbol)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining information for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._symbol_info, 'args': (symbol,), 'error_msg': f'Error in obtaining information for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def symbol_info_tick(self, symbol: str) -> Tick | None:
-        res = await asyncio.to_thread(self._symbol_info_tick, symbol)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining tick for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._symbol_info_tick, 'args': (symbol,), 'error_msg': f'Error in obtaining tick for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def symbol_select(self, symbol: str, enable: bool) -> bool:
-        return await asyncio.to_thread(self._symbol_select, symbol, enable)
+        api = {'func': self._symbol_select, 'args': (symbol, enable), 'error_msg': f'Error in selecting {symbol}'}
+        res = await self._handler(api)
+        return res
 
     async def market_book_add(self, symbol: str) -> bool:
-        return await asyncio.to_thread(self._market_book_add, symbol)
+        api = {'func': self._market_book_add, 'args': (symbol,), 'error_msg': f'Error in adding {symbol} to market book'}
+        res = await self._handler(api)
+        return res
 
     async def market_book_get(self, symbol: str) -> tuple[BookInfo] | None:
-        res = await asyncio.to_thread(self._market_book_get, symbol)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining market depth content for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._market_book_get, 'args': (symbol,), 'error_msg': f'Error in obtaining market depth for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def market_book_release(self, symbol: str) -> bool:
-        return await asyncio.to_thread(self._market_book_release, symbol)
+        api = {'func': self._market_book_release, 'args': (symbol,), 'error_msg': f'Error in releasing market depth for {symbol}'}
+        res = await self._handler(api)
+        return res
 
     async def copy_rates_from(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float, count: int):
-        res = await asyncio.to_thread(self._copy_rates_from, symbol, timeframe, date_from, count)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining rates for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._copy_rates_from, 'args': (symbol, timeframe, date_from, count),
+               'error_msg': f'Error in obtaining rates for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def copy_rates_from_pos(self, symbol: str, timeframe: TimeFrame, start_pos: int, count: int):
-        res = await asyncio.to_thread(self._copy_rates_from_pos, symbol, timeframe, start_pos, count)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining rates for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._copy_rates_from_pos, 'args': (symbol, timeframe, start_pos, count),
+               'error_msg': f'Error in obtaining rates for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def copy_rates_range(self, symbol: str, timeframe: TimeFrame, date_from: datetime | float,
                                date_to: datetime | float):
-        res = await asyncio.to_thread(self._copy_rates_range, symbol, timeframe, date_from, date_to)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining rates for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._copy_rates_range, 'args': (symbol, timeframe, date_from, date_to),
+               'error_msg': f'Error in obtaining rates for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def copy_ticks_from(self, symbol: str, date_from: datetime | float, count: int, flags: CopyTicks):
-        res = await asyncio.to_thread(self._copy_ticks_from, symbol, date_from, count, flags)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining ticks for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._copy_ticks_from, 'args': (symbol, date_from, count, flags),
+               'error_msg': f'Error in obtaining ticks for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def copy_ticks_range(self, symbol: str, date_from: datetime | float, date_to: datetime | float,
                                flags: CopyTicks):
-        res = await asyncio.to_thread(self._copy_ticks_range, symbol, date_from, date_to, flags)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining ticks for {symbol}.{self.error.description}')
-            return res
+        api = {'func': self._copy_ticks_range, 'args': (symbol, date_from, date_to, flags),
+               'error_msg': f'Error in obtaining ticks for {symbol}'}
+        res = await self._handler(api)
         return res
 
     async def orders_total(self) -> int:
-        return await asyncio.to_thread(self._orders_total)
+        api = {'func': self._orders_total, 'error_msg': 'Error in obtaining total orders.'}
+        res = await self._handler(api)
+        return res
 
     async def orders_get(self, group: str = "", ticket: int = 0, symbol: str = "") -> tuple[TradeOrder] | None:
-        """Get active orders with the ability to filter by symbol or ticket. There are three call options.
-           Call without parameters. Return active orders on all symbols
-
-        Keyword Args:
-            symbol (str): Symbol name. Optional named parameter. If a symbol is specified, the ticket parameter is ignored.
-
-            group (str): The filter for arranging a group of necessary symbols. Optional named parameter. If the group is specified, the function
-                returns only active orders meeting a specified criteria for a symbol name.
-
-            ticket (int): Order ticket (ORDER_TICKET). Optional named parameter.
-
-        Returns:
-            tuple[TradeOrder]: A list of active trade orders as TradeOrder objects
-        """
         kwargs = {key: value for key, value in (('group', group), ('ticket', ticket), ('symbol', symbol)) if value}
-        res = await asyncio.to_thread(self._orders_get, **kwargs)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining orders.{self.error.description}')
-            return res
+        api = {'func': self._orders_get, 'kwargs': kwargs, 'error_msg': 'Error in obtaining orders.'}
+        res = await self._handler(api)
         return res
 
     async def order_calc_margin(self, action: OrderType, symbol: str, volume: float, price: float) -> float | None:
-        res = await asyncio.to_thread(self._order_calc_margin, action, symbol, volume, price)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in calculating margin.{self.error.description}')
-            return res
+        api = {'func': self._order_calc_margin, 'args': (action, symbol, volume, price),
+               'error_msg': 'Error in calculating margin.'}
+        res = await self._handler(api)
         return res
 
     async def order_calc_profit(self, action: OrderType, symbol: str, volume: float, price_open: float,
                                 price_close: float) -> float | None:
-        res = await asyncio.to_thread(self._order_calc_profit, action, symbol, volume, price_open, price_close)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in calculating profit.{self.error.description}')
-            return res
+        api = {'func': self._order_calc_profit, 'args': (action, symbol, volume, price_open, price_close),
+               'error_msg': 'Error in calculating profit.'}
+        res = await self._handler(api)
         return res
 
     async def order_check(self, request: dict) -> OrderCheckResult:
-        return await asyncio.to_thread(self._order_check, request)
+        api = {'func': self._order_check, 'args': (request,), 'error_msg': 'Error in checking order.'}
+        res = await self._handler(api)
+        return res
 
     async def order_send(self, request: dict) -> OrderSendResult:
-        return await asyncio.to_thread(self._order_send, request)
+        api = {'func': self._order_send, 'args': (request,), 'error_msg': 'Error in sending order.'}
+        res = await self._handler(api)
+        return res
 
     async def positions_total(self) -> int:
-        return await asyncio.to_thread(self._positions_total)
+        api = {'func': self._positions_total, 'error_msg': 'Error in obtaining total positions.'}
+        res = await self._handler(api)
+        return res
 
     async 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}
-        res = await asyncio.to_thread(self._positions_get, **kwargs)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in obtaining open positions.{self.error.description}')
-            return res
+        api = {'func': self._positions_get, 'kwargs': kwargs,
+               'error_msg': 'Error in obtaining open positions.'}
+        res = await self._handler(api)
         return res
 
     async def history_orders_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
-        return await asyncio.to_thread(self._history_orders_total, date_from, date_to)
+        api = {'func': self._history_orders_total, 'args': (date_from, date_to),
+               'error_msg': 'Error in obtaining total history orders.'}
+        res = await self._handler(api)
+        return res
 
     async 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)
-        res = await asyncio.to_thread(self._history_orders_get, *args, **kwargs)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in getting orders.{self.error.description}')
-            return res
+        api = {'func': self._history_orders_get, 'args': args, 'kwargs': kwargs,
+               'error_msg': 'Error in obtaining history orders'}
+        res = await self._handler(api)
         return res
 
     async def history_deals_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
-        return await asyncio.to_thread(self._history_deals_total, date_from, date_to)
+        api = {'func': self._history_deals_total, 'args': (date_from, date_to),
+               'error_msg': 'Error in obtaining total history deals'}
+        res = await self._handler(api)
+        return res
 
     async 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)
-        res = await asyncio.to_thread(self._history_deals_get, *args, **kwargs)
-        if res is None:
-            err = await self.last_error()
-            self.error = Error(*err)
-            logger.warning(f'Error in getting deals.{self.error}')
-            return res
+        api = {'func': self._history_deals_get, 'args': args, 'kwargs': kwargs,
+               'error_msg': 'Error in obtaining history deals'}
+        res = await self._handler(api)
         return res

src/aiomql/core/models.py

@@ -524,7 +524,6 @@ class OrderSendResult(Base):
     profit: float = None
     loss: float = None
 
-
 class TradePosition(Base):
     """Trade Position
 

src/aiomql/core/task_queue.py

@@ -3,7 +3,6 @@ from typing import Coroutine, Callable, Literal
 from signal import signal, SIGINT
 from logging import getLogger
 
-
 logger = getLogger(__name__)
 
 
@@ -44,7 +43,6 @@ class TaskQueue:
         self.timeout = timeout
         self.stop = False
         self.on_exit = on_exit
-        # signal(SIGINT, self.sigint_handle)
 
     def add(self, *, item: QueueItem, priority=3):
         try:

src/aiomql/lib/strategies/__init__.py

@@ -1,3 +1,3 @@
 from .finger_trap import FingerTrap
 from .tracker import Tracker
-from .finger_trap_back_test import FingerTrapTest
+from .finger_trap_back_test import FingerTrapTest, FingerTrapSingleTest

src/aiomql/lib/strategies/finger_trap.py

@@ -36,7 +36,6 @@ class FingerTrap(Strategy):
 
     async def check_trend(self):
         try:
-
             candles: Candles = await self.symbol.copy_rates_from_pos(timeframe=self.ttf, count=self.tcc)
             if not ((current := candles[-1].time) >= self.tracker.trend_time):
                 self.tracker.update(new=False, order_type=None)
@@ -57,6 +56,7 @@ class FingerTrap(Strategy):
                 self.tracker.update(trend="bearish")
             else:
                 self.tracker.update(trend="ranging", snooze=self.ttf.time, order_type=None)
+            self.tracker.update(trend="bullish") # remove this line
         except Exception as err:
             logger.error(f"{err} for {self.symbol} in {self.__class__.__name__}.check_trend")
             self.tracker.update(snooze=self.ttf.time, order_type=None)
@@ -73,7 +73,7 @@ class FingerTrap(Strategy):
             candles['cae'] = candles.ta_lib.cross(candles.close, candles.ema)
             candles['cbe'] = candles.ta_lib.cross(candles.close, candles.ema, above=False)
             current = candles[-1]
-            if self.tracker.bullish and current.cae:
+            if self.tracker.bullish and True or current.cae: # change True to current.cae
                 sl = find_bullish_fractal(candles).low
                 self.tracker.update(snooze=self.ttf.time, order_type=OrderType.BUY, sl=sl)
             elif self.tracker.bearish and current.cbe:

src/aiomql/lib/strategies/finger_trap_back_test.py

@@ -1,5 +1,24 @@
 from .finger_trap import FingerTrap
-from ...contrib.backtester.test_strategy import TestStrategy
+from ...contrib.backtester.test_strategy import TestStrategy, TestSingleStrategy
+
+
+class FingerTrapSingleTest(TestSingleStrategy, FingerTrap):
+
+    async def test(self):
+        print(f"Backtesting {self.symbol}")
+        while True:
+            try:
+                await self.watch_market()
+                if not self.tracker.new:
+                    continue
+
+                if self.tracker.order_type is not None:
+                    await self.trader.place_trade(order_type=self.tracker.order_type, parameters=self.parameters,
+                                                  sl=self.tracker.sl)
+                await self.sleep(self.tracker.snooze)
+            except Exception as err:
+                print(f"{err} For {self.symbol} in {self.__class__.__name__}.trade")
+                await self.sleep(self.ttf.time)
 
 
 class FingerTrapTest(TestStrategy, FingerTrap):

src/aiomql/utils.py

@@ -22,26 +22,27 @@ def dict_to_string(data: dict, multi=False) -> str:
     return f"{sep}".join(f"{key}: {value}" for key, value in data.items())
 
 
-def backoff_decorator(func=None, *, max_retries: int = 3, retries: int = 0, delay: int = 1, error=None) -> callable:
+def backoff_decorator(func=None, *, max_retries: int = 5, retries: int = 0, error='') -> callable:
     if func is None:
-        return partial(backoff_decorator, max_retries=max_retries, retries=retries, delay=delay, error=error)
+        return partial(backoff_decorator, max_retries=max_retries, retries=retries, error=error)
 
     @wraps(func)
     async def wrapper(*args, **kwargs):
-        nonlocal delay, retries
-
+        nonlocal retries
+        if max_retries == retries:
+            retries = 0
+            await func(*args, **kwargs)
         try:
-            res = await func(*args, **kwargs)
-            if res == error:
-                raise Exception('Invalid return type')
-            return res
-
-        except Exception as _:
-            await asyncio.sleep(delay * 2 ** retries + random.uniform(0, 1))
-            delay += 1
             retries += 1
-            return await wrapper(*args, **kwargs)
-        
+            res = await func(*args, **kwargs)
+            if error != '' and res == error:
+                raise TypeError('Invalid return type')
+            return res
+        except Exception as err:
+             logger.error(f'Error in {func.__name__}: {err}')
+             await asyncio.sleep(retries + random.randint(1, max_retries))
+             await wrapper(*args, **kwargs)
+
     return wrapper
 
 
@@ -86,3 +87,18 @@ def round_off(value: float, step: float, round_down: bool = False) -> float:
     with decimal.localcontext() as ctx:
         ctx.rounding = decimal.ROUND_DOWN if round_down else decimal.ROUND_UP
         return float(decimal.Decimal(str(value)).quantize(decimal.Decimal(str(step))))
+
+
+def async_cache(fun):
+
+    @wraps(fun)
+    async def wrapper(*args, **kwargs):
+        key = (args, frozenset(kwargs.items()))
+        async with wrapper.lock:
+            if key not in wrapper.cache:
+                wrapper.cache[key] = await fun(*args, **kwargs)
+        return wrapper.cache[key]
+
+    wrapper.lock = asyncio.Lock()
+    wrapper.cache = {}
+    return wrapper