Python实现K近邻算法：从原理到实战应用-洪萨配资

1. 从零实现K近邻算法：Python实战指南

K近邻（K-Nearest Neighbors，简称KNN）是机器学习中最直观的算法之一，它的核心思想简单却强大：通过比较新数据与历史数据的相似度来进行预测。本文将带你从零开始实现KNN算法，不使用任何机器学习库，完全基于Python基础功能构建。

1.1 KNN算法核心原理

KNN属于"惰性学习"（lazy learning）算法，因为它不会从训练数据中学习一个明确的模型，而是将整个训练集存储起来，直到需要预测时才进行计算。这种特性使得KNN训练阶段非常快，但预测阶段相对较慢。

算法工作流程分为三个关键步骤：

计算待预测样本与所有训练样本的距离
选择距离最近的K个邻居
根据这些邻居的类别进行投票（分类）或取平均值（回归）

提示：K值的选择对算法性能影响很大。较小的K值会使模型对噪声敏感，较大的K值会使模型过于平滑。通常通过交叉验证来确定最佳K值。

1.2 欧氏距离计算实现

欧氏距离是KNN最常用的距离度量方法，它反映了多维空间中两点之间的直线距离。计算公式为：

distance = √(Σ(x_i - y_i)²)

Python实现如下：

from math import sqrt def euclidean_distance(row1, row2): """计算两个向量之间的欧氏距离 参数: row1 -- 第一个数据向量 row2 -- 第二个数据向量 返回: 两个向量之间的欧氏距离 """ distance = 0.0 for i in range(len(row1)-1): # 忽略最后一列的标签 distance += (row1[i] - row2[i])**2 return sqrt(distance)

在实际应用中，我们需要注意：

数据标准化：不同特征的量纲差异会导致距离计算偏向数值较大的特征
缺失值处理：需要提前处理或采用特殊距离度量
高维诅咒：维度很高时，所有样本的距离会趋同，影响算法效果

1.3 寻找最近邻居

有了距离计算函数后，我们需要找到距离最近的K个邻居。实现思路是：

计算目标样本与所有训练样本的距离
将距离和对应的样本一起存储
按距离排序
取前K个样本作为最近邻居

def get_neighbors(train, test_row, num_neighbors): """查找最近的邻居 参数: train -- 训练数据集 test_row -- 待预测样本 num_neighbors -- K值(邻居数量) 返回: 最近的K个邻居列表 """ distances = [] for train_row in train: dist = euclidean_distance(test_row, train_row) distances.append((train_row, dist)) # 按距离排序 distances.sort(key=lambda tup: tup[1]) # 提取最近的K个邻居 neighbors = [] for i in range(num_neighbors): neighbors.append(distances[i][0]) return neighbors

这个实现的时间复杂度是O(n)，对于大数据集可能会比较慢。在实际项目中，可以考虑使用KD树或球树等数据结构来优化搜索效率。

2. KNN分类预测实现

2.1 分类预测函数

找到最近邻居后，我们需要根据这些邻居的类别进行预测。对于分类问题，最简单的策略是多数投票：

def predict_classification(train, test_row, num_neighbors): """使用KNN进行分类预测 参数: train -- 训练数据集 test_row -- 待预测样本 num_neighbors -- K值 返回: 预测的类别 """ neighbors = get_neighbors(train, test_row, num_neighbors) output_values = [row[-1] for row in neighbors] # 提取邻居的类别 # 找出出现次数最多的类别 prediction = max(set(output_values), key=output_values.count) return prediction

2.2 测试分类预测

我们可以用一个小型数据集测试这个实现：

# 测试数据集 dataset = [ [2.7810836,2.550537003,0], [1.465489372,2.362125076,0], [3.396561688,4.400293529,0], [1.38807019,1.850220317,0], [3.06407232,3.005305973,0], [7.627531214,2.759262235,1], [5.332441248,2.088626775,1], [6.922596716,1.77106367,1], [8.675418651,-0.242068655,1], [7.673756466,3.508563011,1] ] # 预测第一个样本的类别 prediction = predict_classification(dataset, dataset[0], 3) print(f"Expected {dataset[0][-1]}, Got {prediction}")

输出应该是：Expected 0, Got 0，表示预测正确。

2.3 距离加权的KNN

基础KNN算法中，所有邻居的投票权重相同。我们可以改进为距离加权投票，使更近的邻居有更大影响力：

def predict_weighted_classification(train, test_row, num_neighbors): neighbors = get_neighbors(train, test_row, num_neighbors) # 为每个邻居计算权重(距离的倒数) weights = {} for neighbor in neighbors: dist = euclidean_distance(test_row, neighbor) label = neighbor[-1] weights[label] = weights.get(label, 0) + (1.0 / (dist + 1e-5)) # 加小常数避免除零 # 找出权重最大的类别 prediction = max(weights.items(), key=lambda x: x[1])[0] return prediction

