K近邻（KNN）：最简单的机器学习算法

一、KNN要解决什么问题？

1.1 问题场景：物以类聚

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import warnings
warnings.filterwarnings('ignore')

print("=" * 60)
print("KNN的核心思想：物以类聚，人以群分")
print("=" * 60)

# 生成示例数据
np.random.seed(42)
X, y = make_classification(n_samples=200, n_features=2, n_redundant=0,
                           n_clusters_per_class=1, n_classes=3,
                           random_state=42)

plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.scatter(X[y==0, 0], X[y==0, 1], c='blue', alpha=0.6, label='类别A')
plt.scatter(X[y==1, 0], X[y==1, 1], c='red', alpha=0.6, label='类别B')
plt.scatter(X[y==2, 0], X[y==2, 1], c='green', alpha=0.6, label='类别C')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.title('训练数据：已知类别的样本')
plt.legend()
plt.grid(True, alpha=0.3)

# 新样本点
new_point = np.array([0.5, 0.5])
plt.subplot(1, 2, 2)
plt.scatter(X[y==0, 0], X[y==0, 1], c='blue', alpha=0.6, label='类别A')
plt.scatter(X[y==1, 0], X[y==1, 1], c='red', alpha=0.6, label='类别B')
plt.scatter(X[y==2, 0], X[y==2, 1], c='green', alpha=0.6, label='类别C')
plt.scatter(new_point[0], new_point[1], c='black', s=200, marker='*', label='新样本?')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.title('新样本属于哪一类？\n找最近的邻居投票决定')
plt.legend()
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

print("\n💡 直观理解:")
print("   如果一个人周围的朋友都是程序员，那么他也很可能是程序员")
print("   KNN就是基于这个思想：一个样本的类别由它最近的K个邻居决定")

---

二、KNN的核心原理

2.1 KNN的三要素

def explain_knn_principles():
    """解释KNN的三要素"""
    
    fig, axes = plt.subplots(1, 3, figsize=(15, 5))
    
    # 1. K值选择
    ax1 = axes[0]
    ax1.axis('off')
    ax1.set_title('1. K值选择', fontsize=12)
    
    # 模拟邻居
    neighbors = [
        ("邻居1", "类别A", 0.7),
        ("邻居2", "类别A", 0.72),
        ("邻居3", "类别B", 0.75),
        ("邻居4", "类别B", 0.78),
        ("邻居5", "类别C", 0.8),
    ]
    
    y_pos = 0.8
    for name, label, dist in neighbors:
        color = 'blue' if label == '类别A' else ('red' if label == '类别B' else 'green')
        ax1.text(0.1, y_pos, f"{name}: {label}, 距离={dist}", fontsize=9, color=color)
        y_pos -= 0.12
    
    ax1.text(0.5, 0.2, "K=3 → 2个A, 1个B → 预测为A\nK=5 → 2个A, 2个B, 1个C → 平局需处理", 
            ha='center', fontsize=9,
            bbox=dict(boxstyle='round', facecolor='lightyellow'))
    
    # 2. 距离度量
    ax2 = axes[1]
    ax2.axis('off')
    ax2.set_title('2. 距离度量', fontsize=12)
    
    distance_formulas = """
    欧氏距离: d = √[(x₁-y₁)² + (x₂-y₂)² + ...]
    
    曼哈顿距离: d = |x₁-y₁| + |x₂-y₂| + ...
    
    闵可夫斯基距离: d = [Σ|xᵢ-yᵢ|ᵖ]^(1/p)
    • p=1: 曼哈顿距离
    • p=2: 欧氏距离
    • p=∞: 切比雪夫距离
    """
    
    ax2.text(0.05, 0.95, distance_formulas, transform=ax2.transAxes, fontsize=9,
            verticalalignment='top', fontfamily='monospace')
    
    # 3. 决策规则
    ax3 = axes[2]
    ax3.axis('off')
    ax3.set_title('3. 决策规则', fontsize=12)
    
    ax3.text(0.1, 0.8, "分类问题: 多数投票", fontsize=10, fontweight='bold')
    ax3.text(0.15, 0.7, "例: 3个邻居 → 2个A, 1个B → 预测为A", fontsize=9)
    
    ax3.text(0.1, 0.5, "回归问题: 平均值", fontsize=10, fontweight='bold')
    ax3.text(0.15, 0.4, "例: 3个邻居的值 [10, 12, 14] → 预测为12", fontsize=9)
    
    ax3.text(0.1, 0.2, "平局处理:", fontsize=10, fontweight='bold')
    ax3.text(0.15, 0.1, "• 随机选择\n• 减小K值\n• 距离加权投票", fontsize=8)
    
    plt.suptitle('KNN的三要素', fontsize=14)
    plt.tight_layout()
    plt.show()

