import MetaTrader5 as mt5 import pandas as pd from datetime import datetime, timedelta import pytz import time import src.config as config class MT5Interface: def __init__(self): self.connected = False def initialize(self): """Initializes the connection to MetaTrader 5.""" path = getattr(config, 'MT5_PATH', '') login = getattr(config, 'MT5_LOGIN', 0) password = getattr(config, 'MT5_PASSWORD', '') server = getattr(config, 'MT5_SERVER', '') if path: if not mt5.initialize(path=path): print(f"MT5 initialize(path={path}) failed, error code = {mt5.last_error()}") self.connected = False return False else: if not mt5.initialize(): print(f"MT5 initialize() failed, error code = {mt5.last_error()}") self.connected = False return False # Attempt login if credentials provided if login and password and server: authorized = mt5.login(login=login, password=password, server=server) if not authorized: print(f"MT5 login failed, error code = {mt5.last_error()}") mt5.shutdown() self.connected = False return False print("MT5 Initialized successfully.") self.connected = True return True def shutdown(self): """Shuts down the connection.""" mt5.shutdown() self.connected = False def get_ticks(self, symbol, start_time_utc: datetime, end_time_utc: datetime): """ Fetches ticks for a given symbol and time range. Returns a DataFrame with 'time_msc', 'bid', 'ask', 'flags', 'volume'. """ if not self.connected: if not self.initialize(): return pd.DataFrame() # Ensure timestamps are timezone-aware (UTC) if start_time_utc.tzinfo is None: start_time_utc = start_time_utc.replace(tzinfo=pytz.utc) if end_time_utc.tzinfo is None: end_time_utc = end_time_utc.replace(tzinfo=pytz.utc) ticks = mt5.copy_ticks_range(symbol, start_time_utc, end_time_utc, mt5.COPY_TICKS_ALL) if ticks is None or len(ticks) == 0: print(f"No ticks found for {symbol} between {start_time_utc} and {end_time_utc}") return pd.DataFrame() df = pd.DataFrame(ticks) # Convert time_msc to datetime df['datetime'] = pd.to_datetime(df['time_msc'], unit='ms', utc=True) return df def get_last_tick(self, symbol): """Fetches the latest tick for the symbol.""" if not self.connected: return None tick = mt5.symbol_info_tick(symbol) return tick def get_symbol_info(self, symbol): """Returns symbol specification.""" info = mt5.symbol_info(symbol) return info