一、实验介绍

本实验包含两个部分：第一部分使用K近邻(KNN)算法对约会网站数据进行分类，第二部分通过随机生成的数据绘制ROC曲线和PR曲线来评估分类器性能。

二、KNN算法原理

2.1 什么是KNN算法

K近邻(K-Nearest Neighbors)算法是一种简单而有效的监督学习算法，主要用于分类和回归任务。其核心思想是：一个样本的类别可以由其最近邻居的多数投票来决定。

2.2 KNN算法步骤

1. 计算距离：计算测试样本与所有训练样本的距离

2. 寻找邻居：选择距离最近的K个样本

3. 投票决策：根据K个邻居的类别进行多数投票

4. 输出结果：将得票最多的类别作为预测结果

2.3 距离度量

常用的距离度量包括欧氏距离、曼哈顿距离等。本实验使用欧氏距离：

三、KNN分类器实现详解

3.1 实验环境配置

环境搭建步骤：
在VSCode中创建项目目录，新建Python源代码文件，将实验数据集文件放置于项目根目录下，确保运行环境配置正确。

数据集文件：

3.2 核心代码实现

第一步：数据加载功能

def load_data(filename):
    data = []
    labels = []
    with open(filename, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            parts = line.split('\t')
            try:
                # 三个特征
                feature1 = float(parts[0])
                feature2 = float(parts[1])
                feature3 = float(parts[2])
                # 标签
                label = parts[3]
                data.append([feature1, feature2, feature3])
                labels.append(label)
            except:
                continue
    return data, labels

实现解析：该模块负责数据读取与解析，从文本文件中提取特征向量和类别标签，建立结构化数据集。

第二步：特征标准化处理

def normalize_data(data):
    # 找出每个特征的最大值和最小值
    min_vals = [float('inf')] * 3
    max_vals = [float('-inf')] * 3
    
    for sample in data:
        for i in range(3):
            if sample[i] < min_vals[i]:
                min_vals[i] = sample[i]
            if sample[i] > max_vals[i]:
                max_vals[i] = sample[i]
    
    # 归一化：new_value = (old_value - min) / (max - min)
    normalized_data = []
    for sample in data:
        normalized_sample = []
        for i in range(3):
            if max_vals[i] == min_vals[i]:  # 避免除以0
                normalized_value = 0
            else:
                normalized_value = (sample[i] - min_vals[i]) / (max_vals[i] - min_vals[i])
            normalized_sample.append(normalized_value)
        normalized_data.append(normalized_sample)
    
    return normalized_data, min_vals, max_vals

实现解析：通过最小-最大标准化方法消除特征量纲影响，将各维度数值映射到[0,1]区间。记录极值参数便于后续新样本的标准化处理。

第三步：距离计算函数

def euclidean_distance(point1, point2):
    distance = 0
    for i in range(len(point1)):
        distance += (point1[i] - point2[i]) ** 2
    return math.sqrt(distance)

实现解析：实现欧氏距离度量，通过计算各维度差值的平方和再开方，得到样本间的几何距离。

第四步：KNN分类核心逻辑

def knn_classify(train_data, train_labels, test_sample, k=3):
    # 计算测试样本与所有训练样本的距离
    distances = []
    for i in range(len(train_data)):
        dist = euclidean_distance(train_data[i], test_sample)
        distances.append((dist, train_labels[i]))
    
    # 按距离排序，取前k个最近的邻居
    distances.sort(key=lambda x: x[0])
    neighbors = distances[:k]
    
    # 统计k个邻居中每个类别的数量
    class_count = {}
    for neighbor in neighbors:
        label = neighbor[1]
        if label in class_count:
            class_count[label] += 1
        else:
            class_count[label] = 1
    
    # 返回数量最多的类别
    max_count = 0
    predicted_class = None
    for label, count in class_count.items():
        if count > max_count:
            max_count = count
            predicted_class = label
    
    return predicted_class

实现解析：完整实现KNN分类流程，包括距离计算、近邻排序、投票统计和结果决策。通过字典统计类别频数，选择出现次数最多的类别作为最终预测。

第五步：数据集划分策略

def split_data(data, labels, test_ratio=0.2):
    # 随机打乱数据
    combined = list(zip(data, labels))
    random.shuffle(combined)
    data[:], labels[:] = zip(*combined)
    
    # 划分训练集和测试集
    split_index = int(len(data) * (1 - test_ratio))
    train_data = data[:split_index]
    train_labels = labels[:split_index]
    test_data = data[split_index:]
    test_labels = labels[split_index:]
    
    return train_data, train_labels, test_data, test_labels

实现解析：采用随机抽样方式将数据集划分为训练集和测试集。

第六步：性能评估指标

def calculate_accuracy(true_labels, predicted_labels):
    correct = 0
    for i in range(len(true_labels)):
        if true_labels[i] == predicted_labels[i]:
            correct += 1
    return correct / len(true_labels)

实现解析：实现准确率计算功能，通过对比预测结果与真实标签，统计正确分类样本比例，作为模型性能的量化指标。

3.3实现结果

附：完整代码

import math
import random

# 步骤1: 读取数据
def load_data(filename):
    data = []
    labels = []
    with open(filename, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            parts = line.split('\t')
            try:
                # 三个特征
                feature1 = float(parts[0])
                feature2 = float(parts[1])
                feature3 = float(parts[2])
                # 标签
                label = parts[3]
                data.append([feature1, feature2, feature3])
                labels.append(label)
            except:
                continue
    return data, labels

# 步骤2: 数据归一化（重要！）
def normalize_data(data):
    # 找出每个特征的最大值和最小值
    min_vals = [float('inf')] * 3
    max_vals = [float('-inf')] * 3
    
    for sample in data:
        for i in range(3):
            if sample[i] < min_vals[i]:
                min_vals[i] = sample[i]
            if sample[i] > max_vals[i]:
                max_vals[i] = sample[i]
    
    # 归一化：new_value = (old_value - min) / (max - min)
    normalized_data = []
    for sample in data:
        normalized_sample = []
        for i in range(3):
            if max_vals[i] == min_vals[i]:  # 避免除以0
                normalized_value = 0
            else:
                normalized_value = (sample[i] - min_vals[i]) / (max_vals[i] - min_vals[i])
            normalized_sample.append(normalized_value)
        normalized_data.append(normalized_sample)
    
    return normalized_data, min_vals, max_vals

# 步骤3: 计算两个样本之间的欧氏距离
def euclidean_distance(point1, point2):
    distance = 0
    for i in range(len(point1)):
        distance += (point1[i] - point2[i]) ** 2
    return math.sqrt(distance)

# 步骤4: KNN分类器
def knn_classify(train_data, train_labels, test_sample, k=3):
    # 计算测试样本与所有训练样本的距离
    distances = []
    for i in range(len(train_data)):
        dist = euclidean_distance(train_data[i], test_sample)
        distances.append((dist, train_labels[i]))
    
    # 按距离排序，取前k个最近的邻居
    distances.sort(key=lambda x: x[0])
    neighbors = distances[:k]
    
    # 统计k个邻居中每个类别的数量
    class_count = {}
    for neighbor in neighbors:
        label = neighbor[1]
        if label in class_count:
            class_count[label] += 1
        else:
            class_count[label] = 1
    
    # 返回数量最多的类别
    max_count = 0
    predicted_class = None
    for label, count in class_count.items():
        if count > max_count:
            max_count = count
            predicted_class = label
    
    return predicted_class

# 步骤5: 划分训练集和测试集
def split_data(data, labels, test_ratio=0.2):
    # 随机打乱数据
    combined = list(zip(data, labels))
    random.shuffle(combined)
    data[:], labels[:] = zip(*combined)
    
    # 划分训练集和测试集
    split_index = int(len(data) * (1 - test_ratio))
    train_data = data[:split_index]
    train_labels = labels[:split_index]
    test_data = data[split_index:]
    test_labels = labels[split_index:]
    
    return train_data, train_labels, test_data, test_labels

# 步骤6: 计算准确率
def calculate_accuracy(true_labels, predicted_labels):
    correct = 0
    for i in range(len(true_labels)):
        if true_labels[i] == predicted_labels[i]:
            correct += 1
    return correct / len(true_labels)

# 主程序
def main():
    print("=== KNN分类器实现 ===")
    print("步骤1: 读取数据...")
    data, labels = load_data('datingTestSet.txt')
    print(f"总共读取了 {len(data)} 个样本")
    
    # 显示类别分布
    label_count = {}
    for label in labels:
        if label in label_count:
            label_count[label] += 1
        else:
            label_count[label] = 1
    
    print("\n数据类别分布:")
    for label, count in label_count.items():
        chinese_label = {
            'didntLike': '不喜欢',
            'smallDoses': '一般喜欢', 
            'largeDoses': '非常喜欢'
        }.get(label, label)
        print(f"  {chinese_label}: {count} 个样本")
    
    print("\n步骤2: 数据归一化...")
    normalized_data, min_vals, max_vals = normalize_data(data)
    
    print("\n步骤3: 划分训练集和测试集...")
    train_data, train_labels, test_data, test_labels = split_data(normalized_data, labels)
    print(f"训练集: {len(train_data)} 个样本")
    print(f"测试集: {len(test_data)} 个样本")
    
    print("\n步骤4: 使用KNN进行分类...")
    k = 5  # 选择k=5个邻居
    predictions = []
    
    for i in range(len(test_data)):
        predicted_label = knn_classify(train_data, train_labels, test_data[i], k)
        predictions.append(predicted_label)
    
    print("\n步骤5: 计算准确率...")
    accuracy = calculate_accuracy(test_labels, predictions)
    print(f"KNN分类准确率: {accuracy:.2%}")
    
    # 显示部分分类结果
    print("\n=== 部分分类结果 ===")
    print("序号\t真实类别\t预测类别\t是否正确")
    print("-" * 45)
    
    correct_count = 0
    for i in range(min(20, len(test_labels))):  # 显示前20个结果
        is_correct = "✓" if test_labels[i] == predictions[i] else "✗"
        if test_labels[i] == predictions[i]:
            correct_count += 1
        
        # 转换为中文显示，并使用固定宽度
        true_chinese = {
            'didntLike': '不喜欢 ',
            'smallDoses': '一般喜欢', 
            'largeDoses': '非常喜欢'
        }.get(test_labels[i], test_labels[i])
        
        pred_chinese = {
            'didntLike': '不喜欢 ',
            'smallDoses': '一般喜欢', 
            'largeDoses': '非常喜欢'
        }.get(predictions[i], predictions[i])
        
        print(f"{i+1:2d}\t{true_chinese}\t{pred_chinese}\t\t{is_correct}")
    
    print(f"\n前{min(20, len(test_labels))}个样本中正确分类: {correct_count}个")
    
    # 测试不同K值的效果
    print("\n=== 测试不同K值的效果 ===")
    k_values = [1, 3, 5, 7, 9]
    for k_val in k_values:
        k_predictions = []
        for i in range(len(test_data)):
            predicted_label = knn_classify(train_data, train_labels, test_data[i], k_val)
            k_predictions.append(predicted_label)
        k_accuracy = calculate_accuracy(test_labels, k_predictions)
        print(f"K={k_val}时的准确率: {k_accuracy:.2%}")
    
    # 手动测试新样本
    print("\n=== 手动测试新样本 ===")
    print("请输入一个新的约会对象的特征：")
    try:
        feature1 = float(input("每年飞行里程数: "))
        feature2 = float(input("玩游戏时间百分比: "))
        feature3 = float(input("每周冰淇淋消费量(升): "))
        
        # 归一化新样本
        new_sample = [feature1, feature2, feature3]
        normalized_sample = []
        for i in range(3):
            if max_vals[i] == min_vals[i]:
                normalized_value = 0
            else:
                normalized_value = (new_sample[i] - min_vals[i]) / (max_vals[i] - min_vals[i])
            normalized_sample.append(normalized_value)
        
        # 预测
        predicted_label = knn_classify(train_data, train_labels, normalized_sample, k)
        
        chinese_result = {
            'didntLike': '不喜欢',
            'smallDoses': '一般喜欢', 
            'largeDoses': '非常喜欢'
        }.get(predicted_label, predicted_label)
        
        print(f"\n预测结果: {chinese_result}")
        
    except ValueError:
        print("输入无效，请确保输入的是数字。")

if __name__ == "__main__":
    main()

四、生成ROC与PR曲线

4.1 评估指标概念解析

我们通过混淆矩阵定义四个基础指标：

• 真正例(TP)：正类样本被正确预测

• 假正例(FP)：负类样本被错误预测为正类

• 真负例(TN)：负类样本被正确预测

• 假负例(FN)：正类样本被错误预测为负类

4.2 曲线评估原理

ROC曲线以假正例率为横轴，真正例率为纵轴，通过调整分类阈值生成曲线轨迹，曲线下面积AUC值反映分类器整体性能。

PR曲线以召回率为横轴，精确率为纵轴，特别适用于不平衡数据集评估，曲线下面积AP值体现分类器在正类样本上的表现。

五、ROC/PR曲线实现详解

5.1 实验数据生成

第一步：仿真数据构造

import numpy as np

np.random.seed(42)
true_labels = np.random.randint(0, 2, 100)

predicted_probs = []
for label in true_labels:
    if label == 1:
        prob = np.random.normal(0.7, 0.2)
    else:
        prob = np.random.normal(0.3, 0.2)
    prob = max(0, min(1, prob))
    predicted_probs.append(prob)

5.2 曲线数据计算

第二步：ROC曲线数据生成

def calculate_roc_curve(true_labels, predicted_probs):
    thresholds = np.linspace(0, 1, 101)
    tpr_list = []
    fpr_list = []
    
    for threshold in thresholds:
        predicted_labels = [1 if prob >= threshold else 0 for prob in predicted_probs]
        
        tp = fp = tn = fn = 0
        for true, pred in zip(true_labels, predicted_labels):
            if true == 1 and pred == 1:
                tp += 1
            elif true == 0 and pred == 1:
                fp += 1
            elif true == 0 and pred == 0:
                tn += 1
            elif true == 1 and pred == 0:
                fn += 1
        
        tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
        fpr = fp / (fp + tn) if (fp + tn) > 0 else 0
        
        tpr_list.append(tpr)
        fpr_list.append(fpr)
    
    return fpr_list, tpr_list, thresholds

实现解析：遍历从0到1的分类阈值，在每个阈值下计算混淆矩阵并推导出真正例率和假正例率，构建ROC曲线的数据点序列。

第三步：PR曲线数据计算

def calculate_pr_curve(true_labels, predicted_probs):
    thresholds = np.linspace(0, 1, 101)
    precision_list = []
    recall_list = []
    
    for threshold in thresholds:
        predicted_labels = [1 if prob >= threshold else 0 for prob in predicted_probs]
        
        tp = fp = fn = 0
        for true, pred in zip(true_labels, predicted_labels):
            if true == 1 and pred == 1:
                tp += 1
            elif true == 0 and pred == 1:
                fp += 1
            elif true == 1 and pred == 0:
                fn += 1
        
        precision = tp / (tp + fp) if (tp + fp) > 0 else 1
        recall = tp / (tp + fn) if (tp + fn) > 0 else 0
        
        precision_list.append(precision)
        recall_list.append(recall)
    
    return recall_list, precision_list, thresholds

实现解析：同样遍历分类阈值，计算每个阈值对应的精确率和召回率，构建PR曲线的数据点序列。

第四步：性能指标

def calculate_auc(fpr, tpr):
    auc = 0
    for i in range(1, len(fpr)):
        auc += (fpr[i] - fpr[i-1]) * (tpr[i] + tpr[i-1]) / 2
    return auc

def calculate_ap(precision, recall):
    ap = 0
    for i in range(1, len(recall)):
        ap += (recall[i] - recall[i-1]) * precision[i]
    return ap

实现解析：采用数值积分方法计算曲线下面积，AUC值反映ROC曲线与随机分类线的偏离程度，AP值体现PR曲线与基线水平的相对位置。

5.3 实现结果

六、实验结果分析

6.1 KNN分类性能

实验数据显示，KNN分类器在约会数据集上表现优异：

• 成功处理1000个多类别样本

• 分类准确率达到94.50%的高水平

• 参数调优显示K=5时获得最佳性能

6.2 评估曲线分析

ROC与PR曲线生成结果：

• AUC值为0.923，显著优于随机分类基准

• AP值为0.87，在精确率-召回率权衡中表现良好

• 曲线形态直观展示了分类器在不同决策阈值下的性能特征

七、实验总结

通过本实验：我们做到了从数据预处理到KNN核心算法的完整实现，掌握了ROC/PR曲线的生成原理和实现技术。