SVM_Test.py

import numpy as np
import pandas as pd
import os
import joblib
import matplotlib.pyplot as plt
import seaborn as sns
import matplotlib
from sklearn.metrics import accuracy_score, confusion_matrix, classification_report

# 设置全局字体为微软雅黑
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei']  # 指定中文字体
matplotlib.rcParams['axes.unicode_minus'] = False  # 解决保存图像时负号'-'显示为方块的问题

def load_test_data(test_dir='datasets/test'):
    """
    加载测试数据
    
    参数:
    test_dir: 测试数据目录
    
    返回:
    X_test, y_test, subject_test, feature_names
    """
    print("加载测试数据...")
    
    # 加载特征名称
    features_path = os.path.join(os.path.dirname(test_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)
    
    # 加载测试数据
    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)
    
    # 加载真实标签
    try:
        y_test = pd.read_csv(y_test_path, header=None, names=['activity_id'])
    except FileNotFoundError:
        print(f"警告: 未找到真实标签文件 {y_test_path}")
        y_test = None
    
    subject_test = pd.read_csv(subject_test_path, header=None, names=['subject_id'])
    
    print(f"测试数据形状: {X_test.shape}")
    print(f"测试集受试者数量: {subject_test['subject_id'].nunique()}")
    
    return X_test, y_test, subject_test, feature_names

