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

Merge branch 'dev'

Browse Source
This commit is contained in:
Ichinga Samuel
2025-06-07 00:14:24 +01:00
parent 29f8c158bb 0778a354ba
commit 08da117a86
5 changed files with 223 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aiomql/lib/candle.py
+84 -56
View File
@@ -1,12 +1,12 @@
"""Candle and Candles classes for handling bars from the MetaTrader 5 terminal."""
import time
from datetime import datetime
from typing import Type, Self, Iterable
from logging import getLogger
from pandas import DataFrame, Series
import pandas as pd
import pandas_ta as ta
from pandas import DataFrame, Series, DatetimeIndex, Timestamp
from ..core.constants import TimeFrame
@@ -18,17 +18,18 @@ class Candle:
Candlesticks. You can subclass this class for added customization.
Attributes:
time (int): Period start time.
open (int): Open price
time (float): Period start time.
open (float): Open price
high (float): The highest price of the period
low (float): The lowest price of the period
close (float): Close price
tick_volume (float): Tick volume
real_volume (float): Trade volume
spread (float): Spread
Index (int): Custom attribute representing the position of the candle in a sequence.
index (Timestamp): Index of the object in the DataFrame, a timestamp
Index (int): Custom attribute representing the position of the candle for integer-location based indexing
"""
time: int
time: float
open: float
high: float
low: float
@@ -36,6 +37,7 @@ class Candle:
real_volume: float
spread: float
tick_volume: float
index: Timestamp
Index: int
def __init__(self, **kwargs):
@@ -47,8 +49,9 @@ class Candle:
"""
if not all(i in kwargs for i in ["open", "high", "low", "close"]):
raise ValueError("Candle must be instantiated with open, high, low and close prices")
self.time = kwargs.pop("time", int(time.time()))
self.time = kwargs.pop("time", Timestamp.now().timestamp())
self.Index = kwargs.pop("Index", 0)
self.index = kwargs.pop("index", Timestamp(self.time, unit="s", tz=datetime.now().astimezone().tzinfo))
self.real_volume = kwargs.pop("real_volume", 0)
self.spread = kwargs.pop("spread", 0)
self.tick_volume = kwargs.pop("tick_volume", 0)
@@ -56,7 +59,7 @@ class Candle:
def __repr__(self):
return (
"%(class)s(Index=%(Index)s, time=%(time)s, open=%(open)s, high=%(high)s, low=%(low)s, close=%(close)s)"
"%(class)s(Index=%(Index)s, time=%(time)s, open=%(open)s, high=%(high)s, low=%(low)s, close=%(close)s, index=%(index)s)"
% {
"class": self.__class__.__name__,
"open": self.open,
@@ -64,6 +67,7 @@ class Candle:
"low": self.low,
"close": self.close,
"time": self.time,
"index": self.index.isoformat(),
"Index": self.Index,
}
)
@@ -131,13 +135,18 @@ class Candle:
keys = include or set(self.__dict__.keys()).difference(exclude)
return {k: v for k, v in self if k in keys}
def to_series(self) -> Series:
"""Returns a Series Object"""
return Series(self.dict(exclude={"Index", "index"}))
class Candles:
"""An iterable container class of Candle objects in chronological order.
Attributes:
Index (Series['int']): A pandas Series of the indexes of all candles in the object.
time (Series['int']): A pandas Series of the time of all candles in the object.
index (DatetimeIndex): DatetimeIndex of the DataFrame object.
Index (Series['int']): A pandas Series of the indexes of all candles in the object:
time (Series['float']): A pandas Series of the time of all candles in the object.
open (Series[float]): A pandas Series of the opening price of all candles in the object.
high (Series[float]): A pandas Series of the high price of all candles in the object.
low (Series[float]): A pandas Series of the low price of all candles in the object.
@@ -155,7 +164,7 @@ class Candles:
The candle class can be customized by subclassing the Candle class and passing the subclass as the candle
keyword argument, or defining it on the class body as a class attribute.
"""
index: DatetimeIndex
Index: Series
time: Series
open: Series
@@ -189,18 +198,20 @@ class Candles:
raise ValueError(f"Cannot create DataFrame from object of {type(data)}")
self._data = data.loc[::-1] if flip else data
if 'time' in self._data.columns and self._data.index.name != 'time':
self._data.set_index('time', inplace=True, drop=False)
if 'time' in self._data.columns:
dtype = pd.DatetimeTZDtype(unit='s', tz=datetime.now().astimezone().tzinfo)
self._data.index = pd.DatetimeIndex(self._data.time, dtype=dtype)
self.Candle = candle_class or Candle
def __repr__(self):
return self._data.__repr__()
return repr(self._data)
def __len__(self):
return len(self._data.index)
def __contains__(self, item: Candle):
return item.time == self._data.loc[int(item.time)].time
return item.time == self[item.Index].time
def __getitem__(self, index: slice | int | str) -> Self | Series | Candle:
if isinstance(index, slice):
@@ -208,15 +219,18 @@ class Candles:
data = self._data.iloc[index]
return cls(data=data)
elif isinstance(index, str):
if isinstance(index, str):
if index == "index":
return self._data.index
if index == "Index":
return Series(self._data.index)
return Series(range(len(self._data)))
return self._data[index]
elif isinstance(index, int):
if isinstance(index, int):
candle = self._data.iloc[index]
_index = index if index >= 0 else len(self) + index
return self.Candle(**candle, Index=_index)
Index = index if index >= 0 else len(self) + index
_index = self._data.index[index]
return self.Candle(**candle, Index=Index, index=_index)
raise TypeError(f"Expected int, slice or str got {type(index)}")
def __setitem__(self, index, value: Series):
@@ -229,13 +243,27 @@ class Candles:
if item in self._data.columns:
return self._data[item]
if item == "index":
return self._data.index
if item == "Index":
return Series(self._data.index)
return Series(range(len(self._data)))
raise AttributeError(f"Attribute {item} not defined on class {self.__class__.__name__}")
def __reversed__(self):
for index, row in enumerate(iter(self._data[::-1].iloc)):
row = row.to_dict()
index = len(self._data) - index - 1
row["Index"] = index
row["index"] = self._data.index[index]
yield self.Candle(**row)
def __iter__(self):
return (self.Candle(**row.to_dict(), Index=ind) for ind, row in enumerate(self._data.iloc))
# return (self.Candle(**row._asdict()) for row in self._data.itertuples())
for index, row in enumerate(iter(self._data.iloc)):
row = row.to_dict()
row["Index"] = index
row["index"] = self._data.index[index]
yield self.Candle(**row)
@property
def timeframe(self):
@@ -282,39 +310,39 @@ class Candles:
res = self._data.rename(columns=kwargs, inplace=inplace)
return self if inplace else self.__class__(data=res)
def __iadd__(self, row: DataFrame | Series):
"""Add a new row to the candles class."""
# self._data = pd.concat([self._data, row])
if isinstance(row, Series):
self._data.loc[int(row.time)] = row
elif isinstance(row, DataFrame):
for r in row.iloc:
self._data.loc[int(r.time)] = r
def __iadd__(self, other: Self) -> Self:
"""Perform in place addition of candles"""
data_copy = self._data.copy()
other = other._data
for index, row in zip(other.index, iter(other.iloc)):
data_copy.loc[index] = row
self._data = data_copy.sort_index()
return self
def __add__(self, row: DataFrame | Series):
"""Add a new row to the candles class."""
if isinstance(row, Series):
data = self. pd.concat([self._data, pd.DataFrame(row).T])
return self.__class__(data=data)
def __add__(self, other: Self) -> Self:
"""Add two candles object and return a new one"""
data = self._data.copy()
for index, row in zip(other._data.index, iter(other._data.iloc)):
data.loc[index] = row
return self.__class__(data=data.sort_index())
elif isinstance(row, DataFrame):
data = pd.concat([self._data, row])
return self.__class__(data=data)
def add(self, row: DataFrame | Series) -> bool:
"""Add a new row to the candles class."""
new = True
if isinstance(row, Series):
if (index := int(row.time)) in self._data.index:
new = False
self._data.loc[index] = row
elif isinstance(row, DataFrame):
for r in row.iloc:
if (index := int(r.time)) in self._data.index:
new = False
self._data.loc[index] = r
return new
def add(self, obj: DataFrame | Series | Candle) -> Self:
"""Add new row(s) to the candles class."""
if isinstance(obj, Series):
index = Timestamp(obj.time, unit="s", tz=datetime.now().astimezone().tzinfo)
self._data.loc[index] = obj
self._data = self._data.sort_index()
return self
elif isinstance(obj, DataFrame):
data = self._data.copy()
for index, row in zip(obj.index, iter(obj.iloc)):
index = index if isinstance(index, Timestamp) else Timestamp(row.time, unit="s", tz=datetime.now().astimezone().tzinfo)
data.loc[index] = row
self._data = data.sort_index()
return self
elif isinstance(obj, Candle):
self._data.loc[obj.index] = obj.to_series()
self._data = self._data.sort_index()
return self
else:
raise TypeError("Expected Series, DataFrame or Candle, got {}".format(type(obj)))
src/aiomql/lib/symbol.py
+6 -6
View File
@@ -63,7 +63,7 @@ class Symbol(_Base, SymbolInfo):
if tick is not None:
tick = Tick(**tick._asdict())
setattr(self, "tick", tick) if not name else ...
return tick
return tick
except Exception as err:
logger.warning("%s: Unable to get tick for %s", err, self.name)
return None
@@ -92,7 +92,7 @@ class Symbol(_Base, SymbolInfo):
info = await self.mt5.symbol_info(self.name)
if info is not None:
info = info._asdict()
# self.set_attributes(**info)
self.set_attributes(**info)
return SymbolInfo(**info)
return None
@@ -100,7 +100,7 @@ class Symbol(_Base, SymbolInfo):
"""Initialize the symbol by pulling properties from the terminal
Returns:
bool: Returns True if symbol info was successful initialized
bool: Returns True if symbol info was successfully initialized
"""
try:
select = await self.mt5.symbol_select(self.name, True)
@@ -229,7 +229,7 @@ class Symbol(_Base, SymbolInfo):
"""
return self.volume_min
async def convert_currency(self, *, amount: float, from_currency: str, to_currency: str) -> float:
async 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
@@ -245,10 +245,10 @@ class Symbol(_Base, SymbolInfo):
pair = f"{base}{quote}"
tick = await self.info_tick(name=pair)
if tick is not None:
return round(amount / tick.ask, 2)
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
@backoff_decorator
async def copy_rates_from(self, *, timeframe: TimeFrame, date_from: datetime | int, count: int = 500) -> Candles:
src/aiomql/lib/ticks.py
+85 -14
View File
@@ -1,8 +1,9 @@
"""Module for working with price ticks."""
from typing import Iterable, Self
import time
from datetime import datetime
import pandas as pd
from pandas import DataFrame, Series
import pandas_ta as ta
@@ -23,7 +24,6 @@ class Tick:
volume_real (float): Volume for the current Last price
Index (int): Custom attribute representing the position of the tick in a sequence.
"""
time: float
bid: float
ask: float
@@ -32,6 +32,7 @@ class Tick:
time_msc: float
flags: TickFlag
volume_real: float
index: int | float
Index: int
def __init__(self, **kwargs):
@@ -39,14 +40,15 @@ class Tick:
present"""
if not all(key in kwargs for key in ["bid", "ask", "last", "volume"]):
raise ValueError("bid, ask, last and volume, time must be present in the keyword arguments")
self.time = kwargs.pop("time", datetime.now().timestamp())
self.time_msc = kwargs.pop("time_msc", self.time * 1000)
self.Index = kwargs.pop("Index", 0)
self.time = kwargs.pop("time", time.monotonic())
self.time_msc = int(self.time * 1000)
self.index = kwargs.pop("index", self.time_msc)
self.set_attributes(**kwargs)
def __repr__(self):
return (
"%(class)s(Index=%(Index)s, time=%(time)s, bid=%(bid)s, ask=%(ask)s, last=%(last)s, volume=%(volume)s)"
"%(class)s(Index=%(Index)s, time=%(time)s, bid=%(bid)s, ask=%(ask)s, last=%(last)s, volume=%(volume)s, index=%(index)s)"
% {
"class": self.__class__.__name__,
"time": self.time,
@@ -54,18 +56,19 @@ class Tick:
"ask": self.ask,
"last": self.last,
"volume": self.volume,
"index": self.index,
"Index": self.Index,
}
)
def __eq__(self, other: Self):
return self.time == other.time
return self.time_msc == other.time_msc
def __lt__(self, other: Self):
return self.time < other.time
return self.time_msc < other.time_msc
def __hash__(self):
return hash(self.time)
return hash(self.time_msc)
def __getitem__(self, item):
return self.__dict__[item]
@@ -102,10 +105,12 @@ class Tick:
for key, value in kwargs.items():
setattr(self, key, value)
def to_series(self) -> pd.Series:
"""Returns a Series Object"""
return Series(self.dict(exclude={"Index", "index"}))
class Ticks:
"""Container class for price ticks. Arrange in chronological order. Supports iteration, slicing and assignment"""
time: Series
bid: Series
ask: Series
@@ -115,6 +120,7 @@ class Ticks:
flags: Series
volume_real: Series
Index: Series
index: Series
def __init__(self, *, data: DataFrame | Iterable | Self, flip=False):
"""Initialize the Ticks class. Creates a DataFrame of price ticks from the data argument.
@@ -134,8 +140,11 @@ class Ticks:
raise ValueError(f"Cannot create DataFrame from object of {type(data)}")
self._data = data.iloc[::-1] if flip else data
if 'time_msc' in self._data.columns:
self._data.index = self._data.time_msc
def __repr__(self):
return self._data.__repr__()
return repr(self._data)
def __len__(self):
return self._data.shape[0]
@@ -146,20 +155,34 @@ class Ticks:
def __getattr__(self, item):
if item in list(self._data.columns.values):
return self._data[item]
if item == "index":
return self._data.index
if item == "Index":
return Series(range(len(self._data)))
raise AttributeError(f"Attribute {item} not defined on class {self.__class__.__name__}")
def __getitem__(self, index) -> Tick | Self:
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):
if index == "index":
return self._data.index
if index == "Index":
return Series(range(len(self._data)))
return self._data[index]
item = self._data.iloc[index]
return Tick(**item, Index=index)
if isinstance(index, int):
tick = self._data.iloc[index]
Index = index if index >= 0 else len(self) + index
_index = self._data.index[index]
return Tick(**tick, Index=Index, index=_index)
raise TypeError(f"Expected int, slice or str got {type(index)}")
def __setitem__(self, index, value: Series):
if isinstance(value, Series):
@@ -167,8 +190,20 @@ class Ticks:
return
raise TypeError(f"Expected Series got {type(value)}")
def __reversed__(self):
for index, row in enumerate(iter(self._data[::-1].iloc)):
row = row.to_dict()
index = len(self._data) - index - 1
row["Index"] = index
row["index"] = self._data.index[index]
yield Tick(**row)
def __iter__(self):
return (Tick(**row._asdict()) for row in self._data.itertuples())
for index, row in enumerate(iter(self._data.iloc)):
row = row.to_dict()
row["Index"] = index
row["index"] = self._data.index[index]
yield Tick(**row)
@property
def ta(self):
@@ -206,3 +241,39 @@ class Ticks:
"""
res = self._data.rename(columns=kwargs, inplace=inplace)
return res if inplace else self.__class__(data=res)
def __iadd__(self, other: Self) -> Self:
"""Perform in place addition of candles"""
data_copy = self._data.copy()
other = other._data
for index, row in zip(other.index, iter(other.iloc)):
data_copy.loc[index] = row
self._data = data_copy.sort_index()
return self
def __add__(self, other: Self) -> Self:
"""Add two candles object and return a new one"""
data = self._data.copy()
for index, row in zip(other._data.index, iter(other._data.iloc)):
data.loc[index] = row
return self.__class__(data=data.sort_index())
def add(self, obj: DataFrame | Series | Tick) -> Self:
"""Add new row(s) to the candles class."""
if isinstance(obj, Series):
self._data.loc[obj.index] = obj
self._data = self._data.sort_index()
return self
elif isinstance(obj, DataFrame):
data = self._data.copy()
for index, row in zip(obj.index, iter(obj.iloc)):
index = index
data.loc[index] = row
self._data = data.sort_index()
return self
elif isinstance(obj, Tick):
self._data.loc[obj.index] = obj.to_series()
self._data = self._data.sort_index()
return self
else:
raise TypeError("Expected Series, DataFrame or Candle, got {}".format(type(obj)))
tests/live/unit/test_candles.py
+47 -3
View File
@@ -1,8 +1,9 @@
from datetime import datetime
import pytest
import pytz
import pandas as pd
from pandas import Series
from aiomql.lib.candle import Candle, Candles
from aiomql.core.meta_trader import MetaTrader
from aiomql.core.constants import TimeFrame
@@ -11,8 +12,8 @@ from aiomql.core.constants import TimeFrame
class TestCandle:
@classmethod
def setup_class(cls):
cls.bullish_candle = Candle(open=1.3421, high=1.3462, low=1.3405, close=1.3452, time=0, Index=0)
cls.bearish_candle = Candle(open=1.3452, high=1.3405, low=1.3462, close=1.3421, time=1, Index=1)
cls.bullish_candle = Candle(open=1.3421, high=1.3462, low=1.3405, close=1.3452)
cls.bearish_candle = Candle(open=1.3452, high=1.3405, low=1.3462, close=1.3421)
def test_repr(self):
repr_str = repr(self.bearish_candle)
@@ -48,6 +49,10 @@ class TestCandle:
assert self.bearish_candle.is_bearish()
assert self.bullish_candle.is_bullish()
def test_to_series(self):
ser = self.bearish_candle.to_series()
assert isinstance(ser, Series)
class TestCandles:
@pytest.fixture(scope="class")
@@ -57,6 +62,13 @@ class TestCandles:
rates = await mt.copy_rates_from("BTCUSD", mt.TIMEFRAME_H1, start, 200)
return Candles(data=rates)
@pytest.fixture(scope="class")
async def candles_2(self):
mt = MetaTrader()
start = datetime(day=5, month=10, year=2023)
rates = await mt.copy_rates_from("BTCUSD", mt.TIMEFRAME_H1, start, 300)
return Candles(data=rates)
def test_get_series(self, candles):
series = candles["open"]
assert isinstance(series, pd.Series)
@@ -102,3 +114,35 @@ class TestCandles:
assert isinstance(fas, pd.Series)
candles["fas"] = fas
assert "fas" in candles.data.columns
def test_add_candles(self, candles, candles_2):
nc = candles + candles_2
candles += candles_2
assert len(nc) == 300
assert len(candles) == 300
def test_add_candle(self, candles):
length = len(candles)
candle = candles[-1]
now = datetime.now()
candle.time = now.timestamp()
candle.index = pd.Timestamp(candle.time, unit="s", tz=now.astimezone().tzinfo)
candles.add(candle)
assert len(candles) == length + 1
def test_add_series(self, candles):
candle = candles[-1]
length = len(candles)
now = datetime.now()
candle.time = now.timestamp()
series = candle.to_series()
candles.add(series)
assert len(candles) == length + 1
def test_add_dataframe(self, candles, candles_2):
df = candles_2._data.iloc[0:3]
now = datetime.now()
df.index = pd.DatetimeIndex(df.time, tz=now.astimezone().tzinfo)
length = len(candles)
candles.add(df)
assert len(candles) == length + len(df)
tests/live/unit/test_symbol.py
+1 -2
View File
@@ -11,8 +11,7 @@ class TestSymbol:
@pytest.fixture(scope="class", autouse=True)
async def btc(self):
symbol = Symbol(name="BTCUSD")
select = getattr(symbol, "select", False)
if select is False:
if symbol.initialized is False:
await symbol.initialize()
return symbol