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 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 EnhancedQuantumPredictor: def __init__(self, num_qubits=8): # Уменьшаем число кубитов для стабильности self.num_qubits = num_qubits self.simulator = AerSimulator() self.scaler = MinMaxScaler() def create_qpe_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) # Преобразуем в float 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) # Используем только последние 2 знака qc.ry(price_angle, qr[0]) # Измеряем все кубиты qc.measure(qr, cr) return qc def predict(self, market_data, current_price, features=None, shots=2000): """Упрощенное предсказание""" # Обрезаем входные данные if market_data.shape[0] > self.num_qubits: market_data = market_data[-self.num_qubits:] # Создаем и выполняем схему qc = self.create_qpe_circuit(market_data, current_price) compiled_circuit = transpile(qc, self.simulator, optimization_level=3) job = self.simulator.run(compiled_circuit, shots=shots) result = job.result() counts = result.get_counts() # Анализируем результаты predictions = [] total_shots = sum(counts.values()) for bitstring, count in counts.items(): # Используем количество единиц в битстроке для определения направления ones = bitstring.count('1') direction = ones / self.num_qubits # Нормализованное направление # Предсказываем изменение не более чем на 0.1% price_change = (direction - 0.5) * 0.001 predicted_price = current_price * (1 + price_change) predictions.extend([predicted_price] * count) predicted_price = np.mean(predictions) up_probability = sum(1 for p in predictions if p > current_price) / len(predictions) confidence = 1 - np.std(predictions) / current_price return { 'predicted_price': predicted_price, 'up_probability': up_probability, 'down_probability': 1 - up_probability, 'confidence': confidence } 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 = EnhancedQuantumPredictor() 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 predict(self): # Получаем данные df = self.data_loader.get_historical_data(self.window_size + 50) features = self.prepare_features(df) if len(features) < self.window_size: raise ValueError("Недостаточно данных") # Получаем последние данные для прогноза latest_features = features.iloc[-self.window_size:].values current_price = df['close'].iloc[-1] # Делаем прогноз, теперь передаем features как DataFrame prediction = self.quantum_predictor.predict( market_data=latest_features, current_price=current_price, features=features.iloc[-self.window_size:] # Передаем последние записи ) prediction.update({ 'timestamp': datetime.now(), 'current_price': current_price, 'rsi': features['rsi'].iloc[-1], 'sma': features['sma'].iloc[-1], 'ema': features['ema'].iloc[-1] }) return prediction def evaluate_model(symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, test_periods=100): """Оценка точности модели""" predictor = MarketPredictor(symbol, timeframe) predictions = [] actual_movements = [] # Получаем исторические данные df = predictor.data_loader.get_historical_data(test_periods + 50) for i in range(test_periods): try: temp_df = df.iloc[:-(test_periods-i)] predictor_temp = MarketPredictor(symbol, timeframe) features_temp = predictor_temp.prepare_features(temp_df) # Получаем данные для прогноза latest_features = features_temp.iloc[-predictor_temp.window_size:].values current_price = temp_df['close'].iloc[-1] # Делаем прогноз с передачей всех необходимых параметров prediction = predictor_temp.quantum_predictor.predict( market_data=latest_features, current_price=current_price, features=features_temp.iloc[-predictor_temp.window_size:] ) predicted_movement = 1 if prediction['up_probability'] > 0.5 else 0 predictions.append(predicted_movement) actual_price_next = df['close'].iloc[-(test_periods-i)] actual_price_current = df['close'].iloc[-(test_periods-i)-1] actual_movement = 1 if actual_price_next > actual_price_current else 0 actual_movements.append(actual_movement) except Exception as e: print(f"Error in evaluation: {e}") continue if len(predictions) > 0: 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) } else: metrics = { 'accuracy': 0, 'precision': 0, 'recall': 0, 'f1': 0 } return metrics if __name__ == "__main__": if not mt5.initialize(): print("MetaTrader5 initialization failed") mt5.shutdown() else: try: symbol = "EURUSD" timeframe = mt5.TIMEFRAME_H1 print("\nТестирование модели...") metrics = evaluate_model(symbol, timeframe, test_periods=100) 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%}") print("\nТекущий прогноз:") predictor = MarketPredictor(symbol, timeframe) df = predictor.data_loader.get_historical_data(predictor.window_size + 50) features = predictor.prepare_features(df) latest_features = features.iloc[-predictor.window_size:].values current_price = df['close'].iloc[-1] prediction = predictor.predict() # Теперь этот метод правильно передает все параметры print(f"Предсказанная цена: {prediction['predicted_price']:.5f}") print(f"Вероятность роста: {prediction['up_probability']:.2%}") print(f"Вероятность падения: {prediction['down_probability']:.2%}") print(f"Уверенность в прогнозе: {prediction['confidence']:.2%}") print(f"Текущая цена: {prediction['current_price']:.5f}") print(f"RSI: {prediction['rsi']:.2f}") print(f"SMA: {prediction['sma']:.5f}") print(f"EMA: {prediction['ema']:.5f}") finally: mt5.shutdown()