# LIVE TRADING CODE FOR DOUBLE-BARRIER CLASSIFICATION 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 import time import logging import joblib # Setup logging logging.basicConfig( filename='models/saved_models/trading_app_db.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 = 66677507 key = 'ST746$nG38' serv = 'ICMarketsSC-Demo' # Global variables SYMBOL = "EURUSD" LOT_SIZE = 0.01 TIMEFRAME = mt5.TIMEFRAME_D1 N_BARS = 50000 MAGIC_NUMBER = 234003 SLEEP_TIME = 86400 # e.g. 24 hours COMMENT_ML = "DoubleBarrier-ML" 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 classification pipeline here self.last_retrain_time = None def get_data(self, symbol, n, timeframe): """ Fetch the last 'n' bars from MetaTrader 5 for the given 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 '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): """ Loads a pre-trained classification pipeline (e.g., final_production_pipeline.pkl) that was trained on SHIFTED double-barrier labels in {0,1,2}. """ self.pipeline = joblib.load(pipeline_path) logging.info(f"Loaded pipeline from {pipeline_path}") log_and_print(f"Loaded pipeline from {pipeline_path}") def ml_signal_generation(self, symbol, n_bars, timeframe): """ Generate buy/sell signals using the loaded classification pipeline. The pipeline outputs SHIFTED labels in {0,1,2} => SHIFT them back to {-1,0,+1}. We'll interpret +1 => buy, -1 => sell, 0 => no trade. Double-Barrier labeling was used offline to train this pipeline, so we just replicate the same feature engineering steps and let the model predict. """ 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 df = self.add_all_ta_features(df) df.fillna(method='ffill', inplace=True) # 3) Prepare the features X_new = df # 4) Predict SHIFTED classes preds_shifted = self.pipeline.predict(X_new) # SHIFT them back: 0->-1, 1->0, 2->+1 preds = preds_shifted - 1 # Get the latest predicted class latest_pred = preds[-1] # If latest_pred == +1 => buy signal # If latest_pred == -1 => sell signal # If 0 => no trade buy_signal = (latest_pred == 1) sell_signal = (latest_pred == -1) return buy_signal, sell_signal, not buy_signal, not sell_signal def orders(self, symbol, lot, is_buy=True, id_position=None, sl=None, tp=None): """ Send an order (buy/sell) to MetaTrader 5. """ 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 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 = mt5.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}" ) # If you need notifications, you could log or handle them differently here. 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}" ) # If you need notifications, you could log or handle them differently here. log_and_print(success_message) def get_positions_by_magic(self, symbol, magic_number): """ Retrieve positions for a specific 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): """ Decide whether to open a buy or sell order based on signals, close opposite positions if needed, etc. """ 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}" ) 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) 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): """ Close all positions of the opposite type for the given symbol & magic. """ 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: if position.magic == self.magic_number: # if is_buy==True => we want to close SELL positions # if is_buy==False => we want to close BUY positions if ((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) # If you need notifications, you could log or handle them differently here. 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): """ Called in the main loop: generate signals, run strategy, etc. """ 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 the typical Forex trading session, adjusted for CET/CEST. Market closes at Friday 10:00 PM CET and opens at Sunday 11:00 PM CET. """ current_time_utc = datetime.utcnow() current_time_cet = ( current_time_utc + timedelta(hours=2) if time.localtime().tm_isdst else current_time_utc + timedelta(hours=1) ) # Market closes Friday after 10 PM CET if current_time_cet.weekday() == 4 and current_time_cet.hour >= 22: return False # Market opens Sunday after 11 PM CET elif current_time_cet.weekday() == 6 and current_time_cet.hour < 23: return False # Closed 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) # 1) Load the classification pipeline # Make sure this pipeline is a classification model expecting SHIFTED double-barrier labels in {0,1,2} pipeline_path = "models/saved_models/best_rf_db_pipeline.pkl" app.load_pipeline(pipeline_path) while True: log_and_print("Checking market status...") if is_market_open(): log_and_print("Market is open. Executing trades...") # 2) Generate signals using the loaded pipeline # This pipeline is classification-based => SHIFTED labels {0,1,2} # ml_signal_generation() SHIFTs them back to [-1,0,+1] buy_signal, sell_signal, _, _ = app.ml_signal_generation( symbol=app.symbol, n_bars=N_BARS, timeframe=TIMEFRAME ) # 3) Run strategy app.run_strategy(app.symbol, app.lot_size, buy_signal, sell_signal) else: log_and_print("Market is closed. No actions performed.") time.sleep(SLEEP_TIME) except KeyboardInterrupt: log_and_print("Shutdown signal received.") # If you need a notification here, handle it (e.g., log, email, etc.). except Exception as e: error_message = f"An error occurred: {e}" log_and_print(error_message, is_error=True) # If you need a notification here, handle it (e.g., log, email, etc.). finally: mt5.shutdown() log_and_print("MetaTrader 5 shutdown completed.") # If you need a notification here, handle it (e.g., log, email, etc.).