explain_knn_principles()

2.2 K值的影响

def visualize_k_effect():
    """可视化K值对决策边界的影响"""
    
    # 生成数据
    from sklearn.datasets import make_moons
    X, y = make_moons(n_samples=200, noise=0.15, random_state=42)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)
    
    # 不同K值的KNN
    from sklearn.neighbors import KNeighborsClassifier
    k_values = [1, 5, 15, 30]
    
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    
    # 创建网格
    x_min, x_max = X[:, 0].min() - 0.5, X[:, 0].max() + 0.5
    y_min, y_max = X[:, 1].min() - 0.5, X[:, 1].max() + 0.5
    xx, yy = np.meshgrid(np.linspace(x_min, x_max, 200),
                         np.linspace(y_min, y_max, 200))
    
    for idx, k in enumerate(k_values):
        ax = axes[idx // 2, idx % 2]
        
        # 训练KNN
        knn = KNeighborsClassifier(n_neighbors=k)
        knn.fit(X_train, y_train)
        
        # 预测网格
        Z = knn.predict(np.c_[xx.ravel(), yy.ravel()])
        Z = Z.reshape(xx.shape)
        
        # 绘制决策边界
        ax.contourf(xx, yy, Z, alpha=0.3, cmap='RdBu')
        ax.scatter(X_train[y_train==0, 0], X_train[y_train==0, 1], 
                   c='blue', alpha=0.5, s=20, label='类别0')
        ax.scatter(X_train[y_train==1, 0], X_train[y_train==1, 1], 
                   c='red', alpha=0.5, s=20, label='类别1')
        
        train_acc = knn.score(X_train, y_train)
        test_acc = knn.score(X_test, y_test)
        
        ax.set_title(f'K = {k}\n训练准确率={train_acc:.3f}, 测试准确率={test_acc:.3f}')
        ax.set_xlabel('特征1')
        ax.set_ylabel('特征2')
        ax.legend()
        ax.grid(True, alpha=0.3)
    
    plt.suptitle('K值对决策边界的影响', fontsize=14)
    plt.tight_layout()
    plt.show()
    
    print("\n📊 K值选择建议:")
    print("   K太小 (如K=1): 决策边界复杂，容易过拟合")
    print("   K适中 (如K=5-15): 决策边界平滑，泛化好")
    print("   K太大 (如K=30): 决策边界过于平滑，可能欠拟合")
    print("   经验公式: K ≈ √n (n为样本数)")

visualize_k_effect()

---

三、从零实现KNN

3.1 完整实现

class KNNFromScratch:
    """从零实现的KNN分类器"""
    
    def __init__(self, k=3, distance_metric='euclidean'):
        """
        参数:
            k: 邻居数量
            distance_metric: 距离度量方式 ('euclidean', 'manhattan', 'minkowski')
        """
        self.k = k
        self.distance_metric = distance_metric
        self.X_train = None
        self.y_train = None
    
    def fit(self, X, y):
        """
        KNN的训练就是记住所有数据
        """
        self.X_train = X
        self.y_train = y
        return self
    
    def _euclidean_distance(self, a, b):
        """欧氏距离: √Σ(aᵢ - bᵢ)²"""
        return np.sqrt(np.sum((a - b) ** 2))
    
    def _manhattan_distance(self, a, b):
        """曼哈顿距离: Σ|aᵢ - bᵢ|"""
        return np.sum(np.abs(a - b))
    
    def _minkowski_distance(self, a, b, p=3):
        """闵可夫斯基距离: [Σ|aᵢ - bᵢ|ᵖ]^(1/p)"""
        return np.sum(np.abs(a - b) ** p) ** (1/p)
    
    def _compute_distance(self, a, b):
        """根据选择的度量方式计算距离"""
        if self.distance_metric == 'euclidean':
            return self._euclidean_distance(a, b)
        elif self.distance_metric == 'manhattan':
            return self._manhattan_distance(a, b)
        elif self.distance_metric == 'minkowski':
            return self._minkowski_distance(a, b)
        else:
            raise ValueError(f"未知的距离度量: {self.distance_metric}")
    
    def _predict_one(self, x):
        """
        预测单个样本的类别
        
        步骤:
        1. 计算与所有训练样本的距离
        2. 找到最近的K个邻居
        3. 统计邻居的类别
        4. 返回出现最多的类别
        """
        # 步骤1: 计算距离
        distances = []
        for i in range(len(self.X_train)):
            dist = self._compute_distance(x, self.X_train[i])
            distances.append((dist, self.y_train[i]))
        
        # 步骤2: 按距离排序，取前K个
        distances.sort(key=lambda x: x[0])
        k_nearest = distances[:self.k]
        
        # 步骤3: 统计邻居的类别
        votes = {}
        for _, label in k_nearest:
            votes[label] = votes.get(label, 0) + 1
        
        # 步骤4: 返回票数最多的类别
        predicted_label = max(votes, key=votes.get)
        return predicted_label
    
    def predict(self, X):
        """批量预测"""
        predictions = [self._predict_one(x) for x in X]
        return np.array(predictions)
    
    def predict_proba(self, X):
        """预测概率（用于多分类）"""
        probabilities = []
        for x in X:
            # 计算距离
            distances = []
            for i in range(len(self.X_train)):
                dist = self._compute_distance(x, self.X_train[i])
                distances.append((dist, self.y_train[i]))
            
            # 取前K个
            distances.sort(key=lambda x: x[0])
            k_nearest = distances[:self.k]
            
            # 计算每个类别的概率
            unique_labels = np.unique(self.y_train)
            probs = {}
            for label in unique_labels:
                count = sum(1 for _, l in k_nearest if l == label)
                probs[label] = count / self.k
            
            probabilities.append(probs)
        
        return probabilities
    
    def score(self, X, y):
        """计算准确率"""
        y_pred = self.predict(X)
        return np.mean(y_pred == y)

# 测试从零实现的KNN
print("\n" + "=" * 60)
print("从零实现的KNN测试")
print("=" * 60)

# 生成数据
from sklearn.datasets import make_classification
X, y = make_classification(n_samples=300, n_features=2, n_redundant=0,
                           n_clusters_per_class=1, n_classes=3,
                           random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)

# 测试不同K值
k_values = [1, 3, 5, 7, 9, 11, 15]
train_scores = []
test_scores = []

for k in k_values:
    knn = KNNFromScratch(k=k)
    knn.fit(X_train, y_train)
    train_acc = knn.score(X_train, y_train)
    test_acc = knn.score(X_test, y_test)
    train_scores.append(train_acc)
    test_scores.append(test_acc)
    print(f"K={k:2d}: 训练准确率={train_acc:.4f}, 测试准确率={test_acc:.4f}")

# 可视化K值影响
plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.plot(k_values, train_scores, 'bo-', label='训练集', linewidth=2)
plt.plot(k_values, test_scores, 'ro-', label='测试集', linewidth=2)
plt.xlabel('K值')
plt.ylabel('准确率')
plt.title('K值对模型性能的影响')
plt.legend()
plt.grid(True, alpha=0.3)

# 选择最佳K值
best_k = k_values[np.argmax(test_scores)]
knn_best = KNNFromScratch(k=best_k)
knn_best.fit(X_train, y_train)

# 绘制决策边界
plt.subplot(1, 2, 2)
x_min, x_max = X[:, 0].min() - 0.5, X[:, 0].max() + 0.5
y_min, y_max = X[:, 1].min() - 0.5, X[:, 1].max() + 0.5
xx, yy = np.meshgrid(np.linspace(x_min, x_max, 200),
                     np.linspace(y_min, y_max, 200))
Z = knn_best.predict(np.c_[xx.ravel(), yy.ravel()])
Z = Z.reshape(xx.shape)

plt.contourf(xx, yy, Z, alpha=0.3, cmap='tab10')
plt.scatter(X_train[y_train==0, 0], X_train[y_train==0, 1], c='blue', alpha=0.5, s=15)
plt.scatter(X_train[y_train==1, 0], X_train[y_train==1, 1], c='red', alpha=0.5, s=15)
plt.scatter(X_train[y_train==2, 0], X_train[y_train==2, 1], c='green', alpha=0.5, s=15)
plt.title(f'KNN决策边界 (K={best_k})\n测试准确率={knn_best.score(X_test, y_test):.3f}')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

---

四、距离度量的影响

def compare_distance_metrics():
    """对比不同距离度量"""
    
    # 生成数据
    np.random.seed(42)
    X, y = make_classification(n_samples=200, n_features=2, n_redundant=0,
                               n_clusters_per_class=1, random_state=42)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)
    
    metrics = ['euclidean', 'manhattan']
    fig, axes = plt.subplots(1, 2, figsize=(14, 5))
    
    for idx, metric in enumerate(metrics):
        knn = KNNFromScratch(k=5, distance_metric=metric)
        knn.fit(X_train, y_train)
        
        # 绘制决策边界
        x_min, x_max = X[:, 0].min() - 0.5, X[:, 0].max() + 0.5
        y_min, y_max = X[:, 1].min() - 0.5, X[:, 1].max() + 0.5
        xx, yy = np.meshgrid(np.linspace(x_min, x_max, 200),
                             np.linspace(y_min, y_max, 200))
        Z = knn.predict(np.c_[xx.ravel(), yy.ravel()])
        Z = Z.reshape(xx.shape)
        
        axes[idx].contourf(xx, yy, Z, alpha=0.3, cmap='RdBu')
        axes[idx].scatter(X_train[y_train==0, 0], X_train[y_train==0, 1], 
                         c='blue', alpha=0.5, s=15)
        axes[idx].scatter(X_train[y_train==1, 0], X_train[y_train==1, 1], 
                         c='red', alpha=0.5, s=15)
        
        test_acc = knn.score(X_test, y_test)
        axes[idx].set_title(f'距离度量: {metric}\n测试准确率={test_acc:.3f}')
        axes[idx].set_xlabel('特征1')
        axes[idx].set_ylabel('特征2')
        axes[idx].grid(True, alpha=0.3)
    
    plt.suptitle('不同距离度量对KNN的影响', fontsize=14)
    plt.tight_layout()
    plt.show()
    
    print("\n💡 距离度量选择建议:")
    print("   欧氏距离: 特征尺度相近时使用")
    print("   曼哈顿距离: 特征尺度差异大时使用")
    print("   闵可夫斯基: 通过p参数调节")

