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): """Create qunatum circuit - leave original implementation""" qr = QuantumRegister(self.num_qubits, "qr") cr = ClassicalRegister(self.num_qubits, "cr") qc = QuantumCircuit(qr, cr) # Normalize data scaled_data = self.scaler.fit_transform(market_data.reshape(-1, 1)).flatten() # Create superposition for i in range(self.num_qubits): qc.h(qr[i]) # Apply market data as phases for i in range(min(len(scaled_data), self.num_qubits)): angle = float(scaled_data[i] * np.pi) qc.ry(angle, qr[i]) # Create entaglement for i in range(self.num_qubits - 1): qc.cx(qr[i], qr[i + 1]) # Apply the current price 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): """Get quantum features of fixed length""" 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() # Create a fixed feature vector feature_vector = np.zeros( 2**self.num_qubits ) # One element for each possible state total_shots = sum(counts.values()) # Fill in the vector with probabilities for each state for bitstring, count in counts.items(): index = int(bitstring, 2) # Transform the bit string into index feature_vector[index] = count / total_shots return feature_vector def fit(self, X_data, y_data): """Train hybrid model""" quantum_features = [] print("Convert data into quantum features...") 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"Quantum features dimensionality: {np.array(quantum_features).shape}") self.neural_network.fit(np.array(quantum_features), y_data) def predict(self, market_data, current_price): """Forecast using the quantum neural model""" 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): """Prepare technical indicator""" 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() # Get data df = MT5DataLoader(symbol, timeframe).get_historical_data(test_periods + 50) features = MarketPredictor(symbol, timeframe).prepare_features(df) # Prepare data for training X = features.values[:-1] # Everything except the last entry y = ( (df["close"].shift(-1) > df["close"]).values[:-1].astype(int) ) # Movement direction # Divide data into train and test samples 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:] # Train the model print("Training the model...") predictor.fit(X_train, y_train) # Test the model predictions = [] actual_movements = [] print("Testing the model...") 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 # Calculate the metrics 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("\nModel quality 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%}") finally: mt5.shutdown()