机器学习 KNN 算法原理及 C++/Python 实战实现

问题剖析

题目给出了三个特征维度，'每年获得的飞行常客里程数'这一特征维度的数据比另外两个大了 2-3 个数量级，我们选择了欧几里得距离作为距离标准，所以需要对数据进行预处理，让每个特征维度的权重基本一致。我们需要在导入图中数据时，将这些字符串转换为整型数据，方便使用。

具体实现

大致流程示意图

具体步骤

（1）数据预处理

归一化：将所有数据按原范围大小，均匀压缩到 0-1 的范围内，如 25、50、75、125，原范围 25-125，对应 0-1 范围 0、0.25、0.50、1。

给出公式：

洗牌：数据集的数据可能呈现出部分特征集中在某一特定位置，我们需要搅动它让数据均匀地分布在每个地方。对于数据集而言，这种数据集集中分布的情况是会影响生成的模型的能力的，所以我们需要把原来的数据集打乱，让某些过于集中的特征分散开来，提高模型预测的准确性。

（2）最佳 K 值的选择

k 的选取是很重要的，常见的选择方法有经验值法（直接设置参数）和交叉验证法（网格搜索 + 交叉验证）。

在计距离时用到了欧几里得距离距离计算公式。（x 和 y 是同一特征维度的，i 表示不同的特征维度）：

欧几里得距离计算公式：

（3）计算 FPR 和 TPR、AUC 并可视化 ROC 曲线

上面在介绍 ROC 曲线时已经讲过了，这里就不多说了。

C++ 实现

环境准备

本次我们只需要一个额外的可视化第三方库，我们选择 matplot++

matplot++ 下载

官方链接：matplot++发行版

VS2022 环境搭建

把 matplot++放到自己建立的解决方案的其中一个项目的目录下，matplot++的两个静态库，可根据实际位置调整，一般都是一个在 lib 里，一个在 lib\Matplot++里。然后在链接器中链接： nodesoup.lib matplot.lib

链接完成后点击确定，到这里环境就配置好了。

代码实现

头文件

#include <iostream>
#include <fstream>
#include <vector>
#include <memory>
#include <functional>
#include <algorithm>
#include <unordered_map>
#include <random>
#include <numeric>
#include <queue>
#include <matplot/matplot.h>

结构体

struct Data {
    double length;
    double game_time;
    double ice_cream_eating;
    int type;
};

这里 Data 结构体的前三个成员均为 double 类型而不是用 vector，这是因为 vector 是将数据分散存储到堆区，类似链表结构体，后续我们需要频繁地访问这些数据，需要不停使用指针访问内存，会让访问效率下降。

导入数据

std::vector<Data> loadData(const std::string& filepath) {
    std::ifstream ifs;
    ifs.open(filepath, std::ios::in);
    if (!ifs.is_open()) throw std::invalid_argument("Failed to open file: " + filepath);
    std::vector<Data> datalist;
    double trail, game, eating;
    std::string tag;
    std::vector<std::string> searchlist = { "largeDoses", "smallDoses", "didntLike" };
    while (ifs >> trail >> game >> eating >> tag) {
        int type = -1;
        for (int i = 0; i < searchlist.size(); ++i) {
            if (searchlist[i] == tag) {
                type = i;
                break;
            }
        }
        datalist.push_back({ trail, game, eating, type });
    }
    ifs.close();
    return datalist;
}

数据预处理

使用 lamdba 来简化三个特征维度的归一化实现：

// 更新归一化数据
void calculate(std::vector<double>& feature) {
    double max = *max_element(feature.begin(), feature.end());
    double min = *min_element(feature.begin(), feature.end());
    double range = max - min;
    for (double& val : feature) val = (val - min) / range;
}

