3.0.4

src/aiomql/bot_builder.py

@@ -1,14 +1,14 @@
 import asyncio
-from typing import Type, Iterable, TypeVar
+from typing import Type, Iterable, TypeVar, Callable, Coroutine
 import logging
 
 from .executor import Executor
 from .account import Account
 from .symbol import Symbol as _Symbol
 from .strategy import Strategy as _Strategy
-# from
 
 logger = logging.getLogger(__name__)
+
 Strategy = TypeVar('Strategy', bound=_Strategy)
 Symbol = TypeVar('Symbol', bound=_Symbol)
 
@@ -41,11 +41,26 @@ class Bot:
         await self.init_symbols()
         self.executor.remove_workers()
 
-    def add_func(self, func, kwargs):
-        self.executor.add_func(func, kwargs)
+    def add_function(self, func: Callable, **kwargs: dict):
+        """Add a function to the executor.
 
-    def add_coro(self, coro, **kwargs):
-        self.executor.add_coro(coro, kwargs)
+        Args:
+            func (Callable): A function to be executed
+            **kwargs (dict): Keyword arguments for the function
+        """
+        self.executor.add_function(func, kwargs)
+
+    def add_coroutine(self, coro: Coroutine, **kwargs):
+        """Add a coroutine to the executor.
+
+        Args:
+            coro (Coroutine): A coroutine to be executed
+            **kwargs (dict): keyword arguments for the coroutine
+
+        Returns:
+
+        """
+        self.executor.add_coroutine(coro, kwargs)
 
     def execute(self):
         """Execute the bot.

src/aiomql/candle.py

@@ -1,7 +1,7 @@
 """Candle and Candles classes for handling bars from the MetaTrader 5 terminal."""
 
 from typing import Type, TypeVar, Generic, Iterable
-from logging import  getLogger
+from logging import getLogger
 import reprlib
 
 from pandas import DataFrame, Series

src/aiomql/executor.py

@@ -11,23 +11,23 @@ class Executor:
     Attributes:
         executor (ThreadPoolExecutor): The executor object.
         workers (list): List of strategies.
-        coros (dict[Coroutine, dict]): A dictionary of coroutines to run in the executor
-        funcs (dict[Callable, dict]): A dictionary of functions to run in the executor
+        coroutines (dict[Coroutine, dict]): A dictionary of coroutines to run in the executor
+        functions (dict[Callable, dict]): A dictionary of functions to run in the executor
 
     """
 
     def __init__(self, bot=None):
         self.executor = ThreadPoolExecutor
         self.workers: list[type(Strategy)] = []
-        self.coros: dict[Coroutine|Callable: dict] = {}
-        self.funcs: dict[Callable: dict] = {}
+        self.coroutines: dict[Coroutine | Callable: dict] = {}
+        self.functions: dict[Callable: dict] = {}
         self.bot = bot
 
-    def add_func(self, func, kwargs):
-        self.funcs[func] = kwargs | {'bot': self.bot}
+    def add_function(self, func: Callable, kwargs: dict):
+        self.functions[func] = kwargs | {'bot': self.bot}
 
-    def add_coro(self, coro, kwargs):
-        self.coros[coro] = kwargs | {'bot': self.bot}
+    def add_coroutine(self, coro: Coroutine, kwargs: dict):
+        self.coroutines[coro] = kwargs | {'bot': self.bot}
 
     def add_workers(self, strategies: Sequence[type(Strategy)]):
         """Add multiple strategies at once
@@ -38,8 +38,7 @@ class Executor:
         self.workers.extend(strategies)
 
     def remove_workers(self):
-        """Removes any worker running on a symbol not successfully initialized.
-        """
+        """Removes any worker running on a symbol not successfully initialized."""
         self.workers = [worker for worker in self.workers if worker.symbol in self.bot.symbols]
 
     def add_worker(self, strategy: type(Strategy)):
@@ -78,10 +77,10 @@ class Executor:
         Notes:
             No matter the number specified, the executor will always use a minimum of 5 workers.
         """
-        workers = workers or sum([len(self.workers), len(self.funcs), len(self.coros)])
+        workers = workers or sum([len(self.workers), len(self.functions), len(self.coroutines)])
         workers = max(workers, 5)
         loop = asyncio.get_running_loop()
         with self.executor(max_workers=workers) as executor:
             [loop.run_in_executor(executor, self.trade, worker) for worker in self.workers]
