From e8fd4ab4dcea9956d1d69cb0432b6b2f8fb0da69 Mon Sep 17 00:00:00 2001 From: Ichinga Samuel Date: Wed, 11 Sep 2024 06:16:27 +0100 Subject: [PATCH] testdata --- src/aiomql/account.py | 6 ++ src/aiomql/contrib/backtester/check1.py | 45 +--------- .../contrib/backtester/event_manager.py | 16 +++- src/aiomql/contrib/backtester/get_data.py | 64 +++++++-------- src/aiomql/contrib/backtester/meta_tester.py | 12 +-- .../contrib/backtester/strategy_tester.py | 19 +++-- .../contrib/backtester/test_strategy.py | 1 + src/aiomql/core/task_queue.py | 17 ++-- src/aiomql/history.py | 43 +++++----- src/aiomql/order.py | 33 ++++---- src/aiomql/positions.py | 28 ++++--- src/aiomql/records.py | 6 +- src/aiomql/result.py | 12 ++- src/aiomql/sessions.py | 14 +++- src/aiomql/strategy.py | 5 +- src/aiomql/symbol.py | 82 ++++++------------- src/aiomql/terminal.py | 3 +- src/aiomql/trade_records.py | 8 +- src/aiomql/trader.py | 3 +- 19 files changed, 201 insertions(+), 216 deletions(-) diff --git a/src/aiomql/account.py b/src/aiomql/account.py index 47ac91a..c8d8a4d 100644 --- a/src/aiomql/account.py +++ b/src/aiomql/account.py @@ -75,12 +75,18 @@ class Account(AccountInfo): async def _login(self, *, acc: dict, tries=3, **kwargs) -> bool: res = False + if tries == 0: return False + init_args = {**acc} | {'path': self.config.path} | {**kwargs} ini = await self.mt5.initialize(**init_args) + if ini: res = await self.mt5.login(**acc) + if not res: + await self.mt5.shutdown() + if ini and res: return True else: diff --git a/src/aiomql/contrib/backtester/check1.py b/src/aiomql/contrib/backtester/check1.py index db0dd91..43adc46 100644 --- a/src/aiomql/contrib/backtester/check1.py +++ b/src/aiomql/contrib/backtester/check1.py @@ -1,43 +1,4 @@ -from functools import wraps -from dataclasses import dataclass, fields, field -from typing import ClassVar +def fun(a, b=6, *c, **d): + print(f"{a=}, {b=}, {c=}, {d=}") -def dd(func): - - @wraps(func) - def wrapper(*args, **kwargs): - print(func.__name__) - return func(*args, **kwargs) - return wrapper - -class C: - def __new__(cls, *args, **kwargs): - if not hasattr(cls, '_instance'): - cls._instance = super().__new__(cls) - cls._instance.tasks = [] - # cls.__init__(a) - [setattr(cls._instance, k, v) for k, v in kwargs.items()] - return cls._instance - - def __init__(self, *args, **kwargs): - print('receiving args') - - - - -@dataclass -class D: - b: int = 0 - c: str = '' - _fields: list[ClassVar[str]] = field(default_factory=list) - - @dd - def setattrs(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'] - -d = D() -d.setattrs(r=3) +fun(1, 3, 4, 5, six=6, seven=7) diff --git a/src/aiomql/contrib/backtester/event_manager.py b/src/aiomql/contrib/backtester/event_manager.py index bd3ac76..2f246c8 100644 --- a/src/aiomql/contrib/backtester/event_manager.py +++ b/src/aiomql/contrib/backtester/event_manager.py @@ -1,7 +1,7 @@ import asyncio from asyncio import Condition, Task from typing import Self - +from datetime import datetime from ...core import Config @@ -25,11 +25,15 @@ class EventManager: def __init__(self, *, num_tasks: int = 0): self.num_main_tasks = num_tasks or self.num_main_tasks - def add_task(self, *task: Task): - self.tasks.extend(task) + def add_task(self, *tasks: Task): + self.tasks.extend(tasks) def sigint_handler(self, sig, frame): - print(self.config.test_data) + print('KeyboardInterrupt') + print(self.config.test_data.orders) + for task in self.tasks: + task.cancel() + # self.mt.cancel() async def acquire(self): await self.condition.acquire() @@ -45,6 +49,10 @@ 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)}") await asyncio.sleep(0) async def wait(self): diff --git a/src/aiomql/contrib/backtester/get_data.py b/src/aiomql/contrib/backtester/get_data.py index 9af30fd..1cf7468 100644 --- a/src/aiomql/contrib/backtester/get_data.py +++ b/src/aiomql/contrib/backtester/get_data.py @@ -4,7 +4,6 @@ from pathlib import Path import lzma from datetime import datetime from logging import getLogger -import asyncio from typing import Sequence, ClassVar import pytz @@ -13,7 +12,7 @@ from pandas import DataFrame from ...core.meta_trader import MetaTrader from ...core.config import Config -from ...core.constants import TimeFrame, +from ...core.constants import TimeFrame, CopyTicks from ...core.task_queue import TaskQueue, QueueItem from ...utils import backoff_decorator @@ -76,31 +75,6 @@ class GetData: self.mt5 = MetaTrader() self.task_queue = TaskQueue() - async def get_data(self): - """""" - qitems = [QueueItem(self.get_symbol_rates), QueueItem(self.get_symbol_ticks), - QueueItem(self.get_symbol_prices), QueueItem(self.get_symbol_info), - QueueItem(self.get_account_info), QueueItem(self.get_symbols_info), QueueItem(self.get_version)] - [self.task_queue.add(item=item, priority=0) for item in qitems] - await self.task_queue.run() - # terminal, version = await asyncio.gather(self.get_terminal_info(), self.get_version()) - - # self.data.setattrs(account=account, symbols=symbols, prices=prices, ticks=ticks, rates=rates, - # span=self.span, range=self.range, terminal=terminal, version=version, name=self.name) - - def pickle_data(self): - """""" - fh = open(f'{self.config.test_data_dir}/{self.name}', 'wb') - pickle.dump(self.data, fh) - fh.close() - - async def compress_data(self): - """""" - bdata = pickle.dumps(self.data) - name = self.name + 'xz' - with lzma.open(f'{self.config.test_data_dir}/{name}', 'w') as fh: - fh.write(bdata) - @classmethod def dump_data(cls, data: Data, name: str | Path, compress: bool = False): """""" @@ -136,6 +110,32 @@ class GetData: logger.error(f"Error: {err}") return None + async def get_data(self): + """""" + 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), + 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] + 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.close() + + async def compress_data(self): + """""" + bdata = pickle.dumps(self.data) + name = self.name + 'xz' + with lzma.open(f'{self.config.test_data_dir}/{name}', 'w') as fh: + fh.write(bdata) + async def get_terminal_info(self): """""" terminal = await self.mt5.terminal_info() @@ -149,19 +149,19 @@ class GetData: async def get_symbols_info(self): """""" - [self.task_queue.add(QueueItem(self.get_symbol_info, symbol, must_complete=True), priority=4) for symbol in self.symbols] + [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(QueueItem(self.get_symbol_ticks, symbol)) for symbol in self.symbols] + [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(QueueItem(self.get_symbol_prices, symbol)) for symbol in self.symbols] + [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(QueueItem(self.get_symbol_rates, symbol, timeframe, must_complete=True), priority=4) + [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) @@ -203,4 +203,4 @@ class GetData: res = pd.DataFrame(res) res.drop_duplicates(subset=['time'], keep='last', inplace=True) res.set_index('time', inplace=True, drop=False) - self.data.rates[symbol].setdefault(timeframe, res) + self.data.rates.setdefault(symbol, {})[timeframe.name] = res diff --git a/src/aiomql/contrib/backtester/meta_tester.py b/src/aiomql/contrib/backtester/meta_tester.py index 13c58d0..18179b5 100644 --- a/src/aiomql/contrib/backtester/meta_tester.py +++ b/src/aiomql/contrib/backtester/meta_tester.py @@ -62,13 +62,13 @@ 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) + # 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: diff --git a/src/aiomql/contrib/backtester/strategy_tester.py b/src/aiomql/contrib/backtester/strategy_tester.py index 257cf02..20eaac1 100644 --- a/src/aiomql/contrib/backtester/strategy_tester.py +++ b/src/aiomql/contrib/backtester/strategy_tester.py @@ -1,4 +1,5 @@ import asyncio +import signal from .event_manager import EventManager from .get_data import GetData @@ -15,6 +16,7 @@ class StrategyTester: 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}" @@ -27,9 +29,14 @@ class StrategyTester: await self.mt5.login(**acc) async def run(self): - await self.start() - tasks = [*[asyncio.create_task(strategy.test()) for strategy in self.strategies], - asyncio.create_task(self.event_manager.event_monitor())] - self.event_manager.add_task(*tasks) - await asyncio.gather(*tasks) - await self.mt5.shutdown() + try: + await self.start() + tasks = [*[asyncio.create_task(strategy.test()) for strategy in self.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)) + except Exception as err: + print(f"Error {err} occurred in StrategyTester") + finally: + await self.mt5.shutdown() diff --git a/src/aiomql/contrib/backtester/test_strategy.py b/src/aiomql/contrib/backtester/test_strategy.py index f5aedb4..eb88ffc 100644 --- a/src/aiomql/contrib/backtester/test_strategy.py +++ b/src/aiomql/contrib/backtester/test_strategy.py @@ -16,5 +16,6 @@ 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() diff --git a/src/aiomql/core/task_queue.py b/src/aiomql/core/task_queue.py index e03c6c5..922f98f 100644 --- a/src/aiomql/core/task_queue.py +++ b/src/aiomql/core/task_queue.py @@ -8,7 +8,7 @@ logger = getLogger(__name__) class QueueItem: - def __init__(self, task_item: Callable | Coroutine, must_complete=False, *args, **kwargs): + def __init__(self, task_item: Callable | Coroutine, *args, must_complete: bool = False, **kwargs): self.task_item = task_item self.args = args self.kwargs = kwargs @@ -30,7 +30,7 @@ class QueueItem: self.task_item(*self.args, **self.kwargs) except Exception as err: - logger.error(f"Error {err} occurred in {self.func.__name__} with args {self.args} and kwargs {self.kwargs}") + logger.error(f"Error {err} occurred in {self.task_item.__name__} with args {self.args} and kwargs {self.kwargs}") class TaskQueue: @@ -44,7 +44,7 @@ class TaskQueue: self.timeout = timeout self.stop = False self.on_exit = on_exit - signal(SIGINT, self.sigint_handle) + # signal(SIGINT, self.sigint_handle) def add(self, *, item: QueueItem, priority=3): try: @@ -71,19 +71,24 @@ class TaskQueue: self.queue.task_done() self.priority_tasks.discard(item) + if self.stop and len(self.priority_tasks) == 0: + print('All priority tasks completed') + self.cancel() + break + def sigint_handle(self, sig, frame): print('SIGINT received, cleaning up...') - if self.on_exit == 'complete_priority': + if self.on_exit == 'complete_priority' and self.priority_tasks: print(f'Completing {len(self.priority_tasks)} priority tasks...') self.stop = True - else: self.cancel() self.on_exit = 'cancel' # force cancel on exit if SIGINT is received again async def run(self, timeout: int = 0): + signal(SIGINT, self.sigint_handle) loop = asyncio.get_running_loop() start = loop.time() @@ -116,4 +121,4 @@ class TaskQueue: def cancel(self): cancelled = [task.cancel() for task in self.tasks if not task.done()] print(f'Cancelled {len(cancelled)} worker tasks') if cancelled else ... - self.tasks.clear() \ No newline at end of file + self.tasks.clear() diff --git a/src/aiomql/history.py b/src/aiomql/history.py index 7bad2a3..a6d5583 100644 --- a/src/aiomql/history.py +++ b/src/aiomql/history.py @@ -1,4 +1,3 @@ -import asyncio from datetime import datetime from logging import getLogger @@ -9,6 +8,10 @@ from .core.config import Config from .core.meta_trader import MetaTrader, CopyTicks, OrderType from .core.models import TradeDeal, TradeOrder +from .contrib.backtester.meta_tester import MetaTester + +from .utils import backoff_decorator + logger = getLogger(__name__) @@ -26,7 +29,7 @@ class History: mt5 (MetaTrader): MetaTrader instance config (Config): Config instance """ - mt5: MetaTrader + mt5: MetaTrader | MetaTester config: Config def __init__(self, *, date_from: datetime | int = None, date_to: datetime | int = None, @@ -44,7 +47,7 @@ class History: position (int): Filter for selecting history deals by position """ self.config = Config() - self.mt5 = MetaTrader() + self.mt5 = MetaTrader() if self.config.mode == 'live' else MetaTester() self.date_from = date_from self.date_to = date_to self.group = group @@ -67,30 +70,24 @@ class History: self.total_deals = len(self.deals) self.total_orders = len(self.orders) - async def get_deals(self, *, date_from: datetime | int = None, date_to: datetime | int = None, group: str = '', - retries: int = 3) -> tuple[TradeDeal, ...]: + @backoff_decorator + async def get_deals(self, *, date_from: datetime | int = None, date_to: datetime | int = None, group: str = '')\ + -> tuple[TradeDeal, ...]: """Get deals from trading history using the parameters set in the constructor. Returns: tuple[TradeDeal]: A list of trade deals """ - if retries < 1: - logger.warning(f'Failed to get deals: {self.mt5.error}') - return tuple() - date_from, date_to, group = date_from or self.date_from, date_to or self.date_to, group or self.group deals = await self.mt5.history_deals_get(date_from=date_from, date_to=date_to, group=group) if deals is not None: return tuple(TradeDeal(**deal._asdict()) for deal in deals) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.get_deals(date_from=date_from, date_to=date_to, group=group, retries=retries-1) - - logger.warning(f'Failed to get deals: {self.mt5.error}') + logger.warning(f'Failed to get deals') return tuple() + @backoff_decorator async def get_deals_ticket(self, *, ticket: int = None) -> tuple[TradeDeal, ...]: """Call specifying the order ticket. Return all deals having the specified order ticket in the DEAL_ORDER property. @@ -106,6 +103,7 @@ class History: deals = await self.mt5.history_deals_get(ticket=ticket) return tuple(sorted([TradeDeal(**deal._asdict()) for deal in deals or []], key=lambda x: x.time_msc)) + @backoff_decorator async def get_deals_position(self, *, position: int = None) -> tuple[TradeDeal, ...]: """ Get all deals with the specified position ticket in the DEAL_POSITION_ID property @@ -135,6 +133,7 @@ class History: total_deals = await self.mt5.history_deals_total(date_from, date_to) return total_deals + @backoff_decorator async def get_orders(self, *, date_from: datetime | int = None, date_to: datetime | int = None, group: str = '', retries: int = 3) -> tuple[TradeOrder, ...]: """Get orders from trading history using the parameters set in the constructor or the method arguments. @@ -142,30 +141,27 @@ class History: Returns: list[TradeOrder]: A list of trade orders """ - if retries < 1: - logger.warning(f'Failed to get orders: {self.mt5.error}') - return tuple() - date_from, date_to, group = date_from or self.date_from, date_to or self.date_to, group or self.group orders = await self.mt5.history_orders_get(date_from=date_from, date_to=date_to, group=group) + if orders is not None: return tuple(TradeOrder(**order._asdict()) for order in orders) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.get_orders(date_from=date_from, date_to=date_to, group=group, retries=retries - 1) - - logger.warning(f'Failed to get orders: {self.mt5.error}') + logger.warning(f'Failed to get orders') return tuple() + @backoff_decorator async def get_order_ticket(self, ticket: int | None = None) -> TradeOrder | None: ticket = ticket or self.ticket assert isinstance(ticket, int), 'ticket not provided' orders = await self.mt5.history_orders_get(ticket=ticket) + if orders and (order := orders[0]).ticket == ticket: return TradeOrder(**order._asdict()) + return None + @backoff_decorator async def get_orders_position(self, position: int = None) -> tuple[TradeOrder, ...]: """ Call specifying the position ticket. Return all orders with a position ticket specified in the @@ -182,6 +178,7 @@ class History: orders = await self.mt5.history_orders_get(position=position) return tuple(sorted([TradeOrder(**order._asdict()) for order in orders or []], key=lambda x: x.time_done_msc)) + @backoff_decorator async def orders_total(self, date_from: int | datetime = None, date_to: int | datetime = None) -> int: """Get total number of orders within the specified period in the constructor. diff --git a/src/aiomql/order.py b/src/aiomql/order.py index ac7c77a..1edc5df 100644 --- a/src/aiomql/order.py +++ b/src/aiomql/order.py @@ -1,9 +1,10 @@ -import asyncio from logging import getLogger from .core.models import TradeRequest, OrderSendResult, OrderCheckResult, TradeOrder from .core.constants import TradeAction, OrderTime, OrderFilling from .core.exceptions import OrderError +from .utils import backoff_decorator + logger = getLogger(__name__) @@ -38,26 +39,25 @@ class Order(TradeRequest): """ return await self.mt5.orders_total() - async def get_order(self, *, ticket: int, retries: int = 3) -> TradeOrder | None: + @backoff_decorator + async def get_order(self, *, ticket: int) -> TradeOrder | None: """ Get the order by ticket number. + Args: ticket (int): Order ticket number - retries (int): Number of retries + Returns: """ - if retries < 1: - return None orders = await self.mt5.orders_get(ticket=ticket) + if orders and (order := orders[0]).ticket == ticket: return TradeOrder(**order._asdict()) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.get_order(ticket=ticket, retries=retries-1) + return None - async def get_orders(self, *, ticket: int = 0, symbol: str = '', group: str = '', retries=3)\ - -> tuple[TradeOrder, ...]: + @backoff_decorator + async def get_orders(self, *, ticket: int = 0, symbol: str = '', group: str = '') -> tuple[TradeOrder, ...]: """Get the list of active orders for the current symbol. Keyword Args: ticket (int): Order ticket number @@ -66,16 +66,13 @@ class Order(TradeRequest): Returns: tuple[TradeOrder]: A Tuple of active trade orders as TradeOrder objects """ - if retries < 1: - return tuple() symbol = getattr(self, 'symbol', symbol) orders = await self.mt5.orders_get(symbol=symbol, ticket=ticket, group=group) + if orders is not None: orders = (TradeOrder(**order._asdict()) for order in orders) return tuple(orders) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.get_orders(ticket=ticket, symbol=symbol, group=group, retries=retries-1) + return tuple() async def check(self, **kwargs) -> OrderCheckResult: @@ -90,7 +87,7 @@ class Order(TradeRequest): req = self.dict | kwargs res = await self.mt5.order_check(req) if res is None: - raise OrderError(f'Failed to check order due to {self.mt5.error.description}') + raise OrderError(f'Order check failed for {self.symbol}') return OrderCheckResult(**res._asdict()) async def send(self) -> OrderSendResult: @@ -104,7 +101,7 @@ class Order(TradeRequest): """ res = await self.mt5.order_send(self.dict) if res is None: - raise OrderError(f'Failed to send order {self.symbol} due to {self.mt5.error.description}') + raise OrderError(f'Failed to send order {self.symbol}') res = OrderSendResult(**res._asdict()) try: profit = await self.calc_profit() @@ -126,7 +123,7 @@ class Order(TradeRequest): """ res = await self.mt5.order_calc_margin(self.type, self.symbol, self.volume, self.price) if res is None: - raise OrderError(f'Failed to calculate margin for {self.symbol} due to {self.mt5.error.description}') + raise OrderError(f'Failed to calculate margin for {self.symbol}') return res async def calc_profit(self, **kwargs) -> float | None: diff --git a/src/aiomql/positions.py b/src/aiomql/positions.py index 7b796da..ad7d358 100644 --- a/src/aiomql/positions.py +++ b/src/aiomql/positions.py @@ -2,8 +2,15 @@ import asyncio from logging import getLogger -from .core import MetaTrader, TradePosition, TradeAction, OrderType +from .core.meta_trader import MetaTrader +from .core.models import TradePosition, TradeAction +from .core.constants import OrderType +from .core.config import Config + +from .contrib.backtester.meta_tester import MetaTester + from .order import Order +from .utils import backoff_decorator logger = getLogger(__name__) @@ -18,7 +25,7 @@ class Positions: ticket (int): Position ticket. mt5 (MetaTrader): MetaTrader instance. """ - mt5: MetaTrader + mt5: MetaTrader | MetaTester def __init__(self, *, symbol: str = "", group: str = "", ticket: int = 0): """Get Open Positions. @@ -30,7 +37,8 @@ class Positions: ticket (int): Position ticket """ - self.mt5 = MetaTrader() + self.config = Config() + self.mt5 = MetaTrader() if self.config.mode == 'live' else MetaTester() self.symbol = symbol self.group = group self.ticket = ticket @@ -43,7 +51,8 @@ class Positions: """ return await self.mt5.positions_total() - async def positions_get(self, symbol: str = '', group: str = '', ticket: int = 0, retries=3) -> list[TradePosition]: + @backoff_decorator + async def positions_get(self, symbol: str = '', group: str = '', ticket: int = 0) -> list[TradePosition]: """Get open positions with the ability to filter by symbol or ticket. Keyword Args: @@ -55,17 +64,12 @@ class Positions: Returns: list[TradePosition]: A list of open trade positions """ - if retries < 1: - logger.warning(f'Failed to get positions for {symbol or self.symbol}. {self.mt5.error}') - return [] positions = await self.mt5.positions_get(group=group or self.group, symbol=symbol or self.symbol, ticket=ticket or self.ticket) if positions is not None: return [TradePosition(**pos._asdict()) for pos in positions] - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.positions_get(symbol, group, ticket, retries - 1) - logger.warning(f'Failed to get positions for {symbol or self.symbol}. {self.mt5.error}') + + logger.warning(f'Failed to get positions for {symbol or self.symbol}') return [] async def position_get(self, *, ticket: int) -> TradePosition | None: @@ -78,8 +82,10 @@ class Positions: """ positions = await self.positions_get(ticket=ticket) position = positions[0] if positions else None + if position is None or position.ticket != ticket: return None + return position async def close(self, *, ticket: int, symbol: str, price: float, volume: float, order_type: OrderType): diff --git a/src/aiomql/records.py b/src/aiomql/records.py index c9ec5a1..e758329 100644 --- a/src/aiomql/records.py +++ b/src/aiomql/records.py @@ -6,7 +6,9 @@ import csv import logging from typing import Iterable -from .core import Config, MetaTrader +from .contrib.backtester.meta_tester import MetaTester +from .core.config import Config +from .core.meta_trader import MetaTrader logger = logging.getLogger(__name__) @@ -30,7 +32,7 @@ class Records: records_dir (Path): Absolute path to directory containing record of placed trades. """ self.config = Config() - self.mt5 = MetaTrader() + self.mt5 = MetaTrader() if self.config.mode == 'live' else MetaTester() self.records_dir = records_dir or self.config.records_dir async def get_records(self): diff --git a/src/aiomql/result.py b/src/aiomql/result.py index 148ab5b..37d37a5 100644 --- a/src/aiomql/result.py +++ b/src/aiomql/result.py @@ -2,8 +2,9 @@ import csv import json from logging import getLogger from typing import Iterable, Literal +from asyncio import Lock -from .core import Config +from .core.config import Config from .core.models import OrderSendResult logger = getLogger(__name__) @@ -31,6 +32,7 @@ class Result: self.parameters = parameters or {} self.result = result self.name = name or parameters.get('name', 'Trades') + self.lock = Lock() def get_data(self) -> dict: res = self.result.get_dict(exclude={'retcode', 'comment', 'retcode_external', 'request_id', 'request'}) @@ -50,6 +52,7 @@ class Result: async def to_csv(self): """Record trade results and associated parameters as a csv file """ + await self.lock.acquire() try: data = self.get_data() file = self.config.records_dir / f"{self.name}.csv" @@ -67,6 +70,9 @@ class Result: except Exception as err: logger.error(f'Unable to save to csv: {err}') + finally: + self.lock.release() + @staticmethod def serialize(value) -> str: """Serialize the trade records and strategy parameters @@ -79,6 +85,7 @@ class Result: async def to_json(self): """Save trades and strategy parameters in a json file """ + await self.lock.acquire() try: file = self.config.records_dir / f"{self.name}.json" data = self.get_data() @@ -92,3 +99,6 @@ class Result: json.dump(rows, fh, indent=2, skipkeys=True, default=self.serialize) except Exception as err: logger.error(f"Unable to save as json file: {err}") + + finally: + self.lock.release() diff --git a/src/aiomql/sessions.py b/src/aiomql/sessions.py index 77e5c9a..b507dde 100644 --- a/src/aiomql/sessions.py +++ b/src/aiomql/sessions.py @@ -5,6 +5,8 @@ from typing import Literal, Callable from logging import getLogger from .positions import Positions +from .core.config import Config +from.contrib.backtester.event_manager import EventManager logger = getLogger(__name__) @@ -18,6 +20,15 @@ def delta(obj: time) -> timedelta: return timedelta(hours=obj.hour, minutes=obj.minute, seconds=obj.second, microseconds=obj.microsecond) +async def backtest_sleep(secs): + """A custom function to call when the session starts.""" + em = EventManager() + + async with em.condition: + while em.config.test_data.cursor.time < (em.config.test_data.cursor.time + secs): + await em.condition.wait() + + class Session: """A session is a time period between two datetime.time objects specified in utc. @@ -202,6 +213,7 @@ class Sessions: current_session = self.find_next(now) secs = current_session.until() + 10 logger.info(f'sleeping for {secs} seconds until next {current_session} session') - await sleep(secs) + sleep_func = sleep if Config().mode == 'live' else backtest_sleep + await sleep_func(secs) self.current_session = current_session await self.current_session.begin() diff --git a/src/aiomql/strategy.py b/src/aiomql/strategy.py index 29da098..b1becef 100644 --- a/src/aiomql/strategy.py +++ b/src/aiomql/strategy.py @@ -8,6 +8,7 @@ from datetime import time as dtime from .core.meta_trader import MetaTrader from .symbol import Symbol as _Symbol from .core import Config +from .contrib.backtester.meta_tester import MetaTester from .sessions import Sessions, Session Symbol = TypeVar("Symbol", bound=_Symbol) @@ -47,7 +48,7 @@ class Strategy(ABC): self.parameters["name"] = self.name self.sessions = sessions or Sessions(Session(start=0, end=dtime(hour=23, minute=59, second=59))) self.config = Config() - self.mt5 = MetaTrader() + self.mt5 = MetaTrader() if self.config.mode == 'live' else MetaTester() def __repr__(self): return f"{self.name}({self.symbol!r})" @@ -79,4 +80,4 @@ class Strategy(ABC): async def trade(self): """Place trades using this method. This is the main method of the strategy. It will be called by the strategy runner. - """ \ No newline at end of file + """ diff --git a/src/aiomql/symbol.py b/src/aiomql/symbol.py index ab445e4..87dbeb8 100644 --- a/src/aiomql/symbol.py +++ b/src/aiomql/symbol.py @@ -9,7 +9,7 @@ from .ticks import Tick from .account import Account from .candle import Candles from .ticks import Ticks -from .utils import round_off +from .utils import round_off, backoff_decorator logger = getLogger(__name__) @@ -48,6 +48,7 @@ class Symbol(SymbolInfo): """ return self.point * 10 + @backoff_decorator async def info_tick(self, *, name: str = "") -> Tick: """Get the current price tick of a financial instrument. @@ -65,7 +66,7 @@ class Symbol(SymbolInfo): tick = Tick(**tick._asdict()) setattr(self, 'tick', tick) if not name else ... return tick - raise ValueError(f'Could not get tick for {name or self.name}. {self.mt5.error}') + raise ValueError(f'Could not get tick for {name or self.name}.') async def symbol_select(self, *, enable: bool = True) -> bool: """Select a symbol in the MarketWatch window or remove a symbol from the window. @@ -81,7 +82,8 @@ class Symbol(SymbolInfo): self.select = await self.mt5.symbol_select(self.name, enable) return self.select - async def info(self, retries=3) -> SymbolInfo: + @backoff_decorator + async def info(self) -> SymbolInfo: """Get data on the specified financial instrument and update the symbol object properties Returns: @@ -90,18 +92,13 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get info for {self.name}. {self.mt5.error}') info = await self.mt5.symbol_info(self.name) if info: info = info._asdict() info['swap_rollover3days'] = info.get('swap_rollover3days', 0) % 7 self.set_attributes(**info) return SymbolInfo(**info) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.info(retries=retries - 1) - raise ValueError(f'Could not get info for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get info for {self.name}') async def init(self) -> bool: """Initialized the symbol by pulling properties from the terminal @@ -131,7 +128,8 @@ class Symbol(SymbolInfo): """ return await self.mt5.market_book_add(self.name) - async def book_get(self, retries=3) -> tuple[BookInfo, ...]: + @backoff_decorator + async def book_get(self) -> tuple[BookInfo, ...]: """Returns a tuple of BookInfo featuring Market Depth entries for the specified symbol. Returns: @@ -140,16 +138,13 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get book info for {self.name}. {self.mt5.error}') infos = await self.mt5.market_book_get(self.name) + if infos is not None: book_infos = (BookInfo(**info._asdict()) for info in infos) return tuple(book_infos) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.book_get(retries=retries - 1) - raise ValueError(f'Could not get book info for {self.name}. {self.mt5.error}') + + raise ValueError(f'Could not get book info for {self.name}') async def book_release(self) -> bool: """Cancels subscription of the MetaTrader 5 terminal to the Market Depth change events for a specified symbol. @@ -241,8 +236,9 @@ class Symbol(SymbolInfo): else: logger.warning(f'Currency conversion failed: Unable to convert {amount} in {quote} to {base}') + @backoff_decorator async def copy_rates_from(self, *, timeframe: TimeFrame, - date_from: datetime | int, count: int = 500, retries=3) -> Candles: + date_from: datetime | int, count: int = 500) -> Candles: """ Get bars from the MetaTrader 5 terminal starting from the specified date. @@ -260,19 +256,14 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') rates = await self.mt5.copy_rates_from(self.name, timeframe, date_from, count) if rates is not None: return Candles(data=rates) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.copy_rates_from(timeframe=timeframe, date_from=date_from, - count=count, retries=retries - 1) - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get rates for {self.name}.') + @backoff_decorator async def copy_rates_from_pos(self, *, timeframe: TimeFrame, count: int = 500, - start_position: int = 0, retries=3) -> Candles: + start_position: int = 0) -> Candles: """Get bars from the MetaTrader 5 terminal starting from the specified index. Args: @@ -289,19 +280,14 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') rates = await self.mt5.copy_rates_from_pos(self.name, timeframe, start_position, count) if rates is not None: return Candles(data=rates) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.copy_rates_from_pos(timeframe=timeframe, count=count, - start_position=start_position, retries=retries - 1) - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get rates for {self.name}.') + @backoff_decorator async def copy_rates_range(self, *, timeframe: TimeFrame, date_from: datetime | int, - date_to: datetime | int, retries=3) -> Candles: + date_to: datetime | int) -> Candles: """Get bars in the specified date range from the MetaTrader 5 terminal. Args: @@ -321,21 +307,15 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') - rates = await self.mt5.copy_rates_range(symbol=self.name, timeframe=timeframe, date_from=date_from, date_to=date_to) if rates is not None: return Candles(data=rates) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.copy_rates_range(timeframe=timeframe, date_from=date_from, - date_to=date_to, retries=retries - 1) - raise ValueError(f'Could not get rates for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get rates for {self.name}.') + @backoff_decorator async def copy_ticks_from(self, *, date_from: datetime | int, count: int = 100, - flags: CopyTicks = CopyTicks.ALL, retries=3) -> Ticks: + flags: CopyTicks = CopyTicks.ALL) -> Ticks: """ Get ticks from the MetaTrader 5 terminal starting from the specified date. @@ -352,19 +332,14 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned """ - if retries < 1: - raise ValueError(f'Could not get ticks for {self.name}. {self.mt5.error}') - ticks = await self.mt5.copy_ticks_from(self.name, date_from, count, flags) if ticks is not None: return Ticks(data=ticks) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.copy_ticks_from(date_from=date_from, count=count, flags=flags, retries=retries - 1) - raise ValueError(f'Could not get ticks for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get ticks for {self.name}.') + @backoff_decorator async def copy_ticks_range(self, *, date_from: datetime | int, date_to: datetime | int, - flags: CopyTicks = CopyTicks.ALL, retries=3) -> Ticks: + flags: CopyTicks = CopyTicks.ALL) -> Ticks: """Get ticks for the specified date range from the MetaTrader 5 terminal. Args: @@ -383,12 +358,7 @@ class Symbol(SymbolInfo): Raises: ValueError: If request was unsuccessful and None was returned. """ - if retries < 1: - raise ValueError(f'Could not get ticks for {self.name}. {self.mt5.error}') ticks = await self.mt5.copy_ticks_range(self.name, date_from, date_to, flags) if ticks is not None: return Ticks(data=ticks) - if self.mt5.error.is_connection_error(): - await asyncio.sleep(retries) - return await self.copy_ticks_range(date_from=date_from, date_to=date_to, flags=flags, retries=retries - 1) - raise ValueError(f'Could not get ticks for {self.name}. {self.mt5.error}') + raise ValueError(f'Could not get ticks for {self.name}.') diff --git a/src/aiomql/terminal.py b/src/aiomql/terminal.py index 8d84709..3a08115 100644 --- a/src/aiomql/terminal.py +++ b/src/aiomql/terminal.py @@ -1,7 +1,7 @@ """Terminal related functions and properties""" - from typing import NamedTuple from logging import getLogger + from .core.models import TerminalInfo logger = getLogger(__name__) @@ -59,7 +59,6 @@ class Terminal(TerminalInfo): """ info = await self.mt5.terminal_info() self.set_attributes(**info._asdict()) - return self async def symbols_total(self) -> int: """Get the number of all financial instruments in the MetaTrader 5 terminal. diff --git a/src/aiomql/trade_records.py b/src/aiomql/trade_records.py index 55fd395..f0d0a80 100644 --- a/src/aiomql/trade_records.py +++ b/src/aiomql/trade_records.py @@ -7,7 +7,9 @@ import csv import logging from typing import Iterable -from .core import Config, MetaTrader +from .core.config import Config +from .core.meta_trader import MetaTrader +from .contrib.backtester.meta_tester import MetaTester logger = logging.getLogger(__name__) @@ -21,7 +23,7 @@ class TradeRecords: from the config """ config: Config - mt5: MetaTrader + mt5: MetaTrader | MetaTester def __init__(self, *, records_dir: Path | str = ''): """Initialize the Records class. The main method of this class is update_records which you should call to update @@ -31,7 +33,7 @@ class TradeRecords: records_dir (Path): Absolute path to directory containing record of placed trades. """ self.config = Config() - self.mt5 = MetaTrader() + self.mt5 = MetaTrader() if self.config.mode == 'live' else MetaTester() self.records_dir = records_dir or self.config.records_dir async def get_csv_records(self): diff --git a/src/aiomql/trader.py b/src/aiomql/trader.py index 00183d8..753d0fb 100644 --- a/src/aiomql/trader.py +++ b/src/aiomql/trader.py @@ -12,6 +12,7 @@ from .ram import RAM from .core.models import OrderType, OrderSendResult from .core.config import Config from .result import Result +from .core.task_queue import QueueItem logger = getLogger(__name__) Symbol = TypeVar("Symbol", bound=_Symbol) @@ -121,7 +122,7 @@ class Trader(ABC): params["date"] = str(date.date()) params["time"] = str(date.time()) res = Result(result=result, parameters=params, name=name) - self.config.task_queue.add_task(res.save) + self.config.task_queue.add(item=QueueItem(res.save, must_complete=True)) @abstractmethod async def place_trade(self, *args, **kwargs):