SVM.py

import numpy as np
import pandas as pd
import os
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score
from sklearn.model_selection import GridSearchCV
import matplotlib.pyplot as plt
import seaborn as sns
import joblib
import time
import matplotlib

matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei']
matplotlib.rcParams['axes.unicode_minus'] = False

def load_data(train_dir='datasets/train', test_dir='datasets/test'):

    features_path = os.path.join(os.path.dirname(train_dir), 'features.txt')
    features = pd.read_csv(features_path, sep=' ', header=None, names=['index', 'feature_name'])

    feature_names = []
    seen_features = {}
    for feature in features['feature_name'].values:
        if feature in seen_features:
            seen_features[feature] += 1
            feature_names.append(f"{feature}_{seen_features[feature]}")
        else:
            seen_features[feature] = 0
            feature_names.append(feature)

    activity_path = os.path.join(os.path.dirname(train_dir), 'activity_labels.txt')
    activity_labels = pd.read_csv(activity_path, sep=' ', header=None, names=['id', 'activity'])

    X_train_path = os.path.join(train_dir, 'X_train.txt')
    y_train_path = os.path.join(train_dir, 'y_train.txt')
    subject_train_path = os.path.join(train_dir, 'subject_train.txt')
    
    X_train = pd.read_csv(X_train_path, sep='\s+', header=None, names=feature_names)
    y_train = pd.read_csv(y_train_path, header=None, names=['activity_id'])
    subject_train = pd.read_csv(subject_train_path, header=None, names=['subject_id'])

    X_test_path = os.path.join(test_dir, 'X_test.txt')
    y_test_path = os.path.join(test_dir, 'y_test.txt')
    subject_test_path = os.path.join(test_dir, 'subject_test.txt')
    
    X_test = pd.read_csv(X_test_path, sep='\s+', header=None, names=feature_names)
    y_test = pd.read_csv(y_test_path, header=None, names=['activity_id'])
    subject_test = pd.read_csv(subject_test_path, header=None, names=['subject_id'])
    
    print(f"Training set shape: {X_train.shape}, Test Set Shape: {X_test.shape}")
    print(f"Number of subjects in the training set: {subject_train['subject_id'].nunique()}, Number of subjects in the test set: {subject_test['subject_id'].nunique()}")
    
    return X_train, y_train['activity_id'], X_test, y_test['activity_id'], subject_train, subject_test, features, activity_labels

def feature_selection(X_train, X_test, features):
    mean_features = [i for i, name in enumerate(features['feature_name']) if 'mean()' in name]
    std_features = [i for i, name in enumerate(features['feature_name']) if 'std()' in name]
    selected_features = sorted(mean_features + std_features)

    X_train_selected = X_train.iloc[:, selected_features]
    X_test_selected = X_test.iloc[:, selected_features]
    
    print(f"Raw feature count: {X_train.shape[1]}, Number of post-selection features: {X_train_selected.shape[1]}")
    
    return X_train_selected, X_test_selected

def train_svm_model(X_train, y_train, params=None, tune_hyperparams=False, output_dir='train_pic'):
    start_time = time.time()

    os.makedirs(output_dir, exist_ok=True)

    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)

    if params is None:
        params = {
            'C': 1.0,            
            'kernel': 'rbf',
            'degree': 3,
            'gamma': 'scale',
            'coef0': 0.0,
            'shrinking': True,
            'probability': True,
            'tol': 1e-3, 
            'cache_size': 200,
            'class_weight': None,
            'verbose': False,
            'max_iter': -1,
            'decision_function_shape': 'ovr',
            'break_ties': False,
            'random_state': 42
        }
    
    if tune_hyperparams:
        param_grid = {
            'C': [0.1, 1, 10, 100],
            'gamma': ['scale', 'auto', 0.1, 0.01],
            'kernel': ['rbf', 'linear', 'poly']
        }
        
        grid_search = GridSearchCV(
            SVC(), 
            param_grid=param_grid, 
            cv=5, 
            scoring='accuracy', 
            verbose=1,
            n_jobs=-1
        )
        
        grid_search.fit(X_train_scaled, y_train)
        
        best_params = grid_search.best_params_
        print(f"optimal parameter: {best_params}")
        
        model = SVC(**best_params)
    else:
        model = SVC(**params)
        print(f"SVM parameters used: {params}")
    
    model.fit(X_train_scaled, y_train)
    
    training_time = time.time() - start_time
    print(f"Model training completed in: {training_time:.2f} s")
    
    joblib.dump(scaler, os.path.join(output_dir, 'svm_scaler.pkl'))
    joblib.dump(model, os.path.join(output_dir, 'svm_model.pkl'))
    
    return model, scaler

