# TRIFECTA_PATTERN_ENGINE_v1.00.py # # --- v1.12: The Go/No-Go Checklist Integration --- # 1. STRATEGIST AI PROMPT OVERHAUL: The AI's prompt (`get_strategist_prompt`) has # been rebuilt to be a strict "Go/No-Go Checklist" that directly mirrors the # user-defined 7-point trading plan. # 2. EXPLICIT VALIDATION: The AI is now required to sequentially answer and validate # each of the core checklist questions in its rationale before making a decision: # - 1. Session Check # - 2. Market Structure Check # - 3. S/D Zone Identification # - 4. Zone Test Confirmation # - 5. Signal Candle Confirmation # - 6. Risk-to-Reward Assessment # 3. ENHANCED CONTEXT PASSING: The `detect_trifecta_pattern` function now explicitly # passes the preceding trend direction to the AI, enabling a more accurate # assessment of the "Market Structure" checklist item. # 4. AI ROLE CLARIFICATION: The AI's task is now unambiguously to act as the final # gatekeeper for the 7-point plan. It confirms the mechanical signal and the # trade environment against a human-centric checklist before execution. This # makes its reasoning transparent and directly aligned with the strategy. # # --- v1.11: Enhanced AI Confirmation Logic --- # 1. REFINED STRATEGIST PROMPT: Upgraded to include a more nuanced "Strategist's # Confirmation Checklist" for better AI validation. # 2. AI VALIDATION OF TRIFECTA: Tasked the AI with re-validating the core components # of the detected Trifecta signal. # # --- v1.10: Trifecta Strategy & Market Session Intelligence --- # 1. CORE STRATEGY ENHANCED: Integrated the "Trifecta" pattern. # 2. TRIFECTA PATTERN ENGINE: Implemented a mechanical engine to detect the A++ setup. # 3. MARKET SESSION MANAGER: Introduced a class to provide temporal context and # define trading modes (PRIME_SESSION, CAUTIOUS_TRANSITION, LULL_AVOID). # # --- v1.00: Price Action & Market Structure Engine --- # 1. CORE STRATEGY REBUILT: Aligned with a day trading plan based on pure price # action, market structure, and Supply/Demand zones. # ... (previous version history retained) # import MetaTrader5 as mt5 import os import time import json import logging from logging.handlers import TimedRotatingFileHandler from datetime import datetime, timezone, timedelta, time as dt_time import requests from dotenv import load_dotenv import signal import sys import pytz import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from scipy.signal import argrelextrema # --- Script Version --- SCRIPT_VERSION = "TRIFECTA PATTERN ENGINE v1.00" # --- Timezone and Session Configuration --- UTC_TZ = pytz.utc LONDON_TZ = pytz.timezone('Europe/London') NY_TZ = pytz.timezone('America/New_York') SYDNEY_TZ = pytz.timezone('Australia/Sydney') TOKYO_TZ = pytz.timezone('Asia/Tokyo') # --- ======================================================= --- # --- MARKET SESSION MANAGER (v1.10) --- # --- ======================================================= --- class MarketSessionManager: def __init__(self, transition_period_minutes=30): self.transition_delta = timedelta(minutes=transition_period_minutes) # Define market hours in UTC self.sessions = { "SYDNEY": {"open": dt_time(21, 0), "close": dt_time(6, 0)}, "TOKYO": {"open": dt_time(0, 0), "close": dt_time(9, 0)}, "LONDON": {"open": dt_time(7, 0), "close": dt_time(16, 0)}, "NEW_YORK": {"open": dt_time(12, 0), "close": dt_time(21, 0)} } self.prime_sessions = ["LONDON", "NEW_YORK"] self.prime_start_utc = dt_time(7, 0) self.prime_end_utc = dt_time(17, 0) def _is_time_in_range(self, time_to_check, start, end): if start <= end: return start <= time_to_check < end else: # Handles overnight sessions like Sydney return start <= time_to_check or time_to_check < end def get_market_state(self): now_utc = datetime.now(UTC_TZ) now_utc_time = now_utc.time() active_sessions = [name for name, hours in self.sessions.items() if self._is_time_in_range(now_utc_time, hours["open"], hours["close"])] # Determine priority if "NEW_YORK" in active_sessions and "LONDON" in active_sessions: market_priority = "LONDON_NY_OVERLAP" elif "NEW_YORK" in active_sessions: market_priority = "NY_SESSION" elif "LONDON" in active_sessions: market_priority = "LONDON_SESSION" elif "TOKYO" in active_sessions and "SYDNEY" in active_sessions: market_priority = "SYDNEY_TOKYO_OVERLAP" elif "TOKYO" in active_sessions: market_priority = "TOKYO_SESSION" elif "SYDNEY" in active_sessions: market_priority = "SYDNEY_SESSION" else: market_priority = "INTER-SESSION_LULL" # Determine trading mode is_prime_time = self.prime_start_utc <= now_utc_time < self.prime_end_utc is_in_transition = False for session_name, hours in self.sessions.items(): if session_name in self.prime_sessions: open_time = datetime.combine(now_utc.date(), hours["open"]) close_time = datetime.combine(now_utc.date(), hours["close"]) # Handle overnight by adjusting date if hours["open"] > hours["close"]: if now_utc_time < hours["close"]: # We are in the morning part of the session open_time -= timedelta(days=1) else: # We are in the evening part close_time += timedelta(days=1) open_trans_start = (open_time - self.transition_delta).time() open_trans_end = (open_time + self.transition_delta).time() close_trans_start = (close_time - self.transition_delta).time() close_trans_end = (close_time + self.transition_delta).time() if self._is_time_in_range(now_utc_time, open_trans_start, open_trans_end) or \ self._is_time_in_range(now_utc_time, close_trans_start, close_trans_end): is_in_transition = True break if not is_prime_time: trading_mode = "LULL_AVOID" elif is_in_transition: trading_mode = "CAUTIOUS_TRANSITION" else: trading_mode = "PRIME_SESSION" return market_priority, trading_mode # --- Trading Configuration --- class TradingState: def __init__(self, name, min_confidence=0.75, risk_percent=1.0): self.name = name self.min_confidence = min_confidence self.risk_percent = risk_percent PRIME_SESSION_STATE = TradingState("PRIME_SESSION", 0.75, 1.0) CAUTIOUS_TRANSITION_STATE = TradingState("CAUTIOUS_TRANSITION", 0.90, 0.5) # Higher confidence, lower risk LULL_AVOID_STATE = TradingState("LULL_AVOID", 1.01, 0.0) # Risk set to 0 to prevent trading # --- Constants & Configs --- MINIMUM_SL_POINTS = 5.0 MINIMUM_RR_RATIO = 2.0 PIVOT_LOOKBACK = 5 ANALYSIS_BAR_COUNT = 250 TRIFECTA_VOL_LOOKBACK = 10 # Lookback for Trifecta volume average (10 M15 candles = 2.5 hours) TRIFECTA_VOL_SPIKE_MULT = 1.5 # Volume must be 1.5x the average TRIFECTA_RANGE_MIN = 6.5 # Minimum range for a valid Trifecta candle TRIFECTA_BODY_RATIO = 0.25 # Body must be at least 25% of the range # --- Global State & Logging --- g_last_known_trade_id = None TRADE_LOG_FILE = "trade_log.jsonl" ACTIVE_TRADE_CONTEXT_FILE = "active_trade_context.json" # --- Logging Function --- def setup_logging(log_prefix): log_file = f"{log_prefix}_GATE.log" logger = logging.getLogger() logger.setLevel(logging.INFO) if logger.hasHandlers(): logger.handlers.clear() formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) file_handler = TimedRotatingFileHandler(log_file, when="midnight", interval=1, backupCount=14, encoding='utf-8') file_handler.setFormatter(formatter) logger.addHandler(file_handler) # --- Risk Management --- def calculate_risk_adjusted_lot_size(symbol, sl_points, account_equity, risk_percent=1.0): if sl_points <= 0 or account_equity <= 0 or risk_percent == 0: logging.error("Cannot calculate lot size with zero or negative SL, equity, or risk.") return 0.0 symbol_info = mt5.symbol_info(symbol) if not symbol_info: logging.error(f"Could not get symbol info for {symbol}") return 0.0 user_max_lots = float(os.getenv("GATE_MAX_LOTS", 1.0)) value_per_point_per_lot = symbol_info.trade_tick_value / symbol_info.trade_tick_size dollar_risk = account_equity * (risk_percent / 100.0) sl_dollar_value_per_lot = sl_points * value_per_point_per_lot if sl_dollar_value_per_lot <= 1e-9: # Avoid division by zero logging.error("Calculated SL dollar value is zero. Cannot divide.") return 0.0 lot_size = dollar_risk / sl_dollar_value_per_lot min_volume = symbol_info.volume_min effective_max_volume = min(symbol_info.volume_max, user_max_lots) volume_step = symbol_info.volume_step lot_size = max(min_volume, min(lot_size, effective_max_volume)) lot_size = round(lot_size / volume_step) * volume_step logging.info(f"Risk-Adjusted Lot Size: Risking {risk_percent}% of ${account_equity:,.2f} = ${dollar_risk:,.2f}. " f"SL of {sl_points:.2f} points requires lot size of {lot_size:.2f} (Ceiling: {effective_max_volume} lots).") return round(lot_size, 2) # --- ======================================================= --- # --- MARKET STRUCTURE ENGINE (v1.00) --- # --- ======================================================= --- def find_swing_pivots(rates_df: pd.DataFrame, lookback: int) -> pd.DataFrame: """Identifies swing highs and lows from OHLC data in a sandboxed calculation.""" high_indices = argrelextrema(rates_df['high'].values, np.greater_equal, order=lookback)[0] low_indices = argrelextrema(rates_df['low'].values, np.less_equal, order=lookback)[0] pivots = [] for i in high_indices: pivots.append({'index': i, 'type': 'high', 'price': rates_df['high'].iloc[i]}) for i in low_indices: pivots.append({'index': i, 'type': 'low', 'price': rates_df['low'].iloc[i]}) if not pivots: return pd.DataFrame() pivots_df = pd.DataFrame(pivots).sort_values(by='index').reset_index(drop=True) # Filter out consecutive pivots of the same type to get a clean alternating sequence return pivots_df.loc[pivots_df['type'] != pivots_df['type'].shift()] def analyze_trend_and_reversals(pivots_df: pd.DataFrame) -> Dict: """ Analyzes a series of pivots to determine market structure (Trend, HH/HL, CHoCH). This is a sandboxed calculation. """ analysis = { "trend": "RANGING", "last_high": None, "prev_high": None, "last_low": None, "prev_low": None, "change_of_character": None # Will be 'BULLISH' or 'BEARISH' } highs = pivots_df[pivots_df['type'] == 'high'] lows = pivots_df[pivots_df['type'] == 'low'] if len(highs) >= 2: analysis["last_high"] = highs.iloc[-1] analysis["prev_high"] = highs.iloc[-2] if len(lows) >= 2: analysis["last_low"] = lows.iloc[-1] analysis["prev_low"] = lows.iloc[-2] lh, ph = analysis.get("last_high"), analysis.get("prev_high") ll, pl = analysis.get("last_low"), analysis.get("prev_low") if not all([lh, ph, ll, pl]): return analysis # Not enough pivots for a full analysis # Trend Identification is_hh = lh['price'] > ph['price'] is_hl = ll['price'] > pl['price'] is_ll = ll['price'] < pl['price'] is_lh = lh['price'] < ph['price'] if is_hh and is_hl: analysis['trend'] = "UPTREND" elif is_ll and is_lh: analysis['trend'] = "DOWNTREND" # Change of Character (CHoCH) Detection # Bullish CHoCH: Market was in a downtrend, but then broke the last Lower High. if analysis['trend'] != "DOWNTREND" and not (is_ll and is_lh): if ph and pl and lh and ll: if ph.name > pl.name and lh.name > ll.name and lh['price'] > ph['price']: analysis['change_of_character'] = "BULLISH" # Bearish CHoCH: Market was in an uptrend, but then broke the last Higher Low. if analysis['trend'] != "UPTREND" and not (is_hh and is_hl): if ph and pl and lh and ll: if pl.name > ph.name and ll.name > lh.name and ll['price'] < pl['price']: analysis['change_of_character'] = "BEARISH" return analysis def find_supply_demand_zones(rates_df: pd.DataFrame, structure_analysis: Dict) -> Dict: """ Finds the most recent Supply and Demand zones based on the last impulse move. This is a sandboxed calculation. """ zones = {"demand": None, "supply": None} # Find Demand Zone (origin of the move that created the last high) last_high = structure_analysis.get("last_high") if last_high is not None: try: # The impulse move starts from the low *before* the last high impulse_start_pivot = structure_analysis.get("prev_low") if structure_analysis.get("trend") == "UPTREND" else structure_analysis.get("last_low") if impulse_start_pivot is not None and impulse_start_pivot.name < last_high.name: search_df = rates_df.iloc[impulse_start_pivot['index']:last_high['index']] # The demand zone is the last bearish candle before the strong up-move bearish_candles = search_df[search_df['close'] < search_df['open']] if not bearish_candles.empty: origin_candle = bearish_candles.iloc[-1] zones['demand'] = {'top': origin_candle['high'], 'bottom': origin_candle['low']} except Exception as e: logging.warning(f"Could not calculate demand zone: {e}") # Find Supply Zone (origin of the move that created the last low) last_low = structure_analysis.get("last_low") if last_low is not None: try: # The impulse move starts from the high *before* the last low impulse_start_pivot = structure_analysis.get("prev_high") if structure_analysis.get("trend") == "DOWNTREND" else structure_analysis.get("last_high") if impulse_start_pivot is not None and impulse_start_pivot.name < last_low.name: search_df = rates_df.iloc[impulse_start_pivot['index']:last_low['index']] # The supply zone is the last bullish candle before the strong down-move bullish_candles = search_df[search_df['close'] > search_df['open']] if not bullish_candles.empty: origin_candle = bullish_candles.iloc[-1] zones['supply'] = {'top': origin_candle['high'], 'bottom': origin_candle['low']} except Exception as e: logging.warning(f"Could not calculate supply zone: {e}") return zones def get_market_structure_analysis(symbol: str) -> Optional[Dict]: """Main function to get the complete market structure analysis on the M5 chart.""" rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M5, 0, ANALYSIS_BAR_COUNT) if rates is None or len(rates) < 50: logging.warning("Insufficient M5 data for structure analysis.") return None rates_df = pd.DataFrame(rates) rates_df['index'] = rates_df.index pivots_df = find_swing_pivots(rates_df, lookback=PIVOT_LOOKBACK) if pivots_df.empty or len(pivots_df) < 4: logging.info("Not enough clear pivots found, market likely ranging or in strong impulse.") return {"status": "RANGING", "trend": "RANGING"} structure = analyze_trend_and_reversals(pivots_df) zones = find_supply_demand_zones(rates_df, structure) current_price = rates_df.iloc[-1]['close'] # Format for easy consumption by the AI final_context = { "status": "ANALYSIS_COMPLETE", "current_price": round(current_price, 2), "trend": structure.get('trend'), "change_of_character": structure.get('change_of_character'), "last_swing_high": round(structure['last_high']['price'], 2) if structure.get('last_high') is not None else None, "last_swing_low": round(structure['last_low']['price'], 2) if structure.get('last_low') is not None else None, "active_demand_zone": zones.get('demand'), "active_supply_zone": zones.get('supply') } return final_context # --- ======================================================= --- # --- TRIFECTA PATTERN ENGINE (v1.10) --- # --- ======================================================= --- def aggregate_to_m15(m5_df: pd.DataFrame) -> pd.DataFrame: """Aggregates an M5 DataFrame to M15.""" if m5_df.empty: return pd.DataFrame() m5_df['time'] = pd.to_datetime(m5_df['time'], unit='s') m5_df.set_index('time', inplace=True) m15_df = m5_df['open'].resample('15T', label='right').first().to_frame() m15_df['high'] = m5_df['high'].resample('15T', label='right').max() m15_df['low'] = m5_df['low'].resample('15T', label='right').min() m15_df['close'] = m5_df['close'].resample('15T', label='right').last() m15_df['tick_volume'] = m5_df['tick_volume'].resample('15T', label='right').sum() m15_df.dropna(inplace=True) return m15_df.reset_index() def detect_trifecta_pattern(symbol: str) -> Dict: """Detects the full 'Trifecta' reversal pattern on the M15 chart.""" # Fetch enough M5 bars to form at least 15 M15 candles for context num_m5_bars = (TRIFECTA_VOL_LOOKBACK + 5) * 3 rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M5, 0, num_m5_bars) if rates is None or len(rates) < num_m5_bars: return {'signal': 'HOLD', 'reason': 'Insufficient M5 data for aggregation.'} m5_df = pd.DataFrame(rates) m15_df = aggregate_to_m15(m5_df) if len(m15_df) < (TRIFECTA_VOL_LOOKBACK + 3): return {'signal': 'HOLD', 'reason': 'Insufficient M15 data after aggregation.'} # Define our candles of interest trigger_candle = m15_df.iloc[-1] prev1_candle = m15_df.iloc[-2] prev2_candle = m15_df.iloc[-3] # --- Check for a Bullish Reversal (Failed Breakdown) --- is_prev_trend_bearish = (prev1_candle['close'] <= prev1_candle['open']) and (prev2_candle['close'] <= prev2_candle['open']) if is_prev_trend_bearish: # 1. Primary Signal: Failed Breakdown m5_start_time = pd.to_datetime(trigger_candle['time'], unit='s') - timedelta(minutes=15) m5_trigger_bars = m5_df.loc[m5_start_time : pd.to_datetime(trigger_candle['time'], unit='s')] if len(m5_trigger_bars) < 3: return {'signal': 'HOLD', 'reason': 'Incomplete M15 candle data.'} m5_1, m5_2 = m5_trigger_bars.iloc[0], m5_trigger_bars.iloc[1] low_in_first_10_mins = trigger_candle['low'] == m5_1['low'] or trigger_candle['low'] == m5_2['low'] candle_range = trigger_candle['high'] - trigger_candle['low'] close_in_top_third = trigger_candle['close'] > (trigger_candle['high'] - 0.33 * candle_range) if low_in_first_10_mins and close_in_top_third: # 2. Quality Filters if candle_range < TRIFECTA_RANGE_MIN: return {'signal': 'HOLD', 'reason': 'Pattern VETO: Low volatility candle.'} body_size = abs(trigger_candle['close'] - trigger_candle['open']) if body_size < (TRIFECTA_BODY_RATIO * candle_range): return {'signal': 'HOLD', 'reason': 'Pattern VETO: Indecision candle body.'} # 3. Volume Confirmation avg_vol = m15_df.iloc[-(TRIFECTA_VOL_LOOKBACK+1):-1]['tick_volume'].mean() vol_spike = trigger_candle['tick_volume'] > (avg_vol * TRIFECTA_VOL_SPIKE_MULT) # 4. Structure Confirmation one_hour_low = m15_df.iloc[-5:-1]['low'].min() structure_test = trigger_candle['low'] <= one_hour_low if vol_spike and structure_test: logging.info(f"✅ BULLISH TRIFECTA DETECTED: Vol Spike ({trigger_candle['tick_volume']:.0f} > {avg_vol*1.5:.0f}) AND Structure Test ({trigger_candle['low']:.2f} <= {one_hour_low:.2f})") return {'signal': 'BUY', 'trigger_candle': trigger_candle.to_dict(), 'context_trend': 'Bearish'} # --- Check for a Bearish Reversal (Failed Breakout) --- is_prev_trend_bullish = (prev1_candle['close'] > prev1_candle['open']) and (prev2_candle['close'] > prev2_candle['open']) if is_prev_trend_bullish: # 1. Primary Signal: Failed Breakout m5_start_time = pd.to_datetime(trigger_candle['time'], unit='s') - timedelta(minutes=15) m5_trigger_bars = m5_df.loc[m5_start_time : pd.to_datetime(trigger_candle['time'], unit='s')] if len(m5_trigger_bars) < 3: return {'signal': 'HOLD', 'reason': 'Incomplete M15 candle data.'} m5_1, m5_2 = m5_trigger_bars.iloc[0], m5_trigger_bars.iloc[1] high_in_first_10_mins = trigger_candle['high'] == m5_1['high'] or trigger_candle['high'] == m5_2['high'] candle_range = trigger_candle['high'] - trigger_candle['low'] close_in_bottom_third = trigger_candle['close'] < (trigger_candle['low'] + 0.33 * candle_range) if high_in_first_10_mins and close_in_bottom_third: # 2. Quality Filters if candle_range < TRIFECTA_RANGE_MIN: return {'signal': 'HOLD', 'reason': 'Pattern VETO: Low volatility candle.'} body_size = abs(trigger_candle['close'] - trigger_candle['open']) if body_size < (TRIFECTA_BODY_RATIO * candle_range): return {'signal': 'HOLD', 'reason': 'Pattern VETO: Indecision candle body.'} # 3. Volume Confirmation avg_vol = m15_df.iloc[-(TRIFECTA_VOL_LOOKBACK+1):-1]['tick_volume'].mean() vol_spike = trigger_candle['tick_volume'] > (avg_vol * TRIFECTA_VOL_SPIKE_MULT) # 4. Structure Confirmation one_hour_high = m15_df.iloc[-5:-1]['high'].max() structure_test = trigger_candle['high'] >= one_hour_high if vol_spike and structure_test: logging.info(f"✅ BEARISH TRIFECTA DETECTED: Vol Spike ({trigger_candle['tick_volume']:.0f} > {avg_vol*1.5:.0f}) AND Structure Test ({trigger_candle['high']:.2f} >= {one_hour_high:.2f})") return {'signal': 'SELL', 'trigger_candle': trigger_candle.to_dict(), 'context_trend': 'Bullish'} return {'signal': 'HOLD', 'reason': 'No pattern met.'} # --- State Management & Context --- def get_trading_state(mode): if mode == "PRIME_SESSION": return PRIME_SESSION_STATE if mode == "CAUTIOUS_TRANSITION": return CAUTIOUS_TRANSITION_STATE return LULL_AVOID_STATE def save_trade_context(trade_id, context_data): """Saves context for an active trade.""" try: with open(ACTIVE_TRADE_CONTEXT_FILE, 'w') as f: json.dump({str(trade_id): context_data}, f) except IOError as e: logging.error(f"Could not save trade context for {trade_id}: {e}") def load_trade_context(trade_id): """Loads context for an active trade.""" if not os.path.exists(ACTIVE_TRADE_CONTEXT_FILE): return None try: with open(ACTIVE_TRADE_CONTEXT_FILE, 'r') as f: all_contexts = json.load(f) return all_contexts.get(str(trade_id)) except (IOError, json.JSONDecodeError): return None def clear_trade_context(): """Clears the active trade context file.""" if os.path.exists(ACTIVE_TRADE_CONTEXT_FILE): try: os.remove(ACTIVE_TRADE_CONTEXT_FILE) logging.info("Active trade context file cleared.") except OSError as e: logging.error(f"Error clearing trade context file: {e}") # --- AI Interaction Layer --- class PromptManager: def get_strategist_prompt(self, market_structure_context: Dict, market_priority: str, trifecta_signal: Dict) -> str: context_str = json.dumps(market_structure_context, indent=2) signal_str = json.dumps(trifecta_signal, indent=2, default=str) proposed_action = trifecta_signal['signal'] return ( f"You are a Senior Trading Strategist AI. Your role is to provide the final Go/No-Go decision for a high-probability trade setup detected by a mechanical pattern engine.\n\n" f"--- MECHANICALLY DETECTED SIGNAL: 'THE TRIFECTA' REVERSAL ---\n" f"The system has detected a high-probability {proposed_action} signal. Details:\n" f"{signal_str}\n\n" f"--- BROADER MARKET CONTEXT (M5 CHART) ---\n" f"Market Priority: {market_priority}\n" f"Analysis UTC Timestamp: {datetime.now(timezone.utc).isoformat()}\n" f"{context_str}\n\n" f"--- YOUR TASK: THE FINAL GO/NO-GO CHECKLIST ---\n" f"You must methodically answer these questions to form your final rationale. A 'NO' to any question from 1 to 6 should result in a 'HOLD' decision.\n" f"1. **Session Check:** The system has confirmed we are in a PRIME_SESSION or CAUTIOUS_TRANSITION. Is this correct? (Answer YES/NO)\n" f"2. **Market Structure Check:** The mechanical signal is a {proposed_action} against a preceding {trifecta_signal['context_trend']} trend. Does the broader M5 context (trend, CHoCH) support this reversal, or does it present an immediate conflict? (Answer YES/NO, and briefly explain)\n" f"3. **Zone Check & Test:** The signal candle tested a key 1-hour high/low. Does this level appear significant in the broader context (e.g., aligning with a clear swing point or S/D zone)? (Answer YES/NO)\n" f"4. **Confirmation Candle Check:** Review the signal candle's data. Is it a decisive, high-conviction reversal candle with a strong close and notable volume? (Answer YES/NO)\n" f"5. **Risk/Reward Check:** For a {proposed_action}, the Stop Loss would be below/above the signal candle's wick, and the Take Profit would target the last major swing high/low. Does this scenario appear to offer at least a 1:{MINIMUM_RR_RATIO} R:R ratio? (Answer YES/NO)\n" f"6. **Final Veto Check:** Are there any other overriding factors (e.g., imminent high-impact news, extreme market volatility) that make this trade too risky despite the strong technical signal? (Answer NO if clear)\n" f"7. **Final Decision:** If all checks from 1-6 are satisfactory, CONFIRM the trade. Otherwise, VETO by recommending 'HOLD'.\n\n" f"--- CRITICAL OUTPUT FORMAT ---\n" f"Your *entire response* MUST be a single, valid JSON object.\n" f'**Schema:**\n{{"action": "BUY|SELL|HOLD", "confidence": <0.0-1.0 float>, "rationale": "Your concise, point-by-point rationale based on the 7-point checklist."}}' ) class GeminiStrategist: def __init__(self, api_key): self.api_key = api_key self.headers = {"Content-Type": "application/json"} self.model = "gemini-1.5-flash-latest" # Using a capable model for nuanced decisions self.api_url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}" def get_decision(self, prompt): default_response = {"action": "HOLD", "confidence": 0.0, "rationale": "API Error or failed to parse response."} payload = {"contents": [{"parts": [{"text": prompt}]}]} text_response = "" try: response = requests.post(self.api_url, headers=self.headers, json=payload, timeout=90) response.raise_for_status() text_response = response.json()["candidates"][0]["content"]["parts"][0]["text"] start_index = text_response.find('{') end_index = text_response.rfind('}') if start_index != -1 and end_index != -1 and end_index > start_index: json_str = text_response[start_index:end_index+1] return json.loads(json_str) else: raise json.JSONDecodeError("Could not find valid JSON object in response.", text_response, 0) except json.JSONDecodeError as e: logging.error(f"Could not parse JSON from Gemini. Error: {e}. Full response:\n{text_response}") except requests.exceptions.HTTPError as http_err: logging.error(f"Gemini API call failed with HTTP Error: {http_err}", exc_info=True) except Exception as e: logging.error(f"Gemini API request failed unexpectedly: {e}", exc_info=True) return default_response # --- Trade Execution --- def get_active_trade(symbol, magic_number): positions = mt5.positions_get(symbol=symbol, magic=magic_number) return positions[0] if positions else None def open_trade(symbol, direction, lots, magic_number, sl_price, tp_price, context_to_save={}): tick = mt5.symbol_info_tick(symbol) if not tick: logging.error("Failed to get tick for symbol.") return None price = tick.ask if direction == "BUY" else tick.bid digits = mt5.symbol_info(symbol).digits comment = f"{SCRIPT_VERSION}"[:31] request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": float(lots), "type": mt5.ORDER_TYPE_BUY if direction == "BUY" else mt5.ORDER_TYPE_SELL, "price": price, "sl": round(sl_price, digits), "tp": round(tp_price, digits), "magic": magic_number, "comment": comment } logging.info(f"Attempting to open {direction} {lots} lots of {symbol} @ {price:.{digits}f}") result = mt5.order_send(request) if result and result.retcode == mt5.TRADE_RETCODE_DONE: logging.info(f"SUCCESS: Order {result.order} sent. Finding position ID...") time.sleep(1) # Allow time for the position to appear positions = mt5.positions_get(symbol=symbol, magic=magic_number) if positions: new_position = max(positions, key=lambda p: p.ticket) logging.info(f"✅ Position opened with ID: #{new_position.ticket}.") if context_to_save: save_trade_context(new_position.ticket, context_to_save) return new_position.ticket else: logging.error("Order executed but could not find the resulting position.") else: logging.error(f"FAILED: Order send failed. Retcode: {result.retcode if result else 'N/A'}. Error: {mt5.last_error()}") return None # --- Main Analysis & Execution Cycle --- def run_analysis_cycle(strategist, prompter, symbol, magic, current_trading_state, market_priority): global g_last_known_trade_id logging.info("--- Running Analysis Cycle ---") active_trade = get_active_trade(symbol, magic) if active_trade: logging.info(f"Holding active trade #{active_trade.ticket}. Skipping new trade analysis.") return # 1. Detect high-probability pattern first trifecta_signal = detect_trifecta_pattern(symbol) if trifecta_signal['signal'] == "HOLD": logging.info(f"No Trifecta signal detected. Reason: {trifecta_signal['reason']}") return logging.info(f"High-Probability '{trifecta_signal['signal']}' Trifecta signal detected. Proceeding to AI confirmation.") # 2. Get broader market structure context market_structure_context = get_market_structure_analysis(symbol) if not market_structure_context or market_structure_context['status'] != 'ANALYSIS_COMPLETE': logging.warning("Analysis skipped: Market structure context could not be determined for AI confirmation.") return # 3. Get AI Decision (Confirmation/Veto) strategist_prompt = prompter.get_strategist_prompt(market_structure_context, market_priority, trifecta_signal) decision = strategist.get_decision(strategist_prompt) logging.info(f"Strategist AI Decision: {decision.get('action')} (Conf: {decision.get('confidence')}). Rationale: {decision.get('rationale')}") action = decision.get("action", "HOLD").upper() confidence = decision.get('confidence', 0.0) # 4. Execute Trade if all conditions are met if action == trifecta_signal['signal'] and confidence >= current_trading_state.min_confidence: # Calculate SL/TP from the trigger candle and structural context sl_price, tp_price = 0.0, 0.0 trigger_candle = trifecta_signal['trigger_candle'] if action == "BUY": last_high = market_structure_context.get("last_swing_high") if not last_high: logging.warning("VETO: Cannot execute BUY. Missing Last High for TP calculation.") return sl_price = trigger_candle['low'] - (MINIMUM_SL_POINTS / 10) tp_price = last_high elif action == "SELL": last_low = market_structure_context.get("last_swing_low") if not last_low: logging.warning("VETO: Cannot execute SELL. Missing Last Low for TP calculation.") return sl_price = trigger_candle['high'] + (MINIMUM_SL_POINTS / 10) tp_price = last_low # Final Risk Check (This is the code's implementation of checklist item #6) current_price = market_structure_context['current_price'] sl_points = abs(current_price - sl_price) tp_points = abs(tp_price - current_price) if sl_points < MINIMUM_SL_POINTS or (tp_points / sl_points if sl_points > 0 else 0) < MINIMUM_RR_RATIO: logging.warning(f"VETO: Trade does not meet risk criteria. SL Points: {sl_points:.2f}, R:R: {(tp_points / sl_points if sl_points > 0 else 0):.2f}") return # Final Position Sizing Check (Checklist item #7) account_info = mt5.account_info() if not account_info: logging.error("VETO: Could not retrieve account info for lot size calculation.") return lot_size = calculate_risk_adjusted_lot_size(symbol, sl_points, account_info.equity, current_trading_state.risk_percent) if lot_size > 0: context_to_save = { "entry_strategy": f"Trifecta_{action}", "market_context": market_structure_context, "ai_rationale": decision.get('rationale') } trade_id = open_trade(symbol, action, lot_size, magic, sl_price, tp_price, context_to_save) if trade_id: g_last_known_trade_id = trade_id else: logging.error("Trade execution halted due to zero lot size calculation.") def log_closed_trade(trade_id, symbol): """Logs the summary of a closed trade.""" global g_last_known_trade_id try: history_deals = mt5.history_deals_get(position=trade_id) if not history_deals: logging.warning(f"Could not find deal history for closed position #{trade_id}.") return entry_deal = next((d for d in history_deals if d.entry == mt5.DEAL_ENTRY_IN), None) if not entry_deal: logging.warning(f"Could not find entry deal for position #{trade_id}.") return total_profit = sum(d.profit for d in history_deals if d.position_id == trade_id) exit_deal = history_deals[-1] details = { "trade_id": int(trade_id), "symbol": symbol, "direction": "BUY" if entry_deal.type == mt5.ORDER_TYPE_BUY else "SELL", "profit_loss": round(total_profit, 2), "outcome": "WIN" if total_profit > 0 else "LOSS", "entry_time": datetime.fromtimestamp(entry_deal.time, tz=UTC_TZ).isoformat(), "exit_time": datetime.fromtimestamp(exit_deal.time, tz=UTC_TZ).isoformat(), "exit_reason": exit_deal.comment } with open(TRADE_LOG_FILE, "a") as f: f.write(json.dumps(details) + "\n") logging.info(f"Trade Closed: ID #{details['trade_id']} | Outcome: {details['outcome']} | P/L: ${details['profit_loss']:.2f} | Reason: {details['exit_reason']}") except Exception as e: logging.error(f"Failed during closed trade logging for #{trade_id}: {e}", exc_info=True) finally: g_last_known_trade_id = None clear_trade_context() # --- Main Entry Point --- def main(): global g_last_known_trade_id load_dotenv() # Environment Variable Validation required_vars = ["GATE_SYMBOL", "GEMINI_API_KEY", "MT5_TERMINAL_PATH", "MT5_LOGIN", "MT5_PASSWORD", "MT5_BROKER", "GATE_MAGIC"] if any(not os.getenv(var) for var in required_vars): logging.basicConfig() logging.critical(f"FATAL ERROR: Missing one or more required environment variables: {', '.join(required_vars)}") sys.exit(1) SYMBOL = os.getenv("GATE_SYMBOL") GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") TERMINAL_PATH = os.getenv("MT5_TERMINAL_PATH") LOGIN = int(os.getenv("MT5_LOGIN")) PASSWORD = os.getenv("MT5_PASSWORD") BROKER = os.getenv("MT5_BROKER") MAGIC = int(os.getenv("GATE_MAGIC")) setup_logging(SYMBOL) if not mt5.initialize(path=TERMINAL_PATH, timeout=60000) or not mt5.login(login=LOGIN, password=PASSWORD, server=BROKER): logging.error(f"MT5 Connection Failed: {mt5.last_error()}"); sys.exit(1) logging.info(f"--- Starting {SCRIPT_VERSION} for {SYMBOL} ---") strategist = GeminiStrategist(GEMINI_API_KEY) prompter = PromptManager() session_manager = MarketSessionManager() active_trade = get_active_trade(SYMBOL, MAGIC) g_last_known_trade_id = active_trade.ticket if active_trade else None logging.info(f"Initial check: {'Active trade #' + str(g_last_known_trade_id) if g_last_known_trade_id else 'No active trade found'}.") def shutdown_handler(signum, frame): logging.info("Shutdown signal received. Exiting gracefully..."); mt5.shutdown(); sys.exit(0) signal.signal(signal.SIGINT, shutdown_handler) signal.signal(signal.SIGTERM, shutdown_handler) last_bar_time = 0 while True: try: # Check for closed trades if g_last_known_trade_id and not get_active_trade(SYMBOL, MAGIC): log_closed_trade(g_last_known_trade_id, SYMBOL) new_bar = mt5.copy_rates_from_pos(SYMBOL, mt5.TIMEFRAME_M5, 0, 1) if not new_bar: time.sleep(10) continue current_bar_time = new_bar[0]['time'] if current_bar_time != last_bar_time: last_bar_time = current_bar_time market_priority, trading_mode = session_manager.get_market_state() current_trading_state = get_trading_state(trading_mode) logging.info(f"--- New M5 Bar [{datetime.fromtimestamp(last_bar_time, tz=UTC_TZ).strftime('%Y-%m-%d %H:%M:%S UTC')}] ---") logging.info(f"Market Priority: {market_priority} | Trading Mode: {trading_mode}") # Only run analysis if we are in a tradable mode if current_trading_state.risk_percent > 0: # Check if it's the start of a new M15 candle to run the Trifecta check if datetime.fromtimestamp(current_bar_time, tz=UTC_TZ).minute % 15 == 0: run_analysis_cycle(strategist, prompter, SYMBOL, MAGIC, current_trading_state, market_priority) else: logging.info("Trading mode is LULL_AVOID. Analysis cycle skipped.") time.sleep(5) except Exception as e: logging.error(f"Main loop error: {e}", exc_info=True) time.sleep(30) if __name__ == "__main__": main()