import numpy as np import pandas as pd import sklearn from sklearn.neighbors import KNeighborsClassifier from sklearn.model_selection import cross_val_score, GridSearchCV, StratifiedKFold from sklearn.metrics import classification_report, confusion_matrix, accuracy_score from sklearn.preprocessing import StandardScaler, MinMaxScaler, normalize, LabelEncoder from sklearn.ensemble import VotingClassifier from sklearn.decomposition import PCA from sklearn.feature_selection import SelectKBest, f_classif, mutual_info_classif import matplotlib.pyplot as plt import seaborn as sns import joblib from scipy.spatial.distance import cosine import warnings from sklearn.preprocessing import PolynomialFeatures warnings.filterwarnings('ignore') def load_features_from_extraction(output_dir): """ Loads feature files produced by `extract_features_hierarchical.py`. No need to read labels separately - labels are already stored in `fused_features.csv`. """ from pathlib import Path output_dir = Path(output_dir) features_path = output_dir / 'fused_features.npy' if not features_path.exists(): raise FileNotFoundError(f"Feature file not found: {features_path}") print(f"Loading features from {features_path}") features = np.load(features_path) print(f"Loaded {len(features)} features with shape {features.shape}") csv_path = output_dir / 'fused_features.csv' if not csv_path.exists(): raise FileNotFoundError(f"CSV file not found: {csv_path}") print(f"Loading labels from {csv_path}") df = pd.read_csv(csv_path) if 'label' in df.columns: labels = df['label'].values valid_mask = labels != -1 elif 'gleason_grade' in df.columns: from sklearn.preprocessing import LabelEncoder label_encoder = LabelEncoder() grades = df['gleason_grade'].values valid_mask = (grades != 'unknown') & (pd.notna(grades)) valid_grades = grades[valid_mask] labels = np.full(len(grades), -1) if len(valid_grades) > 0: labels[valid_mask] = label_encoder.fit_transform(valid_grades) else: raise ValueError("Could not find the 'label' or 'gleason_grade' column in the CSV!") grades = df['gleason_grade'].values if 'gleason_grade' in df.columns else None patient_ids = df['patient_id'].values if 'patient_id' in df.columns else None features = features[valid_mask] labels = labels[valid_mask] grades = grades[valid_mask] if grades is not None else None patient_ids = patient_ids[valid_mask] if patient_ids is not None else None print(f"Total matched samples: {len(features)}") print(f"Unique labels: {np.unique(labels)}") if grades is not None: print(f"Unique Gleason grades: {np.unique(grades)}") return features, labels, grades, patient_ids def advanced_preprocessing(X_train, X_test, y_train, method='ensemble'): """ Advanced preprocessing methods. """ print(f"Applying advanced preprocessing: {method}") if method == 'ensemble': # Ensemble preprocessing: combine multiple methods # 1. Standard scaling scaler1 = StandardScaler() X_train_scaled1 = scaler1.fit_transform(X_train) X_test_scaled1 = scaler1.transform(X_test) # 2. MinMax scaling scaler2 = MinMaxScaler() X_train_scaled2 = scaler2.fit_transform(X_train) X_test_scaled2 = scaler2.transform(X_test) # 3. L2 normalization (for cosine similarity) X_train_norm = normalize(X_train, norm='l2') X_test_norm = normalize(X_test, norm='l2') # 4. Feature selection selector = SelectKBest(score_func=mutual_info_classif, k=400) X_train_selected = selector.fit_transform(X_train_scaled1, y_train) X_test_selected = selector.transform(X_test_scaled1) # 5. PCA (keep 90% of the variance) pca = PCA(n_components=0.90) X_train_pca = pca.fit_transform(X_train_scaled1) X_test_pca = pca.transform(X_test_scaled1) # Polynomial features poly = PolynomialFeatures(degree=2, include_bias=False) X_poly = poly.fit_transform(X_train_selected) return { 'standard': (X_train_scaled1, X_test_scaled1), 'minmax': (X_train_scaled2, X_test_scaled2), 'normalized': (X_train_norm, X_test_norm), 'selected': (X_train_selected, X_test_selected), 'pca': (X_train_pca, X_test_pca), 'poly': X_poly } elif method == 'optimal': # Find the best preprocessing method methods = ['standard', 'minmax', 'normalized', 'pca', 'feature_selection'] best_score = 0 best_method = None best_data = None for m in methods: if m == 'standard': scaler = StandardScaler() X_train_processed = scaler.fit_transform(X_train) X_test_processed = scaler.transform(X_test) elif m == 'minmax': scaler = MinMaxScaler() X_train_processed = scaler.fit_transform(X_train) X_test_processed = scaler.transform(X_test) elif m == 'normalized': X_train_processed = normalize(X_train, norm='l2') X_test_processed = normalize(X_test, norm='l2') elif m == 'pca': scaler = StandardScaler() X_train_std = scaler.fit_transform(X_train) X_test_std = scaler.transform(X_test) pca = PCA(n_components=0.95) X_train_processed = pca.fit_transform(X_train_std) X_test_processed = pca.transform(X_test_std) elif m == 'feature_selection': scaler = StandardScaler() X_train_std = scaler.fit_transform(X_train) X_test_std = scaler.transform(X_test) selector = SelectKBest(score_func=mutual_info_classif, k=300) X_train_processed = selector.fit_transform(X_train_std, y_train) X_test_processed = selector.transform(X_test_std) # Test with KNN knn = KNeighborsClassifier( n_neighbors=15, weights='distance', # Weight by distance metric='cosine' ) skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) scores = cross_val_score(knn, X_train_processed, y_train, cv=skf, scoring='accuracy') mean_score = scores.mean() print(f" {m}: {mean_score:.4f}") if mean_score > best_score: best_score = mean_score best_method = m best_data = (X_train_processed, X_test_processed) print(f"Best preprocessing method: {best_method} (score: {best_score:.4f})") return best_data def create_ensemble_knn(X_train, X_test, y_train, y_test): """ Create an ensemble KNN classifier. """ print("Creating ensemble KNN classifier...") # Train separate models for different k values k_values = [3, 5, 7, 9, 11, 13, 15] knn_models = [] for k in k_values: knn = KNeighborsClassifier(n_neighbors=k, weights='distance', metric='cosine') knn_models.append(knn) # Combine using voting ensemble = VotingClassifier(estimators=knn_models, voting='soft') return ensemble def optimize_knn_parameters(X_train, y_train): """ KNN parametrelerini optimize et """ print("Optimizing KNN parameters...") # Parametre grid'i param_grid = { 'n_neighbors': [3, 5, 7, 9, 11, 13, 15, 17, 19, 21], 'weights': ['uniform', 'distance'], 'metric': ['cosine', 'euclidean', 'manhattan', 'minkowski'], 'p': [1, 2, 3] } # Grid search knn = KNeighborsClassifier() grid_search = GridSearchCV( knn, param_grid, cv=3, scoring='accuracy', n_jobs=-1, verbose=1 ) grid_search.fit(X_train, y_train) print(f"Best parameters: {grid_search.best_params_}") print(f"Best CV score: {grid_search.best_score_:.4f}") return grid_search.best_estimator_ def train_advanced_knn(X_train, X_test, y_train, y_test): """ Advanced KNN training. """ print("=== ADVANCED KNN TRAINING ===") # 1. Preprocessing preprocessed_data = advanced_preprocessing(X_train, X_test, y_train, method='ensemble') # 2. Train a separate model for each preprocessing method models = {} results = {} for method, (X_train_processed, X_test_processed) in preprocessed_data.items(): print(f"\n--- Training with {method} preprocessing ---") # Optimized KNN best_knn = optimize_knn_parameters(X_train_processed, y_train) # Final model training best_knn.fit(X_train_processed, y_train) # Test predictions y_pred = best_knn.predict(X_test_processed) accuracy = accuracy_score(y_test, y_pred) print(f"Test accuracy: {accuracy:.4f}") print("Classification Report:") print(classification_report(y_test, y_pred)) models[method] = best_knn results[method] = { 'accuracy': accuracy, 'predictions': y_pred, 'model': best_knn } # 3. Build the ensemble model print("\n--- Creating Ensemble Model ---") ensemble = create_ensemble_knn(X_train, X_test, y_train, y_test) # Train the ensemble using the best preprocessing best_method = max(results.keys(), key=lambda x: results[x]['accuracy']) X_train_best, X_test_best = preprocessed_data[best_method] ensemble.fit(X_train_best, y_train) y_pred_ensemble = ensemble.predict(X_test_best) accuracy_ensemble = accuracy_score(y_test, y_pred_ensemble) print(f"Ensemble test accuracy: {accuracy_ensemble:.4f}") print("Ensemble Classification Report:") print(classification_report(y_test, y_pred_ensemble)) results['ensemble'] = { 'accuracy': accuracy_ensemble, 'predictions': y_pred_ensemble, 'model': ensemble } return results, preprocessed_data def augment_features(X, noise_factor=0.01): X_augmented = X + np.random.normal(0, noise_factor, X.shape) return X_augmented def weighted_voting(predictions, weights): weighted_pred = np.average(predictions, weights=weights, axis=0) return np.argmax(weighted_pred, axis=1) def main(): # Load train features (CSV okumaya gerek yok) print("=== LOADING TRAIN FEATURES ===") X_train, y_train, train_grades, train_cases = load_features_from_extraction( 'feature_extraction/extractedfusedfeatures_train' # Train feature extraction output folder ) # Load test features print("\n=== LOADING TEST FEATURES ===") X_test, y_test, test_grades, test_cases = load_features_from_extraction( 'feature_extraction/extractedfusedfeatures_test' # Test feature extraction output folder ) # Remove 2+4 class if present print("\n=== FILTERING DATA ===") if train_grades is not None: train_mask = train_grades != '2+4' X_train = X_train[train_mask] y_train = y_train[train_mask] train_grades = train_grades[train_mask] train_cases = train_cases[train_mask] if train_cases is not None else None if test_grades is not None: test_mask = test_grades != '2+4' X_test = X_test[test_mask] y_test = y_test[test_mask] test_grades = test_grades[test_mask] test_cases = test_cases[test_mask] if test_cases is not None else None print(f"Final training set: {X_train.shape[0]} samples") print(f"Final test set: {X_test.shape[0]} samples") print(f"Training classes: {np.unique(y_train)}") print(f"Test classes: {np.unique(y_test)}") # Select the most important features using mutual information mi_scores = mutual_info_classif(X_train, y_train) top_features = np.argsort(mi_scores)[-400:] # Top 400 features X_selected = X_train[:, top_features] # Train advanced KNN results, preprocessed_data = train_advanced_knn(X_selected, X_test, y_train, y_test) # Find the best method/model best_method = max(results.keys(), key=lambda x: results[x]['accuracy']) best_result = results[best_method] print(f"\n{'='*60}") print(f"BEST MODEL: {best_method}") print(f"Test Accuracy: {best_result['accuracy']:.4f}") print(f"{'='*60}") # Confusion matrix cm = confusion_matrix(y_test, best_result['predictions']) plt.figure(figsize=(12, 10)) sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=np.unique(y_test), yticklabels=np.unique(y_test)) plt.xlabel('Predicted') plt.ylabel('Actual') plt.title(f'Confusion Matrix - Advanced KNN ({best_method}) - 90 Epoch') plt.savefig('confusion_matrix_advanced_knn_90ep.png', dpi=300, bbox_inches='tight') plt.close() # Save best model joblib.dump(best_result['model'], f'advanced_knn_model_90ep_{best_method}.joblib') # Convert labels to grades (if numeric) if test_grades is not None: true_grades = test_grades if isinstance(best_result['predictions'][0], (int, np.integer)): from sklearn.preprocessing import LabelEncoder le = LabelEncoder() le.fit(train_grades if train_grades is not None else test_grades) pred_grades = le.inverse_transform(best_result['predictions']) else: pred_grades = best_result['predictions'] else: true_grades = y_test pred_grades = best_result['predictions'] # Save results results_df = pd.DataFrame({ 'case_id': test_cases if test_cases is not None else range(len(y_test)), 'true_grade': true_grades, 'predicted_grade': pred_grades }) results_df.to_csv('advanced_knn_results_90ep.csv', index=False) # Model comparison print(f"\n{'='*60}") print("MODEL COMPARISON") print(f"{'='*60}") for method, result in sorted(results.items(), key=lambda x: x[1]['accuracy'], reverse=True): print(f"{method}: {result['accuracy']:.4f}") # Save comparison comparison_df = pd.DataFrame({ 'Method': list(results.keys()), 'Accuracy': [results[name]['accuracy'] for name in results.keys()] }) comparison_df = comparison_df.sort_values('Accuracy', ascending=False) comparison_df.to_csv('advanced_knn_comparison_90ep.csv', index=False) print(f"\nResults saved:") print(f"- Best model: advanced_knn_model_90ep_{best_method}.joblib") print(f"- Results: advanced_knn_results_90ep.csv") print(f"- Comparison: advanced_knn_comparison_90ep.csv") print(f"- Confusion matrix: confusion_matrix_advanced_knn_90ep.png") if __name__ == "__main__": main()