def evaluate_model(model, scaler, X_test, y_test, activity_labels, output_dir='train_pic'):
    
    os.makedirs(output_dir, exist_ok=True)
    X_test_scaled = scaler.transform(X_test)
    
    y_pred = model.predict(X_test_scaled)
    
    accuracy = accuracy_score(y_test, y_pred)
    print(f"Accuracy: {accuracy:.4f}")
    
    class_names = activity_labels['activity'].tolist()
    print("\nClassification report:")
    print(classification_report(y_test, y_pred, target_names=class_names))
    
    cm = confusion_matrix(y_test, y_pred)
    
    plt.figure(figsize=(10, 8))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
                xticklabels=class_names, 
                yticklabels=class_names)
    plt.xlabel('Predicted')
    plt.ylabel('Actual')
    plt.title('Confusion Matrix')
    plt.tight_layout()
    plt.savefig(os.path.join(output_dir, 'confusion_matrix.png'))
    print(f"CM saved to '{output_dir}/confusion_matrix.png'")
    
    print("\nAccuracy by category of activity:")
    for i, activity_name in enumerate(class_names):
        class_indices = np.where(y_test == i+1)[0]
        if len(class_indices) > 0:
            class_accuracy = accuracy_score(y_test.iloc[class_indices], y_pred[class_indices])
            print(f"{activity_name}: {class_accuracy:.4f}")
    
    return accuracy, y_pred

def analyze_subject_performance(model, scaler, X_test, y_test, subject_test, activity_labels, output_dir='train_pic'):
    os.makedirs(output_dir, exist_ok=True)
    
    X_test_scaled = scaler.transform(X_test)
    
    y_pred = model.predict(X_test_scaled)
    
    results_df = pd.DataFrame({
        'subject_id': subject_test['subject_id'].values,
        'true_activity': y_test.values,
        'predicted_activity': y_pred
    })
    
    subject_accuracy = {}
    for subject_id in results_df['subject_id'].unique():
        subject_data = results_df[results_df['subject_id'] == subject_id]
        accuracy = accuracy_score(subject_data['true_activity'], subject_data['predicted_activity'])
        subject_accuracy[subject_id] = accuracy
    
    plt.figure(figsize=(12, 6))
    subjects = list(subject_accuracy.keys())
    accuracies = list(subject_accuracy.values())
    
    sorted_indices = np.argsort(accuracies)
    sorted_subjects = [subjects[i] for i in sorted_indices]
    sorted_accuracies = [accuracies[i] for i in sorted_indices]
    
    plt.bar(range(len(sorted_subjects)), sorted_accuracies)
    plt.xlabel('Subject ID')
    plt.ylabel('Accuracy')
    plt.title('Model accuracy across subjects')
    plt.xticks(range(len(sorted_subjects)), sorted_subjects)
    plt.ylim(0, 1.0)
    plt.grid(axis='y', linestyle='--', alpha=0.7)
    plt.tight_layout()
    plt.savefig(os.path.join(output_dir, 'subject_performance.png'))
    print(f"Analysis saved to '{output_dir}/subject_performance.png'")
    
    print(f"Avg. Accuracy: {np.mean(list(subject_accuracy.values())):.4f}")
    print(f"Max. Accuracy: {max(subject_accuracy.values()):.4f} (Subject {subjects[np.argmax(accuracies)]})")
    print(f"Min. Accuracy: {min(subject_accuracy.values()):.4f} (Subject {subjects[np.argmin(accuracies)]})")
    
    return subject_accuracy

def main():

    output_dir = 'train_pic'
    os.makedirs(output_dir, exist_ok=True)
    
    X_train, y_train, X_test, y_test, subject_train, subject_test, features, activity_labels = load_data()
    params = {
        'C': 1.0,
        'kernel': 'rbf',
        'gamma': 'scale',
        'probability': True,
        'random_state': 42
    }
    
    model, scaler = train_svm_model(X_train, y_train, params, tune_hyperparams=False, output_dir=output_dir)
    
    evaluate_model(model, scaler, X_test, y_test, activity_labels, output_dir=output_dir)
    
    analyze_subject_performance(model, scaler, X_test, y_test, subject_test, activity_labels, output_dir=output_dir)
    
    print(f"\nResult saved in '{output_dir}'")

if __name__ == "__main__":
    main()