diegobeyl/backtesting
2
1"""
2MT5 Data Provider
3Fetches historical OHLCV data from MetaTrader 5
4"""
5
6import pandas as pd
7from datetime import datetime, timedelta
8from typing import Optional, List, Dict
9import logging
10
11logger = logging.getLogger(__name__)
12
13# Try to import MetaTrader5 (only available on Windows)
14try:
15 import MetaTrader5 as mt5
16 MT5_AVAILABLE = True
17except ImportError:
18 MT5_AVAILABLE = False
19 logger.warning("MetaTrader5 not available (only works on Windows)")
20 # Create mock constants for other platforms
21 class MT5Mock:
22 TIMEFRAME_M1 = 1
23 TIMEFRAME_M5 = 5
24 TIMEFRAME_M15 = 15
25 TIMEFRAME_M30 = 30
26 TIMEFRAME_H1 = 60
27 TIMEFRAME_H4 = 240
28 TIMEFRAME_D1 = 1440
29 TIMEFRAME_W1 = 10080
30 TIMEFRAME_MN1 = 43200
31 mt5 = MT5Mock()
32
33# Timeframe mapping
34TIMEFRAME_MAP = {
35 "M1": mt5.TIMEFRAME_M1,
36 "M5": mt5.TIMEFRAME_M5,
37 "M15": mt5.TIMEFRAME_M15,
38 "M30": mt5.TIMEFRAME_M30,
39 "H1": mt5.TIMEFRAME_H1,
40 "H4": mt5.TIMEFRAME_H4,
41 "D1": mt5.TIMEFRAME_D1,
42 "W1": mt5.TIMEFRAME_W1,
43 "MN": mt5.TIMEFRAME_MN1,
44 "MN1": mt5.TIMEFRAME_MN1,
45}
46
47# Timeframe to seconds mapping
48TIMEFRAME_SECONDS = {
49 "M1": 60,
50 "M5": 300,
51 "M15": 900,
52 "M30": 1800,
53 "H1": 3600,
54 "H4": 14400,
55 "D1": 86400,
56 "W1": 604800,
57 "MN": 2592000,
58 "MN1": 2592000,
59}
60
61
62class MT5DataProvider:
63 """Provides historical data from MetaTrader 5"""
64
65 def __init__(self):
66 self._connected = False
67 self._cache: Dict[str, pd.DataFrame] = {}
68
69 def connect(self) -> bool:
70 """Initialize connection to MT5"""
71 if not MT5_AVAILABLE:
72 logger.warning("MT5 not available on this platform")
73 return False
74
75 if self._connected:
76 return True
77
78 try:
79 if not mt5.initialize():
80 error = mt5.last_error()
81 logger.error(f"MT5 initialization failed: {error}")
82 return False
83
84 account_info = mt5.account_info()
85 if account_info is None:
86 logger.error("Failed to get account info")
87 return False
88
89 logger.info(f"Connected to MT5 - Account: {account_info.login}")
90 self._connected = True
91 return True
92
93 except Exception as e:
94 logger.error(f"Error connecting to MT5: {e}")
95 return False
96
97 def disconnect(self):
98 """Close MT5 connection"""
99 if self._connected:
100 mt5.shutdown()
101 self._connected = False
102 logger.info("Disconnected from MT5")
103
104 def get_symbols(self) -> List[str]:
105 """Get list of available symbols"""
106 if not self.connect():
107 return []
108
109 try:
110 symbols = mt5.symbols_get()
111 if symbols is None:
112 return []
113 return [s.name for s in symbols]
114 except Exception as e:
115 logger.error(f"Error getting symbols: {e}")
116 return []
117
118 def get_symbol_info(self, symbol: str) -> Optional[dict]:
119 """Get symbol information"""
120 if not self.connect():
121 return None
122
123 try:
124 info = mt5.symbol_info(symbol)
125 if info is None:
126 return None
127 return {
128 "name": info.name,
129 "description": info.description,
130 "digits": info.digits,
131 "point": info.point,
132 "spread": info.spread,
133 "volume_min": info.volume_min,
134 "volume_max": info.volume_max,
135 "volume_step": info.volume_step,
136 }
137 except Exception as e:
138 logger.error(f"Error getting symbol info: {e}")
139 return None
140
141 def get_data(
142 self,
143 symbol: str,
144 timeframe: str,
145 bars: int = 500,
146 start_date: Optional[datetime] = None,
147 end_date: Optional[datetime] = None,
148 use_cache: bool = True
149 ) -> Optional[pd.DataFrame]:
150 """
151 Fetch OHLCV data from MT5
152
153 Args:
154 symbol: Trading symbol (e.g., BTCUSD, EURUSD)
155 timeframe: Timeframe string (D1, W1, MN, etc.)
156 bars: Number of bars to fetch (if no date range)
157 start_date: Start date for data
158 end_date: End date for data
159 use_cache: Whether to use cached data
160
161 Returns:
162 DataFrame with columns: Open, High, Low, Close, Volume
163 """
164 if not self.connect():
165 return None
166
167 # Check cache
168 cache_key = f"{symbol}_{timeframe}_{bars}"
169 if use_cache and cache_key in self._cache:
170 logger.debug(f"Using cached data for {cache_key}")
171 return self._cache[cache_key].copy()
172
173 try:
174 # Select symbol
175 if not mt5.symbol_select(symbol, True):
176 logger.error(f"Failed to select symbol {symbol}")
177 return None
178
179 # Get MT5 timeframe constant
180 mt5_tf = TIMEFRAME_MAP.get(timeframe.upper())
181 if mt5_tf is None:
182 logger.error(f"Unknown timeframe: {timeframe}")
183 return None
184
185 # Fetch data
186 if start_date and end_date:
187 rates = mt5.copy_rates_range(symbol, mt5_tf, start_date, end_date)
188 else:
189 rates = mt5.copy_rates_from_pos(symbol, mt5_tf, 0, bars)
190
191 if rates is None or len(rates) == 0:
192 error = mt5.last_error()
193 logger.error(f"Failed to get data for {symbol}: {error}")
194 return None
195
196 # Convert to DataFrame
197 df = pd.DataFrame(rates)
198 df['time'] = pd.to_datetime(df['time'], unit='s')
199 df.set_index('time', inplace=True)
200
201 # Rename columns for backtesting.py compatibility
202 df.rename(columns={
203 'open': 'Open',
204 'high': 'High',
205 'low': 'Low',
206 'close': 'Close',
207 'tick_volume': 'Volume'
208 }, inplace=True)
209
210 # Keep only required columns
211 df = df[['Open', 'High', 'Low', 'Close', 'Volume']]
212
213 # Cache the data
214 self._cache[cache_key] = df.copy()
215
216 logger.info(f"Fetched {len(df)} bars for {symbol} {timeframe}")
217 return df
218
219 except Exception as e:
220 logger.error(f"Error fetching data: {e}")
221 return None
222
223 def clear_cache(self):
224 """Clear data cache"""
225 self._cache.clear()
226 logger.info("Cache cleared")
227
228 def get_timeframe_seconds(self, timeframe: str) -> int:
229 """Get timeframe duration in seconds"""
230 return TIMEFRAME_SECONDS.get(timeframe.upper(), 86400)
231
232
233# Singleton instance
234_provider = None
235
236def get_data_provider() -> MT5DataProvider:
237 """Get singleton data provider instance"""
238 global _provider
239 if _provider is None:
240 _provider = MT5DataProvider()
241 return _provider
242