File size: 3,048 Bytes
c99df4c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | 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
|