| 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
|
|
|
|
|
| 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()
|
|
|
|
|
| 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)
|
|
|
| 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
|
|
|