def load_model_and_make_predictions(X_test, y_test, subject_test):
    """
    加载模型并进行预测，与真实标签进行比对
    
    参数:
    X_test: 测试特征
    y_test: 测试标签（真实值）
    subject_test: 测试集受试者ID
    
    返回:
    预测结果DataFrame
    """
    print("加载模型并进行预测...")
    
    # 加载标准化器和模型
    try:
        scaler = joblib.load('svm_scaler.pkl')
        model = joblib.load('svm_model.pkl')
    except FileNotFoundError:
        print("错误: 未找到保存的模型文件。请先运行训练脚本!")
        return None
    
    # 加载活动标签
    activity_labels_path = os.path.join('datasets', 'activity_labels.txt')
    activity_labels = pd.read_csv(activity_labels_path, sep=' ', header=None, names=['id', 'activity'])
    
    # 标准化测试数据
    X_test_scaled = scaler.transform(X_test)
    
    # 预测活动
    y_pred = model.predict(X_test_scaled)
    
    # 创建结果DataFrame
    results = pd.DataFrame({
        'subject_id': subject_test['subject_id'].values,
        'predicted_activity_id': y_pred
    })
    
    # 如果有真实标签，添加到结果中并计算准确率
    if y_test is not None:
        results['true_activity_id'] = y_test['activity_id'].values
        
        # 计算总体准确率
        accuracy = accuracy_score(results['true_activity_id'], results['predicted_activity_id'])
        print(f"\n总体准确率: {accuracy:.4f}")
        
        # 计算每个活动类型的准确率
        print("\n各活动类型准确率:")
        for act_id, act_name in zip(activity_labels['id'], activity_labels['activity']):
            mask = results['true_activity_id'] == act_id
            if mask.sum() > 0:
                act_accuracy = accuracy_score(
                    results.loc[mask, 'true_activity_id'], 
                    results.loc[mask, 'predicted_activity_id']
                )
                print(f"{act_name}: {act_accuracy:.4f} ({mask.sum()}个样本)")
        
        # 计算每个受试者的准确率
        print("\n各受试者准确率:")
        subject_accuracies = {}
        for subject in results['subject_id'].unique():
            mask = results['subject_id'] == subject
            subj_accuracy = accuracy_score(
                results.loc[mask, 'true_activity_id'], 
                results.loc[mask, 'predicted_activity_id']
            )
            subject_accuracies[subject] = subj_accuracy
            print(f"受试者 {subject}: {subj_accuracy:.4f}")
        
        # 生成混淆矩阵
        cm = confusion_matrix(results['true_activity_id'], results['predicted_activity_id'])
        plt.figure(figsize=(10, 8))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
                    xticklabels=activity_labels['activity'], 
                    yticklabels=activity_labels['activity'])
        plt.xlabel('Predictive Label')
        plt.ylabel('Actual Label')
        plt.title('Test Set Confusion Matrix')
        plt.tight_layout()
        plt.savefig('test_confusion_matrix.png')
        print("\n测试集混淆矩阵已保存为 'test_confusion_matrix.png'")
        
        # 绘制受试者准确率条形图
        plt.figure(figsize=(12, 6))
        subjects = list(subject_accuracies.keys())
        accuracies = list(subject_accuracies.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 for Different Subjects in the Test Set')
        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('test_subject_performance.png')
        print("测试集受试者性能分析已保存为 'test_subject_performance.png'")
    
    # 添加活动名称
    results = results.merge(activity_labels, left_on='predicted_activity_id', right_on='id')
    results = results.rename(columns={'activity': 'predicted_activity'})
    
    # 如果有真实标签，添加真实活动名称
    if y_test is not None:
        results = results.merge(activity_labels, left_on='true_activity_id', right_on='id', suffixes=('', '_true'))
        results = results.rename(columns={'activity': 'true_activity'})
    
    # 整理列顺序并删除多余的id列
    if y_test is not None:
        results = results[['subject_id', 'true_activity_id', 'true_activity', 
                          'predicted_activity_id', 'predicted_activity']]
    else:
        results = results[['subject_id', 'predicted_activity_id', 'predicted_activity']]
    
    # 保存结果
    results.to_csv('predictions.csv', index=False)
    print("预测结果已保存到 'predictions.csv'")
    
    return results

def analyze_predictions(results):
    """
    分析预测结果
    
    参数:
    results: 预测结果DataFrame
    """
    print("\n分析预测结果...")
    
    # 判断结果中是否包含真实标签
    has_true_labels = 'true_activity_id' in results.columns
    
    # 各活动的预测分布
    activity_counts = results['predicted_activity'].value_counts()
    
    # 绘制活动分布图
    plt.figure(figsize=(10, 6))
    activity_counts.plot(kind='bar')
    plt.xlabel('活动类型')
    plt.ylabel('样本数量')
    plt.title('预测活动分布')
    plt.tight_layout()
    plt.savefig('activity_distribution.png')
    print("活动分布图已保存为 'activity_distribution.png'")
    
    # 各受试者的活动分布
    subject_activity = results.groupby(['subject_id', 'predicted_activity']).size().unstack().fillna(0)
    
    # 绘制热图
    plt.figure(figsize=(12, 8))
    sns.heatmap(subject_activity, cmap='YlGnBu', linewidths=0.5, annot=True, fmt='g')
    plt.xlabel('活动类型')
    plt.ylabel('受试者ID')
    plt.title('各受试者的预测活动分布')
    plt.tight_layout()
    plt.savefig('subject_activity_heatmap.png')
    print("受试者活动热图已保存为 'subject_activity_heatmap.png'")
    
    # 如果有真实标签，计算准确率和错误率
    if has_true_labels:
        # 添加是否预测正确的列
        results['is_correct'] = results['true_activity_id'] == results['predicted_activity_id']
        
        # 计算每个活动的错误率
        error_by_activity = results.groupby('true_activity').apply(
            lambda x: (x['is_correct'] == False).mean()
        ).sort_values(ascending=False)
        
        plt.figure(figsize=(10, 6))
        error_by_activity.plot(kind='bar')
        plt.xlabel('Type of Activity')
        plt.ylabel('Error Rate')
        plt.title('Error Rates by Activity Type')
        plt.ylim(0, 1.0)
        plt.tight_layout()
        plt.savefig('activity_error_rates.png')
        print("活动错误率图已保存为 'activity_error_rates.png'")
        
        # 找出最常见的错误分类
        if not results[~results['is_correct']].empty:
            misclassifications = results[~results['is_correct']].groupby(
                ['true_activity', 'predicted_activity']
            ).size().reset_index(name='count').sort_values('count', ascending=False)
            
            if not misclassifications.empty:
                print("\n最常见的错误分类:")
                for _, row in misclassifications.head(5).iterrows():
                    print(f"真实活动 '{row['true_activity']}' 被错误预测为 '{row['predicted_activity']}' ({row['count']}次)")
                
                # 绘制前10个最常见错误分类的柱状图
                plt.figure(figsize=(12, 6))
                top_errors = misclassifications.head(10)
                labels = [f"{row['true_activity']} → {row['predicted_activity']}" 
                        for _, row in top_errors.iterrows()]
                plt.bar(range(len(top_errors)), top_errors['count'])
                plt.xlabel('Type of Error')
                plt.ylabel('Number of Occurrences')
                plt.title('The Most Common Misclassifications')
                plt.xticks(range(len(top_errors)), labels, rotation=45, ha='right')
                plt.tight_layout()
                plt.savefig('common_errors.png')
                print("常见错误分类图已保存为 'common_errors.png'")

def main():
    """
    主函数
    """
    # 加载测试数据
    X_test, y_test, subject_test, _ = load_test_data()
    
    # 加载模型并进行预测
    results = load_model_and_make_predictions(X_test, y_test, subject_test)
    
    if results is not None:
        # 分析预测结果
        analyze_predictions(results)
    
    print("\n测试数据预测和评估完成!")

if __name__ == "__main__":
    main()