gavindiaz/aiomql
1
0
Fork 0
mirror of https://github.com/Ichinga-Samuel/aiomql.git synced 2026-08-15 12:58:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

4.0.15b

Browse Source
This commit is contained in:
Ichinga Samuel
2025-08-10 04:14:00 +01:00
parent bae2983bde
commit 1dede5515e
33 changed files with 1768 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "aiomql"
version = "4.0.14b"
version = "4.0.15b"
readme = "README.md"
requires-python = ">=3.11"
classifiers = [
@@ -16,7 +16,7 @@ classifiers = [
keywords = ["MetaTrader5", "Asynchronous", "Algorithmic Trading", "Trading Bot", "Backtesting",
"Technical Analysis", "Forex", "Stocks", "Cryptocurrency", "Futures", "Options", "Crypto", "Algo Trading"]
dependencies = ["MetaTrader5>=5.0.37", "pandas>=1.5.0", "pandas-ta>=0.3.14b0", "mplfinance>=0.10.1"]
dependencies = ["MetaTrader5>=5.0.5200", "pandas>=1.5.0", "pandas-ta>=0.3.14b0", "mplfinance>=0.10.1"]
authors = [{name = "Ichinga Samuel", email = "ichingasamuel@gmail.com"}]
requirements.txt
+2 -2
View File
@@ -1,4 +1,4 @@
-e git+ssh://git@github.com/Chimezirim-Bassey/aiomql.git@8e0a5516774894505d8d915bbd510bf88c6e3e48#egg=aiomql
-e git+ssh://git@github.com/Chimezirim-Bassey/aiomql.git@bae2983bded5b3912f7234fd0acaa89ab84fa1c4#egg=aiomql
anyio==4.3.0
argon2-cffi==23.1.0
argon2-cffi-bindings==21.2.0
@@ -71,7 +71,7 @@ MarkupSafe==2.1.3
matplotlib==3.8.4
matplotlib-inline==0.1.6
mdurl==0.1.2
MetaTrader5==5.0.4424
MetaTrader5==5.0.5200
mistune==3.0.2
more-itertools==10.1.0
mplfinance==0.12.10b0
src/aiomql/contrib/quants/change.py
-37
View File
@@ -1,10 +1,3 @@
def relative_difference(first: float, second: float) -> float:
"""Find the relative difference between two values"""
diff = abs(first - second)
div = (first + second) / 2
return (diff / div) * 100
def percentage_difference(first: float, second: float) -> float:
"""Find the percentage difference between two values"""
diff = abs(first - second)
@@ -12,23 +5,11 @@ def percentage_difference(first: float, second: float) -> float:
return (diff / div) * 100
def relative_position(start: float, end: float, value: float):
"""Find the relative position of a value between two values"""
return (value - start) / (end - start)
def percentage_position(start: float, end: float, value: float):
"""Find the percentage position of a value between two values"""
return ((value - start) / (end - start)) * 100
def get_relative_position(start: float, end: float, factor: float):
"""Find position within an interval based on a factor"""
span = end - start
position = start + (factor * span)
return position
def get_percentage_position(start: float, end: float, rate: float):
"""Find position within an interval based on a percentage"""
span = end - start
@@ -36,12 +17,6 @@ def get_percentage_position(start: float, end: float, rate: float):
return position
def extend_interval_by_factor(start: float, end: float, factor: float):
"""Extend the end of an interval by a factor"""
span = end - start
span *= factor
return end + span
def extend_interval_by_percentage(start: float, end: float, rate: float):
"""Extend the end of an interval by a factor"""
span = end - start
@@ -49,28 +24,16 @@ def extend_interval_by_percentage(start: float, end: float, rate: float):
return end + span
def relative_change(start_value: float, end_value: float) -> float:
"""Find the relative change from one value to another"""
return (end_value - start_value) / start_value
def percentage_change(start_value: float, end_value: float) -> float:
"""Find the percentage change from one value to another"""
return ((end_value - start_value) / start_value) * 100
def relative_increase(value: float, factor: float) -> float:
"""Increase the value by a factor"""
return value * (1 + factor)
def percentage_increase(value: float, rate: float) -> float:
"""Increase the value by a factor"""
return value * ((100 + rate) / 100)
def relative_decrease(value: float, factor: float) -> float:
"""Increase the value by a factor"""
return value * (1 - factor)
def percentage_decrease(value: float, rate: float) -> float:
"""Increase the value by a factor"""
return value * ((100 - rate) / 100)
src/aiomql/contrib/strategies/finger_trap.py
+2 -2
View File
@@ -1,6 +1,6 @@
import logging
from ...lib.symbol import Symbol
from ..symbols import ForexSymbol
from ...lib.trader import Trader
from ...lib.candle import Candles
from ...lib.strategy import Strategy
@@ -29,7 +29,7 @@ class FingerTrap(Strategy):
parameters = {"fast_ema": 8, "slow_ema": 20, "etf": TimeFrame.M5, "ttf": TimeFrame.H1,
"entry_ema": 5, "tcc": 720, "ecc": 8640}
def __init__(self, *,symbol: Symbol, params: dict | None = None, trader: Trader = None, sessions: Sessions = None,
def __init__(self, *, symbol: ForexSymbol, params: dict | None = None, trader: Trader = None, sessions: Sessions = None,
name: str = "FingerTrap"):
super().__init__(symbol=symbol, params=params, sessions=sessions, name=name)
self.trader = trader or SimpleTrader(symbol=self.symbol)
src/aiomql/contrib/trackers/open_position.py
+65 -30
View File
@@ -1,11 +1,12 @@
from dataclasses import dataclass, field
from typing import ClassVar
from typing import ClassVar, Self
from logging import getLogger
from ...lib import Symbol, Positions, Order
from ...core.models import TradePosition, TradeAction, OrderSendResult
from ...core.constants import OrderType
from ...core.config import Config
from ..quants import percentage_increase, percentage_decrease
from .position_tracker import PositionTracker
logger = getLogger()
@@ -17,9 +18,13 @@ class OpenPosition:
ticket: int
position: TradePosition
is_open: bool = True
trackers: list[PositionTracker] = field(default_factory=list)
use_checkpoint: bool = False
checkpoint: float = 0
checkpoint: float = None
is_hedged: bool = False
is_a_hedge: bool = False
hedge: Self | None = None
pending_hedge: OrderSendResult | None = None
_trackers: dict[str, PositionTracker] = field(default_factory=dict)
positions: ClassVar[Positions]
state_key: ClassVar[str] = "tracked_positions"
config: ClassVar[Config]
@@ -34,6 +39,16 @@ class OpenPosition:
def __post_init__(self):
self.config.state.setdefault(self.state_key, {}).setdefault(self.ticket, self)
def add_tracker(self, *, tracker: PositionTracker, number: int = None, name: str = ""):
number = number or len(self._trackers) + 1
tracker.set_position(self, number)
self._trackers[name or tracker.function.__name__] = tracker
@property
def trackers(self):
for tracker in sorted(self._trackers, key=lambda key: self._trackers[key].number):
yield self._trackers[tracker]
async def remove_closed(self):
try:
await self.update_position()
@@ -42,13 +57,6 @@ class OpenPosition:
except Exception as exe:
logger.error("%s: Unable to remove closed position from state", exe)
def add_tracker(self, *, tracker: PositionTracker, number: int = None):
number = number or len(self.trackers)
tracker.set_position(self, number)
self.trackers.append(tracker)
self.trackers.sort(key=lambda x: x.number)
async def update_position(self) -> bool:
pos = await self.positions.get_position_by_ticket(ticket=self.ticket)
if pos is not None:
@@ -121,6 +129,16 @@ class OpenPosition:
self.symbol.name, self.ticket)
return False, None
@staticmethod
async def check_pending_order(self):
await self.update_position()
if self.is_open and self.is_hedged and self.pending_hedge is not None:
pos = await self.positions.get_position_by_ticket(ticket=self.pending_hedge.order)
if pos is not None:
self.pending_hedge = None
self.hedge = OpenPosition(symbol=self.symbol, position=pos, ticket=pos.ticket,
is_a_hedge=True, hedge=self)
async def track(self):
try:
for tracker in self.trackers:
@@ -129,35 +147,52 @@ class OpenPosition:
logger.error("%s: Error occurred in track method of Open Position for %d:%s",
exe, self.symbol.name, self.ticket)
async def get_price_from_profit(self, profit):
action = OrderType.BUY if self.position.type == 0 else OrderType.SELL
volume = self.position.volume
price_open = self.position.price_open
price_close = percentage_increase(price_open, 50) if action == 0 else percentage_decrease(price_open, 50)
half_profit = await Order.mt5.order_calc_profit(symbol=self.symbol.name, action=action, volume=volume,
price_open=price_open, price_close=price_close)
rate = profit / half_profit * 50
rate = percentage_increase(price_open, rate) if action == 0 else percentage_decrease(price_open, rate)
return rate
async def hedge_position(self, *, hedge_params: dict = None) -> tuple[bool, OrderSendResult | None]:
async def hedge_order(self, price, **order_params) -> tuple[bool, OrderSendResult | None]:
try:
order_type = OrderType.SELL_STOP if self.position.type == OrderType.BUY else OrderType.BUY_STOP
volume = order_params.get("volume", self.position.volume)
order = Order(action=TradeAction.PENDING, symbol=self.symbol.name, type=order_type, price=price, volume=volume)
res = await order.send()
if res.retcode != 10009:
return False, res
self.pending_hedge = res
self.is_hedged = True
self.add_tracker(tracker=PositionTracker(self.check_pending_order), number=0, name="check_pending_order")
return True, res
except Exception as exe:
logger.error("%s: Error occurred in hedge_order method of Open Position for %d:%s", exe, self.symbol.name, self.ticket)
return False, None
async def hedge_position(self, *, hedge_params: dict = None) -> tuple[bool, Self]:
try:
# hedge_params for customizing the hedge
hedge_params = hedge_params or {}
volume = hedge_params.get("volume", self.position.volume)
order_type = hedge_params.get("type") or OrderType(self.position.type).opposite
tick = await self.symbol.info_tick()
price = hedge_params.get("price") or (tick.ask if order_type == OrderType.BUY else tick.bid)
if (price := hedge_params.get("price", None)) is None:
tick = await self.symbol.info_tick()
price = (tick.ask if order_type == OrderType.BUY else tick.bid)
comment = hedge_params.get("comment", f"Hedge_{self.ticket}")
hedge_order = Order(type=order_type, symbol=self.symbol.name, volume=volume, price=price, comment=comment)
# if position has stops get them from the sl and tp of the position
if self.position.sl and self.position.tp:
osl = abs(self.position.sl - self.position.price_open)
otp = abs(self.position.tp - self.position.price_open)
if order_type == OrderType.BUY:
sl = hedge_params.get("sl") or tick.ask - osl
tp = hedge_params.get("tp") or tick.ask + otp
hedge_order.set_attributes(tp=tp, sl=sl)
if order_type == OrderType.SELL:
sl = hedge_params.get("sl") or tick.bid + osl
tp = hedge_params.get("sl") or tick.bid - otp
hedge_order.set_attributes(tp=tp, sl=sl)
res = await hedge_order.send()
if res.retcode == 10009:
return True, res
return False, res
if res.retcode != 10009:
return False, None
self.is_hedged = True
hedge_pos = await self.positions.get_position_by_ticket(ticket=res.order)
self.hedge = OpenPosition(symbol=self.symbol, ticket=res.order, position=hedge_pos,
hedge=self, is_a_hedge=True)
return True, self.hedge
except Exception as exe:
logger.error("%s: Error occurred in %s.hedge_position for %s:%d", exe, self.__class__.__name__,
self.symbol.name, self.ticket)
src/aiomql/contrib/trackers/position_tracking_functions.py
+11 -12
View File
@@ -1,6 +1,5 @@
from logging import getLogger
from aiomql import OrderType
from .open_position import OpenPosition
from ..quants import extend_interval_by_percentage, get_percentage_position, percentage_position
@@ -8,7 +7,7 @@ from ..quants import extend_interval_by_percentage, get_percentage_position, per
logger = getLogger(__name__)
async def exit_at_price(*, pos: OpenPosition, tp: float = None, sl: float = None):
async def exit_at_price(pos: OpenPosition, /, tp: float = None, sl: float = None):
"""Exit a trade at a particular price"""
if not await pos.update_position():
return
@@ -19,7 +18,7 @@ async def exit_at_price(*, pos: OpenPosition, tp: float = None, sl: float = None
logger.warning("Unable to close %s:%d due to %s in exit_at_price", pos.symbol, pos.ticket, res.comment)
async def extend_take_profit(*, pos: OpenPosition, increase: float = 20, start: float = 80,
async def extend_take_profit(pos: OpenPosition, /, increase: float = 20, start: float = 80,
use_stop_levels: bool = True):
is_open = await pos.update_position()
if is_open is False or pos.position.profit < 0:
@@ -35,13 +34,13 @@ async def extend_take_profit(*, pos: OpenPosition, increase: float = 20, start:
position.symbol, position.ticket, res.comment)
async def extend_stop_loss(*, pos: OpenPosition, increase: float = 20, start: float = 80,
async def extend_stop_loss(pos: OpenPosition, /, increase: float = 20, start: float = 80,
use_stop_levels: bool = True):
is_open = await pos.update_position()
if is_open is False or pos.position.profit > 0:
return
position = pos.position
if percentage_position(position.price_open, position.tp, position.price_current) >= start:
if percentage_position(position.price_open, position.sl, position.price_current) >= start:
new_sl = extend_interval_by_percentage(position.price_open, position.sl, increase)
ok, res = await pos.modify_stops(sl=new_sl, use_stop_levels=use_stop_levels)
if ok:
@@ -52,17 +51,17 @@ async def extend_stop_loss(*, pos: OpenPosition, increase: float = 20, start: fl
position.symbol, position.ticket, res.comment)
async def exit_at_checkpoint(*, pos: OpenPosition, start: float = 80, trail: float = 15):
async def exit_at_checkpoint(pos: OpenPosition, /, start: float = 80, trail: float = 15):
is_open = await pos.update_position()
if is_open is False or pos.position.profit < 0:
if is_open is False:
return
position = pos.position
if percentage_position(position.price_open, position.tp, position.price_current) >= start:
if position.profit > 0 and percentage_position(position.price_open, position.tp, position.price_current) >= start:
new_checkpoint = get_percentage_position(position.price_open, position.price_current, 100-trail)
change_checkpoint = False
if position.type == OrderType.BUY and new_checkpoint > pos.checkpoint:
if position.type.long and new_checkpoint > (pos.checkpoint or position.price_open):
change_checkpoint = True
elif position.type == OrderType.SELL and new_checkpoint < pos.checkpoint:
elif position.type.short and new_checkpoint < (pos.checkpoint or position.price_open):
change_checkpoint = True
if change_checkpoint:
pos.checkpoint = new_checkpoint
@@ -70,10 +69,10 @@ async def exit_at_checkpoint(*, pos: OpenPosition, start: float = 80, trail: flo
logger.info("New checkpoint created for %s:%d at %f:%f",
position.symbol, position.ticket, new_checkpoint, position.profit)
close = False
if position.type == OrderType.BUY and position.price_current <= pos.checkpoint and pos.use_checkpoint:
if position.type.long and position.price_current <= pos.checkpoint and pos.use_checkpoint:
close = True
elif position.type == OrderType.SELL and position.price_current >= pos.checkpoint and pos.use_checkpoint:
elif position.type.short and position.price_current >= pos.checkpoint and pos.use_checkpoint:
close = True
if close:
src/aiomql/contrib/trackers/positions_tracker.py
+8 -7
View File
@@ -2,7 +2,7 @@ import asyncio
from logging import getLogger
from typing import ClassVar
from ...core import Config, State, Store, sleep
from ...core import Config, State, sleep
from ...lib import Positions
@@ -10,19 +10,20 @@ logger = getLogger(__name__)
class OpenPositionsTracker:
state: ClassVar[State]
store: ClassVar[Store]
config: ClassVar[Config]
positions: ClassVar[Positions]
state: ClassVar[State]
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "config"):
cls.config = Config()
if not hasattr(cls, "state"):
cls.state = cls.config.state
if not hasattr(cls, "positions"):
cls.positions = Positions()
if not hasattr(cls, "store"):
cls.store = cls.config.store
if not hasattr(cls, "state"):
cls.state = State()
return super().__new__(cls)
def __init__(self, interval: int = 10, state_key: str = "tracked_positions", autocommit=False):
src/aiomql/core/base.py
+15 -6
View File
@@ -19,8 +19,8 @@ class Base:
include (set [str]): A set of attributes to be included when retrieving attributes
using the get_dict and dict method.
"""
exclude: set[str]
include: set[str]
exclude: set[str] = {"mt5", "config", "exclude", "include", "annotations", "class_vars", "dict", "_instance"}
include: set[str] = {}
def __init__(self, **kwargs):
"""
@@ -29,8 +29,6 @@ class Base:
Args:
**kwargs: Set instance attributes with keyword arguments. Only if they are annotated on the class body.
"""
self.exclude = {"mt5", "config", "exclude", "include", "annotations", "class_vars", "dict", "_instance"}
self.include = set()
self.set_attributes(**kwargs)
def __repr__(self):
@@ -138,7 +136,18 @@ class _Base(Base):
"""
mt5: MetaTrader | MetaBackTester
config: Config
def __new__(cls, *args, **kwargs):
if not hasattr(cls, 'config'):
cls.config = Config()
if not hasattr(cls, 'mt5'):
cls.mt5 = MetaTrader() if cls.config.mode != "backtest" else MetaBackTester()
return super().__new__(cls)
def __getstate__(self):
state = self.__dict__.copy()
state.pop("mt5", None)
return state
def __init__(self, **kwargs):
self.__class__.config = Config()
self.__class__.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester()
super().__init__(**kwargs)
src/aiomql/core/config.py
+8 -4
View File
@@ -45,7 +45,9 @@ class Config:
shutdown: bool
force_shutdown: bool
db_commit_interval: float
auto_commit: bool = False
auto_commit: bool
flush_state: bool
stop_trading: bool
lock: Lock
_defaults = {
"timeout": 60000,
@@ -67,7 +69,9 @@ class Config:
"root": None,
"plots_dir_name": "plots",
"db_commit_interval": 30,
"auto_commit": False
"auto_commit": False,
"flush_state": False,
"stop_trading": False
}
def __new__(cls, *args, **kwargs):
@@ -224,10 +228,10 @@ class Config:
self._store = value
def init_state(self):
self.state = State(db_name=self.db_name)
self.state = State(db_name=self.root / self.db_name, flush=self.flush_state)
def init_store(self):
self.store = Store(db_name=self.db_name)
self.store = Store(db_name=self.root / self.db_name, flush=self.flush_state)
@property
def records_dir(self):
src/aiomql/core/constants.py
+15 -10
View File
@@ -25,15 +25,13 @@ class TradeAction(Repr, IntEnum):
"""TRADE_REQUEST_ACTION Enum.
Attributes:
DEAL (int): Delete the pending order placed previously Place a trade order for an immediate execution with the
specified parameters (market order).
PENDING (int): Delete the pending order placed previously
DEAL (int): Place a trade order for an immediate execution with the specified parameters (market order)
PENDING (int): Place a trade order for the execution under specified conditions (pending order)
SLTP (int): Modify Stop Loss and Take Profit values of an opened position
MODIFY (int): Modify the parameters of the order placed previously
REMOVE (int): Delete the pending order placed previously
CLOSE_BY (int): Close a position by an opposite one
"""
__enum_name__ = "TRADE_ACTION"
DEAL = mt5.TRADE_ACTION_DEAL
PENDING = mt5.TRADE_ACTION_PENDING
@@ -93,12 +91,12 @@ class OrderType(Repr, IntEnum):
Attributes:
BUY (int): Market buy order
SELL (int): Market sell order
BUY_LIMIT (int): Buy Limit pending order
SELL_LIMIT (int): Sell Limit pending order
BUY_STOP (int): Buy Stop pending order
SELL_STOP (int): Sell Stop pending order
BUY_STOP_LIMIT (int): Upon reaching the order price, Buy Limit pending order is placed at StopLimit price
SELL_STOP_LIMIT (int): Upon reaching the order price, Sell Limit pending order is placed at StopLimit price
BUY_LIMIT (int): Buy Limit pending order, buy when price drops to level below the current price
SELL_LIMIT (int): Sell Limit pending order, sell when price rises to a level above the current price
BUY_STOP (int): Buy Stop pending order, buy when price rises to a level above the current price
SELL_STOP (int): Sell Stop pending order, sell when price drops to a level below the current price
BUY_STOP_LIMIT (int): Buy stop limit order, buy when price rises to a level above the current price, with a limit to how far above the current price
SELL_STOP_LIMIT (int): Sell stop limit order, sell when price drops to a level below the current price, with a limit to how far below the current price
CLOSE_BY (int): Order for closing a position by an opposite one
Properties:
@@ -126,6 +124,13 @@ class OrderType(Repr, IntEnum):
_type = {0: 1, 1: 0, 2: 3, 3: 2, 4: 5, 5: 4, 6: 7, 7: 6, 8: 8}[self]
return OrderType(_type)
@property
def long(self):
return self in [0, 2, 4, 6]
@property
def short(self):
return self in [1, 3, 5, 7]
class BookType(Repr, IntEnum):
"""BOOK_TYPE Enum.
src/aiomql/core/meta_trader.py
+5 -1
View File
@@ -28,8 +28,12 @@ logger = getLogger()
class MetaTrader(MetaCore):
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "config"):
cls.config = Config()
return super().__new__(cls)
def __init__(self):
self.config = Config()
self.error: Error = Error(code=1)
self._lock = asyncio.Lock()
src/aiomql/core/models.py
+9
View File
@@ -542,6 +542,15 @@ class OrderSendResult(Base):
profit: float = None
loss: float = None
def __getstate__(self):
state = self.__dict__.copy()
state["request"] = state.pop('request')._asdict()
return state
def __setstate__(self, state):
state['request'] = mt5.TradeRequest(state['request'])
self.__dict__.update(state)
class TradePosition(Base):
"""Trade Position
src/aiomql/core/state.py
+5 -4
View File
@@ -1,6 +1,7 @@
import os
import pickle
import sqlite3
from pathlib import Path
from threading import Lock
from typing import Self
from typing import MutableMapping, Iterable, Any, ClassVar
@@ -27,7 +28,7 @@ class State(MutableMapping):
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, db_name: str = "", data: dict = None, flush: bool = False, autocommit: bool = False):
def __init__(self, db_name: str | Path = "", data: dict = None, flush: bool = False, autocommit: bool = False):
with self._lock:
self.autocommit = autocommit
self.init(data=data, flush=flush, db_name=db_name)
@@ -99,14 +100,14 @@ class State(MutableMapping):
def items(self):
return self.data.items()
def load(self, *, conn = None, db_name: str = "", data: dict = None):
def load(self, *, conn = None, data: dict = None):
try:
db_name = db_name or self.db_name
conn = conn or self.conn
res = conn.execute("SELECT value FROM state where key = 'data'").fetchone()
db_data = pickle.loads(res[0]) if res else {}
db_data |= (data or {})
self.update(db_data)
conn.close()
except Exception as err:
logger.error("%s: Failed to load data from database", err)
@@ -121,7 +122,7 @@ class State(MutableMapping):
self.commit(conn=conn)
else:
self.load(conn=conn, data=data)
conn.close()
return
if flush:
self.data = data or {}
src/aiomql/core/store.py
+2 -1
View File
@@ -2,6 +2,7 @@ import os
import sqlite3
from typing import MutableMapping, Iterable, Any
from logging import getLogger
from pathlib import Path
SENTINEL = object()
@@ -14,7 +15,7 @@ class Store(MutableMapping):
cursor: sqlite3.Cursor
conn: sqlite3.Connection
def __init__(self, db_name: str = "", data: dict = None, flush: bool = False, autocommit: bool = True):
def __init__(self, db_name: str | Path = "", data: dict = None, flush: bool = False, autocommit: bool = True):
self.autocommit = autocommit
self.db_name = db_name or os.environ.get("DB_NAME", "db.sqlite3")
self.conn = sqlite3.connect(self.db_name, check_same_thread=False)
src/aiomql/core/sync/__init__.py
+1
View File
@@ -0,0 +1 @@
from .meta_trader import MetaTrader
src/aiomql/core/sync/meta_trader.py
+344
View File
@@ -0,0 +1,344 @@
from datetime import datetime
from logging import getLogger
from typing import Literal, Self
from pathlib import Path
import numpy as np
from MetaTrader5 import (
BookInfo,
SymbolInfo,
AccountInfo,
Tick,
TerminalInfo,
TradeOrder,
TradeDeal,
TradePosition,
OrderSendResult,
OrderCheckResult,
)
from ..constants import OrderType, CopyTicks
from .._core import MetaCore
from ..errors import Error
from ..config import Config
logger = getLogger()
class MetaTrader(MetaCore):
def __init__(self):
self.config = Config()
self.error: Error = Error(code=1)
def __enter__(self) -> Self:
"""
Async context manager entry point.
Initializes the connection to the MetaTrader terminal.
Returns:
MetaTrader: An instance of the MetaTrader class.
"""
self.initialize()
self.login()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""
Async context manager exit point. Closes the connection to the MetaTrader terminal.
"""
self.shutdown()
def _handler(self, api: dict, retries=3):
func = api["func"]
args = api.get("args", ())
kwargs = api.get("kwargs", {})
error_msg = api.get("error_msg", "An error occurred")
res = func(*args, **kwargs)
if res is not None:
return res
if res is None and self.error.is_connection_error() and retries > 0:
self.initialize()
self.login()
return self._handler(api, retries=retries - 1)
else:
err = self.last_error()
self.error = Error(*err)
logger.warning(f"{error_msg}:{self.error.description}")
return res
def login(self, *, login: int = 0, password: str = "", server: str = "", timeout: int = 60000) -> bool:
"""
Connects to the MetaTrader terminal using the specified login, password and server.
Args:
login (int): The trading account number.
password (str): The trading account password.
server (str): The trading server name.
timeout (int): The timeout for the connection in seconds.
Returns:
bool: True if successful, False otherwise.
"""
acc_details = self.config.account_info
login = login or acc_details.get("login", 0)
password = password or acc_details.get("password", "")
server = server or acc_details.get("server", "")
return self._login(login, password=password, server=server, timeout=timeout)
def initialize(self, path: str = None, login: int = 0, password: str = "", server: str = "",
timeout: int | None = None, portable=False) -> bool:
"""
Initializes the connection to the MetaTrader terminal. All parameters are optional.
Keyword Args:
path (str): The path to the MetaTrader terminal executable.
login (int): The trading account number.
password (str): The trading account password.
server (str): The trading server name.
timeout (int): The timeout for the connection in milliseconds.
portable (bool): If True, the terminal will be launched in portable mode.
Returns:
bool: True if successful, False otherwise.
"""
path = self.config.path if path is None else path
path = "" if Path(path).exists() is False else path
args = (str(path),) if path else ()
acc = self.config.account_info
kwargs = {key: value for key, value in (("login", login or acc.get("login")),
("password", password or acc.get("password")),
("server", server or acc.get("server")), ("timeout", timeout or 45000),
("portable", portable)) if key is not None}
res = self._initialize(*args, **kwargs)
if not res:
err = self._last_error()
self.error = Error(*err)
return res
def shutdown(self) -> None:
"""Closes the connection to the MetaTrader terminal."""
self._shutdown()
def last_error(self) -> tuple[int, str]:
try:
return self.last_error()
except Exception as err:
logger.warning("%s: Error in obtaining last error.", err)
return -1, str(err)
def version(self) -> tuple[int, int, str] | None:
""""""
api = {"func": self._version, "error_msg": "Error in obtaining version."}
return self._handler(api)
def account_info(self) -> AccountInfo | None:
""""""
api = {"func": self._account_info, "error_msg": "Error in obtaining account information"}
return self._handler(api)
def terminal_info(self) -> TerminalInfo | None:
api = {"func": self._terminal_info, "error_msg": "Error in obtaining terminal information"}
return self._handler(api)
def symbols_total(self) -> int:
api = {"func": self._symbols_total, "error_msg": "Error in obtaining total symbols."}
return self._handler(api)
def symbols_get(self, group: str = "") -> tuple[SymbolInfo] | None:
kwargs = {"group": group} if group else {}
api = {"func": self._symbols_get, "kwargs": kwargs, "error_msg": "Error in obtaining symbols."}
return self._handler(api)
def symbol_info(self, symbol: str) -> SymbolInfo | None:
api = {
"func": self._symbol_info,
"args": (symbol,),
"error_msg": f"Error in obtaining information for {symbol}",
}
return self._handler(api)
def symbol_info_tick(self, symbol: str) -> Tick | None:
api = {"func": self._symbol_info_tick, "args": (symbol,), "error_msg": f"Error in obtaining tick for {symbol}"}
return self._handler(api)
def symbol_select(self, symbol: str, enable: bool) -> bool:
api = {"func": self._symbol_select, "args": (symbol, enable), "error_msg": f"Error in selecting {symbol}"}
return self._handler(api)
def market_book_add(self, symbol: str) -> bool:
api = {
"func": self._market_book_add,
"args": (symbol,),
"error_msg": f"Error in adding {symbol} to market book",
}
return self._handler(api)
def market_book_get(self, symbol: str) -> tuple[BookInfo] | None:
api = {
"func": self._market_book_get,
"args": (symbol,),
"error_msg": f"Error in obtaining market depth for {symbol}",
}
return self._handler(api)
def market_book_release(self, symbol: str) -> bool:
api = {
"func": self._market_book_release,
"args": (symbol,),
"error_msg": f"Error in releasing market depth for {symbol}",
}
return self._handler(api)
def copy_rates_from(
self, symbol: str, timeframe: int, date_from: datetime | float, count: int
) -> np.ndarray | None:
api = {
"func": self._copy_rates_from,
"args": (symbol, timeframe, date_from, count),
"error_msg": f"Error in obtaining rates for {symbol}",
}
return self._handler(api)
def copy_rates_from_pos(self, symbol: str, timeframe: int, start_pos: int, count: int) -> np.ndarray | None:
api = {
"func": self._copy_rates_from_pos,
"args": (symbol, timeframe, start_pos, count),
"error_msg": f"Error in obtaining rates for {symbol}",
}
return self._handler(api)
def copy_rates_range(
self, symbol: str, timeframe: int, date_from: datetime | float, date_to: datetime | float
) -> np.ndarray | None:
api = {
"func": self._copy_rates_range,
"args": (symbol, timeframe, date_from, date_to),
"error_msg": f"Error in obtaining rates for {symbol}",
}
return self._handler(api)
def copy_ticks_from(
self, symbol: str, date_from: datetime | float, count: int, flags: CopyTicks
) -> np.ndarray | None:
api = {
"func": self._copy_ticks_from,
"args": (symbol, date_from, count, flags),
"error_msg": f"Error in obtaining ticks for {symbol}",
}
return self._handler(api)
def copy_ticks_range(
self, symbol: str, date_from: datetime | float, date_to: datetime | float, flags: CopyTicks
) -> np.ndarray | None:
api = {
"func": self._copy_ticks_range,
"args": (symbol, date_from, date_to, flags),
"error_msg": f"Error in obtaining ticks for {symbol}",
}
return self._handler(api)
def orders_total(self) -> int:
api = {"func": self._orders_total, "error_msg": "Error in obtaining total orders."}
return self._handler(api)
def orders_get(self, group: str = "", ticket: int = 0, symbol: str = "") -> tuple[TradeOrder] | None:
kwargs = {key: value for key, value in (("group", group), ("ticket", ticket), ("symbol", symbol)) if value}
api = {"func": self._orders_get, "kwargs": kwargs, "error_msg": "Error in obtaining orders."}
return self._handler(api)
def order_calc_margin(
self, action: Literal[OrderType.BUY, OrderType.SELL], symbol: str, volume: float, price: float
) -> float | None:
api = {
"func": self._order_calc_margin,
"args": (action, symbol, volume, price),
"error_msg": "Error in calculating margin.",
}
return self._handler(api)
def order_calc_profit(
self,
action: Literal[OrderType.BUY, OrderType.SELL],
symbol: str,
volume: float,
price_open: float,
price_close: float,
) -> float | None:
api = {
"func": self._order_calc_profit,
"args": (action, symbol, volume, price_open, price_close),
"error_msg": "Error in calculating profit.",
}
return self._handler(api)
def order_check(self, request: dict) -> OrderCheckResult:
api = {"func": self._order_check, "args": (request,), "error_msg": "Error in checking order."}
return self._handler(api)
def order_send(self, request: dict) -> OrderSendResult:
api = {"func": self._order_send, "args": (request,), "error_msg": "Error in sending order."}
return self._handler(api)
def positions_total(self) -> int:
api = {"func": self._positions_total, "error_msg": "Error in obtaining total positions."}
return self._handler(api)
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}
api = {"func": self._positions_get, "kwargs": kwargs, "error_msg": "Error in obtaining open positions."}
return self._handler(api)
def history_orders_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
api = {
"func": self._history_orders_total,
"args": (date_from, date_to),
"error_msg": "Error in obtaining total history orders.",
}
return self._handler(api)
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)
api = {
"func": self._history_orders_get,
"args": args,
"kwargs": kwargs,
"error_msg": "Error in obtaining history orders",
}
return self._handler(api)
def history_deals_total(self, date_from: datetime | float, date_to: datetime | float) -> int:
api = {
"func": self._history_deals_total,
"args": (date_from, date_to),
"error_msg": "Error in obtaining total history deals",
}
return self._handler(api)
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)
api = {
"func": self._history_deals_get,
"args": args,
"kwargs": kwargs,
"error_msg": "Error in obtaining history deals",
}
return self._handler(api)
src/aiomql/lib/account.py
+10 -3
View File
@@ -1,6 +1,6 @@
from threading import Lock
from logging import getLogger
from typing import Self
from typing import Self, ClassVar
from ..core.base import _Base
from ..core.models import AccountInfo
@@ -17,7 +17,7 @@ class Account(_Base, AccountInfo):
connected (bool): Status of connection to MetaTrader 5 Terminal
"""
_instance: Self
_lock: Lock
_lock: ClassVar[Lock]
connected: bool
def __new__(cls, *args, **kwargs):
@@ -26,6 +26,7 @@ class Account(_Base, AccountInfo):
cls._lock = lock
cls._instance = super().__new__(cls)
cls._instance.connected = False
cls._instance.exclude.add("_lock")
return cls._instance
async def __aenter__(self) -> Self:
@@ -51,7 +52,13 @@ class Account(_Base, AccountInfo):
async def refresh(self):
"""Refreshes the account instance with the latest account details from the MetaTrader 5 terminal"""
account_info = await self.mt5.account_info
account_info = await self.mt5.account_info()
acc = account_info._asdict()
self.connected = True
self.set_attributes(**acc)
def refresh_sync(self):
account_info = self.mt5._account_info()
acc = account_info._asdict()
self.connected = True
self.set_attributes(**acc)
src/aiomql/lib/bot.py
+35 -12
View File
@@ -7,6 +7,7 @@ import logging
from .executor import Executor
from ..core.config import Config
from ..core.meta_trader import MetaTrader
from ..core.meta_backtester import MetaBackTester
from .symbol import Symbol as Symbol
from .strategy import Strategy as Strategy
@@ -25,12 +26,16 @@ class Bot:
executor: Executor
mt: MetaTrader
strategies: list[Strategy]
initialized: bool
login: bool
def __init__(self):
self.config = Config(bot=self)
self.executor = Executor()
self.mt5 = MetaTrader()
self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester()
self.strategies = []
self.initialized = False
self.login = False
@classmethod
def process_pool(cls, processes: dict[Callable:dict] = None, num_workers: int = None):
@@ -46,17 +51,36 @@ class Bot:
for bot, kwargs in processes.items():
executor.submit(bot, **kwargs)
async def start_terminal(self):
"""Start terminal and login asynchronously"""
res = await self.mt5.initialize()
if res:
self.initialized = True
res = await self.mt5.login()
if res:
self.login = True
return res
def start_terminal_sync(self):
"""Start terminal and login synchronously"""
res = self.mt5.initialize_sync()
if res:
self.initialized = True
res = self.mt5.login_sync()
if res:
self.login = True
return res
async def initialize(self):
"""Prepares the bot by signing in to the trading account and initializing the symbols for each strategy.
Only strategies with successfully initialized symbols will be added to the executor. Starts the global task queue.
Raises:
SystemExit if sign in was not successful
SystemExit if sign_in was not successful
"""
try:
await self.mt5.initialize()
login = await self.mt5.login()
if not login:
await self.start_terminal()
if not self.login:
logger.critical("Unable to sign in to MetaTrder 5 Terminal")
raise Exception("Unable to sign in to MetaTrader 5 Terminal")
logger.info("Login Successful")
@@ -78,12 +102,11 @@ class Bot:
Starts the global task queue.
Raises:
SystemExit if sign in was not successful
SystemExit if sign_in was not successful
"""
try:
self.mt5.initialize_sync()
login = self.mt5.login_sync()
if not login:
self.start_terminal_sync()
if not self.login:
logger.critical("Unable to sign in to MetaTrder 5 Terminal")
raise Exception("Unable to sign in to MetaTrader 5 Terminal")
logger.info("Login Successful")
@@ -163,7 +186,7 @@ class Bot:
async def init_strategy(self, *, strategy: Strategy) -> bool:
"""Initialize a single strategy. This method is called internally by the bot."""
res = await strategy.symbol.initialize()
res = await strategy.initialize()
if res:
self.executor.add_strategy(strategy=strategy)
return res
@@ -171,11 +194,11 @@ class Bot:
async def init_strategies(self):
"""Initialize the symbols for the current trading session. This method is called internally by the bot."""
tasks = [self.init_strategy(strategy=strategy) for strategy in self.strategies]
await asyncio.gather(*tasks)
await asyncio.gather(*tasks, return_exceptions=True)
def init_strategy_sync(self, *, strategy: Strategy) -> bool:
"""Initialize a single strategy. This method is called internally by the bot."""
res = strategy.symbol.initialize_sync()
res = strategy.initialize_sync()
if res:
self.executor.add_strategy(strategy=strategy)
return res
src/aiomql/lib/history.py
+19 -13
View File
@@ -1,9 +1,8 @@
import asyncio
from datetime import datetime
from typing import ClassVar
from datetime import datetime, UTC
from logging import getLogger
import pytz
from ..core.config import Config
from ..core.meta_trader import MetaTrader
from ..core.models import TradeDeal, TradeOrder
@@ -24,34 +23,41 @@ class History:
mt5 (MetaTrader): MetaTrader instance
config (Config): Config instance
"""
mt5: MetaTrader | MetaBackTester
config: Config
mt5: ClassVar[MetaTrader | MetaBackTester]
config: ClassVar[Config]
deals: tuple[TradeDeal, ...]
orders: tuple[TradeOrder, ...]
total_deals: int
total_orders: int
group: str
def __new__(cls, *args, **kwargs):
instance = super().__new__(cls)
if not hasattr(instance.__class__, 'config'):
instance.__class__.config = Config()
if not hasattr(instance.__class__, 'mt5'):
instance.__class__.mt5 = MetaTrader() if instance.config.mode != "backtest" else MetaBackTester()
return instance
def __init__(
self, *, date_from: datetime | float, date_to: datetime | float, group: str = "", use_utc: bool = True
self, *, date_from: datetime | float, date_to: datetime | float, group: str = "", use_utc: bool = float
):
"""
Args:
date_from (datetime, float): Date the orders are requested from. Set by the 'datetime' object or as a
number of seconds elapsed since 1970.01.01. Defaults to twenty-four hours from the current time in 'utc'
number of seconds elapsed since 1970.01.01.
date_to (datetime, float): Date up to which the orders are requested. Set by the 'datetime' object or as a
number of seconds elapsed since 1970.01.01. Defaults to the current time in "utc"
number of seconds elapsed since 1970.01.01.
use_utc (bool): Convert date_from and date_to to UTC. Default is False.
group (str): Filter for selecting history by symbols. This defaults to an empty string
"""
self.config = Config()
self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester()
date_from = date_from if isinstance(date_from, datetime) else datetime.fromtimestamp(date_from)
date_to = date_to if isinstance(date_to, datetime) else datetime.fromtimestamp(date_to)
self.date_from = date_from.astimezone(pytz.UTC) if use_utc else date_from
self.date_to = date_to.astimezone(pytz.UTC) if use_utc else date_to
self.date_from = date_from.astimezone(UTC) if use_utc else date_from
self.date_to = date_to.astimezone(UTC) if use_utc else date_to
self.group = group
self.deals: tuple[TradeDeal, ...] = ()
self.orders: tuple[TradeOrder, ...] = ()
src/aiomql/lib/order.py
-1
View File
@@ -11,7 +11,6 @@ logger = getLogger(__name__)
class Order(_Base, TradeRequest):
"""Trade order related functions and properties. Subclass of TradeRequest."""
def __init__(self, **kwargs):
"""Initialize the order object with keyword arguments, symbol must be provided.
Provide default values for action, type_time and type_filling if not provided.
src/aiomql/lib/positions.py
+45 -18
View File
@@ -20,22 +20,24 @@ class Positions:
Attributes:
mt5 (MetaTrader): MetaTrader instance.
"""
mt5: MetaTrader | MetaBackTester
positions: tuple[TradePosition, ...]
total_positions: int
config: Config
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "config"):
cls.config = Config()
if not hasattr(cls, "mt5"):
cls.mt5 = MetaTrader()
return super().__new__(cls)
def __init__(self):
"""Get Open Positions"""
self.config = Config()
self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester()
self.positions = ()
self.total_positions = 0
async def get_positions(self, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]:
"""Get open positions with the ability to filter by symbol, ticket or group of symbols.
Args:
symbol (Optional[str]): Financial instrument name. If symbol is provided, ticket is ignored.
symbol (Optional[str]): Financial instrument name. If a symbol is provided, the ticket is ignored.
ticket (Optional[int]): Position ticket.
group (Optional[str]): Group of symbols.
@@ -53,12 +55,28 @@ class Positions:
positions = await self.mt5.positions_get(**kwargs)
if positions is not None:
self.positions = tuple(TradePosition(**pos._asdict()) for pos in positions)
self.total_positions = len(self.positions)
return self.positions
logger.warning("Failed to get open positions")
return ()
async def get_position_by_ticket(self, *, ticket: int) -> TradePosition | None:
@classmethod
async def get_all_positions(cls, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]:
kwargs = {}
if symbol is not None:
kwargs["symbol"] = symbol
ticket = None
if ticket is not None:
kwargs["ticket"] = ticket
if group is not None:
kwargs["group"] = group
positions = await cls.mt5.positions_get(**kwargs)
if positions is not None:
return cls.positions
logger.warning("Failed to get open positions")
return ()
@classmethod
async def get_position_by_ticket(cls, *, ticket: int) -> TradePosition | None:
"""Get an open position by ticket.
Args:
ticket (int): Position ticket.
@@ -66,13 +84,14 @@ class Positions:
Returns:
TradePosition: Return an open position
"""
positions = await self.mt5.positions_get(ticket=ticket)
positions = await cls.mt5.positions_get(ticket=ticket)
position = positions[0] if positions else None
if position is None or position.ticket != ticket:
return None
return TradePosition(**position._asdict())
async def get_positions_by_symbol(self, *, symbol: str) -> tuple[TradePosition, ...]:
@classmethod
async def get_positions_by_symbol(cls, *, symbol: str) -> tuple[TradePosition, ...]:
"""Get open positions by symbol.
Args:
symbol (str): Financial instrument name.
@@ -80,7 +99,7 @@ class Positions:
Returns:
tuple[TradePosition, ...]: A tuple of open trade positions
"""
positions = await self.mt5.positions_get(symbol=symbol)
positions = await cls.mt5.positions_get(symbol=symbol)
return tuple(TradePosition(**pos._asdict()) for pos in (positions or ()))
@staticmethod
@@ -104,9 +123,10 @@ class Positions:
)
return await order.send()
async def close_position_by_ticket(self, *, ticket: int) -> OrderSendResult | None:
@classmethod
async def close_position_by_ticket(cls, *, ticket: int) -> OrderSendResult | None:
"""Close an open position using the ticket."""
position = await self.get_position_by_ticket(ticket=ticket)
position = await cls.get_position_by_ticket(ticket=ticket)
if position is None:
return None
order = Order(
@@ -144,8 +164,15 @@ class Positions:
)
return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)])
async def get_total_positions(self) -> int:
@classmethod
async def close_all_positions(cls):
positions = await cls.mt5.positions_get()
results = await asyncio.gather(
*(cls.close_position(position=position) for position in positions), return_exceptions=True
)
return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)])
@classmethod
async def get_total_positions(cls) -> int:
"""Get the total number of open positions."""
total = await self.mt5.positions_total()
self.total_positions = total or self.total_positions
return self.total_positions
return await cls.mt5.positions_total()
src/aiomql/lib/sessions.py
+5 -5
View File
@@ -2,6 +2,7 @@ import asyncio
from datetime import time, timedelta, datetime, UTC
from typing import Literal, Callable, Iterable, NamedTuple
from logging import getLogger
from time import sleep
from ..core.models import OrderSendResult, TradePosition
from ..core.config import Config
@@ -26,11 +27,11 @@ def delta(obj: time) -> timedelta:
async def backtest_sleep(secs):
"""An async sleep function for use during backtesting."""
"""A sleep function for use during backtesting."""
config = Config()
btc = config.backtest_controller
sleep = config.backtest_engine.cursor.time + secs
while sleep > config.backtest_engine.cursor.time:
sleep_secs = config.backtest_engine.cursor.time + secs
while sleep_secs > config.backtest_engine.cursor.time:
btc.wait()
@@ -59,7 +60,6 @@ class Session:
name: str = "",
):
"""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.
@@ -119,7 +119,7 @@ class Session:
return Duration(hours=hours, minutes=minutes, seconds=seconds)
async def close_positions(self, *, positions: tuple[TradePosition, ...]):
results = asyncio.gather(
results = await asyncio.gather(
*(self.positions_manager.close_position(position=position) for position in positions),
return_exceptions=True,
)
src/aiomql/lib/strategy.py
+16 -24
View File
@@ -1,21 +1,19 @@
"""The base class for creating strategies."""
import asyncio
from time import time
from typing import TypeVar
from abc import ABC, abstractmethod
import time
from abc import ABC
from datetime import time as dtime
from logging import getLogger
from ..core.meta_trader import MetaTrader
from .sessions import Sessions, Session
from .symbol import Symbol
from ..core import Config
from ..core.backtesting.backtest_controller import BackTestController
from ..core.exceptions import StopTrading
from ..core.meta_backtester import MetaBackTester
from ..core.backtesting.backtest_controller import BackTestController
from .sessions import Sessions, Session
from .symbol import Symbol as _Symbol
from ..core.meta_trader import MetaTrader
Symbol = TypeVar("Symbol", bound=_Symbol)
logger = getLogger(__name__)
@@ -93,7 +91,10 @@ class Strategy(ABC):
async def initialize(self):
"""Perform any initialization tasks here."""
...
return await self.symbol.initialize()
def initialize_sync(self):
return self.symbol.initialize_sync()
@staticmethod
async def live_sleep(*, secs: float):
@@ -104,7 +105,7 @@ class Strategy(ABC):
Args:
secs (float): The time in seconds. Usually the timeframe you are trading on.
"""
mod = time() % secs
mod = time.time() % secs
secs = secs - mod if mod != 0 else mod
await asyncio.sleep(secs + 0.1)
@@ -117,26 +118,19 @@ class Strategy(ABC):
secs (float): The time in seconds. Usually the timeframe you are trading on.
"""
if self.config.mode == "backtest":
await self.backtest_sleep(secs=secs)
self.backtest_sleep(secs=secs)
else:
await self.live_sleep(secs=secs)
async def delay(self, *, secs: float):
"""Sleep for the input amount of seconds"""
if self.config.mode == "backtest":
await self._backtest_sleep(secs=secs)
self._backtest_sleep(secs=secs)
else:
await asyncio.sleep(secs)
async def _backtest_sleep(self, *, secs: float):
def _backtest_sleep(self, *, secs: float):
try:
# if self.backtest_controller.parties == 2:
# this is not needed, as the backtest_engine iterator will handle this
# steps = int(secs) // self.config.backtest_engine.speed
# steps = max(steps, 1)
# self.config.backtest_engine.fast_forward(steps=steps)
# self.backtest_controller.wait()
if self.backtest_controller.parties >= 2:
_time = self.config.backtest_engine.cursor.time + secs
while _time > self.config.backtest_engine.cursor.time:
@@ -147,7 +141,7 @@ class Strategy(ABC):
self.backtest_controller.wait()
logger.error("Error: %s in backtest_sleep", err)
async def backtest_sleep(self, *, secs: float):
def backtest_sleep(self, *, secs: float):
"""Sleep for the needed amount of seconds in between requests to the terminal.
Args:
@@ -157,7 +151,7 @@ class Strategy(ABC):
_time = self.config.backtest_engine.cursor.time
mod = _time % secs
secs = secs - mod if mod != 0 else mod
await self._backtest_sleep(secs=secs)
self._backtest_sleep(secs=secs)
except Exception as err:
logger.error("Error: %s in backtest_sleep", err)
@@ -172,7 +166,6 @@ 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()
@@ -208,7 +201,6 @@ class Strategy(ABC):
logger.error(f"Error: {err} in backtest_strategy")
return
@abstractmethod
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.
src/aiomql/lib/symbol.py
+4 -2
View File
@@ -25,7 +25,7 @@ class Symbol(_Base, SymbolInfo):
Notes:
Full properties are on the SymbolInfo Object.
Make sure Symbol is always initialized with a name argument
Make sure a Symbol is always initialized with a name argument
"""
initialized: bool
tick: Tick
@@ -46,7 +46,7 @@ class Symbol(_Base, SymbolInfo):
"""Get the current price tick of a financial instrument.
Args:
name: if name is supplied get price tick of that financial instrument. Optional unnamed parameter.
name: if name is supplied, get price tick of that financial instrument. Optional unnamed parameter.
Returns:
Tick: Return a Tick Object
@@ -112,12 +112,14 @@ class Symbol(_Base, SymbolInfo):
self.set_attributes(**info._asdict())
info_tick = await self.mt5.symbol_info_tick(self.name)
if info_tick:
self.tick = Tick(**info_tick._asdict())
if info is not None and info_tick is not None:
self.initialized = True
return True
logger.warning("Unable to initialize %s", self.name)
return False
except Exception as err:
src/aiomql/lib/sync/__init__.py
+1
View File
@@ -0,0 +1 @@
from .lib import *
src/aiomql/lib/sync/order.py
+148
View File
@@ -0,0 +1,148 @@
from logging import getLogger
from ...core.models import TradeRequest, TradeOrder, OrderCheckResult, OrderSendResult
from ...core.constants import TradeAction, OrderTime, OrderFilling
from ...core.exceptions import OrderError
from ...core.base import _Base
from ...core.sync.meta_trader import MetaTrader
from ...utils import error_handler_sync
logger = getLogger(__name__)
class Order(_Base, TradeRequest):
"""Trade order related functions and properties. Subclass of TradeRequest."""
mt5: MetaTrader
def __init__(self, **kwargs):
"""Initialize the order object with keyword arguments, symbol must be provided.
Provide default values for action, type_time and type_filling if not provided.
Args:
**kwargs: Keyword arguments must match the attributes of TradeRequest as well as the attributes of
Order class as specified in the annotations in the class definition.
Default Values:
action (TradeAction.DEAL): Trade action
type_time (OrderTime.DAY): Order time
type_filling (OrderFilling.FOK): Order filling
"""
kwargs = {"action": TradeAction.DEAL, "type_time": OrderTime.DAY, "type_filling": OrderFilling.FOK, **kwargs}
super().__init__(**kwargs)
def modify(self, **kwargs):
"""Modify the order object with keyword arguments.
Args:
**kwargs: Keyword arguments must match the attributes of TradeRequest as well as the attributes of
Order class as specified in the annotations in the class definition.
"""
self.set_attributes(**kwargs)
def orders_total(self):
"""Get the number of active pending orders.
Returns:
(int): total number of active pending orders
"""
return self.mt5.orders_total()
def get_pending_order(self, *, ticket: int) -> TradeOrder | None:
"""
Get a pending order by ticket number.
Args:
ticket (int): Order ticket number
Returns:
"""
orders = self.mt5.orders_get(ticket=ticket)
order = None
for order_ in orders:
if order_.ticket == ticket:
return TradeOrder(**order_._asdict())
return order
def get_pending_orders(self, *, ticket: int = 0, symbol: str = "", group: str = "") -> tuple[TradeOrder, ...]:
"""Get the list of active pending orders for the current symbol.
Args:
ticket (int): Order ticket number
symbol (str): Symbol name
group (str): Group name
Returns:
tuple[TradeOrder, ...]: A Tuple of active pending trade orders as TradeOrder objects
"""
orders = self.mt5.orders_get(symbol=symbol, ticket=ticket, group=group)
if orders is not None:
return tuple(TradeOrder(**order._asdict()) for order in orders)
return tuple()
def check(self, **kwargs) -> OrderCheckResult:
"""Check funds sufficiency for performing a required trading operation and the possibility of executing it.
Returns:
OrderCheckResult: An OrderCheckResult object
Raises:
OrderError: If not successful
"""
req = self.request | kwargs
res = self.mt5.order_check(req)
if res is None:
raise OrderError(f"Order check failed for {self.symbol}")
return OrderCheckResult(**res._asdict())
def send(self) -> OrderSendResult:
"""Send a request to perform a trading operation from the terminal to the trade server.
Returns:
OrderSendResult: An OrderSendResult object
Raises:
OrderError: If not successful
"""
res = self.mt5.order_send(self.request)
if res is None:
raise OrderError(f"Failed to send order {self.symbol}")
return OrderSendResult(**res._asdict())
@error_handler_sync(log_error_msg=False)
def calc_margin(self) -> float | None:
"""Return the required margin in the account currency to perform a specified trading operation.
Returns:
float: Returns float value if successful
"""
res = self.mt5.order_calc_margin(self.type, self.symbol, self.volume, self.price)
return res
@error_handler_sync(response=0, log_error_msg=False)
def calc_profit(self) -> float:
"""Return profit in the account currency for a specified trading operation.
Returns:
float: Returns float value if successful
None: If not successful
"""
action, symbol, volume, price_open, price_close = (self.type, self.symbol, self.volume, self.price, self.tp)
res = self.mt5.order_calc_profit(action, symbol, volume, price_open, price_close)
return res
@error_handler_sync(response=0, log_error_msg=False)
def calc_loss(self) -> float:
"""Return profit in the account currency for a specified trading operation.
Returns:
float: Returns float value if successful
None: If not successful
"""
action, symbol, volume, price_open, price_close = (self.type, self.symbol, self.volume, self.price, self.sl)
res = self.mt5.order_calc_profit(action, symbol, volume, price_open, price_close)
return res
@property
def request(self) -> dict:
"""Return the order request as a dictionary."""
return {key: value for key, value in self.dict.items() if key in self.mt5.TradeRequest.__match_args__}
src/aiomql/lib/sync/positions.py
+146
View File
@@ -0,0 +1,146 @@
"""Handle Open positions."""
from logging import getLogger
from ...core.meta_trader import MetaTrader
from ...core.models import TradePosition, OrderSendResult
from ...core.constants import OrderType, TradeAction
from ...core.config import Config
from ...core.sync.meta_trader import MetaTrader
from ..order import Order
logger = getLogger(__name__)
class Positions:
"""Get Open Positions.
Attributes:
mt5 (MetaTrader): MetaTrader instance.
"""
mt5: MetaTrader
config: Config
positions: tuple[TradePosition, ...]
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "config"):
cls.config = Config()
if not hasattr(cls, "mt5"):
cls.mt5 = MetaTrader()
return super().__new__(cls)
def __init__(self):
self.positions = ()
def get_positions(self, *, symbol: str = None, ticket: int = None, group: str = None) -> tuple[TradePosition, ...]:
"""Get open positions with the ability to filter by symbol, ticket or group of symbols.
Args:
symbol (Optional[str]): Financial instrument name. If a symbol is provided, the ticket is ignored.
ticket (Optional[int]): Position ticket.
group (Optional[str]): Group of symbols.
Returns:
tuple[TradePosition, ...]: A tuple of open trade positions
"""
kwargs = {}
if symbol is not None:
kwargs["symbol"] = symbol
ticket = None
if ticket is not None:
kwargs["ticket"] = ticket
if group is not None:
kwargs["group"] = group
positions = self.mt5.positions_get(**kwargs)
if positions is not None:
return tuple(TradePosition(**pos._asdict()) for pos in positions)
logger.warning("Failed to get open positions")
return ()
def get_position_by_ticket(self, *, ticket: int) -> TradePosition | None:
"""Get an open position by ticket.
Args:
ticket (int): Position ticket.
Returns:
TradePosition: Return an open position
"""
positions = self.mt5.positions_get(ticket=ticket)
position = positions[0] if positions else None
if position is None or position.ticket != ticket:
return None
return TradePosition(**position._asdict())
def get_positions_by_symbol(self, *, symbol: str) -> tuple[TradePosition, ...]:
"""Get open positions by symbol.
Args:
symbol (str): Financial instrument name.
Returns:
tuple[TradePosition, ...]: A tuple of open trade positions
"""
positions = self.mt5.positions_get(symbol=symbol)
return tuple(TradePosition(**pos._asdict()) for pos in (positions or ()))
@staticmethod
def close(*, ticket: int, symbol: str, price: float, volume: float, order_type: OrderType) -> OrderSendResult:
"""Close an open position for the trading account using the ticket and other parameters.
Args:
ticket (int): Position ticket.
symbol (str): Financial instrument name.
price (float): Closing price.
volume (float): Volume to close.
order_type (OrderType): Order type.
"""
order = Order(
action=TradeAction.DEAL,
price=price,
position=ticket,
symbol=symbol,
volume=volume,
type=order_type.opposite,
)
return order.send()
def close_position_by_ticket(self, *, ticket: int) -> OrderSendResult | None:
"""Close an open position using the ticket."""
position = self.get_position_by_ticket(ticket=ticket)
if position is None:
return None
order = Order(
position=position.ticket,
symbol=position.symbol,
volume=position.volume,
type=position.type.opposite,
price=position.price_current,
action=TradeAction.DEAL,
)
return order.send()
@staticmethod
def close_position(*, position: TradePosition):
"""Close an open position for the trading account. Using a position object."""
order = Order(
position=position.ticket,
symbol=position.symbol,
volume=position.volume,
type=position.type.opposite,
price=position.price_current,
action=TradeAction.DEAL,
)
return order.send()
def close_all(self) -> int:
"""Close all open positions for the trading account. Specify a symbol or group to filter positions.
Returns:
int: Return number of positions closed.
"""
positions = self.get_positions()
results = [self.close_position(position=position) for position in positions]
return len([res for res in results if (isinstance(res, OrderSendResult) and res.retcode == 10009)])
def get_total_positions(self) -> int:
"""Get the total number of open positions."""
return self.mt5.positions_total()
src/aiomql/lib/sync/sessions.py
+254
View File
@@ -0,0 +1,254 @@
from datetime import time, timedelta, datetime, UTC
from typing import Literal, Callable, Iterable, NamedTuple
from logging import getLogger
from time import sleep
from ...core.config import Config
from .positions import Positions
logger = getLogger(__name__)
class Duration(NamedTuple):
hours: int
minutes: int
seconds: int
def delta(obj: time) -> timedelta:
"""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 backtest_sleep(secs):
"""A sleep function for use during backtesting."""
config = Config()
btc = config.backtest_controller
sleep_secs = config.backtest_engine.cursor.time + secs
while sleep_secs > config.backtest_engine.cursor.time:
btc.wait()
class Session:
"""A session is a time period between two datetime.time objects specified in utc.
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.
name (str): A name for the session. Default is a combination of start and end.
"""
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,
name: str = "",
):
"""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.
name (str): A name for the session. Default is a combination of start and end.
"""
self.start = start.replace(tzinfo=UTC) if isinstance(start, time) else time(hour=start, tzinfo=UTC)
self.end = end if isinstance(end, time) else time(hour=end, tzinfo=UTC)
self.on_start = on_start
self.on_end = on_end
self.custom_start = custom_start
self.custom_end = custom_end
self.name = name or f"{self.start}<-->{self.end}"
self.positions_manager = Positions()
self.config = Config()
def __contains__(self, item: time):
span = (delta(self.end) - delta(self.start)).seconds
item_span = (delta(self.end) - delta(item)).seconds
return item_span <= span
def __str__(self):
return f"{self.start}<-->{self.end}"
def __repr__(self):
return f"{self.start}<-->{self.end}"
def __len__(self):
return int((delta(self.end) - delta(self.start)).seconds)
def in_session(self) -> bool:
"""Check if the current time is within the session."""
now = (
datetime.now(tz=UTC).time()
if self.config.mode == "live"
else datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time()
)
return now in self
def begin(self):
"""Call the action specified in on_start or custom_start."""
self.action(action=self.on_start)
def close(self):
"""Call the action specified in on_end or custom_end."""
self.action(action=self.on_end)
def duration(self) -> Duration:
"""Get the duration of the session in seconds."""
hours, seconds = divmod(len(self), 3600)
minutes, seconds = divmod(seconds, 60)
return Duration(hours=hours, minutes=minutes, seconds=seconds)
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:
match action:
case "close_all":
raise NotImplementedError("To be implemented")
case "close_win":
raise NotImplementedError("To be implemented")
case "close_loss":
raise NotImplementedError("To be implemented")
case "custom_end":
raise NotImplementedError("To be implemented")
case "custom_start":
raise NotImplementedError("To be implemented")
case _:
pass
except Exception as exe:
logger.warning(f"Failed to call action {action} due to {exe}")
def until(self):
"""Get the seconds until the session starts from the current time in seconds."""
if self.config.mode == "backtest":
now = datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time()
secs = (delta(self.start) - delta(now)).seconds
else:
secs = (delta(self.start) - delta(datetime.now(tz=UTC).time())).seconds
return secs
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.
"""
sessions: list[Session]
current_session: Session | None
def __init__(self, *, sessions: Iterable[Session]):
self.sessions = list(sessions)
self.sessions.sort(key=lambda x: (x.start.hour, x.end.hour))
self.current_session = None
self.config = Config()
def find(self, *, moment: time = None) -> Session | None:
"""Find a session that contains a datetime.time object, if not found return None.
Keyword Args:
moment (datetime.time | None): A datetime.time object. if not provided, the current time is used.
Returns:
Session | None: A Session object or None if not found.
"""
moment = (
moment or datetime.now(tz=UTC).time()
if self.config.mode == "live"
else (datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time())
)
for session in self.sessions:
if moment in session:
return session
return None
def find_next(self, *, moment: time = None) -> Session:
"""Find the next session that contains a datetime.time object.
Args:
moment (datetime.time | None): A datetime.time object, if not provided, the current time is used.
Returns:
Session: A Session object.
"""
moment = (
moment or datetime.now(tz=UTC).time()
if self.config.mode == "live"
else (datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time())
)
for session in self.sessions:
if delta(moment) < delta(session.start):
return session
return self.sessions[0]
def __contains__(self, moment: time):
return True if self.find(moment=moment) is not None else False
def __enter__(self):
self.check()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.current_session.close() if self.current_session is not None else ...
def check(self):
"""Check if the current session has started and if not, wait until it starts."""
if self.current_session is not None and self.current_session.in_session():
return
if self.config.mode == "backtest":
now = datetime.fromtimestamp(self.config.backtest_engine.cursor.time, tz=UTC).time()
else:
now = datetime.now(tz=UTC).time()
next_session = self.find(moment=now)
if next_session and self.current_session is None:
self.current_session = next_session
self.current_session.begin()
return
if next_session and self.current_session is not None:
self.current_session.close()
self.current_session = next_session
self.current_session.begin()
return
if next_session is None and self.current_session is not None:
self.current_session.close()
next_session = self.find_next(moment=now)
secs = next_session.until() + 10
logger.info(f"sleeping for {secs} seconds until next {next_session} session")
sleep_func = sleep if self.config.mode == "live" else backtest_sleep
sleep_func(secs)
self.current_session = next_session
self.current_session.begin()
src/aiomql/lib/sync/strategy.py
+200
View File
@@ -0,0 +1,200 @@
"""The base class for creating strategies."""
import time
from abc import ABC
from datetime import time as dtime
from logging import getLogger
from .sessions import Sessions, Session
from .symbol import Symbol
from ...core import Config
from ...core.backtesting.backtest_controller import BackTestController
from ...core.exceptions import StopTrading
from ...core.meta_backtester import MetaBackTester
from ...core.meta_trader import MetaTrader
logger = getLogger(__name__)
class Strategy(ABC):
"""The base class for creating strategies.
Attributes:
name (str): The name of the strategy.
symbol (Symbol): The Financial Instrument as a Symbol Object
parameters (Dict): A dictionary of parameters for the strategy.
sessions (Sessions): The sessions to use for the strategy.
running (bool): A flag to indicate if the strategy is running.
backtest_controller (BackTestController): A controller for running the backtester.
current_session (Session): The current session.
mt5 (MetaTrader|MetaBackTester): The MetaTrader object.
config (Config): The config object.
Notes:
Define the name of a strategy as a class attribute. If not provided, the class name will be used as the name.
"""
name: str
symbol: Symbol
sessions: Sessions
mt5: MetaTrader | MetaBackTester
config: Config
running: bool
parameters = {}
backtest_controller = BackTestController
current_session = Session
def __init__(self, *, symbol: Symbol, params: dict = None, sessions: Sessions = None, name=""):
"""Initiate the parameters dict and add name and symbol fields.
Use class name as strategy name if name is not provided
Args:
symbol (Symbol): The Financial instrument
params (Dict): Trading strategy parameters
"""
self.parameters = {**self.parameters} | (params or {})
self.symbol = symbol
self.name = name or self.__class__.__name__
self.parameters["symbol"] = symbol.name
self.parameters["name"] = self.name
self.running = True
self.sessions = sessions or Sessions(sessions=[Session(start=0, end=dtime(hour=23, minute=59, second=59))])
self.config = Config()
self.mt5 = MetaTrader() if self.config.mode != "backtest" else MetaBackTester()
self.backtest_controller = BackTestController()
def __repr__(self):
return f"{self.name}({self.symbol!r})"
def __getattr__(self, item):
if item in self.parameters:
return self.parameters[item]
raise AttributeError(f"{item} not an attribute of {self.name}")
def __setattr__(self, key, value):
if key in self.parameters:
self.parameters[key] = value
super().__setattr__(key, value)
def __enter__(self):
self.sessions.check()
self.running = True
self.current_session = self.sessions.current_session
def __exit__(self, exc_type, exc_val, exc_tb):
try:
self.current_session.close() if self.current_session else ...
self.running = False
except Exception as err:
logger.error(f"Error: {err}")
def initialize(self):
return self.symbol.initialize_sync()
@staticmethod
def live_sleep(*, secs: float):
"""Sleep for the needed amount of seconds in between requests to the terminal.
computes the accurate amount of time needed to sleep ensuring that the next request is made at the start of
a new bar and making cooperative multitasking possible.
Args:
secs (float): The time in seconds. Usually the timeframe you are trading on.
"""
mod = time.time() % secs
secs = secs - mod if mod != 0 else mod
time.sleep(secs + 0.1)
def sleep(self, *, secs: float):
"""Sleep for the needed amount of seconds in between requests to the terminal.
computes the accurate amount of time needed to sleep ensuring that the next request is made at the start of
a new bar and making cooperative multitasking possible.
Args:
secs (float): The time in seconds. Usually the timeframe you are trading on.
"""
if self.config.mode == "backtest":
self.backtest_sleep(secs=secs)
else:
self.live_sleep(secs=secs)
def delay(self, *, secs: float):
"""Sleep for the input amount of seconds"""
if self.config.mode == "backtest":
self._backtest_sleep(secs=secs)
else:
time.sleep(secs)
def _backtest_sleep(self, *, secs: float):
try:
if self.backtest_controller.parties >= 2:
_time = self.config.backtest_engine.cursor.time + secs
while _time > self.config.backtest_engine.cursor.time:
self.backtest_controller.wait()
else:
self.backtest_controller.wait()
except Exception as err:
self.backtest_controller.wait()
logger.error("Error: %s in backtest_sleep", err)
def backtest_sleep(self, *, secs: float):
"""Sleep for the needed amount of seconds in between requests to the terminal.
Args:
secs (float): The time in seconds. Usually the timeframe you are trading on.
"""
try:
_time = self.config.backtest_engine.cursor.time
mod = _time % secs
secs = secs - mod if mod != 0 else mod
self._backtest_sleep(secs=secs)
except Exception as err:
logger.error("Error: %s in backtest_sleep", err)
def run_strategy(self):
"""Run the strategy."""
if self.config.mode == "live":
self.live_strategy()
elif self.config.mode == "backtest":
self.backtest_strategy()
def live_strategy(self):
"""Run the strategy."""
with self as _:
logger.info("Running %s strategy on %s", self.name, self.symbol.name)
while self.running:
try:
self.sessions.check()
self.trade()
except StopTrading:
self.running = False
break
except Exception as err:
logger.error("Error: %s in live_strategy", err)
self.running = False
break
def backtest_strategy(self):
"""Backtest the strategy."""
with self as _:
logger.info("Testing %s strategy on %s with Backtester", self.name, self.symbol.name)
while self.running:
try:
self.sessions.check()
self.backtest_controller.wait()
self.test()
except StopTrading:
self.running = False
break
except Exception as err:
logger.error(f"Error: {err} in backtest_strategy")
return
def trade(self):
"""Place trades using this method. This is the main method of the strategy.
It will be called by the strategy runner.
"""
raise NotImplementedError("Implement this method in your subclass")
def test(self):
self.trade()
src/aiomql/lib/sync/symbol.py
+383
View File
@@ -0,0 +1,383 @@
"""Symbol class for handling a financial instrument."""
from datetime import datetime
from logging import getLogger
from ...core.constants import TimeFrame, CopyTicks
from ...core.base import _Base
from ...core.config import Config
from ...core.sync.meta_trader import MetaTrader
from ...core.models import SymbolInfo, BookInfo
from ...utils import round_off
from ..ticks import Tick
from ..account import Account
from ..candle import Candles
from ..ticks import Ticks
logger = getLogger(__name__)
class Symbol(_Base, SymbolInfo):
"""Main class for handling a financial instrument. A subclass of SymbolInfo it has attributes and methods
for working with a financial instrument.
Attributes:
tick (Tick): Price tick object for instrument
account: An instance of the current trading account
Notes:
Full properties are on the SymbolInfo Object.
Make sure a Symbol is always initialized with a name argument
"""
initialized: bool
tick: Tick
account: Account
def __new__(cls, *args, **kwargs):
instance = super().__new__(cls)
instance.__class__.config = Config()
instance.__class__.mt5 = MetaTrader()
return instance
def __init__(self, **kwargs):
"""Initialize the Symbol object with the name of the financial instrument.
Args:
name (str): Name of the financial instrument
"""
assert "name" in kwargs, "Symbol Object Must be initialized with a name"
super().__init__(**kwargs)
self.account = Account()
self.initialized = False
def info_tick(self, *, name: str = "") -> Tick | None:
"""Get the current price tick of a financial instrument.
Args:
name: if name is supplied, get price tick of that financial instrument. Optional unnamed parameter.
Returns:
Tick: Return a Tick Object
None: If request was unsuccessful
"""
try:
if not name:
tick = self.mt5.symbol_info_tick(self.name)
else:
self.mt5.symbol_select(name, True)
self.mt5.market_book_add(name)
tick = self.mt5.symbol_info_tick(name)
if tick is not None:
tick = Tick(**tick._asdict())
setattr(self, "tick", tick) if not name else ...
return tick
except Exception as err:
logger.warning("%s: Unable to get tick for %s", err, self.name)
return None
def symbol_select(self, *, enable: bool = True) -> bool:
"""Select a symbol in the MarketWatch window or remove a symbol from the window.
Update the select property
Args:
enable (bool): Switch. Optional unnamed parameter. If 'false', a symbol should be removed from
the MarketWatch window.
Returns:
bool: True if successful, otherwise – False.
"""
self.select = self.mt5.symbol_select(self.name, enable)
return self.select
def info(self) -> SymbolInfo | None:
"""Get data on the specified financial instrument and update the symbol object properties
Returns:
(SymbolInfo): SymbolInfo if successful
(None): If request was unsuccessful
"""
info = self.mt5.symbol_info(self.name)
if info is not None:
info = info._asdict()
self.set_attributes(**info)
return SymbolInfo(**info)
return None
def initialize(self) -> bool:
"""Initialize the symbol by pulling properties from the terminal
Returns:
bool: Returns True if symbol info was successfully initialized
"""
try:
select = self.mt5.symbol_select(self.name, True)
self.select = select
self.mt5.market_book_add(self.name)
info = self.mt5.symbol_info(self.name)
if info is not None:
self.set_attributes(**info._asdict())
info_tick = self.mt5.symbol_info_tick(self.name)
if info_tick:
self.tick = Tick(**info_tick._asdict())
if info is not None and info_tick is not None:
self.initialized = True
return True
logger.warning("Unable to initialize %s", self.name)
return False
except Exception as err:
logger.warning("%s: Unable to initialize %s", err, self.name)
return False
def initialize_sync(self) -> bool:
"""Synchronous version of the initialize method"""
try:
select = self.mt5._symbol_select(self.name, True)
self.select = select
self.mt5._market_book_add(self.name)
info = self.mt5._symbol_info(self.name)
if info is not None:
self.set_attributes(**info._asdict())
info_tick = self.mt5._symbol_info_tick(self.name)
if info_tick:
self.tick = Tick(**info_tick._asdict())
if info is not None and info_tick is not None:
self.initialized = True
return True
logger.warning("Unable to initialize %s", self.name)
return False
except Exception as err:
logger.warning("%s: Unable to initialize %s", err, self.name)
return False
def book_add(self) -> bool:
"""Subscribes the MetaTrader 5 terminal to the Market Depth change events for a specified symbol.
If the symbol is not in the list of instruments for the market, This method will return False
Returns:
bool: True if successful, otherwise – False.
"""
res = self.mt5.market_book_add(self.name)
if res is False:
logger.debug("Could not add %s to market book", self.name)
return res
def book_get(self) -> tuple[BookInfo, ...]:
"""Returns a tuple of BookInfo featuring Market Depth entries for the specified symbol.
Returns:
tuple[BookInfo]: Returns the Market Depth contents as a tuples of BookInfo Objects
"""
book_info = self.mt5.market_book_get(self.name)
if book_info is not None:
book_infos = (BookInfo(**info._asdict()) for info in book_info)
return tuple(book_infos)
raise ValueError(f"Could not get book info for {self.name}")
def book_release(self) -> bool:
"""Cancels subscription of the MetaTrader 5 terminal to the Market Depth change events for a specified symbol.
Returns:
bool: True if successful, otherwise – False.
"""
return self.mt5.market_book_release(self.name)
def check_volume(self, *, volume: float) -> tuple[bool, float]:
"""Check if the volume is within the limits of the symbol. If not, return the nearest limit.
Args:
volume (float): Volume to check
Returns: tuple[bool, float]: Returns a tuple of a boolean and a float. The boolean indicates if the volume is
within the limits of the symbol. The float is the volume to use if the volume is not within the limits of the
symbol.
"""
if check := self.volume_min <= volume <= self.volume_max:
return check, volume
else:
return check, self.volume_min if volume <= self.volume_min else self.volume_max
def round_off_volume(self, *, volume: float, round_down: bool = False) -> float:
"""Round off the volume to the nearest volume step.
Args:
volume (float): Volume to round off
round_down (bool): If True, round down. If False, round up. Optional unnamed parameter. Defaults to True.
Returns:
float: Rounded off volume
"""
return round_off(value=volume, step=self.volume_step, round_down=round_down)
def amount_in_quote_currency(self, *, amount: float) -> float:
"""Convert the amount to the quote currency of the symbol."""
if self.currency_profit != self.account.currency:
amount = self.convert_currency(
amount=amount, from_currency=self.account.currency, to_currency=self.currency_profit
)
return amount
def compute_volume(self) -> float:
"""Computes the volume required for a trade usually based on the amount and any other keyword arguments.
This is a dummy method that returns the minimum volume of the symbol. It is meant to be overridden by a subclass
that implements the computation of volume.
Returns:
float: Returns the volume of the trade
"""
return self.volume_min
def convert_currency(self, *, amount: float, from_currency: str, to_currency: str) -> float | None:
"""Convert a given amount from one currency to the other.
Args:
amount: Amount to convert
from_currency: Currency to convert from
to_currency: Currency to convert to
"""
base, quote = to_currency, from_currency
try:
pair = f"{quote}{base}"
tick = self.info_tick(name=pair)
if tick is not None:
return round(amount * tick.bid, 2)
pair = f"{base}{quote}"
tick = self.info_tick(name=pair)
return round(amount / tick.ask, 2)
except Exception as err:
logger.warning(f"{err}: Currency conversion failed: Unable to convert {amount} in {quote} to {base}")
return None
def copy_rates_from(self, *, timeframe: TimeFrame, date_from: datetime | int, count: int = 500) -> Candles:
"""
Get bars from the MetaTrader 5 terminal starting from the specified date.
Args: timeframe (TimeFrame): Timeframe the bars are requested for. Set by a value from the TimeFrame
enumeration. Required unnamed parameter.
date_from (datetime | int): Date of opening of the first bar from the requested sample. Set by the
'datetime' object or as a number of seconds elapsed since 1970.01.01. Required unnamed parameter.
count (int): Number of bars to receive. Required unnamed parameter.
Returns:
Candles: Returns a Candles object as a collection of rates ordered chronologically
Raises:
ValueError: If request was unsuccessful and None was returned
"""
rates = self.mt5.copy_rates_from(self.name, timeframe, date_from, count)
if rates is not None:
return Candles(data=rates)
raise ValueError(f"Could not get rates for {self.name}.")
def copy_rates_from_pos(self, *, timeframe: TimeFrame, count: int = 500, start_position: int = 0) -> Candles:
"""Get bars from the MetaTrader 5 terminal starting from the specified index.
Args:
timeframe (TimeFrame): TimeFrame value from TimeFrame Enum. Required keyword only parameter
count (int): Number of bars to return. Keyword argument defaults to 500
start_position (int): Initial index of the bar the data are requested from. The numbering of bars goes from
present to past. Thus, the zero bar means the current one. Keyword argument defaults to 0.
Returns:
Candles: Returns a Candles object as a collection of rates ordered chronologically.
Raises:
ValueError: If request was unsuccessful and None was returned
"""
rates = self.mt5.copy_rates_from_pos(self.name, timeframe, start_position, count)
if rates is not None:
return Candles(data=rates)
raise ValueError(f"Could not get rates for {self.name}.")
def copy_rates_range(
self, *, timeframe: TimeFrame, date_from: datetime | int, date_to: datetime | int
) -> Candles:
"""Get bars in the specified date range from the MetaTrader 5 terminal.
Args:
timeframe (TimeFrame): Timeframe for the bars using the TimeFrame enumeration. Required unnamed parameter.
date_from (datetime | int): Date the bars are requested from. Set by the 'datetime' object or as a number
of seconds elapsed since 1970.01.01. Bars with the open time >= date_from are returned. Required unnamed
parameter.
date_to (datetime | int): Date, up to which the bars are requested. Set by the 'datetime' object or as a
number of seconds elapsed since 1970.01.01. Bars with the open time <= date_to are returned.
Required unnamed parameter.
Returns:
Candles: Returns a Candles object as a collection of rates ordered chronologically.
Raises:
ValueError: If request was unsuccessful and None was returned
"""
rates = 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)
raise ValueError(f"Could not get rates for {self.name}.")
def copy_ticks_from(self, *, date_from: datetime | int, count: int = 100,
flags: CopyTicks = CopyTicks.ALL) -> Ticks:
"""
Get ticks from the MetaTrader 5 terminal starting from the specified date.
Args: date_from (datetime | int): Date the ticks are requested from. Set by the 'datetime' object or as a
number of seconds elapsed since 1970.01.01.
count (int): Number of requested ticks. Defaults to 100
flags (CopyTicks): A flag to define the type of the requested ticks from CopyTicks enum. INFO is the default
Returns:
Candles: Returns a Candles object as a collection of ticks ordered chronologically.
Raises:
ValueError: If request was unsuccessful and None was returned
"""
ticks = self.mt5.copy_ticks_from(self.name, date_from, count, flags)
if ticks is not None:
return Ticks(data=ticks)
raise ValueError(f"Could not get ticks for {self.name}.")
def copy_ticks_range(
self, *, date_from: datetime | int, date_to: datetime | int, flags: CopyTicks = CopyTicks.ALL
) -> Ticks:
"""Get ticks for the specified date range from the MetaTrader 5 terminal.
Args:
date_from: Date the bars are requested from. Set by the 'datetime' object or as a number of seconds elapsed
since 1970.01.01. Bars with the open time >= date_from are returned. Required unnamed parameter.
date_to: Date, up to which the bars are requested. Set by the 'datetime' object or as a number of
seconds elapsed since 1970.01.01. Bars with the open time <= date_to are returned.
Required unnamed parameter.
flags (CopyTicks):
Returns:
Candles: Returns a Candles object as a collection of ticks ordered chronologically.
Raises:
ValueError: If request was unsuccessful and None was returned.
"""
ticks = self.mt5.copy_ticks_range(self.name, date_from, date_to, flags)
if ticks is not None:
return Ticks(data=ticks)
raise ValueError(f"Could not get ticks for {self.name}.")
tests/conftest.py
+7 -2
View File
@@ -7,6 +7,11 @@ def btc_usd():
return Symbol(name="BTCUSD")
@pytest.fixture(scope="function")
def eth_usd():
return Symbol(name="ETHUSD")
@pytest.fixture(scope="function")
async def buy_order(btc_usd):
sym = btc_usd
@@ -26,8 +31,8 @@ async def buy_order(btc_usd):
@pytest.fixture(scope="function")
async def sell_order(btc_usd):
sym = btc_usd
async def sell_order(eth_usd):
sym = eth_usd
sym_info = await sym.mt5.symbol_info(sym.name)
return {
"action": sym.mt5.TRADE_ACTION_DEAL,
tests/live/unit/test_account.py
+1 -1
View File
@@ -15,6 +15,6 @@ class TestAccount:
assert self.account.connected is True
async def test_account_info(self):
acc_info = await self.account.mt5.account_info
acc_info = await self.account.mt5.account_info()
assert acc_info.login == self.account.login
assert acc_info.server == self.account.server
todo/thread_task_queue.py
View File