intelligent-trading-bot/inputs/utils_mt5.py

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)