这种改进通常能提高模型精度，特别是当不同类别的样本分布不均匀时。

3. 在鸢尾花数据集上应用KNN

3.1 数据准备

我们将使用经典的鸢尾花数据集来测试我们的KNN实现。首先需要加载和预处理数据：

from csv import reader from random import seed, randrange def load_csv(filename): """加载CSV文件""" dataset = [] with open(filename, 'r') as file: csv_reader = reader(file) for row in csv_reader: if not row: continue dataset.append(row) return dataset def str_column_to_float(dataset, column): """将字符串列转换为浮点数""" for row in dataset: row[column] = float(row[column].strip()) def str_column_to_int(dataset, column): """将类别字符串转换为整数编码""" class_values = [row[column] for row in dataset] unique = set(class_values) lookup = {} for i, value in enumerate(unique): lookup[value] = i print(f"[{value}] => {i}") # 打印类别映射 for row in dataset: row[column] = lookup[row[column]] return lookup

3.2 交叉验证评估

为了客观评估模型性能，我们使用5折交叉验证：

def cross_validation_split(dataset, n_folds): """将数据集分割为n折""" dataset_split = [] dataset_copy = list(dataset) fold_size = int(len(dataset) / n_folds) for _ in range(n_folds): fold = [] while len(fold) < fold_size: index = randrange(len(dataset_copy)) fold.append(dataset_copy.pop(index)) dataset_split.append(fold) return dataset_split def accuracy_metric(actual, predicted): """计算准确率""" correct = 0 for i in range(len(actual)): if actual[i] == predicted[i]: correct += 1 return correct / float(len(actual)) * 100.0 def evaluate_algorithm(dataset, algorithm, n_folds, *args): """评估算法性能""" folds = cross_validation_split(dataset, n_folds) scores = [] for fold in folds: train_set = list(folds) train_set.remove(fold) train_set = sum(train_set, []) # 合并其他折作为训练集 test_set = [] for row in fold: row_copy = list(row) test_set.append(row_copy) row_copy[-1] = None # 移除测试集的标签 predicted = algorithm(train_set, test_set, *args) actual = [row[-1] for row in fold] accuracy = accuracy_metric(actual, predicted) scores.append(accuracy) return scores

3.3 完整KNN算法实现

将前面的组件组合成完整的KNN算法：

def k_nearest_neighbors(train, test, num_neighbors): """KNN算法完整实现""" predictions = [] for row in test: output = predict_classification(train, row, num_neighbors) predictions.append(output) return predictions

3.4 运行评估

现在我们可以加载鸢尾花数据集并评估我们的KNN实现：

# 加载数据集 filename = 'iris.csv' dataset = load_csv(filename) # 转换数据类型 for i in range(len(dataset[0])-1): str_column_to_float(dataset, i) str_column_to_int(dataset, len(dataset[0])-1) # 评估参数 n_folds = 5 num_neighbors = 5 seed(1) # 固定随机种子确保结果可复现 # 评估算法 scores = evaluate_algorithm(dataset, k_nearest_neighbors, n_folds, num_neighbors) print(f"Scores: {scores}") print(f"Mean Accuracy: {sum(scores)/float(len(scores)):.3f}%")

典型输出结果：

Scores: [96.667, 96.667, 100.0, 90.0, 100.0] Mean Accuracy: 96.667%

这个结果明显优于33%的基准准确率，说明我们的实现是有效的。

4. 实际应用与优化建议

4.1 进行新样本预测

训练好模型后，我们可以预测新样本的类别：

# 定义新样本 new_sample = [5.7, 2.9, 4.2, 1.3] # 花萼长、花萼宽、花瓣长、花瓣宽 # 预测类别 predicted_label = predict_classification(dataset, new_sample, num_neighbors) print(f"Data={new_sample}, Predicted: {predicted_label}")

4.2 性能优化技巧

数据标准化：不同特征的量纲不同会影响距离计算。可以使用Min-Max标准化或Z-score标准化：

def normalize_dataset(dataset): """Min-Max标准化""" minmax = [] for i in range(len(dataset[0])): col_values = [row[i] for row in dataset] value_min = min(col_values) value_max = max(col_values) minmax.append([value_min, value_max]) for row in dataset: for i in range(len(row)-1): # 不标准化标签列 row[i] = (row[i] - minmax[i][0]) / (minmax[i][1] - minmax[i][0])

KD树优化：对于大数据集，线性搜索最近邻居效率低。可以使用KD树数据结构：

from collections import namedtuple from math import inf class KDNode: def __init__(self, point, left=None, right=None): self.point = point self.left = left self.right = right def build_kdtree(points, depth=0): if not points: return None k = len(points[0]) - 1 # 特征维度(忽略标签) axis = depth % k points.sort(key=lambda x: x[axis]) median = len(points) // 2 return KDNode( point=points[median], left=build_kdtree(points[:median], depth+1), right=build_kdtree(points[median+1:], depth+1) )