-            [loop.run_in_executor(executor, self.run, coro, kwargs) for coro, kwargs in self.coros.items()]
-            [loop.run_in_executor(executor, func, kwargs) for func, kwargs in self.funcs.items()]
+            [loop.run_in_executor(executor, self.run, coro, kwargs) for coro, kwargs in self.coroutines.items()]
+            [loop.run_in_executor(executor, func, kwargs) for func, kwargs in self.functions.items()]

src/aiomql/positions.py

@@ -7,6 +7,7 @@ from .order import Order
 
 logger = getLogger(__name__)
 
+
 class Positions:
     """Get Open Positions.
 
@@ -18,7 +19,7 @@ class Positions:
         mt5 (MetaTrader): MetaTrader instance.
     """
     mt5: MetaTrader = MetaTrader()
-    
+
     def __init__(self, *, symbol: str = "", group: str = "", ticket: int = 0):
         """Get Open Positions.
 
@@ -61,22 +62,29 @@ class Positions:
             return []
         return [TradePosition(**pos._asdict()) for pos in positions]
 
+    async def close(self, *, ticket: int, symbol: str, price: float, volume: float, order_type: OrderType):
+        """Close an open position for the trading account."""
+
+        order = Order(action=TradeAction.DEAL, price=price, position=ticket, symbol=symbol, volume=volume,
+                      type=order_type.opposite)
+        return await order.send()
+
     async def close_all(self, symbol: str = '', group: str = '') -> int:
         """Close all open positions for the trading account.
 
+        Keyword Args:
+            symbol (str): Financial instrument name.
+            group (str): The filter for specifying a group of symbols.
+
         Returns:
             int: Return number of positions closed.
         """
-        orders = [Order(action=TradeAction.DEAL, price=pos.price_current, position=pos.ticket,
-                        type=OrderType(pos.type).opposite,
-                        **pos.get_dict(include={'symbol', 'volume'})) for pos in
-                  (await self.positions_get(symbol=symbol, group=group))]
+        symbol = symbol or self.symbol
+        group = group or self.group
+        positions = [pos for pos in await self.positions_get(symbol=symbol, group=group)]
+        orders = [self.close(price=pos.price_current, ticket=pos.ticket, order_type=pos.type, volume=pos.volume,
+                             symbol=pos.symbol) for pos in positions]
 
-        results = await asyncio.gather(*[order.send() for order in orders], return_exceptions=True)
+        results = await asyncio.gather(*[order for order in orders], return_exceptions=True)
         amount_closed = len([res for res in results if res.retcode == 10009])
-        pos = await self.positions_total()
-        if pos > 0:
-            logger.warning(f'Failed to close {pos} positions')
-        else:
-            logger.info('All positions closed')
         return amount_closed

src/aiomql/ram.py

@@ -11,28 +11,24 @@ class RAM:
     pips: float
     volume: float
 
-    def __init__(self, **kwargs):
-        """Risk Assessment and Management. All provided keyword arguments are set as attributes.
+    def __init__(self, *, risk_to_reward: float = 1, risk: float = 0.01, amount: float = 0, pips: float = 0, volume=0):
+        """Initialize Risk Assessment and Management with the provided keyword arguments.
 
-        Args:
-            kwargs (Dict): Keyword arguments.
-
-        Defaults:
-            risk_to_reward (float): Risk to reward ratio 1
+        Keyword Args:
+            risk_to_reward (float): Risk to reward ratio. Defaults to 1
             risk (float): Percentage of account balance to risk per trade 0.01 # 1%
             amount (float): Amount to risk per trade in terms of account currency 0
-            pips (float): Target pips 0
+            pips (float): Target pips to risk
             volume (float): Volume to trade 0
         """
-        self.risk_to_reward = kwargs.pop('risk_to_reward', 1)
-        self.risk = kwargs.pop('risk', 0.01)
-        self.amount = kwargs.pop('amount', 0)
-        self.pips = kwargs.pop('pips', 0)
-        self.volume = kwargs.pop('volume', 0)
-        [setattr(self, key, value) for key, value in kwargs.items()]
+        self.risk_to_reward = risk_to_reward
+        self.risk = risk
+        self.amount = amount
+        self.pips = pips
+        self.volume = volume
 
     async def get_amount(self, risk: float = 0) -> float:
-        """Calculate the amount to risk per trade as a percentage of free margin.
+        """Calculate the amount to risk per trade as a percentage of equity.
 
         Keyword Args:
             risk (float): Percentage of account balance to risk per trade. Defaults to zero.
@@ -42,16 +38,15 @@ class RAM:
         """
         await self.account.refresh()
         risk = risk or self.risk
-        return self.account.margin_free * risk
+        return self.account.equity * risk
 
     async def get_volume(self, *, symbol: Symbol, pips: float = 0, amount: float = 0) -> float:
         """Calculate the volume to trade. if pips is not provided, the pips attribute is used.
