codes/classification/advanced_knn_classifier.py

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()