MetaTrader5 groq pandas numpy python-dotenv ta # Replace with your actual Groq API key GROQ_API_KEY="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" # Optional: Specify MT5 login details if needed, otherwise uses running terminal # MT5_LOGIN=your_login # MT5_PASSWORD=your_password # MT5_SERVER=your_server # MT5_PATH="C:\Program Files\MetaTrader 5\terminal64.exe" # Example path import MetaTrader5 as mt5 import os from dotenv import load_dotenv load_dotenv() # Trading parameters SYMBOL = "EURUSD" # Trading symbol TIMEFRAME = mt5.TIMEFRAME_M1 # Timeframe for analysis (1 minute) VOLUME = 0.01 # Lot size for trades PROFIT_TARGET_USD = 5.0 # Target profit in USD TRADE_DURATION_MINUTES = 10 # Max duration before re-evaluating trade LOOP_SLEEP_SECONDS = 5 # Reduced frequency to allow for API calls ORDER_MAGIC_NUMBER = 12345 # Magic number to identify trades placed by this script # Technical Indicator parameters RSI_PERIOD = 14 MACD_FAST = 12 MACD_SLOW = 26 MACD_SIGNAL = 9 BB_PERIOD = 20 BB_STDDEV = 2 CCI_PERIOD = 20 STOCH_K = 14 STOCH_D = 3 STOCH_SMOOTH_K = 3 HISTORY_BARS = 100 # Number of bars for indicator calculation # Groq AI parameters GROQ_API_KEY = os.getenv("GROQ_API_KEY") GROQ_MODEL = "llama-3.1-70b-versatile" # Using 3.1 as 3.3 might not be available via API yet MAX_COMPLETION_TOKENS = 1000 TEMPERATURE = 0.7 # Adjust for more creative/deterministic responses # MT5 Connection parameters (optional, loaded from .env if present) MT5_LOGIN = os.getenv("MT5_LOGIN") MT5_PASSWORD = os.getenv("MT5_PASSWORD") MT5_SERVER = os.getenv("MT5_SERVER") MT5_PATH = os.getenv("MT5_PATH") # Logging configuration LOG_FILE = "logs/trading_bot.log" LOG_LEVEL = "INFO" # DEBUG, INFO, WARNING, ERROR, CRITICAL # Ensure logs directory exists if not os.path.exists("logs"): os.makedirs("logs") # Input validation if not GROQ_API_KEY: raise ValueError("GROQ_API_KEY not found in .env file.") if not SYMBOL: raise ValueError("SYMBOL cannot be empty in config.") import MetaTrader5 as mt5 import pandas as pd import numpy as np from ta import add_all_ta_features from ta.utils import dropna import time import logging from datetime import datetime, timedelta, timezone import config logger = logging.getLogger(__name__) class MT5Handler: def __init__(self): self.connected = False self.initialize_mt5() def initialize_mt5(self): """Initializes the MetaTrader 5 connection.""" logger.info("Initializing MetaTrader 5...") args = {} if config.MT5_PATH: args["path"] = config.MT5_PATH if config.MT5_LOGIN: args["login"] = int(config.MT5_LOGIN) if config.MT5_PASSWORD: args["password"] = config.MT5_PASSWORD if config.MT5_SERVER: args["server"] = config.MT5_SERVER if not mt5.initialize(**args): logger.error(f"MT5 initialize() failed, error code: {mt5.last_error()}") self.connected = False return False logger.info(f"MT5 Initialized: Version {mt5.version()}") self.connected = True return True def connect(self): """Ensures MT5 is connected.""" if not self.connected: return self.initialize_mt5() # Check if terminal is still available if not mt5.terminal_info(): logger.warning("MT5 terminal connection lost. Attempting to re-initialize.") self.connected = False return self.initialize_mt5() return True def disconnect(self): """Shuts down the MetaTrader 5 connection.""" if self.connected: logger.info("Shutting down MetaTrader 5 connection.") mt5.shutdown() self.connected = False def get_market_data(self, symbol, timeframe, count): """Fetches historical rates and the latest tick.""" if not self.connect(): return None, None # Get historical data rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count) if rates is None or len(rates) == 0: logger.error(f"Failed to get rates for {symbol}, error: {mt5.last_error()}") return None, None rates_df = pd.DataFrame(rates) # Convert time in seconds into the datetime format rates_df['time'] = pd.to_datetime(rates_df['time'], unit='s', utc=True) rates_df.set_index('time', inplace=True) # Get latest tick tick = mt5.symbol_info_tick(symbol) if tick is None: logger.error(f"Failed to get tick for {symbol}, error: {mt5.last_error()}") return rates_df, None # Return rates even if tick fails return rates_df, tick def calculate_indicators(self, rates_df): """Calculates technical indicators using the 'ta' library.""" if rates_df is None or rates_df.empty: logger.warning("Cannot calculate indicators: DataFrame is empty.") return None try: # Clean column names if necessary (sometimes MT5 returns 'real_volume') rates_df.rename(columns={'real_volume': 'volume'}, inplace=True, errors='ignore') if 'volume' not in rates_df.columns: # If volume is still missing, add a dummy column for ta compatibility rates_df['volume'] = 1 # Ensure required columns exist required_cols = ['open', 'high', 'low', 'close', 'volume'] if not all(col in rates_df.columns for col in required_cols): logger.error(f"DataFrame missing required columns for TA calculation. Found: {rates_df.columns.tolist()}") return None # Calculate indicators df = add_all_ta_features( rates_df.copy(), open="open", high="high", low="low", close="close", volume="volume", fillna=True # Fill NaN values which can occur at the start ) # Select and format specific indicators needed indicators = {} last_row = df.iloc[-1] # RSI rsi_col = f'momentum_rsi' # Default name from ta library if rsi_col in df.columns: indicators['rsi'] = last_row[rsi_col] else: logger.warning("RSI column not found after TA calculation.") indicators['rsi'] = np.nan # MACD macd_line_col = 'trend_macd' macd_signal_col = 'trend_macd_signal' macd_hist_col = 'trend_macd_diff' if all(c in df.columns for c in [macd_line_col, macd_signal_col, macd_hist_col]): indicators['macd_line'] = last_row[macd_line_col] indicators['macd_signal'] = last_row[macd_signal_col] indicators['macd_hist'] = last_row[macd_hist_col] else: logger.warning("MACD columns not found.") indicators.update({'macd_line': np.nan, 'macd_signal': np.nan, 'macd_hist': np.nan}) # Bollinger Bands bb_hband_col = 'volatility_bbh' bb_lband_col = 'volatility_bbl' bb_mavg_col = 'volatility_bbm' if all(c in df.columns for c in [bb_hband_col, bb_lband_col, bb_mavg_col]): indicators['bb_upper'] = last_row[bb_hband_col] indicators['bb_lower'] = last_row[bb_lband_col] indicators['bb_middle'] = last_row[bb_mavg_col] else: logger.warning("Bollinger Band columns not found.") indicators.update({'bb_upper': np.nan, 'bb_lower': np.nan, 'bb_middle': np.nan}) # CCI cci_col = 'trend_cci' if cci_col in df.columns: indicators['cci'] = last_row[cci_col] else: logger.warning("CCI column not found.") indicators['cci'] = np.nan # Stochastic Oscillator stoch_k_col = 'momentum_stoch' stoch_d_col = 'momentum_stoch_signal' if all(c in df.columns for c in [stoch_k_col, stoch_d_col]): indicators['stoch_k'] = last_row[stoch_k_col] indicators['stoch_d'] = last_row[stoch_d_col] else: logger.warning("Stochastic columns not found.") indicators.update({'stoch_k': np.nan, 'stoch_d': np.nan}) # Add recent candlestick data (last 3 bars) indicators['candlesticks'] = rates_df[['open', 'high', 'low', 'close']].tail(3).to_string() return indicators except Exception as e: logger.error(f"Error calculating indicators: {e}", exc_info=True) return None def get_symbol_info(self, symbol): """Gets symbol information.""" if not self.connect(): return None info = mt5.symbol_info(symbol) if info is None: logger.error(f"Failed to get symbol info for {symbol}, error: {mt5.last_error()}") return info def place_order(self, symbol, order_type, volume, price=None, sl=None, tp=None, deviation=10, magic=config.ORDER_MAGIC_NUMBER): """Places a market order.""" if not self.connect(): return None symbol_info = self.get_symbol_info(symbol) if not symbol_info: return None if order_type not in [mt5.ORDER_TYPE_BUY, mt5.ORDER_TYPE_SELL]: logger.error(f"Invalid order type: {order_type}") return None point = symbol_info.point ask_price = mt5.symbol_info_tick(symbol).ask bid_price = mt5.symbol_info_tick(symbol).bid if order_type == mt5.ORDER_TYPE_BUY: order_price = ask_price if price is None else price if sl is not None: sl = float(sl) if tp is not None: tp = float(tp) elif order_type == mt5.ORDER_TYPE_SELL: order_price = bid_price if price is None else price if sl is not None: sl = float(sl) if tp is not None: tp = float(tp) else: logger.error(f"Unsupported order type {order_type}") return None request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": float(volume), "type": order_type, "price": order_price, "sl": sl if sl else 0.0, "tp": tp if tp else 0.0, "deviation": deviation, "magic": magic, "comment": "GroqTrader", "type_time": mt5.ORDER_TIME_GTC, # Good till cancelled "type_filling": mt5.ORDER_FILLING_IOC, # Immediate or Cancel } # Ensure SL/TP are valid relative to current price if order_type == mt5.ORDER_TYPE_BUY: if sl is not None and sl >= ask_price: logger.warning(f"Adjusting SL for BUY order: SL {sl} >= Ask {ask_price}. Setting SL to 0.") request["sl"] = 0.0 if tp is not None and tp <= ask_price: logger.warning(f"Adjusting TP for BUY order: TP {tp} <= Ask {ask_price}. Setting TP to 0.") request["tp"] = 0.0 elif order_type == mt5.ORDER_TYPE_SELL: if sl is not None and sl <= bid_price: logger.warning(f"Adjusting SL for SELL order: SL {sl} <= Bid {bid_price}. Setting SL to 0.") request["sl"] = 0.0 if tp is not None and tp >= bid_price: logger.warning(f"Adjusting TP for SELL order: TP {tp} >= Bid {bid_price}. Setting TP to 0.") request["tp"] = 0.0 logger.info(f"Placing order: {request}") result = mt5.order_send(request) if result is None: logger.error(f"order_send failed, error code: {mt5.last_error()}") return None elif result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"Order failed, retcode: {result.retcode}, comment: {result.comment}") # Log request details for debugging logger.error(f"Failed request details: {request}") return None else: logger.info(f"Order placed successfully: ticket={result.order}, price={result.price}, volume={result.volume}") return result def close_position(self, position, deviation=10): """Closes an open position.""" if not self.connect() or position is None: return False symbol = position.symbol ticket = position.ticket volume = position.volume order_type = position.type # 0 for Buy, 1 for Sell close_type = mt5.ORDER_TYPE_SELL if order_type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY price = mt5.symbol_info_tick(symbol).bid if order_type == mt5.ORDER_TYPE_BUY else mt5.symbol_info_tick(symbol).ask request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": close_type, "position": ticket, "price": price, "deviation": deviation, "magic": config.ORDER_MAGIC_NUMBER, "comment": "GroqTrader Close", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } logger.info(f"Closing position ticket {ticket}: {request}") result = mt5.order_send(request) if result is None: logger.error(f"close_position order_send failed for ticket {ticket}, error code: {mt5.last_error()}") return False elif result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"close_position failed for ticket {ticket}, retcode: {result.retcode}, comment: {result.comment}") return False else: logger.info(f"Position {ticket} closed successfully at price {result.price}") return True def modify_position(self, position_ticket, sl=None, tp=None): """Modifies the SL and TP of an open position.""" if not self.connect(): return False position = mt5.positions_get(ticket=position_ticket) if not position or len(position) == 0: logger.warning(f"Cannot modify position ticket {position_ticket}: Not found.") return False position = position[0] # Get the position object # Get current market prices symbol = position.symbol tick = mt5.symbol_info_tick(symbol) if not tick: logger.error(f"Cannot get tick for {symbol} to modify position.") return False ask_price = tick.ask bid_price = tick.bid # Prepare modification request request = { "action": mt5.TRADE_ACTION_SLTP, "position": position_ticket, "symbol": symbol, "sl": position.sl, # Keep original if not provided "tp": position.tp, # Keep original if not provided } modified = False if sl is not None: sl = float(sl) # Check validity: SL for BUY must be below current Bid, SL for SELL must be above current Ask if position.type == mt5.ORDER_TYPE_BUY and sl < bid_price: request["sl"] = sl modified = True elif position.type == mt5.ORDER_TYPE_SELL and sl > ask_price: request["sl"] = sl modified = True else: logger.warning(f"Invalid SL {sl} for position {position_ticket} (type: {position.type}, bid: {bid_price}, ask: {ask_price}). Modification skipped.") # Keep original SL if new one is invalid request["sl"] = position.sl if tp is not None: tp = float(tp) # Check validity: TP for BUY must be above current Ask, TP for SELL must be below current Bid if position.type == mt5.ORDER_TYPE_BUY and tp > ask_price: request["tp"] = tp modified = True elif position.type == mt5.ORDER_TYPE_SELL and tp < bid_price: request["tp"] = tp modified = True else: logger.warning(f"Invalid TP {tp} for position {position_ticket} (type: {position.type}, bid: {bid_price}, ask: {ask_price}). Modification skipped.") # Keep original TP if new one is invalid request["tp"] = position.tp if not modified: logger.info(f"No valid modifications provided or needed for position {position_ticket}.") return False # No changes made or needed # Only send if there's a valid change if request["sl"] != position.sl or request["tp"] != position.tp: logger.info(f"Modifying position {position_ticket}: SL={request['sl']}, TP={request['tp']}") result = mt5.order_send(request) if result is None: logger.error(f"modify_position order_send failed for ticket {position_ticket}, error code: {mt5.last_error()}") return False elif result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"modify_position failed for ticket {position_ticket}, retcode: {result.retcode}, comment: {result.comment}") return False else: logger.info(f"Position {position_ticket} modified successfully.") return True else: logger.info(f"No modification needed for position {position_ticket} as requested SL/TP are same as current or invalid.") return False def get_open_positions(self, symbol=None, magic=config.ORDER_MAGIC_NUMBER): """Gets all open positions, optionally filtered by symbol and magic number.""" if not self.connect(): return [] positions = mt5.positions_get() if positions is None: logger.error(f"Failed to get positions, error: {mt5.last_error()}") return [] filtered_positions = [] for pos in positions: if pos.magic == magic: if symbol is None or pos.symbol == symbol: filtered_positions.append(pos) return filtered_positions def check_tp_sl_hit(self, position): """ Checks if the current price has hit the TP or SL of a position. MT5 server usually handles this, but this provides a script-level check. Returns 'TP', 'SL', or None. """ if not self.connect() or position is None: return None symbol = position.symbol tick = mt5.symbol_info_tick(symbol) if not tick: logger.warning(f"Could not get tick for {symbol} to check TP/SL for ticket {position.ticket}") return None # Check TP if position.tp > 0: if position.type == mt5.ORDER_TYPE_BUY and tick.bid >= position.tp: logger.info(f"TP hit for BUY position {position.ticket} (TP: {position.tp}, Bid: {tick.bid})") return 'TP' if position.type == mt5.ORDER_TYPE_SELL and tick.ask <= position.tp: logger.info(f"TP hit for SELL position {position.ticket} (TP: {position.tp}, Ask: {tick.ask})") return 'TP' # Check SL if position.sl > 0: if position.type == mt5.ORDER_TYPE_BUY and tick.bid <= position.sl: logger.info(f"SL hit for BUY position {position.ticket} (SL: {position.sl}, Bid: {tick.bid})") return 'SL' if position.type == mt5.ORDER_TYPE_SELL and tick.ask >= position.sl: logger.info(f"SL hit for SELL position {position.ticket} (SL: {position.sl}, Ask: {tick.ask})") return 'SL' return None def get_position_age_minutes(self, position): """Calculates the age of a position in minutes.""" if position is None: return 0 # Position time is usually in UTC seconds entry_time = datetime.fromtimestamp(position.time, tz=timezone.utc) now_time = datetime.now(timezone.utc) age = now_time - entry_time return age.total_seconds() / 60.0 import logging from groq import Groq, GroqError import config import re logger = logging.getLogger(__name__) class GroqAnalyzer: def __init__(self): if not config.GROQ_API_KEY: raise ValueError("Groq API key is not configured.") try: self.client = Groq(api_key=config.GROQ_API_KEY) logger.info("Groq client initialized.") except Exception as e: logger.error(f"Failed to initialize Groq client: {e}", exc_info=True) raise def _parse_recommendation(self, text): """Parses the AI's text response to extract structured data.""" if not text: return None recommendation = { "action": "HOLD", # Default "entry": None, "tp": None, "sl": None, "new_tp": None, # For modification "new_sl": None, # For modification "justification": "N/A" } try: # Extract Action action_match = re.search(r"Action:\s*(BUY|SELL|HOLD|CLOSE|MODIFY)", text, re.IGNORECASE) if action_match: recommendation["action"] = action_match.group(1).upper() # Extract Entry Price (only for BUY/SELL) if recommendation["action"] in ["BUY", "SELL"]: entry_match = re.search(r"Entry:\s*([\d.]+)", text, re.IGNORECASE) if entry_match: recommendation["entry"] = float(entry_match.group(1)) # AI suggests entry, but we use market price # Extract TP tp_match = re.search(r"TP:\s*([\d.]+)", text, re.IGNORECASE) if tp_match: recommendation["tp"] = float(tp_match.group(1)) # Extract SL sl_match = re.search(r"SL:\s*([\d.]+)", text, re.IGNORECASE) if sl_match: recommendation["sl"] = float(sl_match.group(1)) # Extract New TP (for MODIFY) new_tp_match = re.search(r"New TP:\s*([\d.]+)", text, re.IGNORECASE) if new_tp_match: recommendation["new_tp"] = float(new_tp_match.group(1)) # Extract New SL (for MODIFY) new_sl_match = re.search(r"New SL:\s*([\d.]+)", text, re.IGNORECASE) if new_sl_match: recommendation["new_sl"] = float(new_sl_match.group(1)) # Extract Justification justification_match = re.search(r"Justification:\s*(.*)", text, re.IGNORECASE | re.DOTALL) if justification_match: recommendation["justification"] = justification_match.group(1).strip() else: # Fallback if Justification: label is missing lines = text.splitlines() if len(lines) > 1: # Assume everything after the first few lines is justification start_index = 1 for i, line in enumerate(lines): if "Justification:" in line: start_index = i + 1 break if any(kw in line for kw in ["Action:", "Entry:", "TP:", "SL:", "New TP:", "New SL:"]): start_index = i + 1 recommendation["justification"] = "\n".join(lines[start_index:]).strip() except Exception as e: logger.error(f"Error parsing AI recommendation: {e}\nRaw text: {text}", exc_info=True) return None # Indicate parsing failure # Basic validation if recommendation["action"] in ["BUY", "SELL"] and (recommendation["tp"] is None or recommendation["sl"] is None): logger.warning(f"AI recommended {recommendation['action']} but missing TP or SL. Raw: {text}") # Decide how to handle: maybe default to HOLD or reject recommendation # For now, let it pass but log warning. Execution logic should handle missing values. if recommendation["action"] == "MODIFY" and (recommendation["new_tp"] is None and recommendation["new_sl"] is None): logger.warning(f"AI recommended MODIFY but provided no new TP or SL. Raw: {text}") # Treat as HOLD if no changes suggested recommendation["action"] = "HOLD" return recommendation def get_recommendation(self, market_data, indicators, open_position=None, position_age=0, position_pnl=0): """Generates a prompt and gets a trading recommendation from Groq.""" if not market_data or not indicators: logger.error("Cannot get recommendation: Missing market data or indicators.") return None current_time_str = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S %Z') timeframe_str = self._timeframe_to_string(config.TIMEFRAME) # Basic indicator signals for prompt context rsi_val = indicators.get('rsi', float('nan')) rsi_signal = "Neutral" if rsi_val > 70: rsi_signal = "Overbought" elif rsi_val < 30: rsi_signal = "Oversold" macd_hist = indicators.get('macd_hist', float('nan')) macd_cross_signal = "Neutral" if macd_hist > 0: macd_cross_signal = "Bullish Crossover/Momentum" elif macd_hist < 0: macd_cross_signal = "Bearish Crossover/Momentum" cci_val = indicators.get('cci', float('nan')) cci_signal = "Neutral" if cci_val > 100: cci_signal = "Overbought" elif cci_val < -100: cci_signal = "Oversold" stoch_k = indicators.get('stoch_k', float('nan')) stoch_d = indicators.get('stoch_d', float('nan')) stoch_signal = "Neutral" if stoch_k > 80 and stoch_d > 80: stoch_signal = "Overbought" elif stoch_k < 20 and stoch_d < 20: stoch_signal = "Oversold" elif stoch_k > stoch_d: stoch_signal = "Bullish Cross Potential" elif stoch_k < stoch_d: stoch_signal = "Bearish Cross Potential" bb_upper = indicators.get('bb_upper', float('nan')) bb_lower = indicators.get('bb_lower', float('nan')) bb_middle = indicators.get('bb_middle', float('nan')) bb_price_pos = "Inside Bands" current_price = (market_data.bid + market_data.ask) / 2 if current_price > bb_upper: bb_price_pos = "Above Upper Band" elif current_price < bb_lower: bb_price_pos = "Below Lower Band" elif current_price > bb_middle: bb_price_pos = "Above Middle Band" elif current_price < bb_middle: bb_price_pos = "Below Middle Band" if open_position: # Prompt for managing an existing position position_type = "BUY" if open_position.type == 0 else "SELL" prompt = f"""You are a financial market analyst AI. You are managing an open {position_type} position for {config.SYMBOL} entered at {open_position.price_open:.5f} with TP={open_position.tp:.5f} and SL={open_position.sl:.5f}. The position has been open for {position_age:.1f} minutes. The initial goal was to make ${config.PROFIT_TARGET_USD:.2f} profit within {config.TRADE_DURATION_MINUTES} minutes using a lot size of {config.VOLUME}. Current Market Data for {config.SYMBOL} ({timeframe_str}): Current Time: {current_time_str} Bid Price: {market_data.bid:.5f} Ask Price: {market_data.ask:.5f} Last Volume: {market_data.volume} Technical Indicators: RSI ({config.RSI_PERIOD}): {indicators.get('rsi', 'N/A'):.2f} ({rsi_signal}) MACD ({config.MACD_FAST}, {config.MACD_SLOW}, {config.MACD_SIGNAL}): MACD Line = {indicators.get('macd_line', 'N/A'):.5f}, Signal Line = {indicators.get('macd_signal', 'N/A'):.5f}, Histogram = {indicators.get('macd_hist', 'N/A'):.5f} ({macd_cross_signal}) Bollinger Bands ({config.BB_PERIOD}, {config.BB_STDDEV}): Upper = {indicators.get('bb_upper', 'N/A'):.5f}, Middle = {indicators.get('bb_middle', 'N/A'):.5f}, Lower = {indicators.get('bb_lower', 'N/A'):.5f}, Price Position: {bb_price_pos} CCI ({config.CCI_PERIOD}): {indicators.get('cci', 'N/A'):.2f} ({cci_signal}) Stochastic Oscillator ({config.STOCH_K}, {config.STOCH_D}, {config.STOCH_SMOOTH_K}): %K = {indicators.get('stoch_k', 'N/A'):.2f}, %D = {indicators.get('stoch_d', 'N/A'):.2f} ({stoch_signal}) Recent Candlestick Data (OHLC): {indicators.get('candlesticks', 'N/A')} The position is currently at a profit/loss of ${position_pnl:.2f}. Based on the current market conditions and the fact that the position has been open for over {config.TRADE_DURATION_MINUTES} minutes, should we: 1. HOLD the position with the current TP/SL? 2. CLOSE the position now (market order)? 3. MODIFY the TP to [new_tp_price] and SL to [new_sl_price]? (Provide specific prices if modifying) Provide your recommendation in the following format ONLY: Action: [HOLD/CLOSE/MODIFY] New TP: [Price or N/A] New SL: [Price or N/A] Justification: [Your reasoning for holding, closing, or modifying. Be concise.] """ else: # Prompt for a new trade decision prompt = f"""You are a financial market analyst AI. Your goal is to help achieve a profit of ${config.PROFIT_TARGET_USD:.2f} within the next {config.TRADE_DURATION_MINUTES} minutes by trading {config.SYMBOL}. The trade volume (lot size) will be {config.VOLUME}. Analyze the following real-time market data and technical indicators for {config.SYMBOL} on the {timeframe_str} timeframe: Current Time: {current_time_str} Bid Price: {market_data.bid:.5f} Ask Price: {market_data.ask:.5f} Last Volume: {market_data.volume} Technical Indicators: RSI ({config.RSI_PERIOD}): {indicators.get('rsi', 'N/A'):.2f} ({rsi_signal}) MACD ({config.MACD_FAST}, {config.MACD_SLOW}, {config.MACD_SIGNAL}): MACD Line = {indicators.get('macd_line', 'N/A'):.5f}, Signal Line = {indicators.get('macd_signal', 'N/A'):.5f}, Histogram = {indicators.get('macd_hist', 'N/A'):.5f} ({macd_cross_signal}) Bollinger Bands ({config.BB_PERIOD}, {config.BB_STDDEV}): Upper = {indicators.get('bb_upper', 'N/A'):.5f}, Middle = {indicators.get('bb_middle', 'N/A'):.5f}, Lower = {indicators.get('bb_lower', 'N/A'):.5f}, Price Position: {bb_price_pos} CCI ({config.CCI_PERIOD}): {indicators.get('cci', 'N/A'):.2f} ({cci_signal}) Stochastic Oscillator ({config.STOCH_K}, {config.STOCH_D}, {config.STOCH_SMOOTH_K}): %K = {indicators.get('stoch_k', 'N/A'):.2f}, %D = {indicators.get('stoch_d', 'N/A'):.2f} ({stoch_signal}) Recent Candlestick Data (OHLC): {indicators.get('candlesticks', 'N/A')} Market Context: No open positions for {config.SYMBOL}. Based on this data, predict the market direction for {config.SYMBOL} in the next {config.TRADE_DURATION_MINUTES} minutes. Provide a trading recommendation: BUY, SELL, or HOLD. If BUY or SELL: - Provide the entry price (use current ask for BUY {market_data.ask:.5f}, current bid for SELL {market_data.bid:.5f}). - Provide a Take Profit (TP) price aiming for the ${config.PROFIT_TARGET_USD:.2f} goal within {config.TRADE_DURATION_MINUTES} minutes. - Provide a Stop Loss (SL) price considering a reasonable risk/reward ratio. - Justify your decision, referencing the indicators and market context. If HOLD: - Explain why holding is the best action now. Format your response clearly and ONLY in this structure: Action: [BUY/SELL/HOLD] Entry: [Price or N/A - Use current market price if BUY/SELL] TP: [Price or N/A] SL: [Price or N/A] Justification: [Your reasoning. Be concise.] """ logger.debug(f"Sending prompt to Groq:\n{prompt}") try: start_time = time.time() completion = self.client.chat.completions.create( model=config.GROQ_MODEL, messages=[ { "role": "system", "content": "You are a precise financial analyst AI providing trading recommendations based on provided data. Follow the requested output format strictly." }, { "role": "user", "content": prompt } ], temperature=config.TEMPERATURE, max_tokens=config.MAX_COMPLETION_TOKENS, top_p=1, stream=False, # Get full response at once stop=None, ) end_time = time.time() response_time = end_time - start_time ai_response_text = completion.choices[0].message.content logger.info(f"Groq response received in {response_time:.2f} seconds.") logger.debug(f"Groq raw response:\n{ai_response_text}") parsed_recommendation = self._parse_recommendation(ai_response_text) if parsed_recommendation: # Add raw response for logging purposes parsed_recommendation["raw_response"] = ai_response_text logger.info(f"Parsed AI Recommendation: {parsed_recommendation}") return parsed_recommendation else: logger.error("Failed to parse AI recommendation.") return None except GroqError as e: logger.error(f"Groq API error: {e}", exc_info=True) return None except Exception as e: logger.error(f"An unexpected error occurred during Groq API call: {e}", exc_info=True) return None def _timeframe_to_string(self, timeframe): """Converts MT5 timeframe constant to a readable string.""" map = { mt5.TIMEFRAME_M1: "1 Minute", mt5.TIMEFRAME_M5: "5 Minutes", mt5.TIMEFRAME_M15: "15 Minutes", mt5.TIMEFRAME_M30: "30 Minutes", mt5.TIMEFRAME_H1: "1 Hour", mt5.TIMEFRAME_H4: "4 Hours", mt5.TIMEFRAME_D1: "Daily", mt5.TIMEFRAME_W1: "Weekly", mt5.TIMEFRAME_MN1: "Monthly", } return map.get(timeframe, "Unknown Timeframe") import time import logging from logging.handlers import RotatingFileHandler import MetaTrader5 as mt5 import config from mt5_handler import MT5Handler from groq_analyzer import GroqAnalyzer # --- Logging Setup --- log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') log_handler = RotatingFileHandler(config.LOG_FILE, maxBytes=5*1024*1024, backupCount=2) # 5MB per file, 2 backups log_handler.setFormatter(log_formatter) # Configure root logger root_logger = logging.getLogger() root_logger.setLevel(config.LOG_LEVEL) root_logger.addHandler(log_handler) # Add console handler for immediate feedback console_handler = logging.StreamHandler() console_handler.setFormatter(log_formatter) root_logger.addHandler(console_handler) logger = logging.getLogger(__name__) # Logger for this module # --- Main Application --- def run_trader(): logger.info("--- Starting MT5 Groq Trading Bot ---") logger.info(f"Symbol: {config.SYMBOL}, Timeframe: {config.TIMEFRAME}, Lot Size: {config.VOLUME}") logger.info(f"Profit Target: ${config.PROFIT_TARGET_USD}, Max Trade Duration: {config.TRADE_DURATION_MINUTES} min") logger.info(f"Groq Model: {config.GROQ_MODEL}") mt5_handler = None groq_analyzer = None try: mt5_handler = MT5Handler() groq_analyzer = GroqAnalyzer() except Exception as e: logger.critical(f"Failed to initialize handlers: {e}", exc_info=True) return # Cannot proceed if not mt5_handler.connected: logger.critical("Failed to connect to MetaTrader 5. Exiting.") return try: while True: loop_start_time = time.time() logger.info("--- New Trading Cycle ---") # Ensure MT5 connection is active if not mt5_handler.connect(): logger.error("MT5 connection lost. Attempting reconnect...") time.sleep(config.LOOP_SLEEP_SECONDS) continue # Check for open positions managed by this script open_positions = mt5_handler.get_open_positions(symbol=config.SYMBOL, magic=config.ORDER_MAGIC_NUMBER) current_position = open_positions[0] if open_positions else None # --- Position Management --- if current_position: logger.info(f"Managing open position: Ticket={current_position.ticket}, Type={'BUY' if current_position.type == 0 else 'SELL'}, Volume={current_position.volume}, Price={current_position.price_open}") # Check if TP/SL was hit (MT5 should handle this, but good to log) hit_status = mt5_handler.check_tp_sl_hit(current_position) if hit_status: logger.info(f"Position {current_position.ticket} likely closed by {hit_status} on server.") # Position should be closed by MT5, loop will detect it next cycle current_position = None # Assume closed for this cycle time.sleep(1) # Small delay before next cycle check continue # Check position age position_age_min = mt5_handler.get_position_age_minutes(current_position) logger.info(f"Position age: {position_age_min:.2f} minutes.") if position_age_min > config.TRADE_DURATION_MINUTES: logger.info(f"Position {current_position.ticket} exceeded max duration ({config.TRADE_DURATION_MINUTES} min). Asking AI for action.") # Get current data for AI analysis rates_df, tick = mt5_handler.get_market_data(config.SYMBOL, config.TIMEFRAME, config.HISTORY_BARS) if rates_df is None or tick is None: logger.warning("Could not get market data for position re-evaluation. Holding.") time.sleep(config.LOOP_SLEEP_SECONDS) continue indicators = mt5_handler.calculate_indicators(rates_df) if indicators is None: logger.warning("Could not calculate indicators for position re-evaluation. Holding.") time.sleep(config.LOOP_SLEEP_SECONDS) continue # Get AI recommendation for the open position recommendation = groq_analyzer.get_recommendation( market_data=tick, indicators=indicators, open_position=current_position, position_age=position_age_min, position_pnl=current_position.profit # Pass current PnL ) if recommendation: logger.info(f"AI recommendation for open position: {recommendation['action']}") logger.info(f"AI Justification: {recommendation.get('justification', 'N/A')}") if recommendation['action'] == "CLOSE": logger.info(f"AI recommends closing position {current_position.ticket}.") mt5_handler.close_position(current_position) # Reset position state for next cycle current_position = None elif recommendation['action'] == "MODIFY": new_tp = recommendation.get('new_tp') new_sl = recommendation.get('new_sl') logger.info(f"AI recommends modifying position {current_position.ticket}: New TP={new_tp}, New SL={new_sl}") if new_tp is not None or new_sl is not None: mt5_handler.modify_position(current_position.ticket, sl=new_sl, tp=new_tp) else: logger.warning("AI recommended MODIFY but provided no valid TP/SL values.") elif recommendation['action'] == "HOLD": logger.info(f"AI recommends holding position {current_position.ticket}.") else: logger.warning(f"Unknown AI action for open position: {recommendation['action']}") else: logger.warning("Failed to get AI recommendation for open position. Holding.") else: # Position exists but hasn't reached max duration logger.info("Position within max duration. Monitoring.") # --- New Trade Logic --- elif not current_position: # Only look for new trades if no position is open logger.info("No open position found. Looking for new trade opportunities.") # Get data and indicators rates_df, tick = mt5_handler.get_market_data(config.SYMBOL, config.TIMEFRAME, config.HISTORY_BARS) if rates_df is None or tick is None: logger.warning("Could not get market data for new trade analysis. Skipping cycle.") time.sleep(config.LOOP_SLEEP_SECONDS) continue indicators = mt5_handler.calculate_indicators(rates_df) if indicators is None: logger.warning("Could not calculate indicators for new trade analysis. Skipping cycle.") time.sleep(config.LOOP_SLEEP_SECONDS) continue # Get AI recommendation recommendation = groq_analyzer.get_recommendation(market_data=tick, indicators=indicators) if recommendation: logger.info(f"AI Recommendation: {recommendation['action']}") logger.info(f"AI Justification: {recommendation.get('justification', 'N/A')}") action = recommendation['action'] tp = recommendation.get('tp') sl = recommendation.get('sl') if action == "BUY": if tp is not None and sl is not None: logger.info(f"Executing BUY order for {config.SYMBOL} at market price. TP={tp}, SL={sl}") mt5_handler.place_order( symbol=config.SYMBOL, order_type=mt5.ORDER_TYPE_BUY, volume=config.VOLUME, tp=tp, sl=sl ) else: logger.warning("AI recommended BUY but TP or SL is missing. No order placed.") elif action == "SELL": if tp is not None and sl is not None: logger.info(f"Executing SELL order for {config.SYMBOL} at market price. TP={tp}, SL={sl}") mt5_handler.place_order( symbol=config.SYMBOL, order_type=mt5.ORDER_TYPE_SELL, volume=config.VOLUME, tp=tp, sl=sl ) else: logger.warning("AI recommended SELL but TP or SL is missing. No order placed.") elif action == "HOLD": logger.info("AI recommends HOLD. No action taken.") else: logger.warning(f"Unknown AI action: {action}") else: logger.warning("Failed to get AI recommendation. No action taken.") # --- Loop Timing --- loop_end_time = time.time() elapsed_time = loop_end_time - loop_start_time sleep_time = max(0, config.LOOP_SLEEP_SECONDS - elapsed_time) logger.info(f"Cycle finished in {elapsed_time:.2f}s. Sleeping for {sleep_time:.2f}s.") time.sleep(sleep_time) except KeyboardInterrupt: logger.info("Keyboard interrupt received. Shutting down...") except Exception as e: logger.critical(f"An unhandled exception occurred: {e}", exc_info=True) finally: if mt5_handler: mt5_handler.disconnect() logger.info("--- MT5 Groq Trading Bot Stopped ---") if __name__ == "__main__": run_trader()