-        If the amount attribute or amount argument is zero, the amount is calculated using the get_amount method based on the risk.
-
-        Args:
-            symbol (Symbol): Financial instrument
+        If the amount attribute or amount argument is zero, the amount is calculated using the get_amount method based
+        on the risk.
 
         Keyword Args:
+            symbol (Symbol): Financial instrument
             pips (float): Target pips. Defaults to zero.
             amount (float): Amount to risk per trade. Defaults to zero.
 
@@ -61,4 +56,3 @@ class RAM:
         pips = pips or self.pips
         amount = amount or self.amount or await self.get_amount()
         return await symbol.compute_volume(amount=amount, pips=pips)
-

src/aiomql/sessions.py

@@ -1,41 +1,174 @@
+"""Sessions allow you to run code at specific times of the day."""
+import asyncio
 from datetime import time, timedelta, datetime
-from asyncio import sleep
+from asyncio import sleep, iscoroutinefunction
+from typing import Literal, Callable
+from logging import getLogger
+
+from .positions import Positions
+
+logger = getLogger(__name__)
 
 
 class Session:
+    """A session is a time period between two datetime.time objects specified in utc.
 
-    def __init__(self, start: int | time, end: int | time):
+    Attributes:
+        start (datetime.time): The start time of the session.
+        end (datetime.time): The end time of the session.
+        on_start (str): The action to take when the session starts. Default is None.
+        on_end (str): The action to take when the session ends. Default is None.
+        custom_start (Callable): A custom function to call when the session starts. Default is None.
+        custom_end (Callable): A custom function to call when the session ends. Default is None.
+
+    Methods:
+        begin: Call the action specified in on_start or custom_start.
+        close: Call the action specified in on_end or custom_end.
+        action: Used by begin and close to call the action specified.
+        delta: Get the timedelta of a datetime.time object.
+        until: Get the seconds until the session starts from the current time.
+    """
+    def __init__(self, *, start: int | time, end: int | time,
+                 on_start: Literal['close_all', 'close_win', 'close_loss', 'custom_start'] = None,
+                 on_end: Literal['close_all', 'close_win', 'close_loss', 'custom_end'] = None,
+                 custom_start: Callable = None, custom_end: Callable = None):
+        """Create a session.
+
+        Keyword Args:
+            start (int | datetime.time): The start time of the session in UTC.
+            end (int | datetime.time): The end time of the session in UTC.
+            on_start (Literal['close_all', 'close_win', 'close_loss', 'custom_start']): The action to take when the
+                session starts. Default is None.
+            on_end (Literal['close_all', 'close_win', 'close_loss', 'custom_end']): The action to take when the session
+                ends. Default is None.
+            custom_start (Callable): A custom function to call when the session starts. Default is None.
+            custom_end (Callable): A custom function to call when the session ends. Default is None.
+        """
         self.start = start if isinstance(start, time) else time(hour=start)
         self.end = end if isinstance(end, time) else time(hour=end)
         self.__from = self.delta(self.start)
         self.__to = self.delta(self.end)
+        self.on_start = on_start
+        self.on_end = on_end
+        self.custom_start = custom_start
+        self.custom_end = custom_end
 
     def __contains__(self, item: time):
         return self.start <= item < self.end
 
