import numpy as np import MetaTrader5 as mt5 import pandas as pd from datetime import datetime, timedelta from qiskit import QuantumCircuit, transpile, QuantumRegister, ClassicalRegister from qiskit_aer import AerSimulator from sklearn.preprocessing import MinMaxScaler, StandardScaler from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score from sklearn.model_selection import train_test_split from catboost import CatBoostClassifier import warnings warnings.filterwarnings('ignore') class MT5DataLoader: def __init__(self, symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1): if not mt5.initialize(): raise Exception("MetaTrader5 initialization failed") self.symbol = symbol self.timeframe = timeframe def get_historical_data(self, lookback_bars=1000): current_time = datetime.now() rates = mt5.copy_rates_from(self.symbol, self.timeframe, current_time, lookback_bars) if rates is None: raise Exception(f"Failed to get data for {self.symbol}") df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') return df class BinaryPatternGenerator: def __init__(self, df, lookback=10): self.df = df self.lookback = lookback def direction_encoding(self): return (self.df['close'] > self.df['close'].shift(1)).astype(int) def momentum_encoding(self, threshold=0.0001): returns = self.df['close'].pct_change() return (returns.abs() > threshold).astype(int) def volume_encoding(self): return (self.df['tick_volume'] > self.df['tick_volume'].rolling(self.lookback).mean()).astype(int) def convergence_encoding(self): ma_fast = self.df['close'].rolling(5).mean() ma_slow = self.df['close'].rolling(20).mean() return (ma_fast > ma_slow).astype(int) def volatility_encoding(self): volatility = self.df['high'] - self.df['low'] avg_volatility = volatility.rolling(20).mean() return (volatility > avg_volatility).astype(int) def rsi_encoding(self, period=14, threshold=50): delta = self.df['close'].diff() gain = (delta.where(delta > 0, 0)).ewm(alpha=1/period).mean() loss = (-delta.where(delta < 0, 0)).ewm(alpha=1/period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return (rsi > threshold).astype(int) def bollinger_encoding(self, window=20): ma = self.df['close'].rolling(window=window).mean() std = self.df['close'].rolling(window=window).std() upper = ma + (std * 2) lower = ma - (std * 2) return ((self.df['close'] - lower)/(upper - lower) > 0.5).astype(int) def get_all_patterns(self): patterns = { 'direction': self.direction_encoding(), 'momentum': self.momentum_encoding(), 'volume': self.volume_encoding(), 'convergence': self.convergence_encoding(), 'volatility': self.volatility_encoding(), 'rsi': self.rsi_encoding(), 'bollinger': self.bollinger_encoding() } return patterns class QuantumFeatureGenerator: def __init__(self, num_qubits=8): self.num_qubits = num_qubits self.simulator = AerSimulator() self.scaler = MinMaxScaler() def create_quantum_circuit(self, market_data, current_price): qr = QuantumRegister(self.num_qubits, 'qr') cr = ClassicalRegister(self.num_qubits, 'cr') qc = QuantumCircuit(qr, cr) # Нормализуем данные scaled_data = self.scaler.fit_transform(market_data.reshape(-1, 1)).flatten() # Создаем суперпозицию for i in range(self.num_qubits): qc.h(qr[i]) # Применяем рыночные данные как фазы for i in range(min(len(scaled_data), self.num_qubits)): angle = float(scaled_data[i] * np.pi) qc.ry(angle, qr[i]) # Создаем запутанность for i in range(self.num_qubits - 1): qc.cx(qr[i], qr[i + 1]) # Добавляем текущую цену price_angle = float((current_price % 0.01) * 100 * np.pi) qc.ry(price_angle, qr[0]) qc.measure(qr, cr) return qc def get_quantum_features(self, market_data, current_price): qc = self.create_quantum_circuit(market_data, current_price) compiled_circuit = transpile(qc, self.simulator, optimization_level=3) job = self.simulator.run(compiled_circuit, shots=2000) result = job.result() counts = result.get_counts() # Создаем вектор квантовых признаков feature_vector = np.zeros(2**self.num_qubits) total_shots = sum(counts.values()) for bitstring, count in counts.items(): index = int(bitstring, 2) feature_vector[index] = count / total_shots return feature_vector class HybridQuantumBinaryPredictor: def __init__(self, num_qubits=8, lookback=10, forecast_window=5): self.num_qubits = num_qubits self.lookback = lookback self.forecast_window = forecast_window self.quantum_generator = QuantumFeatureGenerator(num_qubits) self.model = CatBoostClassifier( iterations=500, learning_rate=0.03, depth=6, loss_function='Logloss', verbose=False ) def prepare_features(self, df): """Подготовка гибридных признаков""" pattern_generator = BinaryPatternGenerator(df, self.lookback) binary_patterns = pattern_generator.get_all_patterns() features = [] labels = [] # Заполняем NaN в бинарных паттернах for key in binary_patterns: binary_patterns[key] = binary_patterns[key].fillna(0) for i in range(self.lookback, len(df) - self.forecast_window): try: # Квантовые признаки market_data = df['close'].iloc[i-self.lookback:i].values current_price = df['close'].iloc[i] quantum_features = self.quantum_generator.get_quantum_features(market_data, current_price) # Бинарные признаки binary_vector = [] for key in binary_patterns: window = binary_patterns[key].iloc[i-self.lookback:i].values binary_vector.extend([ sum(window), # Общее количество сигналов window[-1], # Последний сигнал sum(window[-3:]) # Последние 3 сигнала ]) # Технические индикаторы rsi = binary_patterns['rsi'].iloc[i] bollinger = binary_patterns['bollinger'].iloc[i] momentum = binary_patterns['momentum'].iloc[i] # Объединяем все признаки feature_vector = np.concatenate([ quantum_features, binary_vector, [rsi, bollinger, momentum] ]) # Метка: направление движения цены future_price = df['close'].iloc[i + self.forecast_window] current_price = df['close'].iloc[i] label = 1 if future_price > current_price else 0 features.append(feature_vector) labels.append(label) except Exception as e: print(f"Error at index {i}: {str(e)}") continue return np.array(features), np.array(labels) def train(self, df): """Обучение гибридной модели""" print("Preparing features...") X, y = self.prepare_features(df) # Разделение на обучающую и тестовую выборки split_point = int(len(X) * 0.8) X_train, X_test = X[:split_point], X[split_point:] y_train, y_test = y[:split_point], y[split_point:] print("Training model...") self.model.fit(X_train, y_train, eval_set=(X_test, y_test)) # Оценка модели predictions = self.model.predict(X_test) probas = self.model.predict_proba(X_test) # Расчет метрик metrics = { 'accuracy': accuracy_score(y_test, predictions), 'precision': precision_score(y_test, predictions), 'recall': recall_score(y_test, predictions), 'f1': f1_score(y_test, predictions) } # Анализ важности признаков feature_importance = self.model.feature_importances_ quantum_importance = np.mean(feature_importance[:2**self.num_qubits]) binary_importance = np.mean(feature_importance[2**self.num_qubits:]) metrics.update({ 'quantum_importance': quantum_importance, 'binary_importance': binary_importance, 'test_predictions': predictions, 'test_probas': probas, 'test_actual': y_test }) return metrics def predict_next(self, df): """Прогноз следующего движения""" X, _ = self.prepare_features(df) if len(X) > 0: last_features = X[-1].reshape(1, -1) prediction_proba = self.model.predict_proba(last_features)[0] prediction = self.model.predict(last_features)[0] return { 'direction': 'UP' if prediction == 1 else 'DOWN', 'probability_up': prediction_proba[1], 'probability_down': prediction_proba[0], 'confidence': max(prediction_proba) } return None def test_hybrid_model(symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, periods=1000): """Полное тестирование гибридной модели""" try: # Инициализация MT5 if not mt5.initialize(): raise Exception("Failed to initialize MT5") # Загрузка данных print(f"Loading {periods} periods of {symbol} {timeframe} data...") loader = MT5DataLoader(symbol, timeframe) df = loader.get_historical_data(periods) # Создание и обучение модели print("Creating hybrid model...") model = HybridQuantumBinaryPredictor() # Обучение и оценка print("Training and evaluating model...") metrics = model.train(df) # Вывод результатов print("\nModel Performance Metrics:") print(f"Accuracy: {metrics['accuracy']:.2%}") print(f"Precision: {metrics['precision']:.2%}") print(f"Recall: {metrics['recall']:.2%}") print(f"F1 Score: {metrics['f1']:.2%}") print("\nFeature Importance Analysis:") print(f"Quantum Features: {metrics['quantum_importance']:.2%}") print(f"Binary Features: {metrics['binary_importance']:.2%}") # Текущий прогноз print("\nCurrent Market Prediction:") prediction = model.predict_next(df) if prediction: print(f"Predicted Direction: {prediction['direction']}") print(f"Up Probability: {prediction['probability_up']:.2%}") print(f"Down Probability: {prediction['probability_down']:.2%}") print(f"Confidence: {prediction['confidence']:.2%}") return model, metrics finally: mt5.shutdown() if __name__ == "__main__": print("Starting Quantum-Binary Hybrid Trading System...") model, metrics = test_hybrid_model() print("\nSystem test completed.")