compare_distance_metrics()

---

五、KNN的优缺点与应用

def knn_pros_cons():
    """KNN的优缺点分析"""
    
    fig, axes = plt.subplots(1, 2, figsize=(14, 6))
    
    # 1. 优缺点对比
    ax1 = axes[0]
    ax1.axis('off')
    ax1.set_title('KNN的优缺点', fontsize=12)
    
    pros_cons = """
    ✅ 优点:
    • 简单直观，易于理解
    • 无需训练（懒惰学习）
    • 对异常值不敏感
    • 可用于分类和回归
    • 非线性决策边界
    
    ❌ 缺点:
    • 预测慢（需要计算所有距离）
    • 内存占用大（存储所有数据）
    • 对特征尺度敏感（需要标准化）
    • 维度灾难（高维数据失效）
    • K值选择困难
    """
    
    ax1.text(0.05, 0.95, pros_cons, transform=ax1.transAxes, fontsize=10,
            verticalalignment='top', fontfamily='monospace')
    
    # 2. 维度灾难演示
    ax2 = axes[1]
    
    # 模拟高维下距离失效
    dimensions = np.arange(1, 50)
    # 随机生成点，计算最小距离与最大距离的比值
    ratios = []
    for d in dimensions:
        points = np.random.rand(100, d)
        # 计算所有点到原点的距离
        distances = np.linalg.norm(points, axis=1)
        ratio = np.min(distances) / np.max(distances)
        ratios.append(ratio)
    
    ax2.plot(dimensions, ratios, 'b-', linewidth=2)
    ax2.set_xlabel('维度')
    ax2.set_ylabel('最小距离 / 最大距离')
    ax2.set_title('维度灾难：高维下所有点距离趋同')
    ax2.grid(True, alpha=0.3)
    ax2.axhline(y=0.5, color='r', linestyle='--', label='随机水平')
    ax2.legend()
    
    plt.suptitle('KNN的优缺点及维度灾难', fontsize=14)
    plt.tight_layout()
    plt.show()
    
    print("\n💡 改进方法:")
    print("   1. 特征缩放: 标准化/归一化")
    print("   2. 降维: PCA/t-SNE")
    print("   3. 使用KD树/球树加速搜索")
    print("   4. 距离加权投票")

