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

version 3.12

Browse Source
This commit is contained in:
Ichinga Samuel
2024-01-01 05:25:41 +01:00
parent 0dad24f38e
commit 4b5f267509
29 changed files with 995 additions and 513 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aiomql/account.py
+5 -8
View File
@@ -27,9 +27,7 @@ class Account(AccountInfo):
return cls._instance
async def refresh(self):
"""
Refreshes the account instance with the latest account details from the MetaTrader 5 terminal
"""
"""Refreshes the account instance with the latest account details from the MetaTrader 5 terminal"""
account_info = await self.mt5.account_info()
acc = account_info._asdict()
self.set_attributes(**acc)
@@ -83,8 +81,8 @@ class Account(AccountInfo):
await self.mt5.shutdown()
return False
def has_symbol(self, symbol: str | Type[SymbolInfo]):
"""Checks to see if a symbol is available for a trading account
def has_symbol(self, symbol: str | SymbolInfo):
"""Checks to see if a symbol is available for a trading account.
Args:
symbol (str | SymbolInfo):
@@ -93,8 +91,7 @@ class Account(AccountInfo):
bool: True if symbol is present otherwise False
"""
try:
symbol = SymbolInfo(name=str(symbol)) if not isinstance(symbol, SymbolInfo) else symbol
return symbol in self.symbols
return str(symbol) in {s.name for s in self.symbols}
except Exception as err:
logger.warning(f'Error: {err}; {symbol} not available in this market')
return False
@@ -106,4 +103,4 @@ class Account(AccountInfo):
set[Symbol]: A set of available symbols.
"""
syms = await self.mt5.symbols_get()
return {SymbolInfo(name=sym.name) for sym in syms}
return {SymbolInfo(name=sym.name) for sym in syms}
src/aiomql/bot_builder.py
+13 -11
View File
@@ -9,8 +9,8 @@ from .strategy import Strategy as _Strategy
logger = logging.getLogger(__name__)
Strategy = TypeVar('Strategy', bound=_Strategy)
Symbol = TypeVar('Symbol', bound=_Symbol)
Strategy = TypeVar("Strategy", bound=_Strategy)
Symbol = TypeVar("Symbol", bound=_Symbol)
class Bot:
@@ -21,6 +21,7 @@ class Bot:
executor: The default thread executor.
symbols (list[Symbols]): A set of symbols for the trading session
"""
account: Account = Account()
def __init__(self):
@@ -34,10 +35,10 @@ class Bot:
SystemExit if sign in was not successful
"""
init = await self.account.sign_in()
logger.info("Login Successful")
if not init:
logger.warning('Unable to sign in to MetaTrder 5 Terminal')
logger.warning("Unable to sign in to MetaTrder 5 Terminal")
raise SystemExit
logger.info("Login Successful")
await self.init_symbols()
self.executor.remove_workers()
@@ -63,13 +64,11 @@ class Bot:
self.executor.add_coroutine(coro, kwargs)
def execute(self):
"""Execute the bot.
"""
"""Execute the bot."""
asyncio.run(self.start())
async def start(self):
"""Starts the bot by calling the initialize method and running the strategies in the executor.
"""
"""Initialize the bot and execute it. Similar to calling `execute` method but is a coroutine."""
await self.initialize()
await self.executor.execute()
@@ -100,7 +99,10 @@ class Bot:
strategy (Strategy): Strategy class
params (dict): A dictionary of parameters for the strategy
"""
[self.add_strategy(strategy(symbol=symbol, params=params)) for symbol in self.symbols]
[
self.add_strategy(strategy(symbol=symbol, params=params))
for symbol in self.symbols
]
async def init_symbols(self):
"""Initialize the symbols for the current trading session. This method is called internally by the bot."""
@@ -123,5 +125,5 @@ class Bot:
if init:
self.symbols.add(symbol)
return symbol
logger.warning(f'Unable to initialize symbol {symbol}')
logger.warning(f'{symbol} not a available for this market')
logger.warning(f"Unable to initialize symbol {symbol}")
logger.warning(f"{symbol} not a available for this market")
src/aiomql/candle.py
+41 -32
View File
@@ -26,6 +26,7 @@ class Candle:
real_volume (float): Trade volume
spread (float): Spread
Index (int): Custom attribute representing the position of the candle in a sequence.
mid (float): The median of the high and low price.
"""
time: float
high: float
@@ -36,6 +37,7 @@ class Candle:
open: float
tick_volume: float
Index: int
mid: float
def __init__(self, **kwargs):
"""Create a Candle object from keyword arguments.
@@ -45,24 +47,30 @@ class Candle:
"""
self.time = kwargs.pop('time', 0)
self.Index = kwargs.pop('Index', 0)
self.mid = kwargs.pop('mid', (kwargs['high'] + kwargs['low']) / 2)
self.set_attributes(**kwargs)
def __repr__(self):
keys = reprlib.repr(', '.join('%s=%s' % (i, j) for i, j in self.__dict__.items()))[1:-1]
return '%(class)s(%(args)s)' % {'class': self.__class__.__name__, 'args': keys}
return ("%(class)s(Index=%(Index)s, time=%(time)s, open=%(open)s, high=%(high)s, low=%(low)s, close=%(close)s,"
" mid=%(mid)s)") % {"class": self.__class__.__name__, "open": self.open, "high": self.high,
"low": self.low, "close": self.close, "time": self.time, "mid": self.mid,
'Index': self.Index}
def __eq__(self, other: 'Candle'):
def __eq__(self, other: "Candle"):
return self.time == other.time
def __hash__(self):
return hash(self.time)
def __lt__(self, other: 'Candle'):
def __lt__(self, other: "Candle"):
return self.time < other.time
def __gt__(self, other: 'Candle'):
def __gt__(self, other: "Candle"):
return self.time > other.time
def __getitem__(self, item):
return self.__dict__[item]
def set_attributes(self, **kwargs):
"""Set keyword arguments as instance attributes
@@ -71,17 +79,8 @@ class Candle:
"""
[setattr(self, i, j) for i, j in kwargs.items()]
@property
def mid(self) -> float:
"""The median of open and close
Returns:
float: The median of open and close
"""
return (self.open + self.close) / 2
def is_bullish(self) -> bool:
""" A simple check to see if the candle is bullish.
"""A simple check to see if the candle is bullish.
Returns:
bool: True or False
@@ -96,8 +95,9 @@ class Candle:
"""
return self.open > self.close
_Candle = TypeVar('_Candle', bound=Candle)
_Candles = TypeVar('_Candles', bound='Candles')
_Candle = TypeVar("_Candle", bound=Candle)
_Candles = TypeVar("_Candles", bound="Candles")
class Candles(Generic[_Candle]):
@@ -132,9 +132,10 @@ class Candles(Generic[_Candle]):
tick_volume: Series
real_volume: Series
spread: Series
mid: Series
Candle: Type[Candle]
timeframe: TimeFrame
def __init__(self, *, data: DataFrame | _Candles | Iterable, flip=False, candle_class: Type[_Candle] = None):
"""A container class of Candle objects in chronological order.
@@ -152,43 +153,52 @@ class Candles(Generic[_Candle]):
elif isinstance(data, Iterable):
data = DataFrame(data)
else:
raise ValueError(f'Cannot create DataFrame from object of {type(data)}')
raise ValueError(f"Cannot create DataFrame from object of {type(data)}")
self._data = data.iloc[::-1] if flip else data
self._data = data.loc[::-1].reset_index(drop=True) if flip else data
if 'mid' not in self._data.columns.values:
mid = (self._data['high'] + self._data['low']) / 2
self._data.insert(0, 'mid', mid)
self.Candle = candle_class or Candle
def __repr__(self):
return self._data.__repr__()
def __len__(self):
return self._data.shape[0]
return len(self._data.index)
def __contains__(self, item: _Candle):
return item.time == self[item.Index].time
def __getitem__(self, index) -> _Candle | _Candles:
def __getitem__(self, index) -> _Candle | _Candles | Series:
if isinstance(index, slice):
cls = self.__class__
data = self._data.iloc[index]
data.reset_index(drop=True, inplace=True)
return cls(data=data)
if isinstance(index, str):
elif isinstance(index, str):
if index == 'Index':
return Series(self._data.index)
return self._data[index]
item = self._data.iloc[index]
return self.Candle(Index=index, **item)
elif isinstance(index, int):
index = index if index >= 0 else len(self) + index
return self.Candle(**self._data.iloc[index])
raise TypeError(f"Expected int, slice or str got {type(index)}")
def __setitem__(self, index, value: Series):
if isinstance(value, Series):
self._data[index] = value
return
raise TypeError(f'Expected Series got {type(value)}')
raise TypeError(f"Expected Series got {type(value)}")
def __getattr__(self, item):
if item in list(self._data.columns.values):
return self._data[item]
raise AttributeError(f'Attribute {item} not defined on class {self.__class__.__name__}')
if item == 'Index':
return Series(self._data.index)
raise AttributeError(f"Attribute {item} not defined on class {self.__class__.__name__}")
def __iter__(self):
return (self.Candle(**row._asdict()) for row in self._data.itertuples())
@@ -213,7 +223,7 @@ class Candles(Generic[_Candle]):
Returns:
ta: The ta library
"""
"""
return ta
@property
@@ -221,7 +231,7 @@ class Candles(Generic[_Candle]):
"""The original data passed to the class as a pandas DataFrame"""
return self._data
def rename(self, inplace=True, **kwargs) -> _Candles | None :
def rename(self, inplace=True, **kwargs) -> _Candles:
"""Rename columns of the candles class.
Keyword Args:
@@ -229,8 +239,7 @@ class Candles(Generic[_Candle]):
**kwargs: The new names of the columns
Returns:
Candles: A new instance of the class with the renamed columns if inplace is False.
None: If inplace is True
Candles: A new instance of the class with the renamed columns if inplace is False else the modified instance
"""
res = self._data.rename(columns=kwargs, inplace=inplace)
return res if inplace else self.__class__(data=res)
return self if inplace else self.__class__(data=res)
src/aiomql/core/base.py
+10 -7
View File
@@ -1,5 +1,5 @@
from functools import cache
import reprlib
import enum
from logging import getLogger
from .config import Config
@@ -29,8 +29,11 @@ class Base:
self.set_attributes(**kwargs)
def __repr__(self):
keys = reprlib.repr(', '.join('%s=%s' % (i, j) for i, j in self.__dict__.items()))[1:-1]
return '%(class)s(%(args)s)' % {'class': self.__class__.__name__, 'args': keys}
kv = [(k, v) for k, v in self.__dict__.items() if not k.startswith('_') and
(type(v) in (int, float, str) or isinstance(v, enum.Enum))]
args = (', '.join('%s=%s' % (i, j) for i, j in kv[:3]))
args = args if len(kv) <= 3 else args + ' ... ' + ', '.join('%s=%s' % (i, j) for i, j in kv[-1:])
return '%(class)s(%(args)s)' % {'class': self.__class__.__name__, 'args': args}
def set_attributes(self, **kwargs):
"""Set keyword arguments as object attributes
@@ -100,7 +103,7 @@ class Base:
clss = self.__class__.__mro__[-3::-1]
cls_dict = {}
for cls in clss:
cls_dict |= cls.__dict__
cls_dict |= cls.__dict__
return {key: value for key, value in cls_dict.items() if key in self.annotations}
@property
@@ -111,7 +114,8 @@ class Base:
dict: A dictionary of instance and class attributes
"""
try:
return {key: value for key, value in (self.class_vars | self.__dict__).items() if key not in self.Meta.filter}
return {key: value for key, value in (self.class_vars | self.__dict__).items() if
key not in self.Meta.filter}
except Exception as err:
logger.warning(err)
@@ -133,5 +137,4 @@ class Base:
Returns:
set: A set of attributes to be excluded
"""
return cls.exclude.difference(cls.include)
return cls.exclude.difference(cls.include)
src/aiomql/core/config.py
+20 -21
View File
@@ -31,43 +31,42 @@ class Config:
You can change this by passing the filename keyword argument to the constructor.
By passing reload=True to the load_config method, you can reload and search again for the config file.
"""
login: int = 0
password: str = ''
server: str = ''
path: str = ''
password: str = ""
server: str = ""
path: str = ""
timeout: int = 60000
record_trades: bool = True
filename: str = 'aiomql.json'
filename: str = "aiomql.json"
win_percentage: float = 0.85
records_dir = Path.home() / 'Documents' / 'Aiomql' / 'Trade Records' if record_trades else None
records_dir = Path.home() / "Documents" / "Aiomql" / "Trade Records"
_load = 1
def __new__(cls, *args, **kwargs):
if not hasattr(cls, '_instance'):
if not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, **kwargs):
self.load_config(reload=False)
[setattr(self, key, value) for key, value in kwargs]
@staticmethod
def walk_to_root(path: str) -> Iterator[str]:
if not os.path.exists(path):
raise IOError('Starting path not found')
raise IOError("Starting path not found")
if os.path.isfile(path):
path = os.path.dirname(path)
last_dir = None
current_dir = os.path.abspath(path)
while last_dir != current_dir:
yield current_dir
parent_dir = os.path.abspath(os.path.join(current_dir, os.path.pardir))
last_dir, current_dir = current_dir, parent_dir
def find_config(self):
current_file = __file__
frame = _getframe()
@@ -77,13 +76,13 @@ class Config:
frame = frame.f_back
frame_filename = frame.f_code.co_filename
path = os.path.dirname(os.path.abspath(frame_filename))
for dirname in self.walk_to_root(path):
check_path = os.path.join(dirname, self.filename)
if os.path.isfile(check_path):
return check_path
return None
def load_config(self, file: str = None, reload: bool = True):
if reload:
self._load = 1
@@ -93,18 +92,18 @@ class Config:
self._load = 0
data = {}
if (file := (file or self.find_config())) is None:
logger.warning('No Config File Found')
logger.warning("No Config File Found")
else:
fh = open(file, mode='r')
fh = open(file, mode="r")
data = json.load(fh)
fh.close()
[setattr(self, key, value) for key, value in data.items()]
self.records_dir.mkdir(parents=True, exist_ok=True) if self.records_dir else ...
def account_info(self) -> dict['login', 'password', 'server']:
def account_info(self) -> dict["login", "password", "server"]:
"""Returns Account login details as found in the config object if available
Returns:
dict: A dictionary of login details
Returns:
dict: A dictionary of login details
"""
return {'login': self.login, 'password': self.password, 'server': self.server}
return {"login": self.login, "password": self.password, "server": self.server}
src/aiomql/core/constants.py
+2 -2
View File
@@ -16,7 +16,7 @@ Examples:
class Repr:
__enum_name__ = ""
def __str__(self):
def __repr__(self):
return f"{self.__enum_name__}_{self.name}"
@@ -783,4 +783,4 @@ class AccountMarginMode(Repr, IntEnum):
__enum_name__ = "ACCOUNT_MARGIN_MODE"
RETAIL_NETTING = mt5.ACCOUNT_MARGIN_MODE_RETAIL_NETTING
EXCHANGE = mt5.ACCOUNT_MARGIN_MODE_EXCHANGE
RETAIL_HEDGING = mt5.ACCOUNT_MARGIN_MODE_RETAIL_HEDGING
RETAIL_HEDGING = mt5.ACCOUNT_MARGIN_MODE_RETAIL_HEDGING
src/aiomql/core/meta_trader.py
+2 -6
View File
@@ -137,7 +137,6 @@ class MetaTrader(metaclass=BaseMeta):
if res is None:
err = await self.last_error()
logger.warning(f'Error in obtaining account information.{Error(*err)}')
return res
async def terminal_info(self) -> TerminalInfo | None:
@@ -210,17 +209,14 @@ class MetaTrader(metaclass=BaseMeta):
err = await self.last_error()
logger.warning(f'Error in obtaining rates for {symbol}.{Error(*err)}')
return res
return res
async def copy_rates_from_pos(self, symbol: str, timeframe: TimeFrame, start_pos: int, count: int):
res = await asyncio.to_thread(self._copy_rates_from_pos, symbol, timeframe, start_pos, count)
if res is None:
err = await self.last_error()
logger.warning(f'Error in obtaining rates for {symbol}.{Error(*err)}')
return res
return res
async def copy_rates_range(self, symbol: str, timeframe: TimeFrame, date_from: datetime | int,
@@ -270,7 +266,7 @@ class MetaTrader(metaclass=BaseMeta):
ticket (int): Order ticket (ORDER_TICKET). Optional named parameter.
Returns:
list[TradeOrder]: A list of active trade orders as TradeOrder objects
tuple[TradeOrder]: A list of active trade orders as TradeOrder objects
"""
kwargs = {key: value for key, value in (('group', group), ('ticket', ticket), ('symbol', symbol)) if value}
res = await asyncio.to_thread(self._orders_get, **kwargs)
@@ -352,4 +348,4 @@ class MetaTrader(metaclass=BaseMeta):
logger.warning(f'Error in getting deals.{Error(*err)}')
return res
return res
return res
src/aiomql/core/models.py
+2 -2
View File
@@ -338,7 +338,7 @@ class SymbolInfo(Base):
super().__init__(**kwargs)
def __repr__(self):
return self.name
return '%(class)s(name=%(name)s)' % {'class': self.__class__.__name__, 'name': self.name}
def __str__(self):
return self.name
@@ -610,4 +610,4 @@ class TradeDeal(Base):
tp: float
symbol: str
comment: str
external_id: str
external_id: str
src/aiomql/lib/strategies/__init__.py
+2 -1
View File
@@ -1 +1,2 @@
from .finger_trap import FingerTrap
from .finger_trap import FingerTrap
from .tracker import Tracker
src/aiomql/lib/strategies/finger_trap.py
+45 -142
View File
@@ -1,8 +1,8 @@
import asyncio
import logging
from typing import Literal
from dataclasses import dataclass
from .tracker import Tracker
from ..traders import SimpleTrader
from ...symbol import Symbol
from ...trader import Trader
from ...candle import Candles
@@ -13,49 +13,6 @@ from ...sessions import Sessions
logger = logging.getLogger(__name__)
@dataclass
class Entry:
"""
Entry class for FingerTrap strategy. Will be used to store entry conditions and other entry related data.
Attributes:
bearish (bool): True if the market is bearish
bullish (bool): True if the market is bullish
ranging (bool): True if the market is ranging
snooze (float): Time to wait before checking for entry conditions
trend (str): The current trend of the market
new (bool): True if the last candle is new
order_type (OrderType): The type of order to place
"""
bearish: bool = False
bullish: bool = False
ranging: bool = True
trending: bool = False
trend: Literal["ranging", "bullish", "bearish"] = "ranging"
snooze: float = 0
last_trend_time: float = 0
last_entry_time: float = 0
new: bool = True
order_type: OrderType | None = None
def update(self, **kwargs):
fields = self.__dict__
for key in kwargs:
if key in fields:
setattr(self, key, kwargs[key])
match self.trend:
case "ranging":
self.ranging = True
self.trending = self.bullish = self.bearish = False
case "bullish":
self.ranging = self.bearish = False
self.bullish = self.trending = True
case "bearish":
self.ranging = self.bullish = False
self.bearish = self.trending = True
class FingerTrap(Strategy):
trend_time_frame: TimeFrame
entry_time_frame: TimeFrame
@@ -64,123 +21,75 @@ class FingerTrap(Strategy):
slow_period: int
entry_period: int
parameters: dict
prices: Candles
name = "FingerTrap"
interval: TimeFrame
entry_candles_count: int
trend_candles_count: int
trader: Trader
tracker: Tracker
_parameters = {"trend": 3, "fast_period": 8, "slow_period": 34, "entry_time_frame": TimeFrame.M5,
"trend_time_frame": TimeFrame.H1, "entry_period": 8,
"trend_candles_count": 48, "entry_candles_count": 50}
def __init__(
self,
*,
symbol: Symbol,
params: dict | None = None,
trader: Trader = None,
sessions: Sessions = None,
):
super().__init__(symbol=symbol, params=params, sessions=sessions)
self.trend = self.parameters.get("trend", 3)
self.fast_period = self.parameters.setdefault("fast_period", 8)
self.slow_period = self.parameters.setdefault("slow_period", 34)
self.entry_time_frame = self.parameters.setdefault(
"entry_time_frame", TimeFrame.M5
)
self.trend_time_frame = self.parameters.setdefault(
"trend_time_frame", TimeFrame.H1
)
self.trader = trader or Trader(symbol=self.symbol)
self.entry: Entry = Entry(snooze=self.trend_time_frame.time)
self.entry_period = self.parameters.setdefault("entry_period", 8)
self.trend_candles_count = self.parameters.setdefault(
"trend_candles_count", 86400 // self.trend_time_frame.time
)
self.trend_candles_count = max(self.trend_candles_count, self.slow_period)
self.entry_candles_count = self.trend_candles_count * (
self.trend_time_frame.time // self.entry_time_frame.time
)
self.entry_candles_count = max(self.entry_candles_count, self.entry_period)
def __init__(self, *, symbol: Symbol, 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)
self.tracker: Tracker = Tracker(snooze=self.trend_time_frame.time)
async def check_trend(self):
try:
candles = await self.symbol.copy_rates_from_pos(
timeframe=self.trend_time_frame, count=self.trend_candles_count
)
current = candles[-1]
if current.time > self.entry.last_trend_time:
self.entry.update(new=True, last_trend_time=current.time)
else:
self.entry.update(new=False)
candles: Candles = await self.symbol.copy_rates_from_pos(timeframe=self.trend_time_frame,
count=self.trend_candles_count)
if not ((current := candles[-1].time) >= self.tracker.trend_time):
self.tracker.new = False
return
self.tracker.update(new=True, trend_time=current)
candles.ta.ema(length=self.slow_period, append=True, fillna=0)
candles.ta.ema(length=self.fast_period, append=True, fillna=0)
candles.rename(
inplace=True,
**{
f"EMA_{self.fast_period}": "fast",
f"EMA_{self.slow_period}": "slow",
},
)
candles.rename(inplace=True, **{f"EMA_{self.fast_period}": "fast", f"EMA_{self.slow_period}": "slow"})
# Compute
candles["fast_A_slow"] = candles.ta_lib.above(candles.fast, candles.slow)
candles["fast_B_slow"] = candles.ta_lib.below(candles.fast, candles.slow)
candles["close_A_fast"] = candles.ta_lib.above(candles.close, candles.fast)
candles["close_B_fast"] = candles.ta_lib.below(candles.close, candles.fast)
trend = candles[-self.trend : -1]
if all(
(c.is_bullish() and c.fast_A_slow and c.close_A_fast) for c in trend
):
self.entry.update(trend="bullish")
elif all(
c.is_bearish() and c.fast_B_slow and c.close_B_fast for c in trend
):
self.entry.update(trend="bearish")
trend = candles[-self.trend: -1]
if all((c.is_bullish() and c.fast_A_slow and c.close_A_fast) for c in trend):
self.tracker.update(trend="bullish")
elif all(c.is_bearish() and c.fast_B_slow and c.close_B_fast for c in trend):
self.tracker.update(trend="bearish")
else:
self.entry.update(trend="ranging", snooze=self.trend_time_frame.time)
self.tracker.update(trend="ranging", snooze=self.trend_time_frame.time)
except Exception as exe:
logger.error(f"{exe}. Error in {self.__class__.__name__}.check_trend")
async def confirm_trend(self):
try:
candles = await self.symbol.copy_rates_from_pos(
timeframe=self.entry_time_frame, count=self.entry_candles_count
)
current = candles[-1]
if current.time > self.entry.last_entry_time:
self.entry.update(new=True, last_entry_time=current.time)
else:
self.entry.update(new=False)
candles = await self.symbol.copy_rates_from_pos(timeframe=self.entry_time_frame,
count=self.entry_candles_count)
if not ((current := candles[-1].time) >= self.tracker.entry_time):
self.tracker.new = False
return
self.tracker.update(new=True, entry_time=current)
candles.ta.ema(length=self.entry_period, append=True, fillna=0)
candles.rename(**{f"EMA_{self.entry_period}": "ema"})
candles["close_A_ema"] = candles.ta_lib.above(candles.close, candles.ema)
candles["close_B_ema"] = candles.ta_lib.below(candles.close, candles.ema)
candles["close_XA_ema"] = candles.ta_lib.cross(candles.close, candles.ema)
candles["close_XB_ema"] = candles.ta_lib.cross(
candles.close, candles.ema, above=False
)
if self.entry.bullish and current.close_XA_ema:
self.entry.update(
snooze=self.entry_time_frame.time, order_type=OrderType.BUY
)
elif self.entry.bearish and current.close_XB_ema:
self.entry.update(
snooze=self.entry_time_frame.time, order_type=OrderType.SELL
)
candles["close_XB_ema"] = candles.ta_lib.cross(candles.close, candles.ema, above=False)
current = candles[-2]
if self.tracker.bullish and current.close_XA_ema:
self.tracker.update(snooze=self.entry_time_frame.time, order_type=OrderType.BUY)
elif self.tracker.bearish and current.close_XB_ema:
self.tracker.update(snooze=self.entry_time_frame.time, order_type=OrderType.SELL)
else:
self.entry.update(snooze=self.entry_time_frame.time, order_type=None)
self.tracker.update(snooze=self.entry_time_frame.time, order_type=None)
except Exception as exe:
logger.error(f"{exe} Error in {self.__class__.__name__}.confirm_trend")
logger.error(f"{exe} Error in {self.name}.confirm_trend")
async def watch_market(self):
await self.check_trend()
if not self.entry.ranging:
if not self.tracker.ranging:
await self.confirm_trend()
async def trade(self):
@@ -190,21 +99,15 @@ class FingerTrap(Strategy):
await sess.check()
try:
await self.watch_market()
if not self.entry.new:
if not self.tracker.new:
await asyncio.sleep(2)
continue
if self.entry.order_type is None:
await self.sleep(self.entry.snooze)
if self.tracker.order_type is None:
await self.sleep(self.tracker.snooze)
continue
await self.trader.place_trade(
order_type=self.entry.order_type, params=self.parameters
)
await self.sleep(self.entry.snooze)
await self.trader.place_trade(order_type=self.tracker.order_type, parameters=self.parameters)
await self.sleep(self.tracker.snooze)
except Exception as err:
logger.error(
f"Error: {err}\t Symbol: {self.symbol} in {self.__class__.__name__}.trade"
)
logger.error(f"Error: {err}\t Symbol: {self.symbol} in {self.__class__.__name__}.trade")
await self.sleep(self.trend_time_frame.time)
continue
continue
src/aiomql/lib/strategies/tracker.py
+35
View File
@@ -0,0 +1,35 @@
from dataclasses import dataclass
from typing import Literal
from ...core.constants import OrderType
@dataclass
class Tracker:
"""Keeps track of a strategy's data and state"""
trend: Literal["ranging", "bullish", "bearish"] = "ranging"
bullish: bool = False
bearish: bool = False
ranging: bool = True
snooze: float = 0
trend_time: float = 0
entry_time: float = 0
new: bool = True
order_type: OrderType = None
def update(self, **kwargs):
fields = self.__dict__
for key in kwargs:
if key in fields:
setattr(self, key, kwargs[key])
if 'trend' in kwargs:
match self.trend:
case "ranging":
self.ranging = True
self.bullish = self.bearish = False
case "bullish":
self.ranging = self.bearish = False
self.bullish = True
case "bearish":
self.ranging = self.bullish = False
self.bearish = True
src/aiomql/lib/symbols/crypto_symbol.py
-30
View File
@@ -1,30 +0,0 @@
from ...symbol import Symbol
from ...core.exceptions import VolumeError
class CryptoSymbol(Symbol):
"""Subclass of Symbol for Crypto/Fiat Symbols. Handles the computation of volume based on the amount to risk."""
async def compute_volume(self, *, amount: float, points, use_limits=False) -> float:
"""Compute volume given an amount to risk and target pips. Round the computed volume to the nearest step.
Args:
amount (float): Amount to risk. Given in terms of the account currency.
points (float): Target pips.
use_limits (bool): If True, the computed volume checked against the maximum and minimum volume.
Returns:
float: volume
Raises:
VolumeError: If the computed volume is less than the minimum volume or greater than the maximum volume.
"""
if self.currency_profit != self.account.currency:
amount = await self.convert_currency(amount=amount, base=self.currency_profit, quote=self.account.currency)
volume = amount / (self.point * points * self.trade_contract_size)
volume = self.round_off_volume(volume)
if self.check_volume(volume)[0]:
return volume
if use_limits:
return self.check_volume(volume)[1]
raise VolumeError(f'Incorrect Volume. Computed Volume outside the range of permitted volumes')
src/aiomql/lib/symbols/forex_symbol.py
+5 -5
View File
@@ -7,12 +7,12 @@ class ForexSymbol(Symbol):
take profit and volume.
"""
async def compute_volume(self, *, amount: float, pips, use_limits=False) -> float:
"""Compute volume given an amount to risk and target pips. Round the computed volume to the nearest step.
async def compute_volume(self, *, amount: float, points, use_limits=False) -> float:
"""Compute volume given an amount to risk and target points. Round the computed volume to the nearest step.
Args:
amount (float): Amount to risk. Given in terms of the account currency.
pips (float): Target pips.
points (float): Target pips.
use_limits (bool): If True, the computed volume checked against the maximum and minimum volume.
Returns:
@@ -23,10 +23,10 @@ class ForexSymbol(Symbol):
"""
if self.currency_profit != self.account.currency:
amount = await self.convert_currency(amount=amount, base=self.currency_profit, quote=self.account.currency)
volume = amount / (self.pip * pips * self.trade_contract_size)
volume = amount / (self.point * points * self.trade_contract_size)
volume = self.round_off_volume(volume)
if self.check_volume(volume)[0]:
return volume
if use_limits:
return self.check_volume(volume)[1]
raise VolumeError(f'Incorrect Volume. Computed Volume outside the range of permitted volumes')
raise VolumeError(f"Incorrect Volume. Computed Volume outside the range of permitted volumes")
src/aiomql/lib/traders/__init__.py
+1
View File
@@ -0,0 +1 @@
from .simple_trader import SimpleTrader
src/aiomql/lib/traders/simple_trader.py
+61
View File
@@ -0,0 +1,61 @@
"""Trader class module. Handles the creation of an order and the placing of trades"""
from logging import getLogger
from ..symbols import ForexSymbol
from ...ram import RAM
from ...core.models import OrderType
from ...positions import Positions
from ...trader import Trader
logger = getLogger(__name__)
class SimpleTrader(Trader):
"""A simple trader class. Limits the number of loosing trades per symbol"""
def __init__(self, *, symbol: ForexSymbol, ram: RAM = None, num_trades: int = 1):
"""Initializes the order object and RAM instance
Args:
symbol (Symbol): Financial instrument
ram (RAM): Risk Assessment and Management instance
num_trades (int): Number of open trades in loosing positions to allow per symbol
"""
super().__init__(symbol=symbol, ram=ram)
self.positions = Positions(symbol=symbol.name)
self.num_trades = num_trades
async def create_order(self, *, order_type: OrderType, points: float = 0):
"""Complete the order object with the required values. Creates a simple order.
Args:
order_type (OrderType): Type of order
points (float): Target points
"""
positions = await self.positions.positions_get()
positions.sort(key=lambda pos: pos.time_msc)
loosing = [trade for trade in positions if trade.profit < 0]
if (losses := len(loosing)) > self.num_trades:
raise RuntimeError(f"Last {losses} trades in a losing position")
points = points or self.symbol.trade_stops_level * 2
amount = self.ram.amount or await self.ram.get_amount()
self.order.volume = await self.symbol.compute_volume(amount=amount, points=points)
self.order.type = order_type
await self.set_trade_stop_levels(points=points)
async def place_trade(self, order_type: OrderType, parameters: dict = None, points: float = 0):
"""Places a trade based on the order_type.
Args:
order_type (OrderType): Type of order
parameters: parameters of the trading strategy used to place the trade
points (float): Target points
"""
try:
self.parameters |= parameters or {}
await self.create_order(order_type=order_type, points=points)
if not await self.check_order():
return
await self.send_order()
except Exception as err:
logger.error(f"{err}. Symbol: {self.order.symbol}\n {self.__class__.__name__}.place_trade")
src/aiomql/order.py
+2 -2
View File
@@ -79,7 +79,7 @@ class Order(TradeRequest):
"""
res = await self.mt5.order_send(self.dict)
if res is None:
raise OrderError(f'Failed to send order {self.symbol} {self.type} {self.volume} {self.price} {res}')
raise OrderError(f'Failed to send order {self.symbol} {self.type} {self.volume} {self.price}')
return OrderSendResult(**res._asdict())
async def calc_margin(self) -> float:
@@ -109,4 +109,4 @@ class Order(TradeRequest):
if res is None:
raise OrderError(
f'Failed to calculate profit for {self.symbol} {self.type} {self.volume} {self.price} {self.tp}')
return res
return res
src/aiomql/positions.py
+3 -5
View File
@@ -54,10 +54,8 @@ class Positions:
Returns:
list[TradePosition]: A list of open trade positions
"""
symbol = symbol or self.symbol
group = group or self.group
ticket = ticket or self.ticket
positions = await self.mt5.positions_get(group=group, symbol=symbol, ticket=ticket)
positions = await self.mt5.positions_get(group=group or self.group, symbol=symbol or self.symbol,
ticket=ticket or self.ticket)
if not positions:
return []
return [TradePosition(**pos._asdict()) for pos in positions]
@@ -87,4 +85,4 @@ class Positions:
results = await asyncio.gather(*[order for order in orders], return_exceptions=True)
amount_closed = len([res for res in results if res.retcode == 10009])
return amount_closed
return amount_closed
src/aiomql/ram.py
+3 -1
View File
@@ -7,6 +7,8 @@ class RAM:
risk_to_reward: float
risk: float
amount: float
points: float
pips: float
def __init__(self, *, risk_to_reward: float = 1, risk: float = 0.01, amount: float = 0, **kwargs):
"""Initialize Risk Assessment and Management with the provided keyword arguments.
@@ -33,4 +35,4 @@ class RAM:
"""
await self.account.refresh()
risk = risk or self.risk
return self.account.equity * risk
return self.account.equity * risk
src/aiomql/records.py
+51 -18
View File
@@ -3,9 +3,11 @@
import asyncio
from pathlib import Path
import csv
import logging
from .history import History
from .core import Config
from .core import Config, MetaTrader
logger = logging.getLogger(__name__)
class Records:
@@ -17,6 +19,7 @@ class Records:
from the config
"""
config: Config = Config()
mt5: MetaTrader = MetaTrader()
def __init__(self, records_dir: Path = ''):
"""Initialize the Records class. The main method of this class is update_records which you should call to update
@@ -43,16 +46,43 @@ class Records:
Args:
file: Trade record file
"""
fr = open(file, mode='r', newline='')
reader = csv.DictReader(fr)
rows = [row for row in reader]
rows = await self.update_rows(rows)
fr.close()
fw = open(file, mode='w', newline='')
writer = csv.DictWriter(fw, fieldnames=reader.fieldnames)
writer.writeheader()
writer.writerows(rows)
fw.close()
try:
fr = open(file, mode='r', newline='')
reader = csv.DictReader(fr)
rows = [row for row in reader]
rows = await self.update_rows(rows)
fr.close()
fw = open(file, mode='w', newline='')
writer = csv.DictWriter(fw, fieldnames=reader.fieldnames, extrasaction='ignore', restval=None)
writer.writeheader()
writer.writerows(rows)
fw.close()
except Exception as err:
logger.error(f'Error: {err}. Unable to read and update trade records')
async def update_row(self, row: dict) -> dict:
"""Update a single row of entered trade in the csv file with the actual profit.
Args:
row: A dictionary from the dictionary writer object of the csv file.
Returns:
dict: A dictionary with the actual profit and win status.
"""
try:
order = int(row['order'])
deals = await self.mt5.history_deals_get(position=order)
if not deals or len(deals) <= 1:
return row
deals = [deal for deal in deals if (deal.order != deal.position_id and deal.position_id == order
and deal.entry == 1)]
deals.sort(key=lambda x: x.time_msc)
deal = deals[-1]
row.update(actual_profit=deal.profit, win=deal.profit > 0, closed=True)
return row
except Exception as err:
logging.error(f'Error: {err}. Unable to update trade record')
return row
async def update_rows(self, rows: list[dict]) -> list[dict]:
"""Update the rows of entered trades in the csv file with the actual profit.
@@ -63,11 +93,14 @@ class Records:
Returns:
list[dict]: A list of dictionaries with the actual profit and win status.
"""
tasks = [History(position=int(row['order'])).get_deals() for row in rows]
deals = [deal for deals in await asyncio.gather(*tasks) for deal in deals]
deals = {str(deal.position_id): deal.profit for deal in deals if deal.order != deal.position_id}
[row.update(actual_profit=(profit := deals[order]), win=profit > 0) for row in rows if (order := row['order']) in deals]
return rows
closed, unclosed = [], []
for row in rows:
if (row.get('closed', 'FALSE')).title() == 'True':
closed.append(row)
else:
unclosed.append(row)
unclosed = await asyncio.gather(*[self.update_row(row) for row in unclosed])
return closed + unclosed
async def update_records(self):
"""Update trade records in the records_dir folder."""
@@ -76,4 +109,4 @@ class Records:
async def update_record(self, file: Path | str):
"""Update a single trade record file."""
await self.read_update(file)
await self.read_update(file)
src/aiomql/result.py
+7 -10
View File
@@ -17,7 +17,6 @@ class Result:
name: Any desired name for the result file object
"""
config = Config()
data: dict
def __init__(self, result: OrderSendResult, parameters: dict = None, name: str = ''):
"""
@@ -29,30 +28,28 @@ class Result:
"""
self.parameters = parameters or {}
self.result = result
self.name = name or parameters.get('name', 'Strategy')
self.name = name or parameters.get('name', 'Trades')
def get_data(self) -> dict:
result = self.result.get_dict(exclude={'retcode', 'retcode_external', 'request_id', 'request'})
return self.parameters | result | {'actual_profit': 0, 'closed': False, 'win': False}
return (self.parameters | self.result.get_dict(exclude={'retcode', 'comment', 'retcode_external', 'request_id', 'request'})
| {'actual_profit': 0, 'closed': False, 'win': False})
def to_csv(self):
"""Record trade results and associated parameters as a csv file
"""
try:
self.data = self.get_data()
data = self.get_data()
file = self.config.records_dir / f"{self.name}.csv"
exists = file.exists()
with open(file, 'a', newline='') as fh:
writer = csv.DictWriter(fh, fieldnames=sorted(list(self.data.keys())), extrasaction='ignore', restval=None)
writer = csv.DictWriter(fh, fieldnames=sorted(list(data.keys())), extrasaction='ignore', restval=None)
if not exists:
writer.writeheader()
writer.writerow(self.data)
writer.writerow(data)
except Exception as err:
logger.error(f'Error: {err}. Unable to save trade results')
async def save_csv(self):
"""Save trade results and associated parameters as a csv file in a separate thread
"""
# exe = self.config.executor
loop = asyncio.get_running_loop()
loop.run_in_executor(None, self.to_csv)
self.to_csv()
src/aiomql/sessions.py
+1 -1
View File
@@ -212,4 +212,4 @@ class Sessions:
print(f'sleeping for {secs} seconds until next {current_session} session')
await sleep(secs)
self.current_session = current_session
await self.current_session.begin()
await self.current_session.begin()
src/aiomql/strategy.py
+24 -10
View File
@@ -11,18 +11,19 @@ from .account import Account
from .core import Config
from .sessions import Sessions, Session
Symbol = TypeVar('Symbol', bound=_Symbol)
Symbol = TypeVar("Symbol", bound=_Symbol)
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.
Class Attributes:
name (str): A name for the strategy.
account (Account): Account instance.
mt5 (MetaTrader): MetaTrader instance.
config (Config): Config instance.
@@ -30,12 +31,15 @@ class Strategy(ABC):
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 = ''
name: str
symbol: Symbol
sessions: Sessions
account = Account()
mt5: MetaTrader()
config = Config()
_parameters = {}
def __init__(self, *, symbol: Symbol, params: dict = None, sessions: Sessions = None):
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
@@ -43,15 +47,26 @@ class Strategy(ABC):
symbol (Symbol): The Financial instrument
params (Dict): Trading strategy parameters
"""
self.parameters = self._parameters | (params or {})
self.symbol = symbol
self.parameters = params.copy() if isinstance(params, dict) else {}
self.parameters['symbol'] = symbol.name
self.parameters['name'] = self.name or self.__class__.__name__
self.name = name or self.__class__.__name__
self.parameters["symbol"] = symbol.name
self.parameters["name"] = self.name
self.sessions = sessions or Sessions(Session(start=0, end=dtime(hour=23, minute=59, second=59)))
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.__dict__.get('parameters', {}):
self.parameters[key] = value
super().__setattr__(key, value)
@staticmethod
async def sleep(secs: float):
"""Sleep for the needed amount of seconds in between requests to the terminal.
@@ -65,9 +80,8 @@ class Strategy(ABC):
secs = secs - mod if mod != 0 else mod
await asyncio.sleep(secs + 0.1)
@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.
"""
It will be called by the strategy runner.
"""
src/aiomql/symbol.py
+5 -4
View File
@@ -79,11 +79,12 @@ class Symbol(SymbolInfo):
Raises:
ValueError: If request was unsuccessful and None was returned
"""
info = await self.mt5.symbol_info(self.name)
if info:
self.set_attributes(**info._asdict())
return SymbolInfo(**info._asdict())
info = info._asdict()
info['swap_rollover3days'] = info.get('swap_rollover3days', 0) % 7
self.set_attributes(**info)
return SymbolInfo(**info)
raise ValueError(f'Could not get info for {self.name}')
async def init(self) -> bool:
@@ -332,4 +333,4 @@ class Symbol(SymbolInfo):
ticks = await 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}')
raise ValueError(f'Could not get ticks for {self.name}')
src/aiomql/trader.py
+42 -55
View File
@@ -1,5 +1,5 @@
"""Trader class module. Handles the creation of an order and the placing of trades"""
from abc import ABC, abstractmethod
from datetime import datetime
from typing import TypeVar
from logging import getLogger
@@ -14,21 +14,18 @@ from .utils import dict_to_string
from .result import Result
logger = getLogger(__name__)
Symbol = TypeVar('Symbol', bound=_Symbol)
Symbol = TypeVar("Symbol", bound=_Symbol)
class Trader:
"""Base class for creating a Trader object. Handles the creation of an order and the placing of trades
class Trader(ABC):
"""Base class for creating a Trader object. Handles the creation of an order and the placing of trades.
Attributes:
symbol (Symbol): Financial instrument class Symbol class or any subclass of it.
symbol (Symbol): The financial instrument.
ram (RAM): RAM instance
order (Order): Trade order
Class Attributes:
name (str): A name for the strategy.
account (Account): Account instance.
mt5 (MetaTrader): MetaTrader instance.
config (Config): Config instance.
"""
config = Config()
@@ -43,21 +40,13 @@ class Trader:
self.symbol = symbol
self.order = Order(symbol=symbol.name)
self.ram = ram or RAM()
self.params = {}
self.parameters = {}
async def create_order(self, *, order_type: OrderType, **kwargs):
"""Complete the order object with the required values. Creates a simple order.
@abstractmethod
async def create_order(self, *args, **kwargs):
"""Complete the order object with the required values. Creates a simple order."""
Args:
order_type (OrderType): Type of order
kwargs: keyword arguments as required for the specific trader
"""
points = kwargs.get('points', self.symbol.trade_stops_level+self.symbol.spread)
self.order.volume = await self.symbol.compute_volume()
self.order.type = order_type
await self.set_trade_stop_levels(points=points)
async def set_order_limits(self, pips: float):
async def set_order_limits(self, *, pips: float):
"""Sets the stop loss and take profit for the order. This method uses pips as defined for forex instruments.
Args:
@@ -67,24 +56,32 @@ class Trader:
sl, tp = pips, pips * self.ram.risk_to_reward
tick = await self.symbol.info_tick()
if self.order.type == OrderType.BUY:
self.order.sl, self.order.tp = tick.ask - sl, tick.ask + tp
self.order.sl, self.order.tp = round(tick.ask - sl, self.symbol.digits), round(tick.ask + tp,
self.symbol.digits)
self.order.price = tick.ask
elif self.order.type == OrderType.SELL:
self.order.sl, self.order.tp = tick.bid + sl, tick.bid - tp
self.order.sl, self.order.tp = round(tick.bid + sl, self.symbol.digits), round(tick.bid - tp,
self.symbol.digits)
self.order.price = tick.bid
else:
raise ValueError(f"Invalid order type: {self.order.type}")
async def set_trade_stop_levels(self, *, points):
"""Set the stop loss and take profit levels of the order based on the points."""
"""Set the stop loss and take profit levels of the order based on the points.
Args:
points: Target points
"""
points = points * self.symbol.point
sl, tp = points, points * self.ram.risk_to_reward
tick = await self.symbol.info_tick()
if self.order.type == OrderType.BUY:
self.order.sl, self.order.tp = tick.ask - sl, tick.ask + tp
self.order.sl, self.order.tp = round(tick.ask - sl, self.symbol.digits), round(tick.ask + tp,
self.symbol.digits)
self.order.price = tick.ask
else:
self.order.sl, self.order.tp = tick.bid + sl, tick.bid - tp
self.order.sl, self.order.tp = round(tick.bid + sl, self.symbol.digits), round(tick.bid - tp,
self.symbol.digits)
self.order.price = tick.bid
async def check_order(self) -> bool:
@@ -95,51 +92,41 @@ class Trader:
"""
check = await self.order.check()
if check.retcode != 0:
logger.warning(
f"Symbol: {self.order.symbol}\nResult:\n{dict_to_string(check.get_dict(include={'comment', 'retcode'}), multi=True)}")
logger.warning(f"Symbol: {self.order.symbol}\nResult:\n"
f"{dict_to_string(check.get_dict(include={'comment', 'retcode'}), multi=True)}")
return False
return True
async def send_order(self):
"""Send the order to the broker."""
parameters = self.parameters.copy()
result = await self.order.send()
if result.retcode != 10009:
logger.warning(
f"Symbol: {self.order.symbol}\nResult:\n{dict_to_string(result.get_dict(include={'comment', 'retcode'}), multi=True)}")
logger.warning(f"Symbol: {self.order.symbol}\nResult:\n"
f"{dict_to_string(result.get_dict(include={'comment', 'retcode'}), multi=True)}")
return
logger.info(f"Symbol: {self.order.symbol}\nOrder: {dict_to_string(result.dict, multi=True)}\n")
await self.record_trade(result)
await self.record_trade(result, parameters)
async def record_trade(self, result: OrderSendResult, parameters: dict):
"""Record the trade in a csv file.
async def record_trade(self, result: OrderSendResult):
"""
Record the trade in a csv file.
Args:
result (OrderSendResult): Result of the order send
parameters: parameters of the trading strategy used to place the trade
"""
if result.retcode != 10009 or not self.config.record_trades:
return
params = parameters
profit = await self.order.calc_profit()
params = self.params
params['expected_profit'] = profit
params["expected_profit"] = profit
date = datetime.utcnow()
date = date.replace(tzinfo=ZoneInfo('UTC'))
params['date'] = date
params['time'] = date.timestamp()
date = date.replace(tzinfo=ZoneInfo("UTC"))
params["date"] = date
params["time"] = date.timestamp()
res = Result(result=result, parameters=params)
await res.save_csv()
async def place_trade(self, order_type: OrderType, params: dict = None, **kwargs):
"""Places a trade based on the order_type.
Args:
order_type (OrderType): Type of order
params: parameters of the trading strategy used to place the trade
kwargs: keyword arguments as required for the specific trader
"""
try:
await self.create_order(order_type=order_type, **kwargs)
if not await self.check_order():
return
self.params |= params or {}
await self.send_order()
except Exception as err:
logger.error(f"{err}. Symbol: {self.order.symbol}\n {self.__class__.__name__}.place_trade")
@abstractmethod
async def place_trade(self, *args, **kwargs):
"""Places a trade based on the order_type."""