import MetaTrader5 as mt5 import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from pathlib import Path import mplfinance as mpf from datetime import datetime, timedelta from sklearn.preprocessing import MinMaxScaler from scipy import stats import plotly.graph_objects as go from plotly.subplots import make_subplots # Path to MetaTrader 5 terminal terminal_path = "C:/Program Files/RannForex MetaTrader 5/terminal64.exe" def normalize_price(price, symbol): symbol_info = mt5.symbol_info(symbol) digits = symbol_info.digits point = symbol_info.point price = round(price, digits) price = round(price / point) * point return price def analyze_pair(symbol, timeframe=mt5.TIMEFRAME_M15, bars=1000): df = pd.DataFrame(mt5.copy_rates_range( symbol, timeframe, datetime.now() - timedelta(days=30), datetime.now() )) if df is None or len(df) < bars: return None df['time'] = pd.to_datetime(df['time'], unit='s') df['trend_move'] = 0 current_trend = 0 trend_moves = [] reversal_moves = [] for i in range(1, len(df)): if df.iloc[i]['close'] > df.iloc[i-1]['close']: if current_trend <= 0: if current_trend < 0: trend_moves.append(abs(current_trend)) current_trend = 0 current_trend += 1 else: if current_trend >= 0: if current_trend > 0: trend_moves.append(current_trend) current_trend = 0 current_trend -= 1 df.loc[df.index[i], 'trend_move'] = current_trend avg_trend_move = np.mean(trend_moves) avg_reversal_size = df['high'].rolling(20).max() - df['low'].rolling(20).min() avg_reversal = avg_reversal_size.mean() df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3 df['price_return'] = df['typical_price'].pct_change() df['volatility'] = df['price_return'].rolling(20).std() df['direction'] = np.sign(df['close'] - df['open']) df['color_intensity'] = df['volatility'] * (df['tick_volume'] / df['tick_volume'].mean()) df['is_yellow'] = df['color_intensity'] > df['color_intensity'].quantile(0.75) yellow_reversals = 0 true_reversals = 0 for i in range(20, len(df)-20): if abs(df.iloc[i]['trend_move']) >= avg_trend_move*0.7: if df.iloc[i]['is_yellow']: future_move = df.iloc[i:i+20]['close'].max() - df.iloc[i:i+20]['close'].min() if future_move >= avg_reversal*0.5: yellow_reversals += 1 true_reversals += 1 reversal_probability = yellow_reversals / true_reversals if true_reversals > 0 else 0 recent_data = df.iloc[-bars:] prev_prices = df['close'].rolling(window=20).mean() clusters = [] current_cluster = {'size': 0, 'start': None} for i in range(len(recent_data)): if recent_data.iloc[i]['is_yellow']: if current_cluster['start'] is None: current_cluster['start'] = i current_cluster['size'] += 1 else: if current_cluster['size'] >= 5: current_cluster['end_price'] = recent_data.iloc[i-1]['close'] clusters.append(current_cluster.copy()) current_cluster = {'size': 0, 'start': None} if not clusters: return None biggest_cluster = max(clusters, key=lambda x: x['size']) cluster_start = biggest_cluster['start'] cluster_size = biggest_cluster['size'] cluster_data = recent_data.iloc[cluster_start:cluster_start+cluster_size] cluster_high = cluster_data['high'].max() cluster_low = cluster_data['low'].min() cluster_close = cluster_data.iloc[-1]['close'] current_trend_size = abs(recent_data.iloc[-1]['trend_move']) trend_completion = current_trend_size / avg_trend_move reversal_chance = min(0.95, trend_completion * reversal_probability) avg_volatility = recent_data['volatility'].mean() optimal_stop = avg_volatility * 2 optimal_take = optimal_stop * 3 # Minimum 1:2 symbol_info = mt5.symbol_info(symbol) point = symbol_info.point is_buy = cluster_close <= prev_prices.iloc[-1] if is_buy: entry = normalize_price(cluster_high, symbol) raw_stop = cluster_low - optimal_stop raw_take = entry + optimal_take stop = normalize_price(min(raw_stop, entry - point * 10), symbol) take = normalize_price(max(raw_take, entry + point * 20), symbol) if (take - entry) < (entry - stop) * 2: take = entry + (entry - stop) * 2 take = normalize_price(take, symbol) else: entry = normalize_price(cluster_low, symbol) raw_stop = cluster_high + optimal_stop raw_take = entry - optimal_take stop = normalize_price(max(raw_stop, entry + point * 10), symbol) take = normalize_price(min(raw_take, entry - point * 20), symbol) if (entry - take) < (stop - entry) * 2: take = entry - (stop - entry) * 2 take = normalize_price(take, symbol) if is_buy: if not (stop < entry < take): return None else: if not (take < entry < stop): return None return { 'symbol': symbol, 'suggested_direction': 1 if is_buy else -1, 'entry_price': entry, 'stop_loss': stop, 'take_profit': take, 'risk_pips': abs(stop - entry) / point, 'reward_pips': abs(take - entry) / point, 'reversal_chance': reversal_chance * 100, 'avg_trend_move': avg_trend_move, 'avg_reversal': avg_reversal, 'cluster_size': cluster_size, 'signal_strength': reversal_chance * trend_completion * cluster_size / 5, 'historical_accuracy': (yellow_reversals / true_reversals * 100) if true_reversals > 0 else 0, 'cluster_start_time': recent_data.iloc[cluster_start]['time'] } def scan_market(): pairs = ['EURUSD.ecn', 'GBPUSD.ecn', 'USDJPY.ecn', 'USDCHF.ecn', 'AUDUSD.ecn', 'USDCAD.ecn', 'NZDUSD.ecn', 'EURGBP.ecn', 'EURJPY.ecn', 'GBPJPY.ecn', 'EURCHF.ecn', 'AUDJPY.ecn', 'CADJPY.ecn', 'NZDJPY.ecn', 'GBPCHF.ecn', 'EURAUD.ecn', 'EURCAD.ecn', 'GBPCAD.ecn', 'AUDNZD.ecn', 'AUDCAD.ecn'] signals = [] for pair in pairs: try: signal = analyze_pair(pair) if signal: signals.append(signal) except Exception as e: print(f"Error analyzing {pair}: {e}") return sorted(signals, key=lambda x: x['signal_strength'], reverse=True) def calculate_var_lots(signals, total_var_percent=1.0, total_lot=500.0): if not signals: return {} account = mt5.account_info() if not account: return {} total_risk_money = account.balance * (total_var_percent) * 10 lots_distribution = {} total_expectancy = sum(s['signal_strength'] * s['reversal_chance'] for s in signals) for signal in signals: symbol_info = mt5.symbol_info(signal['symbol']) if not symbol_info: continue signal_weight = (signal['signal_strength'] * signal['reversal_chance']) / total_expectancy signal_risk_money = total_risk_money * signal_weight pip_value = symbol_info.point * symbol_info.trade_contract_size risk_pips = signal['risk_pips'] lot_by_risk = signal_risk_money / (risk_pips * pip_value) lot_by_weight = total_lot * signal_weight lot_size = min(lot_by_risk, lot_by_weight) min_lot = symbol_info.volume_min max_lot = symbol_info.volume_max step = symbol_info.volume_step normalized_lot = round(lot_size / step) * step normalized_lot = max(min_lot, min(normalized_lot, max_lot)) lots_distribution[signal['symbol']] = { 'lot_size': normalized_lot, 'risk_percent': signal_weight * 100, 'risk_money': signal_risk_money } return lots_distribution def main(): if not mt5.initialize(path=terminal_path): print("Failed to initialize MT5") return try: print("Scanning market for yellow cluster signals...") signals = scan_market() if signals: top_signals = signals[:7] lots_info = calculate_var_lots(top_signals) signal_metrics = [] print("\nTop 7 reversal signals:") for signal in top_signals: direction = "BUY" if signal['suggested_direction'] > 0 else "SELL" lot_data = lots_info[signal['symbol']] symbol_info = mt5.symbol_info(signal['symbol']) contract_size = symbol_info.trade_contract_size point = symbol_info.point if 'JPY' in signal['symbol']: pip_value = (point * 100) * contract_size else: pip_value = point * 10000 * contract_size potential_profit = abs(signal['take_profit'] - signal['entry_price']) * pip_value * lot_data['lot_size'] max_loss = abs(signal['stop_loss'] - signal['entry_price']) * pip_value * lot_data['lot_size'] profit_risk_ratio = potential_profit / max_loss if max_loss != 0 else 0 signal_metrics.append({ 'symbol': signal['symbol'], 'direction': direction, 'profit': potential_profit, 'loss': max_loss, 'profit_risk_ratio': profit_risk_ratio, 'reversal_prob': signal['reversal_chance'], 'historical_accuracy': signal['historical_accuracy'] }) print(f"\n{signal['symbol']}:") print(f"Direction: {direction}") print(f"Position size: {lot_data['lot_size']:.2f} lots ({lot_data['risk_percent']:.1f}% of risk)") print(f"Risk amount: ${lot_data['risk_money']:.2f}") print(f"Entry price: {signal['entry_price']:.5f}") print(f"Stop loss: {signal['stop_loss']:.5f}") print(f"Take profit: {signal['take_profit']:.5f}") print(f"Risk/Reward: 1:{abs(signal['take_profit'] - signal['entry_price'])/abs(signal['stop_loss'] - signal['entry_price']):.1f}") print(f"Reversal probability: {signal['reversal_chance']:.1f}%") print(f"Historical accuracy: {signal['historical_accuracy']:.1f}%") print(f"Signal strength: {signal['signal_strength']:.2f}") print(f"Cluster size: {signal['cluster_size']} bars") print(f"Potential profit: ${potential_profit:.2f}") print(f"Maximum loss: ${max_loss:.2f}") print("\n\nTop 5 signals ranked by Profit/Risk ratio:") print("=" * 50) sorted_signals = sorted(signal_metrics, key=lambda x: x['profit_risk_ratio'], reverse=True)[:5] for i, s in enumerate(sorted_signals, 1): print(f"\n{i}. {s['symbol']} ({s['direction']})") print(f"Profit/Risk ratio: {s['profit_risk_ratio']:.2f}") print(f"Potential profit: ${s['profit']:.2f}") print(f"Maximum loss: ${s['loss']:.2f}") print(f"Reversal probability: {s['reversal_prob']:.1f}%") print(f"Historical accuracy: {s['historical_accuracy']:.1f}%") else: print("\nNo significant yellow clusters found") except Exception as e: print(f"Error during market scan: {e}") finally: mt5.shutdown() if __name__ == "__main__": main() import MetaTrader5 as mt5 from datetime import datetime, timedelta import time import threading import pandas as pd import logging from typing import Dict, List logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def adjust_tp_sl(symbol: str, entry: float, sl: float, tp: float, direction: int) -> tuple: symbol_info = mt5.symbol_info(symbol) if not symbol_info: return None, None min_stop_distance = symbol_info.trade_stops_level * symbol_info.point if direction > 0: sl = min(sl, entry - min_stop_distance) tp = max(tp, entry + min_stop_distance) else: sl = max(sl, entry + min_stop_distance) tp = min(tp, entry - min_stop_distance) return sl, tp class MT5Trader: def __init__(self, total_lot: float = 5.0, var_percent: float = 1.0): self.total_lot = total_lot self.var_percent = var_percent self.active_positions: Dict[str, Dict] = {} self.last_signals: Dict[str, Dict] = {} self._stop_event = threading.Event() def initialize(self) -> bool: if not mt5.initialize(path=terminal_path): logger.error("Failed to initialize MT5") return False return True def place_order(self, symbol: str, direction: int, lot: float, entry: float, sl: float, tp: float) -> bool: sl, tp = adjust_tp_sl(symbol, entry, sl, tp, direction) if not sl or not tp: return False order_type = mt5.ORDER_TYPE_BUY if direction > 0 else mt5.ORDER_TYPE_SELL request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lot, "type": order_type, "price": entry, "sl": sl, "tp": tp, "deviation": 10, "magic": 234000, "comment": "yellow_cluster_signal", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"Order failed: {result.comment}") return False logger.info(f"Order placed: {symbol} {'BUY' if direction > 0 else 'SELL'} {lot} lots") return True def close_position(self, symbol: str) -> bool: positions = mt5.positions_get(symbol=symbol) if not positions: return True position = positions[0] close_type = mt5.ORDER_TYPE_SELL if position.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": position.volume, "type": close_type, "position": position.ticket, "price": mt5.symbol_info_tick(symbol).ask if close_type == mt5.ORDER_TYPE_BUY else mt5.symbol_info_tick(symbol).bid, "deviation": 10, "magic": 234000, "comment": "signal_close", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"Close failed: {result.comment}") return False logger.info(f"Position closed: {symbol}") return True def update_active_positions(self): positions = mt5.positions_get() self.active_positions = { pos.symbol: { 'ticket': pos.ticket, 'type': pos.type, 'volume': pos.volume, 'open_price': pos.price_open } for pos in positions } def should_close_position(self, symbol: str, current_signal: dict) -> bool: if symbol not in self.active_positions: return False position = self.active_positions[symbol] position_type = 1 if position['type'] == mt5.ORDER_TYPE_BUY else -1 return position_type != current_signal['suggested_direction'] def trading_loop(self): while not self._stop_event.is_set(): try: signals = scan_market() if not signals: time.sleep(60) continue top_signals = signals[:5] lots_info = calculate_var_lots(top_signals, self.var_percent, self.total_lot) self.update_active_positions() # Check and close opposite positions for signal in top_signals: symbol = signal['symbol'] if self.should_close_position(symbol, signal): self.close_position(symbol) # Open new positions for signal in top_signals: symbol = signal['symbol'] if symbol not in self.active_positions: lot_data = lots_info[symbol] self.place_order( symbol, signal['suggested_direction'], lot_data['lot_size'], signal['entry_price'], signal['stop_loss'], signal['take_profit'] ) self.last_signals = {s['symbol']: s for s in top_signals} except Exception as e: logger.error(f"Error in trading loop: {e}") time.sleep(900) # 15 minutes def start(self): if not self.initialize(): return self.trading_thread = threading.Thread(target=self.trading_loop) self.trading_thread.start() logger.info("Trading bot started") def stop(self): self._stop_event.set() self.trading_thread.join() mt5.shutdown() logger.info("Trading bot stopped") def main(): # Usage example trader = MT5Trader(total_lot=5.0, var_percent=1.0) try: trader.start() # Can be stopped after a certain time for a test # time.sleep(3600) # Work for an hour # trader.stop() # Or work till Ctrl+C is pressed while True: time.sleep(10) except KeyboardInterrupt: trader.stop() if __name__ == "__main__": main()