knn_pros_cons()

---

六、实战：KNN完整流程

def complete_knn_pipeline():
    """完整的KNN实战流程"""
    
    print("\n" + "=" * 60)
    print("完整KNN实战流程（鸢尾花分类）")
    print("=" * 60)
    
    from sklearn.datasets import load_iris
    from sklearn.preprocessing import StandardScaler
    from sklearn.neighbors import KNeighborsClassifier
    from sklearn.model_selection import cross_val_score
    
    # 1. 加载数据
    iris = load_iris()
    X, y = iris.data, iris.target
    
    print(f"\n数据集: {X.shape[0]}个样本, {X.shape[1]}个特征")
    print(f"类别: {iris.target_names}")
    
    # 2. 划分数据
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
    
    # 3. 特征标准化（KNN对尺度敏感）
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 4. 交叉验证选择K值
    k_range = range(1, 21)
    cv_scores = []
    
    for k in k_range:
        knn = KNeighborsClassifier(n_neighbors=k)
        scores = cross_val_score(knn, X_train_scaled, y_train, cv=5)
        cv_scores.append(scores.mean())
    
    best_k = k_range[np.argmax(cv_scores)]
    print(f"\n交叉验证选择的最佳K值: {best_k}")
    
    # 5. 训练最终模型
    knn_final = KNeighborsClassifier(n_neighbors=best_k)
    knn_final.fit(X_train_scaled, y_train)
    
    # 6. 评估
    train_acc = knn_final.score(X_train_scaled, y_train)
    test_acc = knn_final.score(X_test_scaled, y_test)
    print(f"训练集准确率: {train_acc:.4f}")
    print(f"测试集准确率: {test_acc:.4f}")
    
    # 7. 预测新样本
    new_sample = np.array([[5.1, 3.5, 1.4, 0.2]])  # 山鸢尾的特征
    new_sample_scaled = scaler.transform(new_sample)
    prediction = knn_final.predict(new_sample_scaled)
    print(f"\n新样本预测: {iris.target_names[prediction[0]]}")
    
    # 可视化交叉验证结果
    plt.figure(figsize=(12, 4))
    
    plt.subplot(1, 2, 1)
    plt.plot(k_range, cv_scores, 'bo-', linewidth=2)
    plt.xlabel('K值')
    plt.ylabel('交叉验证准确率')
    plt.title(f'K值选择 (最佳K={best_k})')
    plt.grid(True, alpha=0.3)
    
    plt.subplot(1, 2, 2)
    # 使用PCA降维可视化决策边界
    from sklearn.decomposition import PCA
    pca = PCA(n_components=2)
    X_pca = pca.fit_transform(X_train_scaled)
    
    knn_vis = KNeighborsClassifier(n_neighbors=best_k)
    knn_vis.fit(X_pca, y_train)
    
    x_min, x_max = X_pca[:, 0].min() - 0.5, X_pca[:, 0].max() + 0.5
    y_min, y_max = X_pca[:, 1].min() - 0.5, X_pca[:, 1].max() + 0.5
    xx, yy = np.meshgrid(np.linspace(x_min, x_max, 200),
                         np.linspace(y_min, y_max, 200))
    Z = knn_vis.predict(np.c_[xx.ravel(), yy.ravel()])
    Z = Z.reshape(xx.shape)
    
    plt.contourf(xx, yy, Z, alpha=0.3, cmap='tab10')
    plt.scatter(X_pca[y_train==0, 0], X_pca[y_train==0, 1], c='blue', alpha=0.5, label='山鸢尾')
    plt.scatter(X_pca[y_train==1, 0], X_pca[y_train==1, 1], c='red', alpha=0.5, label='变色鸢尾')
    plt.scatter(X_pca[y_train==2, 0], X_pca[y_train==2, 1], c='green', alpha=0.5, label='维吉尼亚鸢尾')
    plt.xlabel('第一主成分')
    plt.ylabel('第二主成分')
    plt.title('KNN决策边界（PCA降维可视化）')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

