gavindiaz/aiomql
1
0
Fork 0
mirror of https://github.com/Ichinga-Samuel/aiomql.git synced 2026-08-16 05:18:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

v4.0.9

Browse Source
This commit is contained in:
Ichinga Samuel
2025-01-22 15:56:57 +01:00
parent 4601b251ab
commit 12f61f3b6e
4 changed files with 132 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1 -1
View File
@@ -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 = [
src/aiomql/core/backtesting/get_data.py
+6 -6
View File
@@ -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
src/aiomql/core/task_queue.py
+120 -74
View File
@@ -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()
src/aiomql/lib/strategy.py
+5
View File
@@ -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()