// 归一化
void normalized(std::vector<Data>& data) {
    if (data.empty()) {
        std::cout << "nullptr doesn't normalized!" << std::endl;
        __debugbreak();
        return;
    }
    std::vector<std::pair< std::function<double(const Data&)>, std::function<void(Data&, double)> >> field = {
        { [](const Data& d) {return d.game_time; }, [](Data& d, double val) {d.game_time = val; } },
        { [](const Data& d) {return d.ice_cream_eating; }, [](Data& d, double val) {d.ice_cream_eating = val; } },
        { [](const Data& d) {return d.length; }, [](Data& d, double val) {d.length = val; } }
    };
    for (auto& f : field) {
        auto& getVal = f.first;
        auto& setVal = f.second;
        std::vector<double> values;
        for (const auto& d : data) values.push_back(getVal(d));
        calculate(values);
        for (int i = 0; i < data.size(); ++i) setVal(data[i], values[i]);
    }
}

随机生成伪随机数种子，使用 shuffle 洗牌：

// 打乱数据
std::vector<Data> shuffleData(std::vector<Data>& data) {
    auto shuffle_data = std::vector<Data>(data.begin(), data.end());
    std::random_device rd;
    std::mt19937 g(rd());
    std::shuffle(shuffle_data.begin(), shuffle_data.end(), g);
    return shuffle_data;
}

选择最佳 K 值

计算欧几里得距离：

// 计算欧几里得距离
double eucliDistance(Data x, Data y) {
    double trail = x.length - y.length;
    double play = x.game_time - y.game_time;
    double eating = x.ice_cream_eating - y.ice_cream_eating;
    return trail * trail + play * play + eating * eating;
}

获取邻居各类别的数量：

std::vector<int> getKNeighborProb(const Data& test, const std::vector<Data>& trains, int k) {
    if (trains.empty() || k <= 0) throw std::invalid_argument("Invalid input for getNeighborProb!");
    std::priority_queue<std::pair<double, int>> maxHeap; // 使用最大堆存储
    for (const auto& d : trains) {
        double dist = eucliDistance(test, d);
        if (maxHeap.size() < k) maxHeap.emplace(dist, d.type);
        else if (maxHeap.top().first > dist) {
            maxHeap.pop();
            maxHeap.emplace(dist, d.type);
        }
    }
    std::vector<int> neighborProb(3, 0);
    while (!maxHeap.empty()) {
        neighborProb[maxHeap.top().second]++;
        maxHeap.pop();
    }
    return neighborProb;
}

最大的 k 个邻居与它们的位置无关，只需要使用最大堆把最大的 k 个邻居和它们的类别存起来，统计最大堆前 k 个最大邻居出现的类别的数量即可。

选择出现最多的类别作为预测结果：

// 预测类型
int predictType(const Data& test, std::vector<Data>& data, int k) {
    auto vote = getKNeighborProb(test, data, k);
    return max_element(vote.begin(), vote.end()) - vote.begin();
}

使用 k 折交叉验证计算准确率和三分类混淆矩阵

// 使用 K 折交叉验证计算准确率
std::pair<double, std::vector<std::vector<int>>> kFoldCrossVaild(std::vector<Data>& shuffle_data, int kfold, int knn_k) {
    if (shuffle_data.empty()) throw std::invalid_argument("Invalid input for countRoc: Null data is invalid!");
    if (kfold <= 0) throw std::invalid_argument("Invalid input for countRoc: kfold must be bigger than 0!");
    std::vector<std::vector<int>> cntlist(3, std::vector<int>(3, 0));
    int foldSize = shuffle_data.size() / kfold;
    std::vector<double> accuracies;
    for (int fold = 0; fold < kfold; ++fold) {
        int start = fold * foldSize;
        int end = (kfold - 1 == fold) ? shuffle_data.size() : start + foldSize;
        int correct = 0;
        std::vector<Data> trainData(shuffle_data.begin(), shuffle_data.begin() + start);
        trainData.insert(trainData.end(), shuffle_data.begin() + end, shuffle_data.end());
        for(int i = start; i < end; ++i) {
            auto t = shuffle_data[i];
            int res = predictType(t, trainData, knn_k);
            cntlist[t.type][res]++;
            correct += (res == t.type);
        }
        double accuracy = static_cast<double>(correct) / foldSize;
        accuracies.push_back(accuracy);
    }
    return std::make_pair(std::accumulate(accuracies.begin(), accuracies.end(), 0.0) / kfold , cntlist);
}