complete_knn_pipeline()

---

七、代码讲解与总结

def code_explanation():
    """代码讲解"""
    
    print("\n" + "=" * 60)
    print("代码关键点讲解")
    print("=" * 60)
    
    explanations = {
        "距离计算": """
    # 欧氏距离
    def euclidean_distance(a, b):
        return np.sqrt(np.sum((a - b) ** 2))
    
    理解：直线距离，最常用
    注意：特征需要在同一尺度下
    """,
    
        "KNN预测": """
    def predict_one(x):
        # 1. 计算所有距离
        distances = [dist(x, x_train) for x_train in X_train]
        
        # 2. 找最近的K个
        k_indices = np.argsort(distances)[:k]
        
        # 3. 多数投票
        k_labels = y_train[k_indices]
        return np.bincount(k_labels).argmax()
    
    时间复杂度: O(n) 每次预测
    空间复杂度: O(n) 存储所有数据
    """,
    
        "K值选择": """
    交叉验证选择最佳K值:
    for k in range(1, 21):
        scores = cross_val_score(KNN(k), X, y, cv=5)
        avg_score = scores.mean()
    
    选择使交叉验证得分最高的K值
    """
    }
    
    for title, content in explanations.items():
        print(f"\n📌 {title}")
        print(content)

code_explanation()

---

八、总结

KNN核心要点：

要素	作用	选择建议
K值	邻居数量	√n，交叉验证选择
距离度量	计算相似度	欧氏距离最常用
决策规则	确定输出	分类用投票，回归用平均

KNN vs 其他算法：

特性	KNN	决策树	SVM
训练时间	无	快	中等
预测时间	慢	快	快
可解释性	低	高	低
内存占用	高	低	低
非线性	支持	支持	核函数

使用建议：

• 小数据集、低维度 → KNN效果好
• 大数据集、高维度 → 先降维或用其他算法
• 需要可解释性 → 决策树
• 需要快速预测 → 其他算法

记住：

• KNN是"懒惰学习"的代表
• 没有训练过程，直接记住数据
• 特征缩放至关重要
• K值需要谨慎选择