import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta import base58 from sklearn.preprocessing import StandardScaler from collections import Counter def initialize_mt5(): if not mt5.initialize(): print("Initialize() failed") mt5.shutdown() return False return True def get_eurusd_data(start_date, end_date, timeframe): rates = mt5.copy_rates_range("EURUSD", timeframe, start_date, end_date) df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') return df class PriceDecoder: def __init__(self, df): self.df = df self.binary_patterns = [] # Метод 1: Кодирование по направлению движения def direction_encoding(self, window=10): binary = (self.df['close'] > self.df['close'].shift(1)).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 2: Кодирование по отношению к MA def ma_encoding(self, ma_period=20, window=10): ma = self.df['close'].rolling(ma_period).mean() binary = (self.df['close'] > ma).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 3: Кодирование по силе движения def momentum_encoding(self, threshold=0.0001, window=10): returns = self.df['close'].pct_change() binary = (returns.abs() > threshold).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 4: Кодирование по объёмам def volume_encoding(self, window=10): avg_volume = self.df['tick_volume'].rolling(window).mean() binary = (self.df['tick_volume'] > avg_volume).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 5: Фрактальное кодирование def fractal_encoding(self, window=10): highs = self.df['high'].rolling(5, center=True).max() lows = self.df['low'].rolling(5, center=True).min() binary = ((self.df['high'] == highs) | (self.df['low'] == lows)).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 6: Кодирование по волатильности def volatility_encoding(self, window=10): volatility = self.df['high'] - self.df['low'] avg_volatility = volatility.rolling(20).mean() binary = (volatility > avg_volatility).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 7: Кодирование по свечным паттернам def candle_pattern_encoding(self, window=10): body = abs(self.df['close'] - self.df['open']) shadow = self.df['high'] - self.df['low'] binary = (body > shadow/2).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 8: Энтропийное кодирование def entropy_encoding(self, window=10): returns = self.df['close'].pct_change() entropy = returns.rolling(window).apply(lambda x: np.sum(-x[x!=0]*np.log2(abs(x[x!=0])))) binary = (entropy > entropy.mean()).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 9: Кодирование по схождению/расхождению def convergence_encoding(self, window=10): ma_fast = self.df['close'].rolling(5).mean() ma_slow = self.df['close'].rolling(20).mean() binary = (ma_fast > ma_slow).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 10: Кодирование по ценовым уровням def price_level_encoding(self, window=10): pivot = (self.df['high'] + self.df['low'] + self.df['close']) / 3 binary = (self.df['close'] > pivot).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 11: Кодирование по импульсу RSI def rsi_momentum_encoding(self, window=10, rsi_period=14): delta = self.df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=rsi_period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) binary = (rsi > 50).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 12: Кодирование по кластерам def cluster_encoding(self, window=10): prices = self.df['close'].values.reshape(-1, 1) scaler = StandardScaler() prices_scaled = scaler.fit_transform(prices) binary = (prices_scaled > 0).astype(int).flatten() pattern = ''.join(map(str, binary[-window:])) return pattern, self.analyze_pattern(pattern) # Метод 13: Кодирование по максимумам/минимумам def extremum_encoding(self, window=10): highs = self.df['high'].rolling(window).max() lows = self.df['low'].rolling(window).min() mid = (highs + lows) / 2 binary = (self.df['close'] > mid).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 14: Кодирование по тренду def trend_encoding(self, window=10): slope = pd.Series(np.nan, index=self.df.index) for i in range(window, len(self.df)): y = self.df['close'].iloc[i-window:i].values x = np.arange(window) slope[i] = np.polyfit(x, y, 1)[0] binary = (slope > 0).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) # Метод 15: Гибридное кодирование def hybrid_encoding(self, window=10): direction = (self.df['close'] > self.df['close'].shift(1)).astype(int) volume = (self.df['tick_volume'] > self.df['tick_volume'].rolling(window).mean()).astype(int) volatility = ((self.df['high'] - self.df['low']) > (self.df['high'] - self.df['low']).rolling(window).mean()).astype(int) binary = (direction & volume & volatility).astype(int) pattern = ''.join(binary.astype(str).tail(window)) return pattern, self.analyze_pattern(pattern) def analyze_pattern(self, pattern, min_seq_len=2, max_seq_len=8): # Конвертация в base58 для компактности base58_pattern = base58.b58encode(pattern.encode()).decode() # Расширенный анализ повторяемости с разными глубинами repeating_sequences = {} for seq_len in range(min_seq_len, max_seq_len + 1): sequences_at_depth = [] for i in range(len(pattern) - seq_len + 1): sequence = pattern[i:i+seq_len] count = pattern.count(sequence) if count > 1: sequences_at_depth.append({ 'sequence': sequence, 'count': count, 'positions': [j for j in range(len(pattern)) if pattern[j:j+seq_len] == sequence] }) if sequences_at_depth: repeating_sequences[seq_len] = sorted(sequences_at_depth, key=lambda x: (x['count'], len(x['sequence'])), reverse=True) # Анализ энтропии для разных окон entropy_analysis = {} for window_size in range(2, min(9, len(pattern) + 1)): windows = [pattern[i:i+window_size] for i in range(len(pattern)-window_size+1)] window_entropy = {} for window in set(windows): count = windows.count(window) prob = count / len(windows) local_entropy = -prob * np.log2(prob) if prob > 0 else 0 window_entropy[window] = { 'count': count, 'probability': prob, 'entropy': local_entropy } entropy_analysis[window_size] = window_entropy # Поиск стабильных паттернов (с нулевой энтропией) stable_patterns = [] for window_size, patterns in entropy_analysis.items(): zero_entropy_patterns = { pattern: data for pattern, data in patterns.items() if abs(data['entropy']) < 0.01 and data['count'] > 1 } if zero_entropy_patterns: stable_patterns.append({ 'window_size': window_size, 'patterns': zero_entropy_patterns }) # Базовая статистика ones_count = pattern.count('1') zeros_count = pattern.count('0') overall_entropy = 0 if len(pattern) > 0: probabilities = [pattern.count(char)/len(pattern) for char in set(pattern)] overall_entropy = -sum(p * np.log2(p) for p in probabilities if p > 0) return { 'base58': base58_pattern, 'repeating_sequences': repeating_sequences, 'stable_patterns': stable_patterns, 'entropy_analysis': entropy_analysis, 'ones_ratio': ones_count / len(pattern), 'zeros_ratio': zeros_count / len(pattern), 'overall_entropy': overall_entropy } def analyze_pattern_profitability(self, pattern, method_name, forward_bars=5): """Анализ доходности и винрейта для конкретного паттерна""" returns = [] wins = 0 total_trades = 0 window_size = len(pattern) # Получаем бинарные последовательности для всего периода binary = None if method_name == 'direction': binary = (self.df['close'] > self.df['close'].shift(1)).astype(int) elif method_name == 'ma': ma = self.df['close'].rolling(20).mean() binary = (self.df['close'] > ma).astype(int) elif method_name == 'momentum': returns_data = self.df['close'].pct_change() binary = (returns_data.abs() > 0.0001).astype(int) elif method_name == 'volume': avg_volume = self.df['tick_volume'].rolling(window_size).mean() binary = (self.df['tick_volume'] > avg_volume).astype(int) elif method_name == 'fractal': highs = self.df['high'].rolling(5, center=True).max() lows = self.df['low'].rolling(5, center=True).min() binary = ((self.df['high'] == highs) | (self.df['low'] == lows)).astype(int) elif method_name == 'volatility': volatility = self.df['high'] - self.df['low'] avg_volatility = volatility.rolling(20).mean() binary = (volatility > avg_volatility).astype(int) elif method_name == 'candle_pattern': body = abs(self.df['close'] - self.df['open']) shadow = self.df['high'] - self.df['low'] binary = (body > shadow/2).astype(int) elif method_name == 'entropy': returns_data = self.df['close'].pct_change() entropy = returns_data.rolling(window_size).apply( lambda x: np.sum(-x[x!=0]*np.log2(abs(x[x!=0]))) ) binary = (entropy > entropy.mean()).astype(int) elif method_name == 'convergence': ma_fast = self.df['close'].rolling(5).mean() ma_slow = self.df['close'].rolling(20).mean() binary = (ma_fast > ma_slow).astype(int) elif method_name == 'price_level': pivot = (self.df['high'] + self.df['low'] + self.df['close']) / 3 binary = (self.df['close'] > pivot).astype(int) elif method_name == 'rsi_momentum': delta = self.df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) binary = (rsi > 50).astype(int) elif method_name == 'cluster': prices = self.df['close'].values.reshape(-1, 1) scaler = StandardScaler() prices_scaled = scaler.fit_transform(prices) binary = pd.Series((prices_scaled > 0).flatten(), index=self.df.index) elif method_name == 'extremum': highs = self.df['high'].rolling(window_size).max() lows = self.df['low'].rolling(window_size).min() mid = (highs + lows) / 2 binary = (self.df['close'] > mid).astype(int) elif method_name == 'trend': slope = pd.Series(np.nan, index=self.df.index) for i in range(window_size, len(self.df)): y = self.df['close'].iloc[i-window_size:i].values x = np.arange(window_size) slope[i] = np.polyfit(x, y, 1)[0] binary = (slope > 0).astype(int) elif method_name == 'hybrid': direction = (self.df['close'] > self.df['close'].shift(1)).astype(int) volume = (self.df['tick_volume'] > self.df['tick_volume'].rolling(window_size).mean()).astype(int) volatility = ((self.df['high'] - self.df['low']) > (self.df['high'] - self.df['low']).rolling(window_size).mean()).astype(int) binary = (direction & volume & volatility) if binary is None: return None # Ищем все вхождения паттерна for i in range(len(binary) - window_size - forward_bars): current_window = ''.join(binary.iloc[i:i+window_size].astype(str)) if current_window == pattern: entry_price = self.df['close'].iloc[i + window_size] exit_price = self.df['close'].iloc[i + window_size + forward_bars] ret = (exit_price - entry_price) / entry_price * 100 returns.append(ret) if ret > 0: wins += 1 total_trades += 1 if total_trades > 0: return { 'average_return': np.mean(returns), 'win_rate': (wins / total_trades) * 100, 'total_trades': total_trades, 'max_return': max(returns), 'min_return': min(returns), 'return_std': np.std(returns), 'pattern_occurrences': total_trades } return None def analyze_all_patterns_profitability(self): """Комплексный анализ доходности для всех методов кодирования""" profitability_results = {} methods = [ 'direction', 'ma', 'momentum', 'volume', 'fractal', 'volatility', 'candle_pattern', 'entropy', 'convergence', 'price_level', 'rsi_momentum', 'cluster', 'extremum', 'trend', 'hybrid' ] for method in methods: pattern, analysis = getattr(self, f'{method}_encoding')() prof_analysis = self.analyze_pattern_profitability(pattern, method) if prof_analysis: profitability_results[method] = { 'pattern': pattern, 'base58': analysis['base58'], 'profitability': prof_analysis } return profitability_results def main(): if not initialize_mt5(): return # Получаем данные за последний месяц end_date = datetime.now() start_date = end_date - timedelta(days=365) df = get_eurusd_data(start_date, end_date, mt5.TIMEFRAME_H1) decoder = PriceDecoder(df) # Применяем все методы кодирования methods = [ ('Direction', decoder.direction_encoding), ('MA', decoder.ma_encoding), ('Momentum', decoder.momentum_encoding), ('Volume', decoder.volume_encoding), ('Fractal', decoder.fractal_encoding), ('Volatility', decoder.volatility_encoding), ('Candle Pattern', decoder.candle_pattern_encoding), ('Entropy', decoder.entropy_encoding), ('Convergence', decoder.convergence_encoding), ('Price Level', decoder.price_level_encoding), ('RSI Momentum', decoder.rsi_momentum_encoding), ('Cluster', decoder.cluster_encoding), ('Extremum', decoder.extremum_encoding), ('Trend', decoder.trend_encoding), ('Hybrid', decoder.hybrid_encoding) ] print("\nPrice Pattern Analysis Results:") print("-" * 50) # Сначала анализируем паттерны for method_name, method in methods: pattern, analysis = method() print(f"\n{method_name} Encoding:") print(f"Pattern: {pattern}") print(f"Base58: {analysis['base58']}") print("\nStable Patterns with Zero Entropy:") for stable_group in analysis['stable_patterns']: print(f"\nWindow Size {stable_group['window_size']}:") for pattern, data in stable_group['patterns'].items(): print(f" {pattern}: appears {data['count']} times") print("\nRepeating Sequences by Length:") for length, sequences in analysis['repeating_sequences'].items(): print(f"\nLength {length}:") for seq in sequences[:3]: # показываем топ-3 для каждой длины print(f" {seq['sequence']}: appears {seq['count']} times at positions {seq['positions']}") print(f"\nBasic Statistics:") print(f"Ones ratio: {analysis['ones_ratio']:.2f}") print(f"Zeros ratio: {analysis['zeros_ratio']:.2f}") print(f"Overall entropy: {analysis['overall_entropy']:.2f}") # Затем анализируем доходность print("\nProfitability Analysis Results:") print("-" * 50) profitability_results = decoder.analyze_all_patterns_profitability() for method, results in profitability_results.items(): print(f"\n{method.upper()} Pattern Analysis:") print(f"Pattern: {results['pattern']}") print(f"Base58: {results['base58']}") print(f"Average Return: {results['profitability']['average_return']:.2f}%") print(f"Win Rate: {results['profitability']['win_rate']:.2f}%") print(f"Total Trades: {results['profitability']['total_trades']}") print(f"Max Return: {results['profitability']['max_return']:.2f}%") print(f"Min Return: {results['profitability']['min_return']:.2f}%") print(f"Return Std: {results['profitability']['return_std']:.2f}%") mt5.shutdown() if __name__ == "__main__": main() from catboost import CatBoostClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report import numpy as np class BinaryPatternPredictor: def __init__(self, decoder, lookback=10, forecast_window=5): self.decoder = decoder self.lookback = lookback self.forecast_window = forecast_window self.model = CatBoostClassifier( iterations=500, learning_rate=0.03, depth=6, loss_function='Logloss', verbose=False ) def prepare_features(self, method_name): """Подготовка признаков для модели из бинарных паттернов""" features = [] labels = [] # Получаем полные бинарные последовательности binary_sequences = {} # Основная последовательность if method_name == 'direction': base_binary = (self.decoder.df['close'] > self.decoder.df['close'].shift(1)).astype(int) elif method_name == 'momentum': returns = self.decoder.df['close'].pct_change() base_binary = (returns.abs() > 0.0001).astype(int) elif method_name == 'volume': base_binary = (self.decoder.df['tick_volume'] > self.decoder.df['tick_volume'].rolling(self.lookback).mean()).astype(int) elif method_name == 'convergence': ma_fast = self.decoder.df['close'].rolling(5).mean() ma_slow = self.decoder.df['close'].rolling(20).mean() base_binary = (ma_fast > ma_slow).astype(int) elif method_name == 'hybrid': direction = (self.decoder.df['close'] > self.decoder.df['close'].shift(1)).astype(int) volume = (self.decoder.df['tick_volume'] > self.decoder.df['tick_volume'].rolling(self.lookback).mean()).astype(int) volatility = ((self.decoder.df['high'] - self.decoder.df['low']) > (self.decoder.df['high'] - self.decoder.df['low']).rolling(self.lookback).mean()).astype(int) base_binary = (direction & volume & volatility) # Дополнительные последовательности methods = ['momentum', 'volume', 'convergence'] for method in methods: if method != method_name: if method == 'momentum': returns = self.decoder.df['close'].pct_change() binary_sequences[method] = (returns.abs() > 0.0001).astype(int) elif method == 'volume': binary_sequences[method] = (self.decoder.df['tick_volume'] > self.decoder.df['tick_volume'].rolling(self.lookback).mean()).astype(int) elif method == 'convergence': ma_fast = self.decoder.df['close'].rolling(5).mean() ma_slow = self.decoder.df['close'].rolling(20).mean() binary_sequences[method] = (ma_fast > ma_slow).astype(int) # Формируем признаки base_binary = base_binary.fillna(0) # Заполняем NaN for seq in binary_sequences.values(): seq.fillna(0, inplace=True) # Создаём окна для обучения for i in range(self.lookback, len(base_binary) - self.forecast_window): # Основные признаки window = base_binary.iloc[i-self.lookback:i].values # Дополнительные признаки additional_features = [] for method, seq in binary_sequences.items(): current_window = seq.iloc[i-self.lookback:i].values additional_features.extend([ sum(current_window), current_window[-1], sum(current_window[-3:]) ]) feature_vector = np.concatenate([ window, [sum(window)], [sum(window[-3:])], additional_features ]) # Метка: преобладание единиц в будущем окне future_window = base_binary.iloc[i:i+self.forecast_window].values label = 1 if sum(future_window) > len(future_window)/2 else 0 features.append(feature_vector) labels.append(label) return np.array(features), np.array(labels) def train(self, method_name): """Обучение модели""" X, y = self.prepare_features(method_name) X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.5, shuffle=False) self.model.fit(X_train, y_train, eval_set=(X_test, y_test)) # Оценка качества predictions = self.model.predict(X_test) report = classification_report(y_test, predictions, output_dict=True) return { 'accuracy': report['accuracy'], 'precision': report['weighted avg']['precision'], 'recall': report['weighted avg']['recall'], 'f1': report['weighted avg']['f1-score'], 'feature_importance': self.model.feature_importances_ } def predict_next_pattern(self, method_name): """Прогноз на следующий паттерн""" X, _ = self.prepare_features(method_name) if len(X) > 0: # Берем последнее окно данных last_window = X[-1].reshape(1, -1) prediction = self.model.predict_proba(last_window)[0] return { 'probability_more_ones': prediction[1], 'probability_more_zeros': prediction[0], 'prediction': 'More ones' if prediction[1] > 0.5 else 'More zeros', 'confidence': max(prediction) } return None def main(): if not initialize_mt5(): return # Получаем данные за год end_date = datetime.now() start_date = end_date - timedelta(days=365) df = get_eurusd_data(start_date, end_date, mt5.TIMEFRAME_H1) decoder = PriceDecoder(df) predictor = BinaryPatternPredictor(decoder) # Обучаем модели для разных методов кодирования methods = ['direction', 'momentum', 'volume', 'convergence', 'hybrid'] print("\nNeural Network Analysis Results:") print("-" * 50) for method in methods: print(f"\nAnalyzing {method.upper()} patterns:") # Обучаем модель metrics = predictor.train(method) print(f"Model Performance:") print(f"Accuracy: {metrics['accuracy']:.2f}") print(f"Precision: {metrics['precision']:.2f}") print(f"Recall: {metrics['recall']:.2f}") print(f"F1 Score: {metrics['f1']:.2f}") # Делаем прогноз prediction = predictor.predict_next_pattern(method) if prediction: print(f"\nNext Pattern Prediction:") print(f"Forecast: {prediction['prediction']}") print(f"Confidence: {prediction['confidence']:.2f}") print(f"Probability of more ones: {prediction['probability_more_ones']:.2f}") print(f"Probability of more zeros: {prediction['probability_more_zeros']:.2f}") mt5.shutdown() if __name__ == "__main__": main()