网格搜索

从 3 开始，跳过偶数（偶数可能会平票，所以步长为 2），取数据集总量的开方，减少不必要的计算成本。将得到的准确率和混淆矩阵存起来，跑完网格搜素后寻找最大的准确率，换算 k 值，返回 k 和混淆矩阵。

计算 FPR 和 TPR

1. 遍历 vector 容器，将所有数据除以 k：

// 计算测试集在训练集上的多类型概率
std::vector<double> predicProb(const Data& test, std::vector<Data>& data, int k) {
    auto vote = getKNeighborProb(test, data, k);
    std::vector<double> appearance(3, 0);
    for (int i = 0; i < appearance.size(); ++i) appearance[i] = static_cast<double>(vote[i]) / k;
    return appearance;
}

2. 计算某一类型作为正例时单个点的 FPR 和 TPR：

// 使用 K 折交叉验证，计算 ROC 图数据
std::pair<double, double> countRoc(std::vector<Data>& shuffle_data, int kfold, int knn_k, double confidence, int genre) {
    if (shuffle_data.empty()) throw std::invalid_argument("Invalid input for countRoc: Null data is invalid!");
    if (kfold <= 0) throw std::invalid_argument("Invalid input for countRoc: kfold must be bigger than 0!");
    int foldSize = shuffle_data.size() / kfold;
    int TP = 0, FP = 0, FN = 0, TN = 0;
    for (int fold = 0; fold < kfold; ++fold) {
        int start = fold * foldSize;
        int end = (kfold - 1 == fold) ? shuffle_data.size() : start + foldSize;
        int correct = 0;
        std::vector<Data> trainData;
        trainData.reserve(shuffle_data.size() - foldSize);
        trainData.assign(shuffle_data.begin(), shuffle_data.begin() + start);
        trainData.insert(trainData.end(), shuffle_data.begin() + end, shuffle_data.end());
        for (int i = start; i < end; ++i) {
            auto t = shuffle_data[i];
            std::vector<double> res = predicProb(t, trainData, knn_k);
            bool isPred = (res[genre] >= confidence);
            bool isTrue = (genre == t.type);
            TP += isPred && isTrue;
            FP += isPred && (!isTrue);
            FN += (!isPred) && isTrue;
            TN += (!isPred) && (!isTrue);
        }
    }
    double TPR = (TP+FN == 0 ? 0.0 : static_cast<double>(TP) / (TP + FN));
    double FPR = (FP+TN == 0 ? 0.0 : static_cast<double>(FP) / (FP + TN));
    return { TPR, FPR };
}

3. 将某一类别所有点数据整理补全：

// 组织 ROC 图数据并补全
std::pair<std::vector<double>, std::vector<double>> getROC(std::vector<Data>& data, int kfold, int knn_k, int genre) {
    std::vector<double> tprlist;
    std::vector<double> fprlist;
    for (double confd = 1.0; confd > 0; confd -= 0.05) {
        auto [tpr, fpr] = countRoc(data, kfold, knn_k, confd, genre);
        tprlist.push_back(tpr);
        fprlist.push_back(fpr);
    }
    if (!(tprlist.front() == 0 && fprlist.front() == 0)) {
        tprlist.insert(tprlist.begin(), 0);
        fprlist.insert(fprlist.begin(), 0);
    }
    if (!(tprlist.back() == 1 && fprlist.back() == 1)) {
        tprlist.push_back(1);
        fprlist.push_back(1);
    }
    return { fprlist, tprlist };
}

