# ================================================ # ai_trader_quantum_fusion.py # QUANTUM HYBRID: Qiskit + CatBoost + LLM # Version December 9, 2025 — Full integration from the article # ================================================ import os import re import time import json import logging import subprocess from datetime import datetime, timedelta import pandas as pd import numpy as np import matplotlib.pyplot as plt from typing import List, Dict, Tuple try: import MetaTrader5 as mt5 except ImportError: mt5 = None try: import ollama except ImportError: ollama = None try: from catboost import CatBoostClassifier, Pool CATBOOST_AVAILABLE = True except ImportError: CATBOOST_AVAILABLE = False print("⚠️ CatBoost is not installed: pip install catboost") try: from qiskit import QuantumCircuit from qiskit_aer import AerSimulator from scipy.stats import entropy QISKIT_AVAILABLE = True except ImportError: QISKIT_AVAILABLE = False print("⚠️ Qiskit is not installed: pip install qiskit qiskit-aer") # ====================== CONFIG ====================== MODEL_NAME = "koshtenco/quantum-trader-fusion-3b" BASE_MODEL = "llama3.2:3b" SYMBOLS = [ "EURUSD", "GBPUSD", "USDCHF", "USDCAD", "AUDUSD", "NZDUSD", "EURGBP", "AUDCHF", ] TIMEFRAME = mt5.TIMEFRAME_M15 if mt5 else None LOOKBACK = 400 INITIAL_BALANCE = 140.0 RISK_PER_TRADE = 0.08 MIN_PROB = 60 LIVE_LOT = 0.02 MAGIC = 20251209 SLIPPAGE = 10 # Quantum parameters N_QUBITS = 8 N_SHOTS = 2048 # Fine-tuning parameters FINETUNE_SAMPLES = 2000 BACKTEST_DAYS = 30 PREDICTION_HORIZON = 96 # 24 hours on M15 os.makedirs("logs", exist_ok=True) os.makedirs("dataset", exist_ok=True) os.makedirs("models", exist_ok=True) os.makedirs("charts", exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(message)s", handlers=[ logging.FileHandler("logs/quantum_fusion.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger(__name__) # ====================== QUANTUM ENCODER ====================== class QuantumEncoder: """ A Qiskit-based quantum encoder for extracting hidden features Implementation from the article: 8 qubits, entanglement via CZ gates, 2,048 measurements """ def __init__(self, n_qubits: int = 8, n_shots: int = 2048): self.n_qubits = n_qubits self.n_shots = n_shots self.simulator = AerSimulator() def encode_and_measure(self, features: np.ndarray) -> Dict[str, float]: """ Encodes features into a quantum circuit and extracts 4 quantum features: 1. Quantum entropy (a measure of uncertainty) 2. Dominant state (probability of the most frequent basis state) 3. Number of significant states (>3% probability) 4. Quantum variance of probabilities """ if not QISKIT_AVAILABLE: # Fallback to pseudo-quantum features return { "quantum_entropy": np.random.uniform(2.0, 5.0), "dominant_state_prob": np.random.uniform(0.05, 0.20), "significant_states": np.random.randint(3, 20), "quantum_variance": np.random.uniform(0.001, 0.01), } # Normalize features to the range [0, π] normalized = (features - features.min()) / ( features.max() - features.min() + 1e-8 ) angles = normalized * np.pi # Create a quantum circuit qc = QuantumCircuit(self.n_qubits, self.n_qubits) # Encode using RY rotations for i in range(min(len(angles), self.n_qubits)): qc.ry(angles[i], i) # Entanglement via CZ gates (creating second-order correlations) for i in range(self.n_qubits - 1): qc.cz(i, i + 1) # Close the chain qc.cz(self.n_qubits - 1, 0) # Measurement qc.measure(range(self.n_qubits), range(self.n_qubits)) # Running on the simulator job = self.simulator.run(qc, shots=self.n_shots) result = job.result() counts = result.get_counts() # Calculating quantum features total_shots = sum(counts.values()) probabilities = np.array( [ counts.get(format(i, f"0{self.n_qubits}b"), 0) / total_shots for i in range(2**self.n_qubits) ] ) # 1. Quantum Shannon entropy quantum_entropy = entropy(probabilities + 1e-10, base=2) # 2. Dominant state dominant_state_prob = np.max(probabilities) # 3. Number of significant states (>3%) significant_states = np.sum(probabilities > 0.03) # 4. Quantum variance quantum_variance = np.var(probabilities) return { "quantum_entropy": quantum_entropy, "dominant_state_prob": dominant_state_prob, "significant_states": significant_states, "quantum_variance": quantum_variance, } # ====================== TECHNICAL FEATURES ====================== def calculate_features(df: pd.DataFrame) -> pd.DataFrame: """Calculates 33 technical indicators""" d = df.copy() d["close_prev"] = d["close"].shift(1) # ATR tr = pd.concat( [ d["high"] - d["low"], (d["high"] - d["close_prev"]).abs(), (d["low"] - d["close_prev"]).abs(), ], axis=1, ).max(axis=1) d["ATR"] = tr.rolling(14).mean() # RSI delta = d["close"].diff() up = delta.clip(lower=0).rolling(14).mean() down = (-delta.clip(upper=0)).rolling(14).mean() rs = up / down.replace(0, np.nan) d["RSI"] = 100 - (100 / (1 + rs)) # MACD ema12 = d["close"].ewm(span=12, adjust=False).mean() ema26 = d["close"].ewm(span=26, adjust=False).mean() d["MACD"] = ema12 - ema26 d["MACD_signal"] = d["MACD"].ewm(span=9, adjust=False).mean() # Volume d["vol_avg_20"] = d["tick_volume"].rolling(20).mean() d["vol_ratio"] = d["tick_volume"] / d["vol_avg_20"].replace(0, np.nan) # Bollinger Bands d["BB_middle"] = d["close"].rolling(20).mean() bb_std = d["close"].rolling(20).std() d["BB_upper"] = d["BB_middle"] + 2 * bb_std d["BB_lower"] = d["BB_middle"] - 2 * bb_std d["BB_position"] = (d["close"] - d["BB_lower"]) / (d["BB_upper"] - d["BB_lower"]) # Stochastic low_14 = d["low"].rolling(14).min() high_14 = d["high"].rolling(14).max() d["Stoch_K"] = 100 * (d["close"] - low_14) / (high_14 - low_14) d["Stoch_D"] = d["Stoch_K"].rolling(3).mean() # EMA crossover d["EMA_50"] = d["close"].ewm(span=50, adjust=False).mean() d["EMA_200"] = d["close"].ewm(span=200, adjust=False).mean() # Additional features for CatBoost d["price_change_1"] = d["close"].pct_change(1) d["price_change_5"] = d["close"].pct_change(5) d["price_change_21"] = d["close"].pct_change(21) d["log_return"] = np.log(d["close"] / d["close"].shift(1)) d["volatility_20"] = d["log_return"].rolling(20).std() return d.dropna() # ====================== CATBOOST TRAINING ====================== def train_catboost_model( data_dict: Dict[str, pd.DataFrame], quantum_encoder: QuantumEncoder ) -> CatBoostClassifier: """ Trains CatBoost using data from all 8 currency pairs with quantum features Returns the trained model """ print(f"\n{'='*80}") print(f"TRAINING CATBOOST WITH QUANTUM FEATURES") print(f"{'='*80}\n") if not CATBOOST_AVAILABLE: print("❌ CatBoost is unavailable; using a placeholder") return None all_features = [] all_targets = [] all_symbols = [] print("Preparing data and performing quantum encoding...") for symbol, df in data_dict.items(): print(f"\nProcessing {symbol}: {len(df)} bars") df_features = calculate_features(df) # Quantum encoding for each data point quantum_features_list = [] for idx in range(LOOKBACK, len(df_features) - PREDICTION_HORIZON): if idx % 500 == 0: print( f" Quantum encoding: {idx}/{len(df_features) - PREDICTION_HORIZON}" ) row = df_features.iloc[idx] # Take key indicators for quantum encoding feature_vector = np.array( [ row["RSI"], row["MACD"], row["ATR"], row["vol_ratio"], row["BB_position"], row["Stoch_K"], row["price_change_1"], row["volatility_20"], ] ) # Quantum encoding quantum_feats = quantum_encoder.encode_and_measure(feature_vector) quantum_features_list.append(quantum_feats) # Target variable: price 24 hours ahead future_idx = idx + PREDICTION_HORIZON future_price = df_features.iloc[future_idx]["close"] current_price = row["close"] target = 1 if future_price > current_price else 0 # 1=UP, 0=DOWN # Collect features: technical features + quantum features + symbol features = { "RSI": row["RSI"], "MACD": row["MACD"], "ATR": row["ATR"], "vol_ratio": row["vol_ratio"], "BB_position": row["BB_position"], "Stoch_K": row["Stoch_K"], "Stoch_D": row["Stoch_D"], "EMA_50": row["EMA_50"], "EMA_200": row["EMA_200"], "price_change_1": row["price_change_1"], "price_change_5": row["price_change_5"], "price_change_21": row["price_change_21"], "volatility_20": row["volatility_20"], "quantum_entropy": quantum_feats["quantum_entropy"], "dominant_state_prob": quantum_feats["dominant_state_prob"], "significant_states": quantum_feats["significant_states"], "quantum_variance": quantum_feats["quantum_variance"], "symbol": symbol, } all_features.append(features) all_targets.append(target) all_symbols.append(symbol) print(f"\n✓ Total examples: {len(all_features)}") # Create a DataFrame X = pd.DataFrame(all_features) y = np.array(all_targets) # One-hot encoding of symbols X = pd.get_dummies(X, columns=["symbol"], prefix="sym") print(f"✓ Number of features: {len(X.columns)}") print( f"✓ Class balance: UP={np.sum(y==1)} ({np.sum(y==1)/len(y)*100:.1f}%), DOWN={np.sum(y==0)} ({np.sum(y==0)/len(y)*100:.1f}%)" ) # CatBoost training print("\nTraining CatBoost...") model = CatBoostClassifier( iterations=3000, learning_rate=0.03, depth=8, loss_function="Logloss", eval_metric="Accuracy", random_seed=42, verbose=500, ) # Use TimeSeriesSplit for proper validation from sklearn.model_selection import TimeSeriesSplit tscv = TimeSeriesSplit(n_splits=3) accuracies = [] for fold_idx, (train_idx, val_idx) in enumerate(tscv.split(X)): print(f"\n--- Fold {fold_idx + 1}/3 ---") X_train, X_val = X.iloc[train_idx], X.iloc[val_idx] y_train, y_val = y[train_idx], y[val_idx] model.fit(X_train, y_train, eval_set=(X_val, y_val), verbose=False) accuracy = model.score(X_val, y_val) accuracies.append(accuracy) print(f"Fold {fold_idx + 1} Accuracy: {accuracy*100:.2f}%") print(f"\n{'='*80}") print(f"CROSS-VALIDATION RESULTS") print(f"{'='*80}") print( f"Average accuracy: {np.mean(accuracies)*100:.2f}% ± {np.std(accuracies)*100:.2f}%" ) # Final training on all data print("\nTraining the final model on all data...") model.fit(X, y, verbose=500) # Saving the model model_path = "models/catboost_quantum.cbm" model.save_model(model_path) print(f"\n✓ Model saved: {model_path}") # Feature importance feature_importance = model.get_feature_importance() feature_names = X.columns importance_df = pd.DataFrame( {"feature": feature_names, "importance": feature_importance} ).sort_values("importance", ascending=False) print(f"\nTOP 10 FEATURES BY IMPORTANCE:") print(importance_df.head(10).to_string(index=False)) return model # ====================== GENERATING A HYBRID DATASET ====================== def generate_hybrid_dataset( data_dict: Dict[str, pd.DataFrame], catboost_model: CatBoostClassifier, quantum_encoder: QuantumEncoder, num_samples: int = 2000, ) -> List[Dict]: """ Generates a dataset for an LLM with embedded CatBoost predictions and quantum features Each example contains: - Technical indicators - Quantum features (in human-readable format) - CatBoost prediction (direction + confidence) - Actual result after 24 hours """ print(f"\n{'='*80}") print(f"GENERATING A HYBRID DATASET FOR AN LLM") print(f"{'='*80}\n") print( f"Goal: {num_samples} examples with CatBoost predictions and quantum features\n" ) dataset = [] up_count = 0 down_count = 0 target_per_symbol = num_samples // len(SYMBOLS) for symbol, df in data_dict.items(): print(f"Processing {symbol}...") df_features = calculate_features(df) candidates = [] for idx in range(LOOKBACK, len(df_features) - PREDICTION_HORIZON): row = df_features.iloc[idx] future_idx = idx + PREDICTION_HORIZON future_row = df_features.iloc[future_idx] # Quantum encoding feature_vector = np.array( [ row["RSI"], row["MACD"], row["ATR"], row["vol_ratio"], row["BB_position"], row["Stoch_K"], row["price_change_1"], row["volatility_20"], ] ) quantum_feats = quantum_encoder.encode_and_measure(feature_vector) # Preparing features for CatBoost X_features = { "RSI": row["RSI"], "MACD": row["MACD"], "ATR": row["ATR"], "vol_ratio": row["vol_ratio"], "BB_position": row["BB_position"], "Stoch_K": row["Stoch_K"], "Stoch_D": row["Stoch_D"], "EMA_50": row["EMA_50"], "EMA_200": row["EMA_200"], "price_change_1": row["price_change_1"], "price_change_5": row["price_change_5"], "price_change_21": row["price_change_21"], "volatility_20": row["volatility_20"], "quantum_entropy": quantum_feats["quantum_entropy"], "dominant_state_prob": quantum_feats["dominant_state_prob"], "significant_states": quantum_feats["significant_states"], "quantum_variance": quantum_feats["quantum_variance"], } # Creating a DataFrame for CatBoost (with one-hot encoding) X_df = pd.DataFrame([X_features]) for s in SYMBOLS: X_df[f"sym_{s}"] = 1 if s == symbol else 0 # CatBoost prediction if catboost_model: proba = catboost_model.predict_proba(X_df)[0] catboost_prob_up = proba[1] * 100 catboost_direction = "UP" if proba[1] > 0.5 else "DOWN" catboost_confidence = max(proba) * 100 else: catboost_prob_up = 50.0 catboost_direction = "UP" catboost_confidence = 50.0 # Actual result actual_price_24h = future_row["close"] price_change = actual_price_24h - row["close"] price_change_pips = int(price_change / 0.0001) actual_direction = "UP" if price_change > 0 else "DOWN" candidates.append( { "symbol": symbol, "row": row, "future_row": future_row, "quantum_feats": quantum_feats, "catboost_direction": catboost_direction, "catboost_confidence": catboost_confidence, "catboost_prob_up": catboost_prob_up, "actual_direction": actual_direction, "price_change_pips": price_change_pips, "current_time": df.index[idx], } ) # Balancing: take an equal number of UP and DOWN samples up_candidates = [c for c in candidates if c["actual_direction"] == "UP"] down_candidates = [c for c in candidates if c["actual_direction"] == "DOWN"] target_up = target_per_symbol // 2 target_down = target_per_symbol // 2 selected_up = ( np.random.choice( len(up_candidates), size=min(target_up, len(up_candidates)), replace=False, ) if up_candidates else [] ) selected_down = ( np.random.choice( len(down_candidates), size=min(target_down, len(down_candidates)), replace=False, ) if down_candidates else [] ) for idx in selected_up: candidate = up_candidates[idx] example = create_hybrid_training_example(candidate) dataset.append(example) up_count += 1 for idx in selected_down: candidate = down_candidates[idx] example = create_hybrid_training_example(candidate) dataset.append(example) down_count += 1 print( f" {symbol}: {len(selected_up)} UP + {len(selected_down)} DOWN = {len(selected_up) + len(selected_down)}" ) print(f"\n{'='*80}") print(f"HYBRID DATASET CREATED") print(f"{'='*80}") print(f"Total: {len(dataset)} examples") print(f" UP: {up_count} ({up_count/len(dataset)*100:.1f}%)") print(f" DOWN: {down_count} ({down_count/len(dataset)*100:.1f}%)") print(f"{'='*80}\n") return dataset def create_hybrid_training_example(candidate: Dict) -> Dict: """Creates a training example with a CatBoost prediction and quantum features""" row = candidate["row"] future_row = candidate["future_row"] quantum_feats = candidate["quantum_feats"] # Interpretation of quantum features entropy_level = ( "high uncertainty" if quantum_feats["quantum_entropy"] > 4.0 else ( "moderate uncertainty" if quantum_feats["quantum_entropy"] > 3.0 else "low uncertainty (market direction is clear)" ) ) dominant_strength = ( "strong" if quantum_feats["dominant_state_prob"] > 0.15 else "moderate" if quantum_feats["dominant_state_prob"] > 0.10 else "weak" ) market_complexity = ( "high" if quantum_feats["significant_states"] > 15 else "medium" if quantum_feats["significant_states"] > 8 else "low" ) # Checking CatBoost correctness catboost_correct = ( "CORRECT" if candidate["catboost_direction"] == candidate["actual_direction"] else "ERROR" ) prompt = f"""{candidate['symbol']} {candidate['current_time'].strftime('%Y-%m-%d %H:%M')} Current price: {row['close']:.5f} TECHNICAL INDICATORS: RSI: {row['RSI']:.1f} MACD: {row['MACD']:.6f} ATR: {row['ATR']:.5f} Volume: {row['vol_ratio']:.2f}x BB position: {row['BB_position']:.2f} Stochastic K: {row['Stoch_K']:.1f} QUANTUM FEATURES: Quantum entropy: {quantum_feats['quantum_entropy']:.2f} ({entropy_level}) Dominant state: {quantum_feats['dominant_state_prob']:.3f} ({dominant_strength} dominance) Significant states: {quantum_feats['significant_states']} (market complexity: {market_complexity}) Quantum variance: {quantum_feats['quantum_variance']:.6f} CATBOOST+QUANTUM PREDICTION: Direction: {candidate['catboost_direction']} Confidence: {candidate['catboost_confidence']:.1f}% UP probability: {candidate['catboost_prob_up']:.1f}% Source: catboost_quantum Analyze the situation taking into account the quantum model's prediction and provide an accurate price prediction 24 hours from now.""" response = f"""DIRECTION: {candidate['actual_direction']} CONFIDENCE: {min(98, max(65, candidate['catboost_confidence'] + np.random.randint(-5, 10)))}% 24-HOUR PRICE PREDICTION: {future_row['close']:.5f} ({candidate['price_change_pips']:+d} pips) CATBOOST PREDICTION ANALYSIS: The quantum model predicted {candidate['catboost_direction']} with {candidate['catboost_confidence']:.1f}% confidence. Actual result: {candidate['actual_direction']} ({catboost_correct}). QUANTUM ANALYSIS: Entropy {quantum_feats['quantum_entropy']:.2f} indicates {entropy_level}. {'The market has collapsed into a specific state — movement is predictable.' if quantum_feats['quantum_entropy'] < 3.0 else 'The market is in uncertainty mode — multiple scenarios are equally likely.' if quantum_feats['quantum_entropy'] > 4.5 else 'Moderate uncertainty — there is a preferred direction.'} The dominant state {quantum_feats['dominant_state_prob']:.3f} indicates {dominant_strength} dominance of a single quantum state. {quantum_feats['significant_states']} significant states indicate {market_complexity} market-structure complexity. 24-HOUR TECHNICAL ANALYSIS: {'RSI ' + str(round(row["RSI"], 1)) + ' — oversold, I expect a rebound' if row['RSI'] < 30 else 'RSI ' + str(round(row["RSI"], 1)) + ' — overbought, a correction is possible' if row['RSI'] > 70 else 'RSI ' + str(round(row["RSI"], 1)) + ' — neutral zone'}. {'MACD is positive — bullish momentum continues' if row['MACD'] > 0 else 'MACD is negative — bearish pressure continues'}. {'Volume is above average — the move is supported' if row['vol_ratio'] > 1.3 else 'Low volume — weak momentum'}. {'Price is near the lower BB — statistically, I expect a return to the mean' if row['BB_position'] < 0.25 else 'Price is near the upper BB — a pullback is possible' if row['BB_position'] > 0.75 else 'Price is in the middle of the BB — direction is not defined by the levels'}. CONCLUSION: The quantum CatBoost model {'correctly identified' if catboost_correct == 'CORRECT' else 'incorrectly predicted'} the direction. {'Quantum entropy confirms the predictability of the movement.' if quantum_feats['quantum_entropy'] < 3.5 else 'High quantum entropy indicates that the prediction is complex.'} Actual movement over 24 hours: {abs(candidate['price_change_pips'])} pips {candidate['actual_direction']}. Final price: {future_row['close']:.5f}. IMPORTANT: The quantum model has an accuracy of 62–68% on the validation set. This is an additional factor, not the absolute truth. {'In this case, the quantum features showed high confidence and turned out to be correct.' if catboost_correct == 'CORRECT' and quantum_feats['quantum_entropy'] < 3.5 else 'The next prediction may be the opposite—the market is unpredictable.'}""" return { "prompt": prompt, "response": response, "direction": candidate["actual_direction"], } # ====================== SAVING THE DATASET ====================== def save_dataset( dataset: List[Dict], filename: str = "dataset/quantum_fusion_data.jsonl" ) -> str: """Save the hybrid dataset""" with open(filename, "w", encoding="utf-8") as f: for item in dataset: f.write(json.dumps(item, ensure_ascii=False) + "\n") print(f"✓ Dataset saved: {filename}") print(f" Size: {os.path.getsize(filename) / 1024:.1f} KB") return filename # ====================== LLM FINE-TUNING ====================== def finetune_llm_with_catboost(dataset_path: str): """LLM fine-tuning with built-in CatBoost predictions""" print(f"\n{'='*80}") print(f"LLM FINE-TUNING WITH CATBOOST PREDICTIONS") print(f"{'='*80}\n") try: subprocess.run(["ollama", "--version"], check=True, capture_output=True) except: print("❌ Ollama is not installed!") print("Install: https://ollama.com/download") return print("Loading training data...") with open(dataset_path, "r", encoding="utf-8") as f: training_data = [json.loads(line) for line in f] training_sample = training_data[: min(500, len(training_data))] print(f"✓ {len(training_sample)} examples loaded") print("\nCreating a Modelfile with quantum examples...") modelfile_content = f"""FROM {BASE_MODEL} PARAMETER temperature 0.55 PARAMETER top_p 0.92 PARAMETER top_k 30 PARAMETER num_ctx 8192 PARAMETER num_predict 768 PARAMETER repeat_penalty 1.1 SYSTEM \"\"\" You are QuantumTrader-3B-Fusion, an elite analyst enhanced by quantum capabilities. UNIQUE CAPABILITIES: 1. You see the CatBoost model's predictions using quantum features (accuracy: 62–68%) 2. You understand quantum entropy, dominant states, and market complexity 3. You integrate quantum predictions with classical technical analysis STRICT RULES: 1. Only UP or DOWN—no FLAT 2. Confidence: 65–98% 3. REQUIRED: 24h price prediction: X.XXXXX (±NN pips) 4. Analyze the CatBoost prediction and quantum features 5. Explain why the quantum model is right or where it was wrong RESPONSE FORMAT: DIRECTION: UP/DOWN CONFIDENCE: XX% PRICE PREDICTION IN 24H: X.XXXXX (±NN pips) CATBOOST PREDICTION ANALYSIS: [Evaluation of the quantum model's prediction] QUANTUM ANALYSIS: [Interpretation of quantum entropy and dominant states] 24-HOUR TECHNICAL ANALYSIS: [RSI, MACD, volume, levels] CONCLUSION: [Synthesis of quantum and technical signals with a specific target] \"\"\" """ for i, example in enumerate(training_sample, 1): modelfile_content += f""" MESSAGE user \"\"\"{example['prompt']}\"\"\" MESSAGE assistant \"\"\"{example['response']}\"\"\" """ modelfile_path = "Modelfile_quantum_fusion" with open(modelfile_path, "w", encoding="utf-8") as f: f.write(modelfile_content) print(f"✓ Modelfile created with {len(training_sample)} examples") print(f"\nCreating model {MODEL_NAME}...") print("This will take 2–5 minutes...\n") try: result = subprocess.run( ["ollama", "create", MODEL_NAME, "-f", modelfile_path], check=True, capture_output=True, text=True, ) print(result.stdout) print(f"\n✓ Model {MODEL_NAME} was created successfully!") print("\nTesting the model...") test_prompt = """EURUSD 2025-12-09 10:00 Current price: 1.0850 TECHNICAL INDICATORS: RSI: 32.5 MACD: -0.00015 ATR: 0.00085 Volume: 1.8x BB position: 0.15 Stochastic K: 25.0 QUANTUM FEATURES: Quantum entropy: 2.8 (low uncertainty—the market direction is clear) Dominant state: 0.187 (strong dominance) Significant states: 5 (market complexity: low) Quantum variance: 0.003421 CATBOOST+QUANTUM PREDICTION: Direction: UP Confidence: 87.3% Probability of UP: 87.3% Source: catboost_quantum Analyze it.""" test_result = ollama.generate(model=MODEL_NAME, prompt=test_prompt) print("\n" + "=" * 80) print("TEST ANSWER:") print("=" * 80) print(test_result["response"]) print("=" * 80) os.remove(modelfile_path) print(f"\n{'='*80}") print(f"FINE-TUNING COMPLETE!") print(f"{'='*80}") print(f"✓ Model ready: {MODEL_NAME}") print(f"✓ Integration: CatBoost + Qiskit + LLM") print(f"✓ To publish: ollama push {MODEL_NAME}") except subprocess.CalledProcessError as e: print(f"❌ Error: {e}") print(f"Output: {e.output}") # ====================== PARSING LLM RESPONSES ====================== def parse_answer(text: str) -> dict: """Parsing the LLM response containing a price prediction""" prob = re.search(r"(?:CONFIDENCE|PROBABILITY)[\s:]*(\d+)", text, re.I) direction = re.search(r"\b(UP|DOWN)\b", text, re.I) price_pred = re.search(r"PRICE PREDICTION.*?(\d+\.\d+)", text, re.I) p = int(prob.group(1)) if prob else 50 d = direction.group(1).upper() if direction else "DOWN" target_price = float(price_pred.group(1)) if price_pred else None return {"prob": p, "dir": d, "target_price": target_price} # ====================== VISUALIZATION ====================== def plot_results(balance_hist, equity_hist, slots): """Equity chart with exact dimensions""" DPI = 100 WIDTH_PX = 700 HEIGHT_PX = 350 fig = plt.figure(figsize=(WIDTH_PX / DPI, HEIGHT_PX / DPI), dpi=DPI) min_length = min(len(equity_hist), len(slots)) dates = [s["datetime"] for s in slots[:min_length]] equity_to_plot = equity_hist[:min_length] plt.plot(dates, equity_to_plot, color="#1E90FF", linewidth=3.5, label="Equity") plt.title( "Equity Curve (Quantum Fusion)", fontsize=16, fontweight="bold", color="white" ) plt.xlabel("Time", color="white") plt.ylabel("Balance ($)", color="white") ax = plt.gca() ax.set_facecolor("#0a0a0a") ax.spines["bottom"].set_color("white") ax.spines["top"].set_color("none") ax.spines["right"].set_color("none") ax.spines["left"].set_color("white") ax.tick_params(colors="white") plt.grid(alpha=0.2, color="gray") plt.xticks(rotation=45) plt.legend(facecolor="#0a0a0a", edgecolor="white", labelcolor="white") plt.tight_layout(pad=2.0) filename = f"charts/equity_quantum_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png" plt.savefig( filename, dpi=DPI, facecolor="#0a0a0a", edgecolor="none", bbox_inches="tight", pad_inches=0.1, ) print(f"\n✓ Chart saved: {filename} ({WIDTH_PX}×{HEIGHT_PX} px)") plt.show() def calculate_max_drawdown(equity): """Calculate maximum drawdown""" if len(equity) == 0: return 0 peak = np.maximum.accumulate(equity) dd = (peak - equity) / (peak + 1e-8) return np.max(dd) * 100 # ====================== BACKTEST ====================== def backtest(): """ Backtest of a quantum hybrid system (CatBoost + Quantum + LLM) Uses a CatBoost model trained with quantum features """ print(f"\n{'='*80}") print(f"BACKTEST OF A QUANTUM HYBRID SYSTEM") print(f"{'='*80}\n") # Check that models exist if not os.path.exists("models/catboost_quantum.cbm"): print("❌ CatBoost model not found!") print("First, train the model (mode 1) or run the full cycle (mode 6)") return # Loading the CatBoost model print("Loading the CatBoost model...") if not CATBOOST_AVAILABLE: print("❌ CatBoost is not available") return catboost_model = CatBoostClassifier() catboost_model.load_model("models/catboost_quantum.cbm") print("✓ CatBoost model loaded") # Checking Ollama and the LLM model use_llm = False if ollama: try: ollama.list() # Check whether our model is available models = ollama.list() if any(MODEL_NAME in str(m) for m in models.get("models", [])): use_llm = True print("✓ LLM model found; using hybrid mode") else: print(f"⚠️ LLM model {MODEL_NAME} not found") print("We are operating in CatBoost+Quantum-only mode") except: print("⚠️ Ollama is unavailable; running with CatBoost+Quantum only") # Loading data if not mt5 or not mt5.initialize(): print("❌ MT5 is not connected") return end = datetime.now().replace(second=0, microsecond=0) start = end - timedelta(days=BACKTEST_DAYS) data = {} print( f"\nLoading data from {start.strftime('%Y-%m-%d')} to {end.strftime('%Y-%m-%d')}..." ) for sym in SYMBOLS: rates = mt5.copy_rates_range(sym, TIMEFRAME, start, end) if rates is None or len(rates) == 0: print(f" ⚠️ {sym}: no data") continue df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s") df.set_index("time", inplace=True) if len(df) > LOOKBACK + PREDICTION_HORIZON: data[sym] = df print(f" ✓ {sym}: {len(df)} bars") if not data: print("\n❌ No data for the backtest!") mt5.shutdown() return # Initialization balance = INITIAL_BALANCE equity = INITIAL_BALANCE trades = [] balance_hist = [balance] equity_hist = [equity] slots = [{"datetime": start}] SPREAD_PIPS = 2 SWAP_LONG = -0.5 SWAP_SHORT = -0.3 print(f"\n{'='*80}") print(f"BACKTEST PARAMETERS") print(f"{'='*80}") print(f"Initial balance: ${balance:,.2f}") print(f"Risk per trade: {RISK_PER_TRADE * 100}%") print(f"Minimum confidence: {MIN_PROB}%") print(f"Spread: {SPREAD_PIPS} pips") print(f"Long/short swap: {SWAP_LONG}/{SWAP_SHORT} USD/day") print(f"Mode: {'CatBoost + Quantum + LLM' if use_llm else 'CatBoost + Quantum'}") print(f"{'='*80}\n") # Quantum encoder quantum_encoder = QuantumEncoder(N_QUBITS, N_SHOTS) # Determining analysis points main_symbol = list(data.keys())[0] main_data = data[main_symbol] total_bars = len(main_data) analysis_points = list( range(LOOKBACK, total_bars - PREDICTION_HORIZON, PREDICTION_HORIZON) ) print(f"Analysis points: {len(analysis_points)} (every 24 hours)\n") print("Starting trading...\n") # Main backtest loop for point_idx, current_idx in enumerate(analysis_points): current_time = main_data.index[current_idx] print(f"{'='*80}") print( f"Analysis #{point_idx + 1}/{len(analysis_points)}: {current_time.strftime('%Y-%m-%d %H:%M')}" ) print(f"{'='*80}") for sym in SYMBOLS: if sym not in data: continue # Historical data up to the current moment historical_data = data[sym].iloc[: current_idx + 1].copy() if len(historical_data) < LOOKBACK: continue # Calculate technical features df_with_features = calculate_features(historical_data) if len(df_with_features) == 0: continue row = df_with_features.iloc[-1] # Get symbol information symbol_info = mt5.symbol_info(sym) if symbol_info is None: continue point = symbol_info.point contract_size = symbol_info.trade_contract_size # ===== QUANTUM ENCODING ===== feature_vector = np.array( [ row["RSI"], row["MACD"], row["ATR"], row["vol_ratio"], row["BB_position"], row["Stoch_K"], row["price_change_1"], row["volatility_20"], ] ) quantum_feats = quantum_encoder.encode_and_measure(feature_vector) # ===== CATBOOST PREDICTION ===== X_features = { "RSI": row["RSI"], "MACD": row["MACD"], "ATR": row["ATR"], "vol_ratio": row["vol_ratio"], "BB_position": row["BB_position"], "Stoch_K": row["Stoch_K"], "Stoch_D": row["Stoch_D"], "EMA_50": row["EMA_50"], "EMA_200": row["EMA_200"], "price_change_1": row["price_change_1"], "price_change_5": row["price_change_5"], "price_change_21": row["price_change_21"], "volatility_20": row["volatility_20"], "quantum_entropy": quantum_feats["quantum_entropy"], "dominant_state_prob": quantum_feats["dominant_state_prob"], "significant_states": quantum_feats["significant_states"], "quantum_variance": quantum_feats["quantum_variance"], } X_df = pd.DataFrame([X_features]) for s in SYMBOLS: X_df[f"sym_{s}"] = 1 if s == sym else 0 proba = catboost_model.predict_proba(X_df)[0] catboost_prob_up = proba[1] * 100 catboost_direction = "UP" if proba[1] > 0.5 else "DOWN" catboost_confidence = max(proba) * 100 # Interpretation of quantum features entropy_level = ( "low" if quantum_feats["quantum_entropy"] < 3.0 else "medium" if quantum_feats["quantum_entropy"] < 4.5 else "high" ) print(f"\n{sym}:") print( f" Quantum: entropy={quantum_feats['quantum_entropy']:.2f} ({entropy_level}), " f"dominant={quantum_feats['dominant_state_prob']:.3f}" ) print(f" CatBoost: {catboost_direction} {catboost_confidence:.1f}%") # ===== LLM PREDICTION (if available) ===== final_direction = catboost_direction final_confidence = catboost_confidence if use_llm: try: prompt = f"""{sym} {current_time.strftime('%Y-%m-%d %H:%M')} Current price: {row['close']:.5f} TECHNICAL INDICATORS: RSI: {row['RSI']:.1f} MACD: {row['MACD']:.6f} ATR: {row['ATR']:.5f} Volume: {row['vol_ratio']:.2f}x BB position: {row['BB_position']:.2f} Stochastic K: {row['Stoch_K']:.1f} QUANTUM FEATURES: Quantum entropy: {quantum_feats['quantum_entropy']:.2f} ({entropy_level} uncertainty) Dominant state: {quantum_feats['dominant_state_prob']:.3f} Significant states: {quantum_feats['significant_states']} Quantum variance: {quantum_feats['quantum_variance']:.6f} CATBOOST+QUANTUM PREDICTION: Direction: {catboost_direction} Confidence: {catboost_confidence:.1f}% Probability of UP: {catboost_prob_up:.1f}% Analyze the situation and provide a 24-hour prediction.""" resp = ollama.generate( model=MODEL_NAME, prompt=prompt, options={"temperature": 0.3} ) result = parse_answer(resp["response"]) final_direction = result["dir"] final_confidence = result["prob"] print( f" LLM: {final_direction} {final_confidence}% (correction: {final_confidence - catboost_confidence:+.1f}%)" ) except Exception as e: log.error(f"LLM error for {sym}: {e}") final_direction = catboost_direction final_confidence = catboost_confidence # ===== CONFIDENCE CHECK ===== if final_confidence < MIN_PROB: print( f" ❌ Confidence {final_confidence:.1f}% < {MIN_PROB}%, skipping" ) continue # ===== CALCULATING THE RESULT AFTER 24 HOURS ===== exit_idx = current_idx + PREDICTION_HORIZON if exit_idx >= len(data[sym]): continue exit_row = data[sym].iloc[exit_idx] # Entry price, including the spread entry_price = ( row["close"] + SPREAD_PIPS * point if final_direction == "UP" else row["close"] ) # Exit price, including the spread exit_price = ( exit_row["close"] if final_direction == "UP" else exit_row["close"] + SPREAD_PIPS * point ) # Price Movement in Pips price_move_pips = ( (exit_price - entry_price) / point if final_direction == "UP" else (entry_price - exit_price) / point ) # ===== CALCULATING POSITION SIZE ===== risk_amount = balance * RISK_PER_TRADE atr_pips = row["ATR"] / point stop_loss_pips = max(20, atr_pips * 2) lot_size = risk_amount / (stop_loss_pips * point * contract_size) lot_size = max(0.01, min(lot_size, 10.0)) # ===== PROFIT CALCULATION ===== profit_pips = price_move_pips profit_usd = profit_pips * point * contract_size * lot_size # 24-Hour Swap swap_cost = SWAP_LONG if final_direction == "UP" else SWAP_SHORT swap_cost = swap_cost * (lot_size / 0.01) profit_usd -= swap_cost # Slippage profit_usd -= SLIPPAGE * point * contract_size * lot_size # ===== BALANCE UPDATE ===== balance += profit_usd equity = balance # ===== CORRECTNESS CHECK ===== actual_direction = "UP" if (exit_row["close"] > row["close"]) else "DOWN" correct = final_direction == actual_direction # ===== RECORD TRADE ===== trades.append( { "time": current_time, "symbol": sym, "direction": final_direction, "confidence": final_confidence, "catboost_confidence": catboost_confidence, "quantum_entropy": quantum_feats["quantum_entropy"], "entry_price": entry_price, "exit_price": exit_price, "lot_size": lot_size, "profit_pips": profit_pips, "profit_usd": profit_usd, "balance": balance, "correct": correct, } ) # ===== OUTPUT ===== status = "✓ CORRECT" if correct else "✗ ERROR" color = "\033[92m" if correct else "\033[91m" reset = "\033[0m" print( f" {color}{status}{reset} | Entry: {entry_price:.5f} → Exit: {exit_price:.5f}" ) print( f" Lot: {lot_size:.2f} | Profit: {profit_pips:+.1f} pips = ${profit_usd:+.2f}" ) print(f" Balance: ${balance:,.2f}") balance_hist.append(balance) equity_hist.append(equity) slots.append({"datetime": current_time}) mt5.shutdown() # ===== STATISTICS ===== print(f"\n{'='*80}") print(f"BACKTEST RESULTS") print(f"{'='*80}\n") print( f"Period: {start.strftime('%Y-%m-%d')} → {end.strftime('%Y-%m-%d')} ({BACKTEST_DAYS} days)" ) print( f"Mode: {'CatBoost + Quantum + LLM (Hybrid)' if use_llm else 'CatBoost + Quantum'}" ) print(f"\nTRADES:") print(f" Total: {len(trades)}") print(f" Initial balance: ${INITIAL_BALANCE:,.2f}") print(f" Final balance: ${balance:,.2f}") print(f" Profit/Loss: ${balance - INITIAL_BALANCE:+,.2f}") print(f" Return: {((balance/INITIAL_BALANCE - 1) * 100):+.2f}%") if trades: wins = sum(1 for t in trades if t["profit_usd"] > 0) losses = len(trades) - wins win_rate = wins / len(trades) * 100 print(f"\nSTATISTICS:") print(f" Profitable: {wins} ({win_rate:.2f}%)") print(f" Losing: {losses} ({100 - win_rate:.2f}%)") if wins > 0: avg_win = np.mean([t["profit_usd"] for t in trades if t["profit_usd"] > 0]) print(f" Average profit: ${avg_win:.2f}") if losses > 0: avg_loss = np.mean([t["profit_usd"] for t in trades if t["profit_usd"] < 0]) print(f" Average loss: ${avg_loss:.2f}") if wins > 0 and losses > 0: total_profit = sum(t["profit_usd"] for t in trades if t["profit_usd"] > 0) total_loss = abs( sum(t["profit_usd"] for t in trades if t["profit_usd"] < 0) ) profit_factor = total_profit / total_loss if total_loss > 0 else 0 print(f" Profit Factor: {profit_factor:.2f}") max_dd = calculate_max_drawdown(np.array(equity_hist)) print(f" Max. drawdown: {max_dd:.2f}%") # Quantum statistics print(f"\nQUANTUM ANALYSIS:") low_entropy_trades = [t for t in trades if t["quantum_entropy"] < 2.5] high_entropy_trades = [t for t in trades if t["quantum_entropy"] > 4.5] if low_entropy_trades: low_entropy_wins = sum(1 for t in low_entropy_trades if t["correct"]) print( f" Low entropy (<2.5): {len(low_entropy_trades)} trades, " f"win rate {low_entropy_wins/len(low_entropy_trades)*100:.1f}%" ) if high_entropy_trades: high_entropy_wins = sum(1 for t in high_entropy_trades if t["correct"]) print( f" High entropy (>4.5): {len(high_entropy_trades)} trades, " f"win rate {high_entropy_wins/len(high_entropy_trades)*100:.1f}%" ) # LLM corrections if use_llm: corrections = [ t for t in trades if abs(t["confidence"] - t["catboost_confidence"]) > 3 ] if corrections: correct_corrections = sum(1 for t in corrections if t["correct"]) print(f"\nLLM CORRECTIONS:") print(f" Total corrections (>3%): {len(corrections)}") print( f" Successful: {correct_corrections} ({correct_corrections/len(corrections)*100:.1f}%)" ) best_trade = max(trades, key=lambda x: x["profit_usd"]) worst_trade = min(trades, key=lambda x: x["profit_usd"]) print(f"\nBEST TRADE:") print( f" {best_trade['time'].strftime('%Y-%m-%d %H:%M')} | {best_trade['symbol']} " f"{best_trade['direction']} | ${best_trade['profit_usd']:+.2f}" ) print(f"\nWORST TRADE:") print( f" {worst_trade['time'].strftime('%Y-%m-%d %H:%M')} | {worst_trade['symbol']} " f"{worst_trade['direction']} | ${worst_trade['profit_usd']:+.2f}" ) # Chart if len(equity_hist) > 1: print(f"\n{'='*80}") print("Plotting the equity chart...") plot_results(balance_hist, equity_hist, slots) # Save detailed report trades_df = pd.DataFrame(trades) report_path = ( f"logs/backtest_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" ) trades_df.to_csv(report_path, index=False) print(f"\n✓ Detailed report saved: {report_path}") print(f"\n{'='*80}") print("BACKTEST COMPLETED") print(f"{'='*80}\n") # ====================== DATA LOADING ====================== def load_mt5_data(days: int = 180) -> Dict[str, pd.DataFrame]: """Loading real MT5 data""" if not mt5 or not mt5.initialize(): print("⚠️ MT5 is unavailable") return {} end = datetime.now() start = end - timedelta(days=days) data = {} print(f"\nLoading MT5 data for {days} days...") for symbol in SYMBOLS: rates = mt5.copy_rates_range(symbol, TIMEFRAME, start, end) if rates is None or len(rates) < LOOKBACK + PREDICTION_HORIZON: print(f" ⚠️ {symbol}: Insufficient data") continue df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s") df.set_index("time", inplace=True) data[symbol] = df print(f" ✓ {symbol}: {len(df)} bars") mt5.shutdown() return data # ====================== MAIN MENU ====================== def main(): """Main Menu""" print(f"\n{'='*80}") print(f" QUANTUM TRADER FUSION — Qiskit + CatBoost + LLM") print(f" Version: December 9, 2025 (Full Integration)") print(f"{'='*80}\n") print(f"MODES:") print(f"-" * 80) print(f"1 → Train CatBoost with quantum features") print(f"2 → Generate a hybrid dataset (CatBoost + Quantum)") print(f"3 → Fine-tuning an LLM with CatBoost predictions") print(f"4 → Hybrid system backtest") print(f"5 → Live trading (MT5)") print(f"6 → FULL CYCLE (all together)") print(f"-" * 80) choice = input("\nSelect a mode (1-6): ").strip() if choice == "1": # Mode 1: CatBoost training data = load_mt5_data(180) if not data: print("❌ No data for training") return quantum_encoder = QuantumEncoder(N_QUBITS, N_SHOTS) model = train_catboost_model(data, quantum_encoder) elif choice == "2": # Mode 2: Dataset generation data = load_mt5_data(180) if not data: print("❌ No data") return # Loading the CatBoost model if os.path.exists("models/catboost_quantum.cbm"): print("Loading the CatBoost model...") model = CatBoostClassifier() model.load_model("models/catboost_quantum.cbm") else: print("❌ CatBoost model not found; train it first (mode 1)") return quantum_encoder = QuantumEncoder(N_QUBITS, N_SHOTS) dataset = generate_hybrid_dataset( data, model, quantum_encoder, FINETUNE_SAMPLES ) save_dataset(dataset, "dataset/quantum_fusion_data.jsonl") elif choice == "3": # Mode 3: LLM fine-tuning dataset_path = "dataset/quantum_fusion_data.jsonl" if not os.path.exists(dataset_path): print(f"❌ Dataset not found: {dataset_path}") print("First, generate the dataset (mode 2)") return finetune_llm_with_catboost(dataset_path) elif choice == "4": # Mode 4: Backtest backtest() elif choice == "5": print("⚠️ Live trading is under development — use the original script") elif choice == "6": # Mode 6: FULL CYCLE print(f"\n{'='*80}") print(f"FULL CYCLE: QUANTUM FUSION") print(f"{'='*80}\n") print("This process will take 2–3 hours:") print("1. Loading MT5 data (180 days)") print("2. Quantum encoding (~60 min)") print("3. CatBoost training (~15 min)") print("4. Generating the dataset (~45 min)") print("5. LLM fine-tuning (~20 min)") confirm = input("\nContinue? (YES): ").strip() if confirm != "YES": print("Canceled") return # Step 1: Loading data print(f"\n{'='*80}") print("STEP 1/5: LOADING MT5 DATA") print(f"{'='*80}") data = load_mt5_data(180) if not data: print("❌ Unable to load data") return # Steps 2–3: CatBoost training print(f"\n{'='*80}") print("STEP 2-3/5: QUANTUM ENCODING + CATBOOST TRAINING") print(f"{'='*80}") quantum_encoder = QuantumEncoder(N_QUBITS, N_SHOTS) model = train_catboost_model(data, quantum_encoder) # Step 4: Dataset generation print(f"\n{'='*80}") print("STEP 4/5: GENERATING A HYBRID DATASET") print(f"{'='*80}") dataset = generate_hybrid_dataset( data, model, quantum_encoder, FINETUNE_SAMPLES ) dataset_path = save_dataset(dataset, "dataset/quantum_fusion_data.jsonl") # Step 5: Fine-tuning the LLM print(f"\n{'='*80}") print("STEP 5/5: LLM FINE-TUNING") print(f"{'='*80}") finetune_llm_with_catboost(dataset_path) print(f"\n{'='*80}") print("🎉 THE FULL CYCLE IS COMPLETE!") print(f"{'='*80}") print("✓ The CatBoost model has been trained using quantum features") print("✓ The LLM has been fine-tuned with CatBoost predictions") print("✓ The system is ready for use") print(f"\nModel: {MODEL_NAME}") print(f"CatBoost: models/catboost_quantum.cbm") print(f"Dataset: {dataset_path}") else: print("❌ Invalid selection") if __name__ == "__main__": main()