import sys import os import warnings from pathlib import Path # --------------------------------------------------------------------------- # 1) SET PROJECT ROOT AND UPDATE PATH/WORKING DIRECTORY # --------------------------------------------------------------------------- project_root = Path.cwd().parent.parent # Adjust if your notebook is in notebooks/time_series sys.path.append(str(project_root)) os.chdir(str(project_root)) warnings.filterwarnings("ignore") import warnings warnings.filterwarnings("ignore") import MetaTrader5 as mt5 import pandas as pd import numpy as np import ta from datetime import datetime, timedelta from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.ensemble import RandomForestRegressor from sklearn.feature_selection import SelectFromModel import time import logging import joblib # Setup logging logging.basicConfig( filename='models/saved_models/trading_app.log', level=logging.INFO, format='%(asctime)s %(levelname)s:%(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) def log_and_print(message, is_error=False): """ Logs and prints a message. If is_error=True, logs at the ERROR level; otherwise logs at INFO level. """ if is_error: logging.error(message) else: logging.info(message) print(message) # Update the login credentials and server information accordingly name = 52868686 key = 'kkk7s$zzz6' serv = 'ICMarketsSC-Demo' # Global variables SYMBOL = "EURUSD" LOT_SIZE = 0.01 TIMEFRAME = mt5.TIMEFRAME_D1 N_BARS = 50000 MAGIC_NUMBER = 234003 SLEEP_TIME = 86400 # 4 hours in seconds COMMENT_ML = "regression return" def select_features_rf_reg(X, y, estimator, max_features=20): """ Use a RandomForest (or similar) to select top 'max_features' features. """ selector = SelectFromModel(estimator=estimator, threshold=-np.inf, max_features=max_features).fit(X, y) X_transformed = selector.transform(X) selected_features_mask = selector.get_support() return X_transformed, selected_features_mask class TradingApp: def __init__(self, symbol, lot_size, magic_number): self.symbol = symbol self.lot_size = lot_size self.magic_number = magic_number self.pipeline = None # We'll store the loaded pipeline here self.last_retrain_time = None def get_data(self, symbol, n, timeframe): """ Fetch 'n' bars of historical data for the given symbol and timeframe. """ rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n) rates_frame = pd.DataFrame(rates) rates_frame['time'] = pd.to_datetime(rates_frame['time'], unit='s') rates_frame.set_index('time', inplace=True) return rates_frame def add_all_ta_features(self, df): """ Add technical analysis features to the DataFrame using the 'ta' library. """ df = ta.add_all_ta_features( df, open="open", high="high", low="low", close="close", volume="tick_volume", fillna=True ) return df def load_pipeline(self, pipeline_path): """ Load a pre-trained pipeline (scaler + model + possibly feature selection) from disk, e.g. 'best_rf_pipeline.pkl'. """ self.pipeline = joblib.load(pipeline_path) logging.info(f"Loaded pipeline from {pipeline_path}") def ml_signal_generation(self, symbol, n_bars, timeframe): """ Generate buy/sell signals using the loaded pipeline. Make sure the pipeline expects the same features as we create below. """ if self.pipeline is None: logging.error("No pipeline loaded. Call load_pipeline(...) first.") return False, False, True, True # 1) Fetch new data df = self.get_data(symbol, n_bars, timeframe) # 2) Add TA features (if your pipeline doesn't handle feature eng, do it here) df = self.add_all_ta_features(df) df.fillna(method='ffill', inplace=True) # 3) Prepare the features (the pipeline will do scaling/selection if included) X_new = df # If your pipeline expects specific columns, subset accordingly. # 4) Predict with the pipeline predictions = self.pipeline.predict(X_new) latest_pred = predictions[-1] # Get the most recent bar's prediction buy_signal = latest_pred > 0 sell_signal = latest_pred < 0 return buy_signal, sell_signal, not buy_signal, not sell_signal def calculate_future_returns(self, df): """ (Optional) Example function to calculate future returns for labeling. """ df["future_returns"] = df["close"].pct_change().shift(-1) return df.dropna() def orders(self, symbol, lot, is_buy=True, id_position=None, sl=None, tp=None): """ Place an order (BUY or SELL) for the specified symbol and lot size. """ symbol_info = mt5.symbol_info(symbol) if symbol_info is None: log_and_print(f"Symbol {symbol} not found, can't place order.", is_error=True) return "Symbol not found" # Make sure symbol is selected/visible if not symbol_info.visible: if not mt5.symbol_select(symbol, True): log_and_print(f"Failed to select symbol {symbol}", is_error=True) return "Symbol not visible or could not be selected." tick_info = mt5.symbol_info_tick(symbol) if tick_info is None: log_and_print(f"Could not get tick info for {symbol}.", is_error=True) return "Tick info unavailable" # Check for valid bid/ask if tick_info.bid <= 0 or tick_info.ask <= 0: log_and_print( f"Zero or invalid bid/ask for {symbol}: bid={tick_info.bid}, ask={tick_info.ask}", is_error=True ) return "Invalid prices" # ----------- LOT SIZE VALIDATION ----------- lot = max(lot, symbol_info.volume_min) step = symbol_info.volume_step if step > 0: remainder = lot % step if remainder != 0: lot = lot - remainder + step if lot > symbol_info.volume_max: lot = symbol_info.volume_max log_and_print( f"Adjusted lot size to {lot} (min={symbol_info.volume_min}, " f"step={symbol_info.volume_step}, max={symbol_info.volume_max})" ) # ----------- FORCE ORDER_FILLING_IOC ----------- filling_mode = 1 # ORDER_FILLING_IOC order_type = mt5.ORDER_TYPE_BUY if is_buy else mt5.ORDER_TYPE_SELL deviation = 20 request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lot, "type": order_type, "deviation": deviation, "magic": self.magic_number, "comment": COMMENT_ML, "type_time": mt5.ORDER_TIME_GTC, "type_filling": filling_mode, } if sl is not None: request["sl"] = sl if tp is not None: request["tp"] = tp if id_position is not None: request["position"] = id_position log_and_print(f"Sending order request: {request}") result = mt5.order_send(request) order_type_str = "BUY" if is_buy else "SELL" if result is None or result.retcode != mt5.TRADE_RETCODE_DONE: error_message = f"Order failed for {symbol}" if result: error_message += f", retcode={result.retcode}, comment={result.comment}" additional_info = ( f"Date/Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"Order Type: {order_type_str}\n" f"Lot Size: {lot}\n" f"SL: {sl if sl else 'None'}\n" f"TP: {tp if tp else 'None'}\n" f"Comment: {COMMENT_ML}\n" f"Request: {request}\n" f"Result: {result}" ) log_and_print(f"Order failed details: {additional_info}", is_error=True) else: success_message = f"Order successful for {symbol}, comment={result.comment}" additional_info = ( f"Date/Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"Order Type: {order_type_str}\n" f"Lot Size: {lot}\n" f"SL: {sl if sl else 'None'}\n" f"TP: {tp if tp else 'None'}\n" f"Comment: {COMMENT_ML}" ) log_and_print(success_message) def get_positions_by_magic(self, symbol, magic_number): """ Retrieve open positions for the specified symbol and magic number. """ all_positions = mt5.positions_get(symbol=symbol) if not all_positions: log_and_print("No positions found.", is_error=False) return [] return [pos for pos in all_positions if pos.magic == magic_number] def run_strategy(self, symbol, lot, buy_signal, sell_signal): """ Based on buy/sell signals, decide whether to open or close positions. """ log_and_print("------------------------------------------------------------------") log_and_print( f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}, " f"SYMBOL: {symbol}, BUY SIGNAL: {buy_signal}, SELL SIGNAL: {sell_signal}" ) # Retrieve positions based on the magic number to manage trades specific to this instance positions = self.get_positions_by_magic(symbol, self.magic_number) has_buy = any(pos.type == mt5.POSITION_TYPE_BUY for pos in positions) has_sell = any(pos.type == mt5.POSITION_TYPE_SELL for pos in positions) # Decision making based on current signals and existing positions if buy_signal and not has_buy: if has_sell: log_and_print("Existing sell positions found. Attempting to close...") if self.close_position(symbol, is_buy=True): log_and_print("Sell positions closed. Placing new buy order.") self.orders(symbol, lot, is_buy=True) else: log_and_print("Failed to close sell positions.") else: self.orders(symbol, lot, is_buy=True) elif sell_signal and not has_sell: if has_buy: log_and_print("Existing buy positions found. Attempting to close...") if self.close_position(symbol, is_buy=False): log_and_print("Buy positions closed. Placing new sell order.") self.orders(symbol, lot, is_buy=False) else: log_and_print("Failed to close buy positions.") else: self.orders(symbol, lot, is_buy=False) else: log_and_print("Appropriate position already exists or no signal to act on.") def close_position(self, symbol, is_buy): """ Closes positions of the opposite type (BUY/SELL) for this app's magic number. """ positions = mt5.positions_get(symbol=symbol) if not positions: log_and_print(f"No positions to close for symbol: {symbol}") return False initial_balance = mt5.account_info().balance closed_any = False for position in positions: # Close positions of the opposite type with the same magic number if position.magic == self.magic_number and ( (is_buy and position.type == mt5.POSITION_TYPE_SELL) or (not is_buy and position.type == mt5.POSITION_TYPE_BUY) ): close_request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": position.volume, "type": mt5.ORDER_TYPE_BUY if position.type == mt5.POSITION_TYPE_SELL else mt5.ORDER_TYPE_SELL, "position": position.ticket, "deviation": 20, "magic": self.magic_number, "comment": COMMENT_ML, "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_RETURN, } result = mt5.order_send(close_request) if result.retcode != mt5.TRADE_RETCODE_DONE: error_message = ( f"Failed to close position {position.ticket} for {symbol}: {result.retcode}" ) log_and_print(error_message, is_error=True) else: log_and_print(f"Successfully closed position {position.ticket} for {symbol}") closed_any = True if closed_any: final_balance = mt5.account_info().balance profit = final_balance - initial_balance success_message = f"Closed positions successfully, Profit: {profit}" log_and_print(success_message) return True return False def check_and_execute_trades(self): """ Convenience method to perform the entire flow: generate signals, run strategy, and deselect symbol. """ mt5.symbol_select(self.symbol, True) buy, sell, _, _ = self.ml_signal_generation(self.symbol, N_BARS, TIMEFRAME) self.run_strategy(self.symbol, self.lot_size, buy, sell) mt5.symbol_select(self.symbol, False) log_and_print("Waiting for new signals...") def is_market_open(): """ Check if the current time is within typical Forex trading session hours (CET/CEST). - Closes: Friday 10:00 PM CET - Opens: Sunday 11:00 PM CET - Closed all day Saturday """ current_time_utc = datetime.utcnow() # Adjust for CET (UTC+1) or CEST (UTC+2) current_time_cet = current_time_utc + timedelta(hours=2) if time.localtime().tm_isdst else current_time_utc + timedelta(hours=1) # Friday after 10 PM CET if current_time_cet.weekday() == 4 and current_time_cet.hour >= 22: return False # Sunday before 11 PM CET elif current_time_cet.weekday() == 6 and current_time_cet.hour < 23: return False # All day Saturday elif current_time_cet.weekday() == 5: return False return True if __name__ == "__main__": try: if not mt5.initialize(login=name, server=serv, password=key): log_and_print("Failed to initialize MetaTrader 5", is_error=True) exit() app = TradingApp(symbol=SYMBOL, lot_size=LOT_SIZE, magic_number=MAGIC_NUMBER) # Load a previously trained pipeline (scaler + model, etc.) pipeline_path = "models/saved_models/best_rf_pipeline.pkl" app.load_pipeline(pipeline_path) log_and_print(f"Loaded final pipeline for {app.symbol}") while True: log_and_print("Checking market status...") if is_market_open(): log_and_print("Market is open. Executing trades...") # Generate signals using the loaded pipeline buy_signal, sell_signal, _, _ = app.ml_signal_generation( symbol=app.symbol, n_bars=N_BARS, timeframe=TIMEFRAME ) # Run strategy app.run_strategy(app.symbol, app.lot_size, buy_signal, sell_signal) else: log_and_print("Market is closed. No actions performed.") # Sleep for the configured interval (e.g., 4 hours) time.sleep(SLEEP_TIME) except KeyboardInterrupt: log_and_print("Shutdown signal received.") except Exception as e: error_message = f"An error occurred: {e}" log_and_print(error_message, is_error=True) finally: mt5.shutdown() log_and_print("MetaTrader 5 shutdown completed.")