-    def delta(self, obj):
+    def __repr__(self):
+        return f'{self.start}<-{len(self)}->{self.end}'
+
+    async def begin(self):
+        """Call the action specified in on_start or custom_start."""
+        await self.action(self.on_start)
+
+    async def close(self):
+        """Call the action specified in on_end or custom_end."""
+        await self.action(self.on_end)
+
+    async def action(self, action):
+        """Used by begin and close to call the action specified.
+
+        Args:
+            action (Literal['close_all', 'close_win', 'close_loss', 'custom_start', 'custom_end']): The action to take.
+        """
+        try:
+            position = Positions()
+            positions = await position.positions_get()
+
+            match action:
+                case 'close_all':
+                    await asyncio.gather(*(position.close(price=pos.price_current, ticket=pos.ticket,
+                                                          order_type=pos.type, volume=pos.volume,
+                                                          symbol=pos.symbol) for pos in positions),
+                                         return_exceptions=True)
+
+                case 'close_win':
+                    await asyncio.gather(
+                        *(position.close(price=pos.price_current, ticket=pos.ticket, order_type=pos.type,
+                                         volume=pos.volume, symbol=pos.symbol) for pos in positions if pos.profit > 0),
+                        return_exceptions=True)
+
+                case 'close_loss':
+                    await asyncio.gather(
+                        *(position.close(price=pos.price_current, ticket=pos.ticket, order_type=pos.type,
+                                         volume=pos.volume, symbol=pos.symbol) for pos in positions if
+                          pos.profit < 0), return_exceptions=True)
+
+                case 'custom_end':
+                    if iscoroutinefunction(self.custom_end):
+                        await self.custom_end()
+                    self.custom_end()
+
+                case 'custom_start':
+                    if iscoroutinefunction(self.custom_start):
+                        await self.custom_start()
+                    self.custom_start()
+
+                case _:
+                    pass
+        except Exception as exe:
+            logger.warning(f'Failed to call action {action} due to {exe}')
+
+    @staticmethod
+    def delta(obj: time):
+        """Get the timedelta of a datetime.time object.
+
+        Args:
+            obj (datetime.time): A datetime.time object.
+        """
         return timedelta(hours=obj.hour, minutes=obj.minute, seconds=obj.second, microseconds=obj.microsecond)
 
     def __len__(self):
         return (self.__to - self.__from).seconds
 
     def until(self):
-        now = lambda: datetime.utcnow().time()
-        return (self.__from - self.delta(now())).seconds
+        """Get the seconds until the session starts from the current time."""
+        return (self.__from - self.delta(datetime.utcnow().time())).seconds
 
 
 class Sessions:
+    """Sessions allow you to run code at specific times of the day. It is a collection of Session objects.
+    Sessions are sorted by start time. The sessions object is an asynchronous context manager.
+
+    Attributes:
+        sessions (list[Session]): A list of Session objects.
+        current_session (Session): The current session.
+
+    Methods:
+        find: Find a session that contains a datetime.time object.
+        find_next: Find the next session that contains a datetime.time object.
+        check: Check if the current session has started and if not, wait until it starts.
+    """
     def __init__(self, *sessions: Session):
         self.sessions = list(sessions)
         self.sessions.sort(key=lambda x: x.start)
+        self.current_session = sessions[0]
 
-    def find(self, obj):
+    def find(self, obj: time) -> Session | None:
+        """Find a session that contains a datetime.time object.
+
+        Args:
+            obj (datetime.time): A datetime.time object.
+
+        Returns:
+            Session | None: A Session object or None if not found.
+        """
         for session in self.sessions:
             if obj in session:
                 return session
         return None
 
-    def find_next(self, obj):
+    def find_next(self, obj: time) -> Session:
+        """Find the next session that contains a datetime.time object.
+
+        Args:
+            obj (datetime.time): A datetime.time object.
+
+        Returns:
+            Session: A Session object.
+        """
         for session in self.sessions:
             if obj < session.start:
                 return session
@@ -45,16 +178,24 @@ class Sessions:
         return True if self.find(item) is not None else False
 
     async def __aenter__(self):
+        await self.check()
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        pass
+        await self.current_session.close()
 
     async def check(self):
+        """Check if the current session has started and if not, wait until it starts."""
         now = datetime.utcnow().time()
-        if now in self:
+        if now in self.current_session:
             return
-        next_session = self.find_next(now)
-        secs = next_session.until()
-        print(f'sleeping for {secs} seconds')
-        await sleep(secs)
+        await self.current_session.close()
+
+        current_session = self.find(now)
+        if current_session is None:
+            current_session = self.find_next(now)
+            secs = current_session.until() + 10
+            print(f'sleeping for {secs} seconds until next session')
+            await sleep(secs)
+        self.current_session = current_session
+        await self.current_session.begin()

src/aiomql/strategy.py

@@ -35,7 +35,7 @@ class Strategy(ABC):
     mt5: MetaTrader()
     config = Config()
 
-    def __init__(self, *, symbol: Symbol, params: dict = None, session: Session):
+    def __init__(self, *, symbol: Symbol, params: dict = None, sessions: Sessions = None):
         """Initiate the parameters dict and add name and symbol fields.
         Use class name as strategy name if name is not provided
 
@@ -47,7 +47,7 @@ class Strategy(ABC):
         self.parameters = params.copy() if isinstance(params, dict) else {}
         self.parameters['symbol'] = symbol.name
         self.parameters['name'] = self.name or self.__class__.__name__
-        self.session = session or Sessions(Session(8, 13))
+        self.sessions = sessions or Sessions(Session(start=0, end=23))
 
     def __repr__(self):
         return f"{self.name}({self.symbol!r})"