import numpy as np import pandas as pd import os import sys import time from datetime import datetime from sklearn.svm import SVC from sklearn.model_selection import GridSearchCV, train_test_split, KFold from sklearn.preprocessing import StandardScaler from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score, accuracy_score, precision_score, recall_score, f1_score from sklearn.inspection import permutation_importance from sklearn.decomposition import PCA import json import gc # Import bibliotek kwantowych from qiskit import Aer from qiskit.circuit.library import ZZFeatureMap from qiskit_machine_learning.kernels import QuantumKernel from qiskit_machine_learning.algorithms import QSVC # Import funkcji z głównego modułu import qsvm def run_experiment(): """ Eksperyment 1: ZZ1 i ZZ2 Feature Maps Testuje klasyczny SVM i kwantowy SVM z mapami cech ZZ1 i ZZ2 """ print("======= EKSPERYMENT 1: ZZ1 i ZZ2 FEATURE MAPS =======") # Konfiguracja eksperymentu FEATURE_MAPS = { 'ZZ1': {'reps': 1, 'enabled': True}, 'ZZ2': {'reps': 2, 'enabled': True} } # Dla każdego pliku danych for data_file in qsvm.DATA_FILES: if not os.path.exists(data_file): print(f"Pominięto {data_file} - plik nie istnieje") continue print(f"\n======= PRZETWARZANIE PLIKU: {data_file} =======") # Utwórz nazwę pliku wyjściowego file_base_name = os.path.basename(data_file).split('.')[0] output_file = os.path.join(qsvm.OUTPUT_DIR, f'wyniki_zz_{file_base_name}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt') # Utwórz plik cache cache_file = os.path.join(qsvm.OUTPUT_DIR, f'qsvm_zz_cache_{file_base_name}.json') # Przekierowanie wyjścia logger = qsvm.Logger(output_file) sys.stdout = logger try: # Przygotowanie danych data_dict = qsvm.prepare_data(data_file) X_train = data_dict['X_train'] X_test = data_dict['X_test'] X_train_reduced = data_dict['X_train_reduced'] X_test_reduced = data_dict['X_test_reduced'] y_train = data_dict['y_train'] y_test = data_dict['y_test'] data_processed = data_dict['data_processed'] # Inicjalizacja backendu ibm_service, ibm_backend, ibm_success = qsvm.initialize_ibm_quantum() # ----------------- KLASYCZNY SVM ----------------- if qsvm.RUN_CLASSIC_SVM: print("\n======= KLASYCZNY SVM (BASELINE) =======") start_time_classic = time.time() # Trenowanie modelu grid = GridSearchCV(SVC(), qsvm.SVM_PARAM_GRID, cv=qsvm.SVM_CV, scoring='accuracy') grid.fit(X_train, y_train) print("Najlepsze parametry klasycznego SVM:", grid.best_params_) print("Dokładność klasycznego SVM:", grid.best_score_) # Ewaluacja modelu classic_pred = grid.predict(X_test) print("Raport klasyfikacji (klasyczny SVM):") print(classification_report(y_test, classic_pred, zero_division=0)) # Zapisz szczegółowe metryki classic_metrics = qsvm.save_metrics(y_test, classic_pred, "Klasyczny SVM") end_time_classic = time.time() classic_svm_time = end_time_classic - start_time_classic print(f"\nCzas trenowania i ewaluacji klasycznego SVM: {classic_svm_time:.2f} sekund") else: print("\n======= KLASYCZNY SVM (BASELINE) - POMINIĘTY =======") classic_svm_time = 0 classic_metrics = None # ----------------- KWANTOWY SVM ----------------- if qsvm.RUN_QUANTUM_SVM: print("\n======= KWANTOWY SVM Z ZZ FEATURE MAPS =======") start_time_quantum = time.time() # Wczytaj cache cache = qsvm.load_results_cache(cache_file) quantum_results = cache.get('quantum_results', []) # Tworzenie map cech feature_maps = [] feature_dimension = X_train_reduced.shape[1] for name, config in FEATURE_MAPS.items(): if config['enabled']: feature_map = ZZFeatureMap(feature_dimension=feature_dimension, reps=config['reps']) feature_maps.append({'name': name, 'map': feature_map}) print(f"Testowanie {len(feature_maps)} map cech: {[fm['name'] for fm in feature_maps]}") # Testowanie każdej mapy cech for fm in feature_maps: for C in qsvm.C_VALUES: # Sprawdź cache already_tested = False for name, c_val, _ in quantum_results: if name == fm['name'] and c_val == C: already_tested = True break if already_tested: print(f"Pomijanie już przetestowanej kombinacji: {fm['name']}, C={C}") continue fm_start_time = time.time() try: print(f"Testowanie {fm['name']} z C={C}...") # Debugowanie danych print(f" Wymiary danych: X_train_reduced {X_train_reduced.shape}") print(f" Sprawdzenie NaN: {np.isnan(X_train_reduced).sum()}") print(f" Sprawdzenie inf: {np.isinf(X_train_reduced).sum()}") print(f" Zakres danych: [{X_train_reduced.min():.4f}, {X_train_reduced.max():.4f}]") # Utworzenie quantum kernel z debugowaniem quantum_kernel = QuantumKernel( feature_map=fm['map'], quantum_instance=ibm_backend ) # Test quantum kernel print(f" Testowanie quantum kernel...") try: test_kernel = quantum_kernel.evaluate(X_train_reduced[:2], X_train_reduced[:2]) print(f" Test kernel shape: {test_kernel.shape}") print(f" Test kernel range: [{test_kernel.min():.4f}, {test_kernel.max():.4f}]") if np.isnan(test_kernel).any() or np.isinf(test_kernel).any(): print(f" BŁĄD: Kernel zawiera NaN lub inf!") continue except Exception as e: print(f" BŁĄD testowania kernel: {str(e)}") continue # Utworzenie SVM z niestandardowym jądrem i debugowaniem def custom_kernel(X, Y): try: kernel_matrix = quantum_kernel.evaluate(X, Y) # Sprawdź czy kernel jest poprawny if np.isnan(kernel_matrix).any() or np.isinf(kernel_matrix).any(): print(f" BŁĄD: Kernel matrix zawiera NaN lub inf!") return np.eye(len(X), len(Y)) # Fallback return kernel_matrix except Exception as e: print(f" BŁĄD kernel evaluation: {str(e)}") return np.eye(len(X), len(Y)) # Fallback qsvm_model = SVC(kernel=custom_kernel, C=C, random_state=qsvm.RANDOM_STATE) # Walidacja krzyżowa z debugowaniem cv_start_time = time.time() scores = [] kf = KFold(n_splits=qsvm.QSVM_CV, shuffle=True, random_state=qsvm.RANDOM_STATE) for fold, (train_idx, val_idx) in enumerate(kf.split(X_train_reduced)): X_cv_train, X_cv_val = X_train_reduced[train_idx], X_train_reduced[val_idx] y_cv_train, y_cv_val = y_train.iloc[train_idx], y_train.iloc[val_idx] print(f" Fold {fold+1}/{qsvm.QSVM_CV}: train {X_cv_train.shape}, val {X_cv_val.shape}") try: qsvm_model.fit(X_cv_train, y_cv_train) score = qsvm_model.score(X_cv_val, y_cv_val) scores.append(score) print(f" Fold {fold+1} score: {score:.4f}") except Exception as e: print(f" BŁĄD fold {fold+1}: {str(e)}") scores.append(0.0) # Fallback if len(scores) > 0: mean_score = np.mean(scores) std_score = np.std(scores) print(f" Wszystkie scores: {scores}") print(f" Mean score: {mean_score:.4f} ± {std_score:.4f}") else: mean_score = 0.0 print(f" BŁĄD: Brak poprawnych scores!") cv_end_time = time.time() cv_time = cv_end_time - cv_start_time quantum_results.append((fm['name'], C, mean_score)) fm_end_time = time.time() fm_time = fm_end_time - fm_start_time print(f"Dokładność kwantowego SVM z {fm['name']}, C={C}: {mean_score:.4f} (czas: {fm_time:.2f} s)") # Zapisz wyniki pośrednie cache['quantum_results'] = quantum_results qsvm.save_results_cache(cache, cache_file) except Exception as e: print(f"BŁĄD dla {fm['name']}, C={C}: {str(e)}") # Dodaj fallback wynik quantum_results.append((fm['name'], C, 0.0)) cache['quantum_results'] = quantum_results qsvm.save_results_cache(cache, cache_file) continue # Znajdź najlepszy model kwantowy if quantum_results: best_qsvm = max(quantum_results, key=lambda x: x[2]) print(f"\nNajlepszy kwantowy SVM: {best_qsvm[0]} z C={best_qsvm[1]}, dokładność: {best_qsvm[2]:.4f}") # Ewaluacja najlepszego modelu best_feature_map = None for fm in feature_maps: if fm['name'] == best_qsvm[0]: best_feature_map = fm['map'] break if best_feature_map: quantum_kernel_best = QuantumKernel( feature_map=best_feature_map, quantum_instance=ibm_backend ) qsvm_best = SVC(kernel=quantum_kernel_best.evaluate, C=best_qsvm[1]) qsvm_best.fit(X_train_reduced, y_train) quantum_pred = qsvm_best.predict(X_test_reduced) print("Raport klasyfikacji (najlepszy kwantowy SVM):") print(classification_report(y_test, quantum_pred, zero_division=0)) quantum_metrics = qsvm.save_metrics(y_test, quantum_pred, f"Kwantowy SVM {best_qsvm[0]}") else: print("Nie udało się wytrenować żadnego modelu kwantowego.") quantum_metrics = None end_time_quantum = time.time() quantum_svm_time = end_time_quantum - start_time_quantum print(f"\nCałkowity czas dla kwantowego SVM: {quantum_svm_time:.2f} sekund") else: print("\n======= KWANTOWY SVM - POMINIĘTY =======") quantum_svm_time = 0 quantum_metrics = None # ----------------- ANALIZA WYNIKÓW ----------------- print("\n======= PORÓWNANIE WYNIKÓW =======") if classic_metrics: print(f"Klasyczny SVM: {classic_metrics['accuracy']:.4f}") if quantum_metrics: print(f"Kwantowy SVM: {quantum_metrics['accuracy']:.4f}") # Analiza znaczenia cech (tylko dla klasycznego SVM) if qsvm.RUN_CLASSIC_SVM and classic_metrics: print("\n======= ANALIZA ZNACZENIA CECH =======") importance_start_time = time.time() result = permutation_importance(grid.best_estimator_, X_test, y_test, n_repeats=10, random_state=qsvm.RANDOM_STATE) important_features = [] feature_columns = list(data_processed.columns) for i in range(len(feature_columns)): if result.importances_mean[i] > qsvm.IMPORTANCE_THRESHOLD: important_features.append((feature_columns[i], result.importances_mean[i])) print("Najważniejsze cechy dla klasyfikacji:") for feature, importance in sorted(important_features, key=lambda x: x[1], reverse=True): print(f" {feature}: {importance:.4f}") importance_end_time = time.time() importance_time = importance_end_time - importance_start_time print(f"\nCzas analizy znaczenia cech: {importance_time:.2f} sekund") # Podsumowanie print("\n======= PODSUMOWANIE EKSPERYMENTU ZZ =======") print(f"Data i czas zakończenia: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") total_time = time.time() - data_dict['preparation_time'] print(f"Całkowity czas eksperymentu: {total_time:.2f} sekund") except Exception as e: print(f"BŁĄD podczas przetwarzania {data_file}: {str(e)}") finally: # Zamknięcie pliku wyjściowego logger.close() sys.stdout = logger.terminal # Czyszczenie pamięci gc.collect() print("\n======= EKSPERYMENT 1 ZAKOŃCZONY =======") if __name__ == "__main__": run_experiment()