147 lines
6 KiB
Python
147 lines
6 KiB
Python
import re
| |
from datetime import datetime, timezone, timedelta
| |
import logging
| |
| |
import MetaTrader5 as mt5
| |
| |
logger = logging.getLogger(__name__)
| |
| |
| |
def mt5_freq_from_pandas(freq: str) -> int:
| |
"""
| |
Dynamically map pandas frequency strings to MetaTrader5 API timeframe constants.
| |
| |
Handles inputs like '1min', '15min', '1h', '4h', '1D', 'D', '1W', 'W', '1MS', 'MS'.
| |
| |
:param freq: pandas frequency string (e.g., '5min', '1h', '1D').
| |
See https://pandas.pydata.org/docs/user_guide/timeseries.html#timeseries-offset-aliases
| |
:return: Corresponding MetaTrader5 TIMEFRAME_* constant (integer).
| |
See https://www.mql5.com/en/docs/integration/python_metatrader5/
| |
:raises ValueError: If the frequency string is not recognized or the corresponding
| |
MT5 constant cannot be found.
| |
"""
| |
# Map Pandas units (lowercase) to MT5 prefixes and whether they always imply '1'
| |
unit_map = {
| |
'min': ('M', False),
| |
'h': ('H', False),
| |
'd': ('D', True),
| |
'w': ('W', True),
| |
'ms': ('MN', True), # Month Start maps to MN1
| |
}
| |
| |
# Try to match pattern: optional number + unit letters
| |
match = re.fullmatch(r"(\d+)?([A-Za-z]+)", str(freq))
| |
| |
if not match:
| |
raise ValueError(f"Input frequency '{freq}' does not match expected format (e.g., '1min', '4h', '1D').")
| |
| |
num_str, unit_pandas_raw = match.groups()
| |
unit_pandas = unit_pandas_raw.lower() # Normalize unit to lower case for map lookup
| |
| |
# Find the corresponding MT5 unit info
| |
mt5_prefix, is_always_one = None, False
| |
found_unit = False
| |
if unit_pandas == 'min':
| |
mt5_prefix, is_always_one = unit_map['min']
| |
found_unit = True
| |
elif unit_pandas == 'h':
| |
mt5_prefix, is_always_one = unit_map['h']
| |
found_unit = True
| |
# Use original case for D, W, MS check as they are distinct in Pandas
| |
elif unit_pandas_raw == 'D':
| |
mt5_prefix, is_always_one = unit_map['d'] # map key is lowercase
| |
found_unit = True
| |
elif unit_pandas_raw == 'W':
| |
mt5_prefix, is_always_one = unit_map['w'] # map key is lowercase
| |
found_unit = True
| |
elif unit_pandas_raw == 'MS':
| |
mt5_prefix, is_always_one = unit_map['ms'] # map key is lowercase
| |
found_unit = True
| |
| |
if not found_unit:
| |
raise ValueError(f"Unsupported Pandas frequency unit '{unit_pandas_raw}' in '{freq}'.")
| |
| |
# Determine the number part
| |
if is_always_one:
| |
number = 1
| |
elif num_str:
| |
number = int(num_str)
| |
else:
| |
# If number is missing for min/h (e.g., 'h'), assume 1
| |
number = 1
| |
| |
# Construct the MT5 constant name (e.g., "TIMEFRAME_M15", "TIMEFRAME_H4", "TIMEFRAME_D1")
| |
mt5_constant_name = f"TIMEFRAME_{mt5_prefix}{number}"
| |
| |
# Retrieve the constant value from the mt5 module
| |
try:
| |
return getattr(mt5, mt5_constant_name)
| |
except AttributeError:
| |
# Provide a more informative error if the constant doesn't exist
| |
supported_timeframes = [tf for tf_name, tf in mt5.__dict__.items() if tf_name.startswith('TIMEFRAME_')]
| |
raise ValueError(
| |
f"Could not find or map MetaTrader5 constant '{mt5_constant_name}' for frequency '{freq}'. "
| |
f"Check if this timeframe is supported by the MetaTrader5 library/API. "
| |
f"Available TIMEFRAME constants might include: {sorted(list(set(supported_timeframes)))}"
| |
)
| |
| |
| |
def get_timedelta_for_mt5_timeframe(mt5_timeframe: int, count: int) -> timedelta:
| |
"""
| |
Calculate the total duration corresponding to 'count' bars
| |
of the specified MT5 timeframe constant.
| |
| |
Internally maintains a cache of parsed timeframe details
| |
and a compiled regex for parsing attribute names.
| |
| |
:param mt5_timeframe: MT5 constant (e.g., mt5.TIMEFRAME_M15)
| |
:param count: Number of bars
| |
:return: timedelta representing the aggregated duration
| |
:raises ValueError: If the timeframe is unknown or unsupported
| |
"""
| |
# Initialize static attributes on the function for cache and pattern
| |
if not hasattr(get_timedelta_for_mt5_timeframe, "_pattern"):
| |
# Compile regex once
| |
get_timedelta_for_mt5_timeframe._pattern = re.compile(r"TIMEFRAME_([A-Z]+)(\d+)$")
| |
# Build cache mapping MT5 timeframe constants to (name, unit, number)
| |
cache: dict[int, tuple[str, str, int]] = {}
| |
for attr_name, attr_value in mt5.__dict__.items():
| |
if not attr_name.startswith("TIMEFRAME_") or not isinstance(attr_value, int):
| |
continue
| |
match = get_timedelta_for_mt5_timeframe._pattern.match(attr_name)
| |
if match:
| |
unit_prefix, number_str = match.groups()
| |
cache[attr_value] = (attr_name, unit_prefix, int(number_str))
| |
elif attr_name == "TIMEFRAME_MN1":
| |
# Special case for monthly timeframe without explicit number
| |
cache[attr_value] = (attr_name, "MN", 1)
| |
get_timedelta_for_mt5_timeframe._cache = cache
| |
logger.debug("Initialized MT5 timeframe pattern and cache")
| |
| |
# Retrieve static attributes
| |
pattern = get_timedelta_for_mt5_timeframe._pattern
| |
cache = get_timedelta_for_mt5_timeframe._cache
| |
| |
details = cache.get(mt5_timeframe)
| |
if details is None:
| |
raise ValueError(f"Unknown MetaTrader5 timeframe constant: {mt5_timeframe}")
| |
| |
name, unit_prefix, number = details
| |
| |
# Mapping of unit prefix to a factory function returning a timedelta
| |
unit_to_timedelta = {
| |
'M': lambda n, c: timedelta(minutes=n * c),
| |
'H': lambda n, c: timedelta(hours=n * c),
| |
'D': lambda n, c: timedelta(days=n * c),
| |
'W': lambda n, c: timedelta(weeks=n * c),
| |
'MN': lambda n, c: timedelta(days=n * c * 30.5), # approximate month
| |
}
| |
| |
factory = unit_to_timedelta.get(unit_prefix)
| |
if factory is None:
| |
raise ValueError(f"Unsupported timeframe unit '{unit_prefix}' derived from {name}")
| |
| |
if unit_prefix == 'MN':
| |
logger.warning("Using approximate duration of 30.5 days for monthly timeframes.")
| |
| |
return factory(number, count)
|