src/core/mt5_interface.py

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