From 12f61f3b6e858d57d1accf86888657a0c6d4fb3d Mon Sep 17 00:00:00 2001 From: Ichinga Samuel Date: Wed, 22 Jan 2025 15:56:57 +0100 Subject: [PATCH] v4.0.9 --- pyproject.toml | 2 +- src/aiomql/core/backtesting/get_data.py | 12 +- src/aiomql/core/task_queue.py | 194 +++++++++++++++--------- src/aiomql/lib/strategy.py | 5 + 4 files changed, 132 insertions(+), 81 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c7c96a0..68ade5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "aiomql" -version = "4.0.7" +version = "4.0.8" readme = "README.md" requires-python = ">=3.11" classifiers = [ diff --git a/src/aiomql/core/backtesting/get_data.py b/src/aiomql/core/backtesting/get_data.py index 5c68261..3e0c604 100644 --- a/src/aiomql/core/backtesting/get_data.py +++ b/src/aiomql/core/backtesting/get_data.py @@ -200,7 +200,7 @@ class GetData: terminal = await self.mt5.terminal_info() if terminal is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True terminal = terminal._asdict() self.data.set_attrs(terminal=terminal) @@ -208,7 +208,7 @@ class GetData: version = await self.mt5.version() if version is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True self.data.set_attrs(version=version) @backoff_decorator @@ -216,7 +216,7 @@ class GetData: res = await self.mt5.account_info() if res is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True res = res._asdict() self.data.set_attrs(account=res) @@ -247,7 +247,7 @@ class GetData: res = await self.mt5.symbol_info(symbol) if res is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True self.data.symbols[symbol] = res._asdict() @backoff_decorator @@ -255,7 +255,7 @@ class GetData: res = await self.mt5.copy_ticks_range(symbol, self.start, self.end, MetaTrader5.COPY_TICKS_ALL) if res is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True self.data.ticks[symbol] = res @backoff_decorator @@ -263,5 +263,5 @@ class GetData: res = await self.mt5.copy_rates_range(symbol, timeframe, self.start, self.end) if res is None: self.data.fully_loaded = False - self.task_queue.stop_queue() + self.task_queue.stop = True self.data.rates.setdefault(symbol, {})[int(timeframe)] = res diff --git a/src/aiomql/core/task_queue.py b/src/aiomql/core/task_queue.py index d94d137..f9ce06a 100644 --- a/src/aiomql/core/task_queue.py +++ b/src/aiomql/core/task_queue.py @@ -1,5 +1,6 @@ import asyncio import time +import random from typing import Coroutine, Callable, Literal from logging import getLogger from signal import SIGINT, signal @@ -10,37 +11,45 @@ logger = getLogger(__name__) class QueueItem: """A class to represent a task item in the queue. - Attributes: - - `task_item` (Callable | Coroutine): The task to run. + Attributes: + - `task_item` (Callable | Coroutine): The task to run. - - `args` (tuple): The arguments to pass to the task + - `args` (tuple): The arguments to pass to the task - - `kwargs` (dict): The keyword arguments to pass to the task + - `kwargs` (dict): The keyword arguments to pass to the task - - `must_complete` (bool): A flag to indicate if the task must complete before the queue stops. Default is False. + - `must_complete` (bool): A flag to indicate if the task must complete before the queue stops. Default is False. - - `time` (int): The time the task was added to the queue. + - `time` (int): The time the task was added to the queue. """ - def __init__(self, task_item: Callable | Coroutine, *args, **kwargs): self.task_item = task_item self.args = args self.kwargs = kwargs self.must_complete = False - self.time = time.monotonic_ns() + self.time = time.time_ns() def __hash__(self): - return id(self) + return self.time def __lt__(self, other): return self.time < other.time + def __eq__(self, other): + return self.time == other.time + + def __le__(self, other): + return self.time <= other.time + async def run(self): try: if asyncio.iscoroutinefunction(self.task_item): - await self.task_item(*self.args, **self.kwargs) + return await self.task_item(*self.args, **self.kwargs) else: - await asyncio.to_thread(self.task_item, *self.args, **self.kwargs) + return await asyncio.to_thread(self.task_item, *self.args, **self.kwargs) + except asyncio.CancelledError: + logger.debug("Task %s with args %s and %s was cancelled", + self.task_item.__name__, self.args, self.kwargs) except Exception as err: logger.error("Error %s occurred in %s with args %s and %s", err, self.task_item.__name__, self.args, self.kwargs) @@ -48,20 +57,21 @@ class QueueItem: class TaskQueue: queue_task: asyncio.Task + start_time: float - def __init__(self, size: int = 0, workers: int = 500, timeout: int = None, queue: asyncio.Queue = None, - on_exit: Literal['cancel', 'complete_priority'] = 'complete_priority', - mode: Literal['finite', 'infinite'] = 'finite', worker_timeout: int = 60): - + def __init__(self, *, size: int = 0, workers: int = 10, queue: asyncio.Queue = None, queue_timeout: int = 0, + on_exit: Literal['cancel', 'complete_priority'] = 'complete_priority', absolute_timeout: int = 0, + mode: Literal['finite', 'infinite'] = 'finite', worker_timeout: int = 1): self.queue = queue or asyncio.PriorityQueue(maxsize=size) self.workers = workers - self.worker_tasks = [] - self.priority_tasks = set() # tasks that must complete - self.timeout = timeout + self.worker_tasks = {} + self.queue_timeout = queue_timeout + self.absolute_timeout = absolute_timeout self.stop = False self.on_exit = on_exit self.mode = mode self.worker_timeout = worker_timeout + self.queue_task_cancelled = False signal(SIGINT, self.sigint_handle) def add(self, *, item: QueueItem, priority=3, must_complete=False): @@ -76,23 +86,27 @@ class TaskQueue: if self.stop: return item.must_complete = must_complete - self.priority_tasks.add(item) if item.must_complete else ... if isinstance(self.queue, asyncio.PriorityQueue): item = (priority, item) self.queue.put_nowait(item) except asyncio.QueueFull: logger.error("Queue is full") - async def worker(self): + async def worker(self, wid: int = None): """Worker function to run tasks in the queue.""" while True: try: + if self.queue_task_cancelled or not self.check_timeout(): + self.remove_worker(wid) + break + if self.mode == 'infinite' and self.queue.qsize() <= 1: dummy = QueueItem(self.dummy_task) self.add(item=dummy) if isinstance(self.queue, asyncio.PriorityQueue): _, item = self.queue.get_nowait() + else: item = self.queue.get_nowait() @@ -101,98 +115,130 @@ class TaskQueue: self.queue.task_done() - self.priority_tasks.discard(item) - - if self.stop and (self.on_exit == 'cancel' or len(self.priority_tasks) == 0): + if self.stop and (self.on_exit == 'cancel' or len(self.worker_tasks) <= 1): self.cancel() - break + + await self.add_workers() except asyncio.QueueEmpty: if self.stop: + self.remove_worker(wid) break if self.mode == 'finite': - self.stop = True + self.remove_worker(wid) break + + except asyncio.CancelledError: + break + except Exception as err: logger.error("%s: Error occurred in worker", err) + self.remove_worker(wid) break + def start_timer(self, *, queue_timeout: int = None, absolute_timeout: int = None, start=False): + self.queue_timeout = queue_timeout or self.queue_timeout + self.absolute_timeout = absolute_timeout or self.absolute_timeout + if start: + self.start_time = time.perf_counter() + + def check_timeout(self): + if self.queue_timeout and (time.perf_counter() - self.start_time) > self.queue_timeout: + if self.on_exit == 'cancel': + self.stop = True + self.cancel() + return False + else: + self.stop = True + self.queue_timeout = 0 + return True + if self.absolute_timeout and (time.perf_counter() - self.start_time) > self.absolute_timeout: + self.stop = True + self.cancel() + return False + return True + async def dummy_task(self): await asyncio.sleep(self.worker_timeout) - async def run(self, timeout: int = 0): + def remove_worker(self, wid: int): + try: + task = self.worker_tasks.pop(wid, None) + if task is not None: + task.cancel() + except Exception as err: + logger.debug("%s: Error occurred in removing worker %d", err, wid) + except asyncio.CancelledError as _: + ... + + async def add_workers(self, no_of_workers: int = None): + """Create workers for running queue tasks.""" + if no_of_workers is None: + queue_size = self.queue.qsize() + req_workers = queue_size - len(self.worker_tasks) + if req_workers > 1: + no_of_workers = req_workers + else: + return + + ri = lambda : random.randint(999, 999_999_999) # random id + ct = lambda ti: asyncio.create_task(self.worker(wid=ti), name=ti) # create task + wr = range(no_of_workers) + [self.worker_tasks.setdefault(wi:=ri(), ct(wi)) for _ in wr] + + async def run(self, queue_timeout: int = None, absolute_timeout: int = None): """Run the queue until all tasks are completed or the timeout is reached. Args: - timeout (int): The maximum time to wait for the queue to complete. Default is 0. If timeout is provided - the queue is joined using `asyncio.wait_for` with the timeout. If the timeout is reached, the queue is - stopped and the remaining tasks are handled based on the `on_exit` attribute. - If the timeout is 0, the queue will run until all tasks are completed or the queue is stopped. + queue_timeout (int): The maximum time to wait for the queue to complete. Default is 0. + absolute_timeout (int): The maximum time to run the queue. + This timeout overrides the timeout attribute of the queue instance. + The queue stops when the timeout is reached, and the remaining tasks are handled based on the + `on_exit` attribute. If the timeout is 0, the queue will run until all tasks are completed or the queue + is stopped. """ - start = time.perf_counter() try: - self.worker_tasks.extend(asyncio.create_task(self.worker()) for _ in range(self.workers)) - timeout = timeout or self.timeout + self.start_timer(queue_timeout=queue_timeout, absolute_timeout=absolute_timeout, start=True) + await self.add_workers(no_of_workers=self.workers) self.queue_task = asyncio.create_task(self.queue.join()) + await self.queue_task - if timeout: - await asyncio.wait_for(self.queue_task, timeout=timeout) - - else: - await self.queue_task - - except TimeoutError: - logger.warning("Timed out after %d seconds, %d tasks remaining", - time.perf_counter() - start, self.queue.qsize()) - self.stop = True - await self.clean_up() + except asyncio.TimeoutError: + logger.warning("Timeout occurred after %d seconds, %d tasks remaining", + time.perf_counter() - self.start_time, self.queue.qsize()) except asyncio.CancelledError: - self.stop = True - await self.clean_up() - + logger.warning("Task Queue Cancelled after %d seconds, %d tasks remaining", + time.perf_counter() - self.start_time, self.queue.qsize()) except Exception as err: - logger.warning("%s: An error occurred in %s.run", err, self.__class__.__name__) - self.stop = True - await self.clean_up() + logger.warning("%s occurred after %d seconds, %d tasks remaining", + err, time.perf_counter() - self.start_time, self.queue.qsize()) + finally: + logger.info("Tasks completed after %d seconds, %d tasks remaining", + time.perf_counter() - self.start_time, self.queue.qsize()) - - async def clean_up(self): - """Clean up tasks in the queue, completing priority tasks if `on_exit` is `complete_priority`""" - self.stop = True + def cancel_all_workers(self): try: - logger.info('cleaning up tasks...') - if self.on_exit == 'complete_priority' and (pt := len(self.priority_tasks)) > 0: - logger.info('Completing %d priority tasks...', pt) - self.queue_task = asyncio.create_task(self.queue.join()) - await self.queue_task - logger.info('Cleaning up tasks done...') - self.cancel() - - except asyncio.CancelledError: - self.cancel() - + wids = list(self.worker_tasks.keys()) + [self.remove_worker(wid) for wid in wids] except Exception as err: - logger.error(f"%s: Error occurred in %s", err, self.__class__.__name__) - self.cancel() + logger.error("%s: Error occurred in cancelling workers", err) def cancel(self): - """Cancel all tasks in the queue""" try: - self.queue_task.cancel() - except asyncio.CancelledError: + self.queue_task.cancel() + self.queue_task_cancelled = True + self.cancel_all_workers() + except asyncio.CancelledError as _: ... - except Exception as err: - logger.error("%s: occurred in canceling all tasks", err) + logger.error("%s: Error occurred in cancelling queue", err) def sigint_handle(self, sig, frame): - logger.info('SIGINT received, cleaning up...') if self.stop is False: self.stop = True else: - self.stop = True self.cancel() diff --git a/src/aiomql/lib/strategy.py b/src/aiomql/lib/strategy.py index 9d5bb26..b57e4b0 100644 --- a/src/aiomql/lib/strategy.py +++ b/src/aiomql/lib/strategy.py @@ -91,6 +91,10 @@ class Strategy(ABC): except Exception as err: logger.error(f"Error: {err}") + async def initialize(self): + """Perform any initialization tasks here.""" + ... + @staticmethod async def live_sleep(*, secs: float): """Sleep for the needed amount of seconds in between requests to the terminal. @@ -168,6 +172,7 @@ class Strategy(ABC): """Run the strategy.""" async with self as _: logger.info("Running %s strategy on %s", self.name, self.symbol.name) + await self.initialize() while self.running: try: await self.sessions.check()