diff --git a/core/bot.py b/core/bot.py index 3c11068..3da2757 100644 --- a/core/bot.py +++ b/core/bot.py @@ -183,6 +183,10 @@ class TradingBot: hour=21, minute=0 ) + self.scheduler.add_model_retrain_job( + self._retrain_model_task, + interval_hours=12 + ) self.scheduler.start() # Register shutdown handlers @@ -379,6 +383,17 @@ class TradingBot: logger.error(f"Error in main loop cycle #{self._cycle_count}: {e}") await asyncio.sleep(loop_interval) + async def _retrain_model_task(self): + """Asynchronous background task to retrain XGBoost model on latest market candles.""" + try: + from scripts.train_ai import run_training_pipeline + logger.info("🧠 Scheduled periodic AI model retraining started...") + await asyncio.to_thread(run_training_pipeline, "M15", 3000) + self.signal_classifier.load_model() + logger.success("βœ… Scheduled AI model retraining completed & reloaded") + except Exception as e: + logger.error(f"Error during scheduled model retraining: {e}") + async def _process_news(self): """Fetch and analyze news.""" try: diff --git a/core/mt5_connector.py b/core/mt5_connector.py index 1cd4535..f028d5c 100644 --- a/core/mt5_connector.py +++ b/core/mt5_connector.py @@ -260,6 +260,9 @@ class MT5Connector: symbol = symbol or self.settings.symbol tick = mt5.symbol_info_tick(symbol) + if tick is None or getattr(tick, 'bid', 0.0) == 0.0: + mt5.symbol_select(symbol, True) + tick = mt5.symbol_info_tick(symbol) if tick is None: logger.error(f"Failed to get tick for {symbol}") @@ -448,6 +451,9 @@ class MT5Connector: symbol = symbol or self.settings.symbol tick = mt5.symbol_info_tick(symbol) + if tick is None or getattr(tick, 'bid', 0.0) == 0.0: + mt5.symbol_select(symbol, True) + tick = mt5.symbol_info_tick(symbol) if tick is None: logger.error(f"Cannot get price for {symbol}") @@ -507,6 +513,101 @@ class MT5Connector: return result_dict @synchronized + @synchronized + def send_pending_order( + self, + order_type: str, + price: float, + symbol: str | None = None, + volume: float = 0.01, + sl: float = 0.0, + tp: float = 0.0, + comment: str = "XAUUSD_AI_PENDING", + magic: int = 123456, + ) -> dict | None: + """ + Send a pending order (BUY_LIMIT, SELL_LIMIT, BUY_STOP, SELL_STOP). + + Args: + order_type: "BUY_LIMIT", "SELL_LIMIT", "BUY_STOP", "SELL_STOP" + price: Order execution trigger price + symbol: Trading symbol (defaults to settings.symbol) + volume: Lot size + sl: Stop loss price + tp: Take profit price + comment: Order comment + magic: Magic number + + Returns: + Order result dict or None on failure + """ + if not self.ensure_connected(): + return None + + symbol = symbol or self.settings.symbol + type_map = { + "BUY_LIMIT": mt5.ORDER_TYPE_BUY_LIMIT, + "SELL_LIMIT": mt5.ORDER_TYPE_SELL_LIMIT, + "BUY_STOP": mt5.ORDER_TYPE_BUY_STOP, + "SELL_STOP": mt5.ORDER_TYPE_SELL_STOP, + } + mt5_type = type_map.get(order_type.upper()) + if mt5_type is None: + logger.error(f"Invalid pending order type: {order_type}") + return None + + request = { + "action": mt5.TRADE_ACTION_PENDING, + "symbol": symbol, + "volume": volume, + "type": mt5_type, + "price": round(price, 2), + "sl": round(sl, 2) if sl else 0.0, + "tp": round(tp, 2) if tp else 0.0, + "deviation": 30, + "magic": magic, + "comment": comment, + "type_time": mt5.ORDER_TIME_GTC, + "type_filling": self._get_filling_mode(symbol), + } + + result = mt5.order_send(request) + if result is None: + logger.error(f"Pending order send returned None: {mt5.last_error()}") + return None + + res_dict = { + "retcode": result.retcode, + "order": result.order, + "price": result.price, + "volume": result.volume, + "comment": result.comment, + } + + if result.retcode != mt5.TRADE_RETCODE_DONE: + logger.error(f"❌ Pending order failed: {result.retcode} β€” {result.comment}") + return res_dict + + logger.success(f"πŸ“Œ {order_type} {volume} lots {symbol} @ {price:.2f} | SL: {sl:.2f} | TP: {tp:.2f} | Order #{result.order}") + return res_dict + + @synchronized + def cancel_order(self, order_ticket: int) -> bool: + """Cancel a pending order by ticket number.""" + if not self.ensure_connected(): + return False + + request = { + "action": mt5.TRADE_ACTION_REMOVE, + "order": order_ticket, + } + result = mt5.order_send(request) + if result and result.retcode == mt5.TRADE_RETCODE_DONE: + logger.success(f"πŸ—‘οΈ Pending order #{order_ticket} canceled successfully") + return True + logger.error(f"Failed to cancel order #{order_ticket}: {result.comment if result else mt5.last_error()}") + return False + def modify_position( self, ticket: int, diff --git a/core/scheduler.py b/core/scheduler.py index 90f0e23..a8bffa3 100644 --- a/core/scheduler.py +++ b/core/scheduler.py @@ -70,3 +70,14 @@ class BotScheduler: replace_existing=True, ) logger.info(f"Scheduled economic calendar refresh every {interval_hours} hour(s)") + + def add_model_retrain_job(self, func: Callable, interval_hours: int = 12): + """Schedule periodic AI model retraining on fresh market data.""" + self.scheduler.add_job( + func, + trigger=IntervalTrigger(hours=interval_hours), + id="model_retrain", + name="Periodic AI Model Retraining", + replace_existing=True, + ) + logger.info(f"Scheduled AI model retraining every {interval_hours} hours") diff --git a/main.py b/main.py index ccfc60d..b752793 100644 --- a/main.py +++ b/main.py @@ -25,6 +25,11 @@ def parse_args(): action="store_true", help="Train AI models before starting the bot", ) + parser.add_argument( + "--train-only", + action="store_true", + help="Train AI models and exit without starting the bot", + ) parser.add_argument( "--backtest", action="store_true", @@ -65,26 +70,20 @@ async def train_models(): return try: - # Fetch 6 months of H1 data for training - df = mt5.get_rates(timeframe="H1", count=5000) + from scripts.train_ai import run_training_pipeline + run_training_pipeline(timeframe="M15", count=3000) - if df is None or len(df) < 1000: - logger.error("Insufficient historical data for training") - return + # Optional LSTM if TensorFlow is installed + try: + import tensorflow + df = mt5.get_rates(timeframe="H1", count=2000) + if df is not None: + lstm = PricePredictor() + lstm.train(df, epochs=10) + except Exception: + pass - logger.info(f"Training data: {len(df)} H1 candles") - - # Train LSTM - lstm = PricePredictor() - lstm_result = lstm.train(df, epochs=50) - logger.info(f"LSTM result: {lstm_result}") - - # Train XGBoost - xgb = SignalClassifier() - xgb_result = xgb.train(df) - logger.info(f"XGBoost result: {xgb_result}") - - logger.success("βœ… AI models trained successfully") + logger.success("βœ… AI models trained and ready for live execution") finally: mt5.disconnect() @@ -181,6 +180,10 @@ def main(): run_backtest() return + if args.train_only: + asyncio.run(train_models()) + return + if args.train: asyncio.run(train_models()) diff --git a/notifications/telegram_bot.py b/notifications/telegram_bot.py index 7f50c27..c9551f3 100644 --- a/notifications/telegram_bot.py +++ b/notifications/telegram_bot.py @@ -214,9 +214,15 @@ class TelegramNotifier: account = None if self.bot_instance and hasattr(self.bot_instance, "mt5"): mt5_obj = self.bot_instance.mt5 - is_conn = mt5_obj.is_connected() if callable(getattr(mt5_obj, "is_connected", None)) else getattr(mt5_obj, "is_connected", False) - if is_conn: - account = mt5_obj.get_account_info(auto_reconnect=False) + else: + from core.mt5_connector import MT5Connector + mt5_obj = MT5Connector() + + is_conn = mt5_obj.is_connected() if callable(getattr(mt5_obj, "is_connected", None)) else getattr(mt5_obj, "is_connected", False) + if not is_conn and hasattr(mt5_obj, "connect"): + is_conn = mt5_obj.connect(max_retries=1, retry_delay=0.1) + if is_conn: + account = mt5_obj.get_account_info(auto_reconnect=False) paused = getattr(self.bot_instance, "_trading_paused", False) state_str = "⏸️ PAUSED" if paused else "🟒 RUNNING" @@ -263,14 +269,21 @@ class TelegramNotifier: return positions = [] + pending = [] is_conn = False if self.bot_instance and hasattr(self.bot_instance, "mt5"): mt5_obj = self.bot_instance.mt5 - is_conn = mt5_obj.is_connected() if callable(getattr(mt5_obj, "is_connected", None)) else getattr(mt5_obj, "is_connected", False) - if not is_conn and hasattr(mt5_obj, "connect"): - is_conn = mt5_obj.connect(max_retries=1, retry_delay=0.1) - if is_conn: - positions = mt5_obj.get_open_positions(self.settings.symbol, auto_reconnect=False) + else: + from core.mt5_connector import MT5Connector + mt5_obj = MT5Connector() + + is_conn = mt5_obj.is_connected() if callable(getattr(mt5_obj, "is_connected", None)) else getattr(mt5_obj, "is_connected", False) + if not is_conn and hasattr(mt5_obj, "connect"): + is_conn = mt5_obj.connect(max_retries=1, retry_delay=0.1) + if is_conn: + positions = mt5_obj.get_open_positions(self.settings.symbol, auto_reconnect=False) + if hasattr(mt5_obj, "get_pending_orders"): + pending = mt5_obj.get_pending_orders(self.settings.symbol, auto_reconnect=False) keyboard_rows = [] @@ -281,11 +294,11 @@ class TelegramNotifier: f"⚠️ MT5 Terminal is currently Offline.\n\n" f"Cannot query live positions. Please ensure MetaTrader 5 is launched on your desktop with your Exness account logged in." ) - elif not positions: + elif not positions and not pending: msg = ( - f"πŸ“ˆ OPEN POSITIONS\n" + f"πŸ“ˆ OPEN POSITIONS & ORDERS\n" f"{'━' * 25}\n\n" - f"No open positions currently active on {self.settings.symbol}.\n" + f"No open positions or pending orders active on {self.settings.symbol}.\n" f"The bot is scanning market regimes and waiting for high-confluence entry signals." ) else: @@ -294,8 +307,8 @@ class TelegramNotifier: ticket = pos.get("ticket") direction = pos.get("type", "BUY") volume = pos.get("volume", 0.0) - entry = pos.get("open_price", 0.0) - current = pos.get("current_price", 0.0) + entry = pos.get("price_open", pos.get("open_price", 0.0)) + current = pos.get("price_current", pos.get("current_price", 0.0)) profit = pos.get("profit", 0.0) p_emoji = "🟒" if profit >= 0 else "πŸ”΄" dir_emoji = "πŸ“ˆ" if direction == "BUY" else "πŸ“‰" diff --git a/scripts/train_ai.py b/scripts/train_ai.py new file mode 100644 index 0000000..fcc5915 --- /dev/null +++ b/scripts/train_ai.py @@ -0,0 +1,219 @@ +""" +Institutional AI Model Training Pipeline for XAUUSD +Trains the XGBoost Signal Classifier using historical MT5 market data, +multi-timeframe engineered technical features, and TimeSeriesSplit cross-validation. +""" + +import os +import sys +if hasattr(sys.stdout, 'reconfigure'): + sys.stdout.reconfigure(encoding='utf-8') +if hasattr(sys.stderr, 'reconfigure'): + sys.stderr.reconfigure(encoding='utf-8') +import time +from pathlib import Path +import joblib +import numpy as np +import pandas as pd +from loguru import logger +import xgboost as xgb +from sklearn.model_selection import TimeSeriesSplit +from sklearn.metrics import accuracy_score, precision_recall_fscore_support + +# Ensure project root is in sys.path +PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from core.mt5_connector import MT5Connector +from analysis.technical import TechnicalAnalyzer +from config.settings import get_settings, MODELS_DIR + + +def fetch_training_data(timeframe: str = "M15", count: int = 3000) -> pd.DataFrame: + """Fetch high-quality historical candle data directly from MT5.""" + logger.info(f"Connecting to MT5 to fetch {count} candles on {timeframe}...") + connector = MT5Connector() + if not connector.connect(): + raise RuntimeError("Failed to connect to MT5. Ensure MT5 is running and logged in.") + + df = connector.get_rates(timeframe=timeframe, count=count) + connector.disconnect() + + if df is None or df.empty: + raise RuntimeError(f"Failed to fetch {timeframe} rates from MT5.") + + logger.success(f"Fetched {len(df)} historical candles from MT5 ({df.index[0]} -> {df.index[-1]})") + return df + + +def engineer_and_label(df: pd.DataFrame, horizon: int = 5, threshold_pct: float = 0.0018) -> tuple[np.ndarray, np.ndarray, list[str], pd.DataFrame]: + """ + Generate 48+ institutional quantitative features and future return target labels. + + Target: + 0: BUY (Forward return > +threshold_pct) + 1: HOLD (Forward return within [-threshold_pct, +threshold_pct]) + 2: SELL (Forward return < -threshold_pct) + """ + logger.info("Computing multi-indicator technical feature matrix...") + ta = TechnicalAnalyzer() + feature_df = ta.create_ml_features(df) + + # Future return labeling + future_returns = feature_df["close"].pct_change(horizon).shift(-horizon) + feature_df["target"] = 1 # Default HOLD + feature_df.loc[future_returns > threshold_pct, "target"] = 0 # BUY + feature_df.loc[future_returns < -threshold_pct, "target"] = 2 # SELL + + # Drop NaNs created by rolling indicators & future shift + clean_df = feature_df.dropna() + + numeric_df = clean_df.select_dtypes(include=[np.number]) + exclude_cols = ["target", "open", "high", "low", "close", "volume", "spread", "real_volume"] + feature_cols = [c for c in numeric_df.columns if c not in exclude_cols] + + X = clean_df[feature_cols].astype(float).values + y = clean_df["target"].astype(int).values + + counts = pd.Series(y).value_counts().to_dict() + logger.info(f"Feature matrix engineered: {X.shape[0]} samples, {len(feature_cols)} features") + logger.info(f"Label distribution: BUY (0)={counts.get(0, 0)}, HOLD (1)={counts.get(1, 0)}, SELL (2)={counts.get(2, 0)}") + + return X, y, feature_cols, clean_df + + +def train_model(X: np.ndarray, y: np.ndarray, feature_cols: list[str]) -> tuple[xgb.XGBClassifier, dict]: + """ + Train XGBoost model with TimeSeriesSplit to prevent lookahead bias. + """ + logger.info("Splitting data chronologically via TimeSeriesSplit (5 folds)...") + tscv = TimeSeriesSplit(n_splits=5) + last_train_idx, last_val_idx = list(tscv.split(X))[-1] + + X_train, X_val = X[last_train_idx], X[last_val_idx] + y_train, y_val = y[last_train_idx], y[last_val_idx] + + logger.info(f"Training set: {len(X_train)} samples | Validation set: {len(X_val)} samples") + + # Balanced class weights + classes = np.unique(y_train) + weights = len(y_train) / (len(classes) * np.bincount(y_train)) + sample_weight = np.array([weights[yi] for yi in y_train]) + + classifier = xgb.XGBClassifier( + n_estimators=180, + max_depth=4, + learning_rate=0.04, + subsample=0.85, + colsample_bytree=0.85, + gamma=0.2, + reg_alpha=0.1, + reg_lambda=1.0, + eval_metric="mlogloss", + random_state=42, + verbosity=0, + n_jobs=1, + ) + + logger.info("Fitting XGBoost Classifier with regularization...") + classifier.fit( + X_train, + y_train, + sample_weight=sample_weight, + eval_set=[(X_val, y_val)], + verbose=False, + ) + + y_pred = classifier.predict(X_val) + accuracy = accuracy_score(y_val, y_pred) + prec, rec, f1, _ = precision_recall_fscore_support(y_val, y_pred, average="weighted", zero_division=0) + + # Feature importances + importances = classifier.feature_importances_ + top_indices = np.argsort(importances)[::-1][:10] + top_features = [(feature_cols[i], float(importances[i])) for i in top_indices] + + metrics = { + "accuracy": float(accuracy), + "precision": float(prec), + "recall": float(rec), + "f1_score": float(f1), + "train_samples": len(X_train), + "val_samples": len(X_val), + "top_features": top_features, + } + + logger.success(f"XGBoost Training Complete! Accuracy: {accuracy:.2%} | F1-Score: {f1:.2%}") + return classifier, metrics + + +def save_artifacts(model: xgb.XGBClassifier, feature_cols: list[str]) -> Path: + """Save model and feature columns.""" + MODELS_DIR.mkdir(parents=True, exist_ok=True) + model_path = MODELS_DIR / "xgboost_classifier.joblib" + feat_path = MODELS_DIR / "xgb_feature_columns.joblib" + + joblib.dump(model, str(model_path)) + joblib.dump(feature_cols, str(feat_path)) + logger.success(f"Production artifacts saved to:\n β€’ {model_path}\n β€’ {feat_path}") + return model_path + + +def run_training_pipeline(timeframe: str = "M15", count: int = 3000) -> dict: + """Execute complete end-to-end model training.""" + start_time = time.time() + print("=" * 60) + print(" πŸš€ XAUUSD AI TRADING BOT β€” MODEL TRAINING PIPELINE") + print("=" * 60) + + # 1. Fetch data + df = fetch_training_data(timeframe=timeframe, count=count) + + # 2. Featurize & label + X, y, feature_cols, clean_df = engineer_and_label(df) + + # 3. Train & evaluate + model, metrics = train_model(X, y, feature_cols) + + # 4. Save + save_artifacts(model, feature_cols) + + # 5. Live inference test + test_sample = X[-1:].astype(float) + t0 = time.perf_counter() + probs = model.predict_proba(test_sample)[0] + pred_class = int(model.predict(test_sample)[0]) + inference_ms = (time.perf_counter() - t0) * 1000 + + class_names = {0: "BUY", 1: "HOLD", 2: "SELL"} + print("\n" + "─" * 60) + print(" πŸ“Š MODEL TRAINING REPORT") + print("─" * 60) + print(f" β€’ Timeframe: {timeframe}") + print(f" β€’ Total Candles: {len(df)}") + print(f" β€’ Features Engineered: {len(feature_cols)}") + print(f" β€’ Validation Accuracy: {metrics['accuracy']:.2%}") + print(f" β€’ Weighted F1-Score: {metrics['f1_score']:.2%}") + print(f" β€’ Inference Latency: {inference_ms:.2f} ms") + print(f" β€’ Live Test Direction: {class_names[pred_class]}") + print(f" - BUY Probability: {probs[0]:.2%}") + print(f" - HOLD Probability: {probs[1]:.2%}") + print(f" - SELL Probability: {probs[2]:.2%}") + print("\n πŸ” Top 5 Feature Drivers:") + for rank, (feat, imp) in enumerate(metrics["top_features"][:5], 1): + print(f" {rank}. {feat:25s} ({imp*100:.1f}%)") + print("─" * 60) + print(f" ✨ Training pipeline finished in {time.time() - start_time:.2f}s\n") + + return metrics + + +if __name__ == "__main__": + count = 3000 + tf = "M15" + if len(sys.argv) > 1 and sys.argv[1].isdigit(): + count = int(sys.argv[1]) + if len(sys.argv) > 2: + tf = sys.argv[2] + run_training_pipeline(timeframe=tf, count=count)