import pandas as pd import numpy as np from datetime import datetime, timedelta import platform # Only attempt to import MetaTrader5 on Windows systems try: if platform.system() == 'Windows': import MetaTrader5 as mt5 else: # On non-Windows systems, set mt5 to None mt5 = None print("MetaTrader5 library not available on this platform, using mock data") except ImportError: # If import fails even on Windows, use mock data mt5 = None print("Failed to import MetaTrader5, using mock data") class MT5DataProcessor: def __init__(self): """Initialize the MT5 data processor""" self.initialized = False if mt5 is not None: self.initialized = self._initialize_mt5() def _initialize_mt5(self): """Initialize MT5 connection""" if mt5 is None: return False if not mt5.initialize(): print("Failed to initialize MT5") return False return True def get_historical_data(self, symbol, timeframe, start_date, end_date): """Get historical data, using mock data if MT5 is unavailable Args: symbol (str): 交易品种，如 'EURUSD' timeframe (int): 时间周期，如 mt5.TIMEFRAME_H1 start_date (datetime): 开始日期 end_date (datetime): 结束日期 Returns: pd.DataFrame: 包含OHLCV数据的DataFrame """ # Try to get real data if MT5 is available if self.initialized: try: rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date) if rates is not None: df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) return df except Exception as e: print(f"Error getting MT5 data: {e}") # Generate mock data if real data unavailable print(f"Generating mock data for {symbol}") dates = pd.date_range(start=start_date, end=end_date, freq='H') n = len(dates) np.random.seed(42) close = 1900 + np.cumsum(np.random.randn(n) * 2) high = close + np.random.rand(n) * 3 low = close - np.random.rand(n) * 3 # 使用numpy实现shift功能 open_price = np.roll(close, 1) open_price[0] = 1900 # 第一个元素设为初始值 volume = np.random.randint(100000, 1000000, n) df = pd.DataFrame({ 'open': open_price, 'high': high, 'low': low, 'close': close, 'volume': volume }, index=dates) df.index.name = 'time' return df def calculate_ema(self, df, period, price_column='close'): """Calculate Exponential Moving Average Args: df (pd.DataFrame): 价格数据 period (int): EMA周期 price_column (str): 价格列名 Returns: pd.Series: EMA值 """ return df[price_column].ewm(span=period, adjust=False).mean() def calculate_atr(self, df, period=14): """Calculate Average True Range Args: df (pd.DataFrame): 价格数据 period (int): ATR周期 Returns: pd.Series: ATR值 """ df = df.copy() df['tr1'] = abs(df['high'] - df['low']) df['tr2'] = abs(df['high'] - df['close'].shift(1)) df['tr3'] = abs(df['low'] - df['close'].shift(1)) df['tr'] = df[['tr1', 'tr2', 'tr3']].max(axis=1) atr = df['tr'].ewm(span=period, adjust=False).mean() return atr def calculate_rsi(self, df, period=14, price_column='close'): """Calculate Relative Strength Index Args: df (pd.DataFrame): 价格数据 period (int): RSI周期 price_column (str): 价格列名 Returns: pd.Series: RSI值 """ delta = df[price_column].diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def generate_features(self, df, fast_ema=12, slow_ema=26, atr_period=14, rsi_period=14): """Generate trading features Args: df (pd.DataFrame): 原始价格数据 fast_ema (int): 快速EMA周期 slow_ema (int): 慢速EMA周期 atr_period (int): ATR周期 rsi_period (int): RSI周期 Returns: pd.DataFrame: 包含原始数据和特征的DataFrame """ df = df.copy() # Calculate EMA df['ema_fast'] = self.calculate_ema(df, fast_ema) df['ema_slow'] = self.calculate_ema(df, slow_ema) # Calculate ATR df['atr'] = self.calculate_atr(df, atr_period) # Calculate RSI df['rsi'] = self.calculate_rsi(df, rsi_period) # Calculate EMA crossover signal df['ema_crossover'] = 0 df.loc[df['ema_fast'] > df['ema_slow'], 'ema_crossover'] = 1 df.loc[df['ema_fast'] < df['ema_slow'], 'ema_crossover'] = -1 # Calculate volatility (ATR percentage) df['volatility'] = (df['atr'] / df['close']) * 100 # Calculate price change rate df['price_change'] = df['close'].pct_change() * 100 # Instead of dropping rows with NaN, fill forward from first valid value # This preserves all original data points while handling missing indicator values indicators = ['ema_fast', 'ema_slow', 'atr', 'rsi', 'ema_crossover', 'volatility', 'price_change'] for indicator in indicators: if indicator in df.columns: # Fill NaN values with the first valid value of the indicator first_valid = df[indicator].first_valid_index() if first_valid is not None: # Fix FutureWarning: use assignment instead of inplace=True df[indicator] = df[indicator].fillna(df.loc[first_valid, indicator]) else: # If no valid values, fill with neutral value if indicator in ['ema_crossover']: df[indicator] = df[indicator].fillna(0) # Neutral signal elif indicator in ['rsi']: df[indicator] = df[indicator].fillna(50) # Neutral RSI else: df[indicator] = df[indicator].fillna(0) # Zero for other indicators return df def prepare_model_input(self, df, lookback_period=20): """Prepare input data for ML models Args: df (pd.DataFrame): 包含特征的数据 lookback_period (int): 回溯期 Returns: np.array: 模型输入数据 """ features = ['close', 'high', 'low', 'volume', 'ema_fast', 'ema_slow', 'atr', 'rsi', 'volatility', 'price_change'] X = [] for i in range(lookback_period, len(df)): X.append(df[features].iloc[i-lookback_period:i].values.flatten()) return np.array(X) def convert_to_dataframe(self, rates_data: list) -> pd.DataFrame: """将MT5发送的rates数据转换为DataFrame Args: rates_data (list): MT5发送的K线数据列表 Returns: pd.DataFrame: 包含OHLCV数据的DataFrame """ if not rates_data: return pd.DataFrame() # 转换为DataFrame df = pd.DataFrame(rates_data) # 确保必要的列存在 required_columns = ['time', 'open', 'high', 'low', 'close', 'tick_volume'] for col in required_columns: if col not in df.columns: # 如果缺少必要列，尝试使用默认值 if col == 'tick_volume': df[col] = 1000 # 默认成交量 else: df[col] = 1900 # 默认价格值 # 转换时间戳为datetime df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) # 重命名列以保持一致性 df.rename(columns={'tick_volume': 'volume'}, inplace=True) # 确保数据按时间排序 df.sort_index(inplace=True) return df def close(self): """Close MT5 connection if open""" if self.initialized and mt5 is not None: mt5.shutdown() def main(): """Test the data processor""" processor = MT5DataProcessor() end_date = datetime.now() start_date = end_date - timedelta(days=30) df = processor.get_historical_data('GOLD', None, start_date, end_date) print(f"Raw data shape: {df.shape}") df_with_features = processor.generate_features(df) print(f"Features data shape: {df_with_features.shape}") print(df_with_features.head()) processor.close() if __name__ == "__main__": main()