由于可能出现 FPR 不为 0 和不为 1 的情况，这样面积就可能会少算一部分（最小 FPR 的左边区域和最大 FPR 的右边区域），所以需要补全。

计算 AUC 和可视化实现

// 累加 AUC 得到最终值 结合 8 使用 data 十分有序，无需处理即可使用
double getAUC(std::pair<std::vector<double>, std::vector<double>>& data) {
    double auc = 0.0;
    auto& fprs = data.first;
    auto& tprs = data.second;
    for (int i = 0; i < fprs.size()-1; ++i) {
        auc += (fprs[i + 1] - fprs[i]) * (tprs[i + 1] + tprs[i]) / 2;
    }
    std::cout << auc << std::endl;
    return auc;
}

void plotEvalution(std::vector<std::pair<std::vector<double>, std::vector<double>>> data, std::vector<double> aucs, std::vector<std::vector<int>>& list) {
    using namespace matplot;
    std::vector<std::string> colorList({ "blue", "yellow", "green", "red" });
    std::vector<std::string> tagName({ "largeDoses", "smallDoses", "didntLike", "Random Guess" });
    auto fig = figure(true); // 设置为 true 是能正常使用 figure 的关键
    fig->size(2560, 1000);
    subplot(1, 2, 1);
    hold(on);
    for (int i = 0; i < data.size(); ++i) {
        plot(data[i].first, data[i].second)->line_width(2).line_style("-").color(colorList[i]);
    }
    plot(std::vector{ 0,1 }, std::vector{ 0,1 })->line_width(1).line_style("--").color(colorList[3]);
    hold(off);
    xlabel("False Positive Rate");
    ylabel("True Positive Rate");
    title("ROC");
    xlim({ 0, 1 });
    ylim({ 0, 1 });
    for (int i = 0; i < aucs.size(); ++i) text(0.02, 0.1 - i * 0.025, tagName[i] + " AUC: " + std::to_string(aucs[i]))->font_size(8);
    auto l = ::matplot::legend(tagName);
    l->location(legend::general_alignment::bottomright);
    l->num_rows(2);
    l->font_size(5);
    subplot(1, 2, 2);
    heatmap(list)->normalization(matrix::color_normalization::columns);
    title("Three distribution");
    auto ax = gca();
    ax->x_axis().ticklabels({ "largeDoses", "smallDoses" , "didntLike" });
    ax->y_axis().ticklabels({ "largeDoses", "smallDoses" , "didntLike" });
    ax->x_axis().label_font_size(5);
    ax->y_axis().label_font_size(5);
    //save("ROC.png");
    show();
}

subplot(1, 2, 1) 下面的是 ROC 曲线可视化 subplot(1, 2, 2) 下面的是热力图可视化

最后调用 main 函数即可：

int main() {
    auto data = loadData("source\\datingTestSet.txt");
    normalized(data);
    auto shuffle_Data = shuffleData(data);
    auto [best_k, list] = getBestK(data);
    std::vector<std::pair<std::vector<double>, std::vector<double>>> results;
    std::vector<double> AUCs;
    for (int i = 0; i < 3; ++i) {
        auto res = getROC(shuffle_Data, 10, best_k, i);
        auto auc = getAUC(res);
        results.push_back(res);
        AUCs.push_back(auc);
    }
    plotEvalution(results, AUCs, list);
    return 0;
}

结果展示

Python 实现

python 也需要配置环境，但是 python 的环境配置简单很多

环境配置

在你的新的虚拟环境中，安装 numpy 和 matplotlib：

conda install numpy
conda install matplotlib

matplotlib 官方使用说明 numpy 官方使用说明

然后在创建项目时选择 conda 环境，到你的 anaconda 的文件夹中找到虚拟环境对应的 python.exe 即可。

代码实现

import numpy as np
import matplotlib.pyplot as plt

tag_map = {
    "largeDoses": 0,
    "smallDoses": 1,
    "didntLike": 2
}

