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 from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score from sklearn.neural_network import MLPClassifier 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 QuantumNeuralPredictor: def __init__(self, num_qubits=8): self.num_qubits = num_qubits self.simulator = AerSimulator() self.scaler = MinMaxScaler() self.neural_network = MLPClassifier( hidden_layer_sizes=(100, 50), activation='relu', solver='adam', max_iter=1000, random_state=42 ) 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 def fit(self, X_data, y_data): """Обучение гибридной модели""" quantum_features = [] print("Преобразование данных в квантовые признаки...") for i in range(len(X_data)): market_data = X_data[i] current_price = market_data[-1] features = self.get_quantum_features(market_data, current_price) quantum_features.append(features) print(f"Размерность квантовых признаков: {np.array(quantum_features).shape}") self.neural_network.fit(np.array(quantum_features), y_data) def predict(self, market_data, current_price): """Предсказание с использованием квантово-нейронной модели""" quantum_features = self.get_quantum_features(market_data, current_price) prediction_proba = self.neural_network.predict_proba([quantum_features])[0] predicted_price = current_price * (1 + (prediction_proba[1] - 0.5) * 0.001) return { 'predicted_price': predicted_price, 'up_probability': prediction_proba[1], 'down_probability': prediction_proba[0], 'confidence': max(prediction_proba) } class MarketPredictor: def __init__(self, symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, window_size=14): self.symbol = symbol self.timeframe = timeframe self.window_size = window_size self.quantum_predictor = QuantumNeuralPredictor() self.data_loader = MT5DataLoader(symbol, timeframe) def prepare_features(self, df): """Подготовка технических индикаторов""" df['sma'] = df['close'].rolling(window=self.window_size).mean() df['ema'] = df['close'].ewm(span=self.window_size).mean() df['std'] = df['close'].rolling(window=self.window_size).std() df['upper_band'] = df['sma'] + (df['std'] * 2) df['lower_band'] = df['sma'] - (df['std'] * 2) df['rsi'] = self.calculate_rsi(df['close']) df['momentum'] = df['close'] - df['close'].shift(self.window_size) df['rate_of_change'] = (df['close'] / df['close'].shift(1) - 1) * 100 features = df[['sma', 'ema', 'std', 'upper_band', 'lower_band', 'rsi', 'momentum', 'rate_of_change']].dropna() return features def calculate_rsi(self, prices, period=14): delta = prices.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 return 100 - (100 / (1 + rs)) def evaluate_quantum_neural_model(symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, test_periods=100): predictor = QuantumNeuralPredictor() # Получаем данные df = MT5DataLoader(symbol, timeframe).get_historical_data(test_periods + 50) features = MarketPredictor(symbol, timeframe).prepare_features(df) # Готовим данные для обучения X = features.values[:-1] # Все кроме последней записи y = (df['close'].shift(-1) > df['close']).values[:-1].astype(int) # Направление движения # Разделяем данные на обучающую и тестовую выборки train_size = int(0.8 * len(X)) X_train, X_test = X[:train_size], X[train_size:] y_train, y_test = y[:train_size], y[train_size:] # Обучаем модель print("Обучение модели...") predictor.fit(X_train, y_train) # Тестируем модель predictions = [] actual_movements = [] print("Тестирование модели...") for i in range(len(X_test)): try: market_data = X_test[i] current_price = df['close'].iloc[train_size + i] prediction = predictor.predict(market_data, current_price) predicted_movement = 1 if prediction['up_probability'] > 0.5 else 0 actual_movement = y_test[i] predictions.append(predicted_movement) actual_movements.append(actual_movement) except Exception as e: print(f"Error in evaluation: {e}") continue # Рассчитываем метрики metrics = { 'accuracy': accuracy_score(actual_movements, predictions), 'precision': precision_score(actual_movements, predictions), 'recall': recall_score(actual_movements, predictions), 'f1': f1_score(actual_movements, predictions) } return metrics, predictor if __name__ == "__main__": if not mt5.initialize(): print("MetaTrader5 initialization failed") mt5.shutdown() else: try: metrics, model = evaluate_quantum_neural_model() print("\nМетрики качества модели:") print(f"Точность (Accuracy): {metrics['accuracy']:.2%}") print(f"Точность прогнозов (Precision): {metrics['precision']:.2%}") print(f"Полнота (Recall): {metrics['recall']:.2%}") print(f"F1-мера: {metrics['f1']:.2%}") finally: mt5.shutdown()