import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime from itertools import combinations import matplotlib matplotlib.use('Agg') # Set the Agg backend for non-GUI operation import matplotlib.pyplot as plt import seaborn as sns from matplotlib.patches import Rectangle import io import base64 import matplotlib.pyplot as plt import matplotlib.dates as mdates from matplotlib.ticker import AutoMinorLocator def generate_fibonacci_sequence(n): """Generate an extended Fibonacci sequence""" fib = [1, 1] while len(fib) < n: fib.append(fib[-1] + fib[-2]) return fib def generate_fibonacci_ratios(): """Generate important Fibonacci ratios""" ratios = { '0.236': 0.236, '0.382': 0.382, '0.500': 0.500, '0.618': 0.618, '0.786': 0.786, '1.000': 1.000, '1.618': 1.618, '2.000': 2.000, '2.618': 2.618, '3.618': 3.618, '4.236': 4.236 } return ratios def calculate_price_movements(df, min_movement=0.0001): """Calculate significant price movements considering a minimum threshold""" movements = [] current_direction = None start_price = df['close'].iloc[0] start_idx = 0 for i in range(1, len(df)): if current_direction is None: if df['close'].iloc[i] > df['close'].iloc[i-1]: current_direction = 'up' elif df['close'].iloc[i] < df['close'].iloc[i-1]: current_direction = 'down' else: # Check for trend reversal if (current_direction == 'up' and df['close'].iloc[i] < df['close'].iloc[i-1]) or \ (current_direction == 'down' and df['close'].iloc[i] > df['close'].iloc[i-1]): movement = abs(df['close'].iloc[i-1] - start_price) # Record only significant movements if movement >= min_movement: movements.append({ 'start_time': df.index[start_idx], 'end_time': df.index[i-1], 'start_price': start_price, 'end_price': df['close'].iloc[i-1], 'movement': movement, 'direction': current_direction, 'duration': (df.index[i-1] - df.index[start_idx]).total_seconds() / 3600 # in hours }) current_direction = 'down' if current_direction == 'up' else 'up' start_price = df['close'].iloc[i-1] start_idx = i-1 return movements def find_fibonacci_patterns(movements, tolerance=0.01): """Find Fibonacci patterns in price and time movements""" fib_sequence = generate_fibonacci_sequence(15) fib_ratios = generate_fibonacci_ratios() patterns = [] time_patterns = [] # Find sequential patterns (3 movements) for i in range(len(movements) - 2): moves = [movements[i]['movement'], movements[i+1]['movement'], movements[i+2]['movement']] # Calculate actual distance in time between points times = [] for j in range(3): start_price = movements[i+j]['start_price'] end_price = movements[i+j]['end_price'] time_distance = abs(end_price - start_price) times.append(time_distance) # Normalize price movements min_move = min(moves) normalized_moves = [m/min_move for m in moves] # Normalize time distances min_time_dist = min(times) if min_time_dist > 0: normalized_times = [t/min_time_dist for t in times] # Find time patterns for j in range(len(fib_sequence)-2): fib_pattern = [fib_sequence[j], fib_sequence[j+1], fib_sequence[j+2]] time_matches = all(abs(normalized_times[k] - fib_pattern[k]) <= tolerance for k in range(3)) if time_matches: time_patterns.append({ 'type': 'time_sequence', 'start_time': movements[i]['start_time'], 'end_time': movements[i+2]['end_time'], 'price_distances': times, 'fibonacci_numbers': fib_pattern, 'ratio_accuracy': [abs(1 - normalized_times[k]/fib_pattern[k]) for k in range(3)], 'movements': moves, 'durations': [movements[i+k]['duration'] for k in range(3)] }) # Find price patterns for j in range(len(fib_sequence)-2): fib_pattern = [fib_sequence[j], fib_sequence[j+1], fib_sequence[j+2]] matches = all(abs(normalized_moves[k] - fib_pattern[k]) <= tolerance for k in range(3)) if matches: patterns.append({ 'type': 'price_sequence', 'start_time': movements[i]['start_time'], 'end_time': movements[i+2]['end_time'], 'movements': moves, 'fibonacci_numbers': fib_pattern, 'ratio_accuracy': [abs(1 - normalized_moves[k]/fib_pattern[k]) for k in range(3)], 'durations': [movements[i+k]['duration'] for k in range(3)] }) # Find ratios between any two movements for i, j in combinations(range(len(movements)), 2): # Price ratios price_ratio = movements[j]['movement'] / movements[i]['movement'] # Time ratios (by size of movement in time) time_dist1 = abs(movements[i]['end_price'] - movements[i]['start_price']) time_dist2 = abs(movements[j]['end_price'] - movements[j]['start_price']) time_ratio = time_dist2 / time_dist1 if time_dist1 > 0 else 0 for ratio_name, fib_ratio in fib_ratios.items(): # Check price ratios if abs(price_ratio - fib_ratio) <= tolerance: patterns.append({ 'type': 'price_ratio', 'start_time': movements[i]['start_time'], 'end_time': movements[j]['end_time'], 'movement1': movements[i]['movement'], 'movement2': movements[j]['movement'], 'actual_ratio': price_ratio, 'fibonacci_ratio': fib_ratio, 'ratio_name': ratio_name, 'accuracy': abs(1 - price_ratio/fib_ratio), 'duration1': movements[i]['duration'], 'duration2': movements[j]['duration'] }) # Check time ratios if time_ratio > 0 and abs(time_ratio - fib_ratio) <= tolerance: time_patterns.append({ 'type': 'time_ratio', 'start_time': movements[i]['start_time'], 'end_time': movements[j]['end_time'], 'price_distance1': time_dist1, 'price_distance2': time_dist2, 'actual_ratio': time_ratio, 'fibonacci_ratio': fib_ratio, 'ratio_name': ratio_name, 'accuracy': abs(1 - time_ratio/fib_ratio), 'movement1': movements[i]['movement'], 'movement2': movements[j]['movement'], 'duration1': movements[i]['duration'], 'duration2': movements[j]['duration'] }) return patterns, time_patterns def create_visualizations(df, movements, patterns, time_patterns): """Create visualizations for analysis""" figures = [] # 1. Price chart with marked movements plt.figure(figsize=(700/96, 8)) plt.plot(df.index, df['close'], label='Price', color='blue', alpha=0.6) for m in movements: if m['direction'] == 'up': color = 'green' else: color = 'red' plt.plot([m['start_time'], m['end_time']], [m['start_price'], m['end_price']], color=color, linewidth=2) plt.title('Price Movements') plt.xlabel('Time') plt.ylabel('Price') plt.grid(True) plt.legend() plt.xticks(rotation=45) figures.append(plt.gcf()) plt.close() # 2. Heatmap of Fibonacci ratios ratio_counts = {k: 0 for k in generate_fibonacci_ratios().keys()} for p in patterns: if p['type'] == 'price_ratio': ratio_counts[p['ratio_name']] += 1 plt.figure(figsize=(700/96, 6)) plt.bar(ratio_counts.keys(), ratio_counts.values()) plt.title('Frequency of Fibonacci Ratios') plt.xlabel('Ratio') plt.ylabel('Count') plt.xticks(rotation=45) figures.append(plt.gcf()) plt.close() # 3. Time patterns time_ratio_counts = {k: 0 for k in generate_fibonacci_ratios().keys()} for p in time_patterns: if p['type'] == 'time_ratio': time_ratio_counts[p['ratio_name']] += 1 plt.figure(figsize=(700/96, 6)) plt.bar(time_ratio_counts.keys(), time_ratio_counts.values(), color='orange') plt.title('Frequency of Fibonacci Time Ratios') plt.xlabel('Ratio') plt.ylabel('Count') plt.xticks(rotation=45) figures.append(plt.gcf()) plt.close() # 4. Scatter plot of movements and their duration plt.figure(figsize=(700/96, 10)) movements_df = pd.DataFrame(movements) plt.scatter(movements_df['duration'], movements_df['movement']) plt.xlabel('Duration (hours)') plt.ylabel('Movement Size') plt.title('Relationship between Movement Size and Duration') plt.grid(True) figures.append(plt.gcf()) plt.close() # 5. Accuracy of patterns accuracies = [] for p in patterns: if 'accuracy' in p: accuracies.append(1 - p['accuracy']) plt.figure(figsize=(700/96, 6)) plt.hist(accuracies, bins=20, color='green', alpha=0.6) plt.title('Distribution of Pattern Accuracy') plt.xlabel('Accuracy') plt.ylabel('Count') figures.append(plt.gcf()) plt.close() return figures def analyze_price_fibonacci(symbol, timeframe=mt5.TIMEFRAME_H1, bars_count=1000, min_movement=0.0001): """Main analysis function""" if not mt5.initialize(): print("MT5 initialization error") return None rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars_count) if rates is None: print(f"Failed to get data for {symbol}") mt5.shutdown() return None df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) movements = calculate_price_movements(df, min_movement) patterns, time_patterns = find_fibonacci_patterns(movements) # Create visualizations figures = create_visualizations(df, movements, patterns, time_patterns) mt5.shutdown() return patterns, time_patterns, movements, figures def print_fibonacci_patterns(patterns, time_patterns, movements): """Extended output of found patterns""" price_sequences = [p for p in patterns if p['type'] == 'price_sequence'] price_ratios = [p for p in patterns if p['type'] == 'price_ratio'] time_sequences = [p for p in time_patterns if p['type'] == 'time_sequence'] time_ratios = [p for p in time_patterns if p['type'] == 'time_ratio'] print(f"\nTotal price movements found: {len(movements)}") print(f"Price sequences found: {len(price_sequences)}") print(f"Price ratios found: {len(price_ratios)}") print(f"Time sequences found: {len(time_sequences)}") print(f"Time ratios found: {len(time_ratios)}") # Output statistics print("\n=== Analysis Statistics ===") print("Average movement duration:", np.mean([m['duration'] for m in movements]), "hours") print("Average price movement:", np.mean([m['movement'] for m in movements])) # Output pattern details print("\n=== Price Sequences ===") for i, pattern in enumerate(price_sequences, 1): print(f"\nSequence {i}:") print(f"Period: from {pattern['start_time']} to {pattern['end_time']}") print("Price movements:", [f"{m:.5f}" for m in pattern['movements']]) print("Fibonacci numbers:", pattern['fibonacci_numbers']) print("Ratio accuracy:", [f"{acc:.2%}" for acc in pattern['ratio_accuracy']]) print("Movement durations (hours):", [f"{d:.1f}" for d in pattern['durations']]) print("-" * 50) print("\n=== Price Ratios ===") for i, pattern in enumerate(price_ratios, 1): print(f"\nRatio {i}:") print(f"Period: from {pattern['start_time']} to {pattern['end_time']}") print(f"Movement 1: {pattern['movement1']:.5f} (duration: {pattern['duration1']:.1f}h)") print(f"Movement 2: {pattern['movement2']:.5f} (duration: {pattern['duration2']:.1f}h)") print(f"Fibonacci ratio: {pattern['ratio_name']} ({pattern['fibonacci_ratio']:.3f})") print(f"Actual ratio: {pattern['actual_ratio']:.3f}") print(f"Accuracy: {(1 - pattern['accuracy']):.2%}") print("-" * 50) print("\n=== Time Sequences ===") for i, pattern in enumerate(time_sequences, 1): print(f"\nSequence {i}:") print(f"Period: from {pattern['start_time']} to {pattern['end_time']}") print("Price distances:", [f"{d:.5f}" for d in pattern['price_distances']]) print("Fibonacci numbers:", pattern['fibonacci_numbers']) print("Ratio accuracy:", [f"{acc:.2%}" for acc in pattern['ratio_accuracy']]) print("Related movements:", [f"{m:.5f}" for m in pattern['movements']]) print("Durations (hours):", [f"{d:.1f}" for d in pattern['durations']]) print("-" * 50) print("\n=== Time Ratios ===") for i, pattern in enumerate(time_ratios, 1): print(f"\nRatio {i}:") print(f"Period: from {pattern['start_time']} to {pattern['end_time']}") print(f"Distance 1: {pattern['price_distance1']:.5f} (movement: {pattern['movement1']:.5f})") print(f"Distance 2: {pattern['price_distance2']:.5f} (movement: {pattern['movement2']:.5f})") print(f"Fibonacci ratio: {pattern['ratio_name']} ({pattern['fibonacci_ratio']:.3f})") print(f"Actual ratio: {pattern['actual_ratio']:.3f}") print(f"Accuracy: {(1 - pattern['accuracy']):.2%}") print(f"Duration 1: {pattern['duration1']:.1f}h") print(f"Duration 2: {pattern['duration2']:.1f}h") print("-" * 50) def predict_next_movement(movements, patterns, time_patterns, confidence_threshold=0.95): """Predict the next price movement based on Fibonacci patterns""" if not movements or len(movements) < 2: return None predictions = [] last_movement = movements[-1] last_price = last_movement['end_price'] last_movement_size = last_movement['movement'] # Analyze patterns with high accuracy high_accuracy_patterns = [p for p in patterns if p['type'] == 'price_ratio' and (1 - p['accuracy']) >= confidence_threshold] # Group patterns by ratios ratio_groups = {} for pattern in high_accuracy_patterns: ratio = pattern['ratio_name'] if ratio not in ratio_groups: ratio_groups[ratio] = [] ratio_groups[ratio].append(pattern) # Analyze each Fibonacci ratio fib_ratios = generate_fibonacci_ratios() for ratio_name, ratio_value in fib_ratios.items(): patterns_with_ratio = ratio_groups.get(ratio_name, []) if not patterns_with_ratio: continue # Analyze movement direction up_count = sum(1 for p in patterns_with_ratio if p['movement2'] > p['movement1']) down_count = len(patterns_with_ratio) - up_count # Calculate potential target levels target_levels = [] for pattern in patterns_with_ratio: if pattern['movement1'] > 0: level = last_movement_size * pattern['movement2'] / pattern['movement1'] target_levels.append(level) if target_levels: avg_target = np.mean(target_levels) # Predict up if up_count > 0: confidence_up = (up_count / len(patterns_with_ratio)) * (1 - np.std(target_levels) / avg_target) if confidence_up >= confidence_threshold: up_target = last_price + avg_target predictions.append({ 'direction': 'up', 'target_price': up_target, 'ratio': ratio_name, 'confidence': confidence_up, 'pattern_count': up_count, 'avg_movement': avg_target, 'expected_duration': np.mean([p['duration2'] for p in patterns_with_ratio]) }) # Predict down if down_count > 0: confidence_down = (down_count / len(patterns_with_ratio)) * (1 - np.std(target_levels) / avg_target) if confidence_down >= confidence_threshold: down_target = last_price - avg_target predictions.append({ 'direction': 'down', 'target_price': down_target, 'ratio': ratio_name, 'confidence': confidence_down, 'pattern_count': down_count, 'avg_movement': avg_target, 'expected_duration': np.mean([p['duration2'] for p in patterns_with_ratio]) }) # Additional analysis of time patterns time_patterns_high_accuracy = [p for p in time_patterns if (1 - p['accuracy']) >= confidence_threshold] # Adjust predictions based on time patterns for pred in predictions: matching_time_patterns = [p for p in time_patterns_high_accuracy if p['ratio_name'] == pred['ratio']] if matching_time_patterns: avg_time_accuracy = np.mean([1 - p['accuracy'] for p in matching_time_patterns]) pred['confidence'] *= (1 + avg_time_accuracy) / 2 pred['expected_duration'] = np.mean([p['duration2'] for p in matching_time_patterns]) # Sort predictions by confidence level predictions.sort(key=lambda x: x['confidence'], reverse=True) return predictions def visualize_predictions(df, movements, predictions, num_predictions=5): """Visualize predictions with improved layout and styling""" # Create figure with controlled width (700px = 7 inches at 100 dpi) plt.figure(figsize=(7, 4), dpi=100) # Style improvements plt.style.use('seaborn-v0_8-darkgrid') # Historical price chart with improved styling plt.plot(df.index, df['close'], label='Historical Prices', color='#1f77b4', alpha=0.7, linewidth=1.5) # Mark last movements with enhanced visibility for m in movements[-5:]: color = '#2ecc71' if m['direction'] == 'up' else '#e74c3c' plt.plot([m['start_time'], m['end_time']], [m['start_price'], m['end_price']], color=color, linewidth=2.5, marker='o', markersize=6, markerfacecolor='white', markeredgewidth=2) # Mark predictions with improved styling last_time = df.index[-1] time_delta = df.index[-1] - df.index[-2] colors = ['#c0392b', '#27ae60', '#f39c12', '#8e44ad', '#d35400'] for i, pred in enumerate(predictions[:num_predictions]): linestyle = '--' if pred['direction'] == 'up' else ':' expected_duration = pred.get('expected_duration', 5) future_time = last_time + time_delta * expected_duration plt.plot([last_time, future_time], [movements[-1]['end_price'], pred['target_price']], color=colors[i % len(colors)], linestyle=linestyle, linewidth=2, label=f"{pred['direction'].title()} {pred['ratio']} ({pred['confidence']:.1%})") # Enhanced annotation styling plt.annotate(f"{pred['target_price']:.5f}", (future_time, pred['target_price']), textcoords="offset points", xytext=(10, -10), ha='left', va='top', color=colors[i % len(colors)], fontsize=9, bbox=dict(facecolor='white', edgecolor=colors[i % len(colors)], alpha=0.8, pad=3, boxstyle='round,pad=0.5')) # Improve date formatting ax = plt.gca() ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d\n%H:%M')) ax.xaxis.set_major_locator(mdates.AutoDateLocator(minticks=5, maxticks=10)) # Rotate and align the tick labels so they look better plt.setp(ax.get_xticklabels(), ha='center') # Add minor gridlines for better readability ax.grid(True, which='major', linestyle='-', alpha=0.2) ax.grid(True, which='minor', linestyle=':', alpha=0.1) ax.yaxis.set_minor_locator(AutoMinorLocator()) # Style improvements for title and labels plt.title('Price Movement Prediction Based on Fibonacci Patterns', fontsize=14, pad=20, fontweight='bold') plt.xlabel('Time', fontsize=12, labelpad=10) plt.ylabel('Price', fontsize=12, labelpad=10) # Improved legend positioning and styling plt.legend(loc='center left', bbox_to_anchor=(1.02, 0.5), fontsize=10, framealpha=0.9, edgecolor='none', fancybox=True) # Adjust layout to prevent label cutoff plt.tight_layout() return plt.gcf() def print_predictions(predictions, last_price, num_predictions=5): """Extended output of predictions""" print("\n=== Price Movement Prediction ===") print(f"Current price: {last_price:.5f}") print("\nMost likely scenarios:") for i, pred in enumerate(predictions[:num_predictions], 1): movement = abs(pred['target_price'] - last_price) movement_percent = (movement / last_price) * 100 print(f"\nScenario {i}:") print(f"Direction: {pred['direction']}") print(f"Target price: {pred['target_price']:.5f}") print(f"Potential movement: {movement:.5f} ({movement_percent:.1f}%)") print(f"Fibonacci ratio: {pred['ratio']}") print(f"Confidence: {pred['confidence']:.1%}") print(f"Number of confirming patterns: {pred['pattern_count']}") print(f"Expected movement duration: {pred['expected_duration']:.1f} hours") print(f"Average movement size: {pred['avg_movement']:.5f}") print("-" * 30) def analyze_and_predict(symbol, timeframe=mt5.TIMEFRAME_H1, bars_count=1000, min_movement=0.0001, confidence_threshold=0.95): """Analyze and predict price""" # Initialize MT5 if not mt5.initialize(): print("MT5 initialization error") return None # Get data rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars_count) if rates is None: print(f"Failed to get data for {symbol}") mt5.shutdown() return None # Convert to DataFrame try: df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) # Get patterns movements = calculate_price_movements(df, min_movement) if not movements: print("No significant price movements found") mt5.shutdown() return None patterns, time_patterns = find_fibonacci_patterns(movements) if not patterns and not time_patterns: print("No Fibonacci patterns found") mt5.shutdown() return None # Create visualizations figures = create_visualizations(df, movements, patterns, time_patterns) # Get predictions predictions = predict_next_movement( movements, patterns, time_patterns, confidence_threshold ) if predictions: prediction_fig = visualize_predictions(df, movements, predictions) figures.append(prediction_fig) print_predictions(predictions, movements[-1]['end_price']) else: print("Failed to create predictions") mt5.shutdown() return patterns, time_patterns, movements, figures, predictions except Exception as e: print(f"An error occurred during analysis: {str(e)}") mt5.shutdown() return None # Example usage if __name__ == "__main__": symbol = "EURUSD" try: print(f"Starting analysis for {symbol}...") print("Connecting to MetaTrader5...") if not mt5.initialize(): print("MT5 initialization error") raise Exception("Failed to initialize MT5") # Check symbol availability symbols = mt5.symbols_get() symbol_names = [s.name for s in symbols] if symbol not in symbol_names: print(f"Symbol {symbol} not found") raise Exception(f"Symbol {symbol} not available") print("Running analysis...") results = analyze_and_predict( symbol, timeframe=mt5.TIMEFRAME_H1, bars_count=1000, min_movement=0.0001, confidence_threshold=0.95 ) if results: patterns, time_patterns, movements, figures, predictions = results print("\nAnalysis completed successfully") # Save visualizations for i, fig in enumerate(figures): filename = f'fibonacci_analysis_{i}.png' fig.savefig(filename) print(f"Saved chart: {filename}") else: print("Failed to perform analysis") except Exception as e: print(f"Execution error: {str(e)}") finally: mt5.shutdown() print("Connection to MT5 closed")