# 导入数据
def load_data(filepath):
    # 读取数据
    data = []
    with open(filepath, 'r', encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            parts = line.split('\t')
            if len(parts) != 4:
                print(f"特征数据无法转换：{line}")
                continue
            tag = parts[3]
            data.append({
                "feature": [parts[0], parts[1], parts[2]],
                "tag": tag_map[tag]
            })
    return data

# 归一化 利用 numpy 的广播机制
def normalized(features: np.ndarray):
    # 获取每一列的最大值和最小值
    max_val = features.max(axis=0)
    min_val = features.min(axis=0)
    ranges = max_val-min_val
    ranges[ranges < 1e-9] = 1.0 # 比较每个列的相减结果，排除 0 的情况
    return (features-min_val)/ranges

# 计算欧式距离
def euclidean_dist(test, train):
    return np.sum((test - train) ** 2, axis=1)

# 计算各类标签的概率
def predict_prob(test, train_data, k):
    train_features = train_data[:,:3]
    dist = euclidean_dist(test, train_features)
    k_indices = np.argpartition(dist, k)[:k] # 完成一次快速排序
    k_tags = train_data[k_indices, 3].astype(int)
    prob = np.bincount(k_tags, minlength=3) / k
    return prob

# 选择最大概率作为当前测试样本的预测类型
def predict_type(test, train_data, k):
    return max(enumerate(predict_prob(test, train_data, k)), key=lambda x:x[1])[0]

# 使用 k 折交叉验证，计算准确率和三分类特征混淆矩阵
def k_folds_cross_valid_acc(features: np.ndarray, k, k_fold):
    # 计算每一折的大小
    fold_size = int(len(features)/k_fold)
    fold_accuracies = 0.0
    confusion_mat = np.zeros([3,3], dtype=np.int32)
    for i in range(k_fold):
        start = i*fold_size # 防止最后一折不够一折的大小
        end = start+fold_size if i != k_fold-1 else len(features)
        train_data = np.concatenate([features[:start], features[end:]])
        correct = 0
        test_data = features[start:end]
        for t in test_data:
            pred_type = predict_type(t[:3], train_data, k)
            if pred_type == t[3]:
                correct += 1
            confusion_mat[int(t[3])][pred_type] += 1
        fold_accuracies += correct/fold_size
    return fold_accuracies/k_fold, confusion_mat

# 计算模型 roc 曲线的单个点
def count_roc(features: np.ndarray, k_fold, k_neighbor, confidence, genre):
    fold_size = int(len(features)/k_fold)
    tp, tn, fp, fn = 0, 0, 0, 0
    for i in range(k_fold):
        start = i*fold_size
        end = start+fold_size if i != k_fold-1 else len(features)
        train_data = np.concatenate([features[:start], features[end:]])
        test_data = features[start:end]
        for t in test_data:
            prob = predict_prob(t[:3], train_data, k_neighbor)
            if genre == t[3] and prob[genre] >= confidence:
                tp += 1
            elif genre != t[3] and prob[genre] < confidence:
                tn += 1
            elif genre != t[3] and prob[genre] >= confidence:
                fp += 1
            elif genre == t[3] and prob[genre] < confidence:
                fn += 1
    tpr = tp / (tp+fn) if (tp+fn) else 0.0
    fpr = fp / (fp+tn) if (fp+tn) else 0.0
    return fpr, tpr

# 整理并补全 roc 曲线的两个点集
def get_roc(features: np.ndarray, k_fold, k_neighbor, genre):
    # 生成 1 到 0 步长为 -0.05 的列表
    confidence = np.arange(1.0, 0, -0.05)
    fpr_list = list()
    tpr_list = list()
    for conf in confidence:
        fpr, tpr = count_roc(features, k_fold, k_neighbor, conf, genre)
        fpr_list.append(fpr)
        tpr_list.append(tpr)
    if fpr_list[0] != 0.0:
        fpr_list.insert(0, 0.0)
        tpr_list.insert(0, 0.0)
    if fpr_list[len(fpr_list)-1] != 1.0:
        fpr_list.append(1.0)
        tpr_list.append(1.0)
    return [fpr_list, tpr_list]

# 获取 AUC
def get_auc(result):
    auc = 0.0
    fpr_list, tpr_list = result
    if len(fpr_list) < 2:
        return 0.0
    for i in range(len(fpr_list)-1): # 梯形法算 AUC 面积
        auc += (fpr_list[i+1]-fpr_list[i])*(tpr_list[i]+tpr_list[i+1])/2
    return auc

# 通过值寻找键列表
def get_value(dictionary: dict, target_value: int):
    return [key for key, value in dictionary.items() if value == target_value]

# 绘制 roc 曲线
def plot_roc(results, ax):
    class_names = list(tag_map.keys())
    for i in range(len(results)):
        ax.plot(results[i][0], results[i][1], linewidth=2, label=f"{class_names[i]} AUC: {results[i][2]:0.5f}")
    ax.plot((0, 1), (0, 1), "--", linewidth=1, label="predicted line")
    ax.legend(loc="lower right")
    ax.set_title("ROC Curve")
    ax.set_xlabel("False Positive Rate")
    ax.set_ylabel("True Positive Rate")
    ax.set_xlim(0,1)
    ax.set_ylim(0,1)

# 绘制热力图
def plot_heatmap(heat_conf, ax):
    class_names = list(tag_map.keys())
    im = ax.imshow(heat_conf)
    ax.set_xticks(range(len(class_names)))
    ax.set_xticklabels(labels=tag_map.keys(), rotation=45, ha="right", rotation_mode="anchor")
    ax.set_yticks(range(len(class_names)))
    ax.set_yticklabels(labels=tag_map.keys())
    ax.set_title("Three Distribution")
    for i in range(len(tag_map)):
        for j in range(len(tag_map)):
            value = heat_conf[i, j]
            # 提高可视化观感
            text_color = "black" if im.norm(value) > 0.5 else "white"
            ax.text(j, i, value, ha="center", va="center", color=text_color, fontweight="bold")
    cbar = plt.colorbar(im, ax=ax)
    cbar.set_label("Sample Count")

if __name__ == "__main__":
    # 导入数据
    dataset = load_data("datingTestSet.txt")
    # 检查数据是否导入
    if not dataset:
        print("没有有效数据")
        exit()
    # 创建 numpy 数组
    feature_list = np.array([d["feature"] for d in dataset], dtype=np.float64)
    tag_list = np.array([d["tag"] for d in dataset], dtype=np.int32)
    # 归一化数组
    normalized_feature = normalized(feature_list)
    # 将特征和标签拼接
    tag_list = tag_list.reshape(-1, 1)
    # 将 numpy 数组转置
    feature_tag = np.hstack((normalized_feature, tag_list))
    # 洗牌
    np.random.seed(42)
    np.random.shuffle(feature_tag)
    # 选择最大准确率，选择最佳 k 值，并计算出最佳 k 值下的三分类特征混淆矩阵
    best_k = 0
    best_acc = 0
    best_confusion = np.zeros([3, 3], dtype=np.int32)
    for i in range(3, int(np.sqrt(len(feature_tag))), 2):
        acc, confusion = k_folds_cross_valid_acc(feature_tag, i, 10)
        if best_acc < acc:
            best_k = i
            best_acc = acc
            best_confusion = confusion
    print(f"十折交叉验证的最佳 k 值为：{best_k}，对应的准确率为：{best_acc:0.5f}")
    # 评估模型：获取 ROC 曲线和 AUC 值
    results = list()
    for i in range(3):
        res = get_roc(feature_tag, 10, best_k, i)
        auc = get_auc(res)
        results.append([*res, auc])
    # 评估数据可视化
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
    plot_roc(results, ax1)
    plot_heatmap(best_confusion, ax2)
    plt.tight_layout()
    plt.show()

运行结果