---
title: "KNN算法笔记"
description: "K近邻（K-Nearest Neighbors, KNN）算法是机器学习中最直观、最易于理解的算法之一，完美体现了“物以类聚”这一朴素思想。这个笔记对KNN算法进行学习与探讨，从核心原理到数学公式，从基础实现到优化，并提供可视化工具帮助直观理解算法行为。"
---

> Documentation Index
> Fetch the complete documentation index at: https://petwan.tech/llms.txt
> Use this file to discover all available pages before exploring further.

# KNN算法笔记

K近邻（K-Nearest Neighbors, KNN）算法是机器学习中最直观、最易于理解的算法之一，完美体现了“物以类聚”这一朴素思想。这个笔记对KNN算法进行学习与探讨，从核心原理到数学公式，从基础实现到优化，并提供可视化工具帮助直观理解算法行为。

## 1. 概述与核心特性

K近邻算法是一种**监督学习**方法，可用于分类和回归任务。其本质是一种**非参数化**方法，意味着它不对数据的潜在分布做任何假设。同时，它属于**惰性学习**（lazy learning）或**基于实例的学习**（instance-based learning），与急切学习（eager learning）方法（如决策树、神经网络）形成对比。

### 1.1 核心特点

| 特征         | 说明                           | 影响                         |
| ------------ | ------------------------------ | ---------------------------- |
| **惰性学习** | 训练阶段只存储数据，不进行计算 | 训练快，预测慢               |
| **非参数化** | 不假设数据分布形式             | 适应复杂分布，但需要更多数据 |
| **基于实例** | 模型就是训练数据本身           | 需要存储全部训练数据         |
| **局部近似** | 基于局部邻居做决策             | 对局部结构敏感，全局结构忽略 |

### 1.2 算法流程概览

对于一个待预测样本，KNN的基本流程如下：

1. **计算距离**：计算查询点与所有训练样本的距离
2. **排序选择**：按距离升序排列，选择前K个最近邻
3. **聚合决策**：
   - 分类：多数投票
   - 回归：加权平均

> 💡 **重要启示**：KNN的核心假设是**局部一致性**——相近的样本具有相似的属性。这一假设在大多数现实问题中成立，但在边界不规则或噪声多的场景下可能失效。

## 2. 算法原理与数学表达

### 2.1 分类任务：多数投票机制

#### 2.1.1 基本形式

对于一个查询点，其预测类别由K个最近邻的多数类别决定：

其中：

- ：所有可能类别的集合
- ：第个最近邻的真实标签
- ：指示函数（当条件为真时值为1，否则为0）
- ：最近邻数量（超参数）

#### 2.1.2 加权投票机制

基本KNN中每个邻居的投票权重相等，但距离越近的邻居通常应具有更高的话语权。引入距离加权：

其中权重通常定义为距离的倒数：

这里是一个极小正值（如），防止除零错误。距离越近，权重越大，对决策的影响也越大。

### 2.2 回归任务：局部加权平均

#### 2.2.1 基本回归公式

对于回归问题，预测值是K个最近邻目标值的平均：

#### 2.2.2 加权回归公式

更常见的是距离加权平均，距离越近的邻居贡献越大：

权重函数可以有不同的选择：
1. **反距离加权**：，通常取1或2
2. **高斯核加权**：，其中是带宽参数
3. **三角核加权**：，其中是窗口宽度

### 2.3 算法实现示例

以下是一个基础的KNN分类器Python实现，展示算法核心逻辑：

```python title="算法实现示例"
import numpy as np
from collections import Counter
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

class BasicKNN:
    """
    基础的KNN分类器实现
    """
    def __init__(self, k=5, distance_metric='euclidean'):
        """
        初始化KNN分类器
        
        参数:
        - k: 最近邻数量
        - distance_metric: 距离度量方式 ('euclidean', 'manhattan')
        """
        self.k = k
        self.distance_metric = distance_metric
        self.X_train = None
        self.y_train = None
        
    def _calculate_distance(self, x1, x2):
        """计算两个样本之间的距离"""
        if self.distance_metric == 'euclidean':
            return np.sqrt(np.sum((x1 - x2) ** 2))
        elif self.distance_metric == 'manhattan':
            return np.sum(np.abs(x1 - x2))
        else:
            raise ValueError(f"不支持的距离度量: {self.distance_metric}")
    
    def fit(self, X, y):
        """
        训练KNN模型（惰性学习，仅存储数据）
        
        注意：实际应用中应对特征进行标准化处理
        """
        self.X_train = np.array(X)
        self.y_train = np.array(y)
        return self
    
    def predict(self, X):
        """预测单个或多个样本的类别"""
        X = np.array(X)
        # 处理单个样本情况
        if X.ndim == 1:
            X = X.reshape(1, -1)
            
        predictions = []
        for x_query in X:
            # 1. 计算与所有训练样本的距离
            distances = []
            for i, x_train in enumerate(self.X_train):
                dist = self._calculate_distance(x_query, x_train)
                distances.append((dist, i))
            
            # 2. 按距离排序，选择K个最近邻
            distances.sort(key=lambda x: x[0])
            k_nearest_indices = [idx for _, idx in distances[:self.k]]
            k_nearest_labels = self.y_train[k_nearest_indices]
            
            # 3. 多数投票
            label_counts = Counter(k_nearest_labels)
            predicted_label = label_counts.most_common(1)[0][0]
            predictions.append(predicted_label)
            
        return np.array(predictions) if len(predictions) > 1 else predictions[0]
    
    def predict_proba(self, X):
        """预测类别概率"""
        X = np.array(X)
        if X.ndim == 1:
            X = X.reshape(1, -1)
            
        proba_list = []
        for x_query in X:
            distances = []
            for i, x_train in enumerate(self.X_train):
                dist = self._calculate_distance(x_query, x_train)
                distances.append((dist, i))
            
            distances.sort(key=lambda x: x[0])
            k_nearest_indices = [idx for _, idx in distances[:self.k]]
            k_nearest_labels = self.y_train[k_nearest_indices]
            
            # 计算各类别比例作为概率估计
            label_counts = Counter(k_nearest_labels)
            total = sum(label_counts.values())
            proba = {label: count/total for label, count in label_counts.items()}
            proba_list.append(proba)
            
        return proba_list
```

### 2.4 使用scikit-learn的实现

实际应用中，推荐使用优化过的库实现：

```python title="使用scikit-learn的实现"
from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.datasets import load_iris
from sklearn.metrics import classification_report, confusion_matrix

# 加载数据
iris = load_iris()
X, y = iris.data, iris.target

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42, stratify=y
)

# 特征标准化 - 对KNN至关重要！
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 创建KNN分类器
knn = KNeighborsClassifier(
    n_neighbors=5,
    weights='distance',  # 使用距离加权
    algorithm='auto',    # 自动选择最优搜索算法
    p=2,                 # 欧氏距离
    metric='minkowski'
)

# 训练模型
knn.fit(X_train_scaled, y_train)

# 预测与评估
y_pred = knn.predict(X_test_scaled)
y_pred_proba = knn.predict_proba(X_test_scaled)

print("分类报告:")
print(classification_report(y_test, y_pred, target_names=iris.target_names))
print("\n混淆矩阵:")
print(confusion_matrix(y_test, y_pred))
```

> **重要实践提示**：即使是一维特征，也建议进行标准化处理；对于多维特征，**标准化是必须的**！否则，量纲大的特征将主导距离计算，导致模型性能下降。

## 3. 可视化理解：KNN决策边界

交互式可视化展示了KNN算法在不同K值下的决策边界变化。可以通过调整参数观察K值如何影响模型的复杂度与决策边界平滑度，通过调整K值，我们可以直观看到：
- **K=1**：每个训练样本都形成一个"势力范围"，决策边界极其复杂，对噪声敏感
- **K=3~10**：决策边界开始平滑，局部模式被捕捉
- **K>10**：决策边界过于平滑，可能丢失重要细节，尤其在类别边界复杂时

## 4. 距离度量：KNN的核心引擎

距离度量定义了"相似性"的数学表达，是KNN算法的核心。不同的距离度量适用于不同的数据特性和问题领域。

### 4.1 常用距离度量

#### 4.1.1 闵可夫斯基距离

**闵可夫斯基距离**是一个广义的距离：

- **曼哈顿距离**（，L1范数）：

  适用于网格状路径或特征独立的场景，对异常值比欧氏距离更鲁棒。

- **欧氏距离**（，L2范数）：

  最常用的距离度量，具有旋转不变性，但对特征尺度敏感。

- **切比雪夫距离**（，L∞范数）：

  适用于只关心最大差异分量的场景，如棋盘移动。

#### 4.1.2 余弦相似度与余弦距离

对于文本、推荐系统等高维稀疏数据：

**余弦相似度**：

**余弦距离**：

余弦相似度关注方向而非大小，适用于文本TF-IDF向量等场景。

#### 4.1.3 汉明距离

对于分类特征或二进制数据：

即不匹配的特征数量，适用于DNA序列比对、错误检测等。

#### 4.1.4 马氏距离

考虑特征相关性和方距离：

其中是协方差矩阵。当特征高度相关或尺度差异大时，马氏距离比欧氏距离更合适。

### 4.2 距离度量选择

| 数据/问题类型          | 推荐距离度量 | 原因                 |
| ---------------------- | ------------ | -------------------- |
| 连续数值特征，各向同性 | 欧氏距离     | 自然、直观、计算高效 |
| 网格数据、路径规划     | 曼哈顿距离   | 反映实际移动距离     |
| 高维稀疏数据（文本）   | 余弦相似度   | 关注方向而非大小     |
| 二进制/分类数据        | 汉明距离     | 计算不匹配数量       |
| 特征相关性强           | 马氏距离     | 考虑特征相关性       |

## 5. K值选择：偏差-方差权衡的艺术

K值是KNN中最重要的超参数，控制着模型的复杂度，直接影响偏差-方差权衡。

### 5.1 K值的影响分析

| K值范围     | 模型复杂度 | 决策边界         | 偏差 | 方差 | 适用场景               |
| ----------- | ---------- | ---------------- | ---- | ---- | ---------------------- |
| **K=1**     | 非常高     | 极其复杂，过拟合 | 低   | 高   | 数据非常干净，噪声极少 |
| **K=3~5**   | 高         | 复杂，捕捉细节   | 较低 | 较高 | 小数据集，类别边界清晰 |
| **K=7~15**  | 中等       | 平滑，平衡       | 中等 | 中等 | 一般情况，最常用范围   |
| **K=20~50** | 低         | 非常平滑，欠拟合 | 高   | 低   | 大数据集，噪声较多     |
| **K>50**    | 非常低     | 过于平滑         | 很高 | 很低 | 需要强正则化的场景     |

### 5.2 数学视角：K值与决策边界

从统计学习理论角度，KNN的泛化误差可以分解为：

对于KNN分类器：
- **偏差**：随着K增大而增加（模型更简单，可能欠拟合）
- **方差**：随着K增大而减小（对训练数据细节不敏感）
- **噪声**：数据固有噪声，与K无关

最优K值位于偏差和方差的平衡点。

### 5.3 交叉验证选择K值

<details>
<summary>Python 代码</summary>

```python title="交叉验证选择K值"
import numpy as np
import matplotlib.pyplot as plt
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.datasets import make_classification
from sklearn.preprocessing import StandardScaler

# 设置中文字体
plt.rcParams["font.sans-serif"] = ["SimHei"]  # 用黑体显示中文
plt.rcParams["axes.unicode_minus"] = False  # 正常显示负号


def find_optimal_k(X, y, k_range=(1, 50), cv_folds=10):
    """
    使用交叉验证寻找最优K值

    参数:
    - X: 特征矩阵
    - y: 目标向量
    - k_range: K值搜索范围 (start, end)
    - cv_folds: 交叉验证折数

    返回:
    - best_k: 最优K值
    - scores_dict: 各K值对应的平均准确率
    """
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    k_values = list(range(k_range[0], k_range[1] + 1))
    cv_mean_scores = []
    cv_std_scores = []

    for k in k_values:
        knn = KNeighborsClassifier(n_neighbors=k, weights="distance")
        cv = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=42)
        scores = cross_val_score(knn, X_scaled, y, cv=cv, scoring="accuracy")
        cv_mean_scores.append(scores.mean())
        cv_std_scores.append(scores.std())

    best_idx = np.argmax(cv_mean_scores)
    best_k = k_values[best_idx]
    best_score = cv_mean_scores[best_idx]

    # ===== 绘图美化 =====
    plt.figure(figsize=(6, 4))

    # 左图：整体K值趋势
    plt.subplot(1, 1, 1)
    plt.plot(k_values, cv_mean_scores, "b-", linewidth=2, label="平均准确率")
    plt.fill_between(
        k_values,
        np.array(cv_mean_scores) - np.array(cv_std_scores),
        np.array(cv_mean_scores) + np.array(cv_std_scores),
        alpha=0.2,
        color="skyblue",
    )
    plt.axvline(
        x=best_k,
        color="red",
        linestyle="--",
        label=f"最优K={best_k}\n准确率={best_score:.3f}",
    )
    plt.xlim(0, k_range[1])  # 横轴从0到最大K值
    plt.ylim(0.7, 1.0)  # 纵轴0～1
    plt.xlabel("K值")
    plt.ylabel("交叉验证准确率")
    plt.title("不同K值与模型准确率的关系")
    plt.legend(loc="upper right")
    plt.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

    # 打印结果
    print(f"最优K值: {best_k}")
    print(f"对应准确率: {best_score:.4f} ± {cv_std_scores[best_idx]:.4f}")
    print(f"K=1时的准确率: {cv_mean_scores[0]:.4f} ± {cv_std_scores[0]:.4f}")
    print(
        f"K={k_range[1]}时的准确率: {cv_mean_scores[-1]:.4f} ± {cv_std_scores[-1]:.4f}"
    )

    return best_k, dict(zip(k_values, cv_mean_scores))


# 示例使用
if __name__ == "__main__":
    X, y = make_classification(
        n_samples=1000,
        n_features=20,
        n_informative=15,
        n_redundant=5,
        n_classes=3,
        n_clusters_per_class=2,
        random_state=42,
    )

    best_k, scores = find_optimal_k(X, y, k_range=(1, 60))
```
</details>
### 5.4 经验法则与高级选择策略

#### 5.4.1 经验法则

1. **奇数K值**：避免二分类中的平票情况
2. **类别平衡时**：K值应大于类别数量

#### 5.4.2 自适应K值选择

传统KNN使用固定的K值，但实际数据密度可能不均匀。自适应KNN根据局部密度调整K值：

```python title="自适应K值选择"
class AdaptiveKNN:
    """自适应KNN：根据局部密度调整K值"""
    
    def __init__(self, k_min=3, k_max=15, density_threshold=0.1):
        self.k_min = k_min
        self.k_max = k_max
        self.density_threshold = density_threshold
        
    def estimate_local_density(self, x_query, distances):
        """估计查询点的局部密度"""
        # 使用K近邻的平均距离作为密度估计
        # 距离越小，密度越高
        avg_distance = np.mean(distances[:self.k_min])
        density = 1 / (avg_distance + 1e-8)
        return density
    
    def adaptive_k_selection(self, density):
        """根据密度选择K值"""
        # 密度高时使用小K值，密度低时使用大K值
        if density > self.density_threshold:
            return self.k_min
        else:
            # 线性插值
            k = int(self.k_min + (self.k_max - self.k_min) * 
                   (1 - density/self.density_threshold))
            return min(max(k, self.k_min), self.k_max)
    
    def predict(self, X_train, y_train, x_query):
        """自适应KNN预测"""
        # 计算所有距离
        distances = []
        for i, x_train in enumerate(X_train):
            dist = np.linalg.norm(x_query - x_train)
            distances.append((dist, i))
        
        distances.sort(key=lambda x: x[0])
        
        # 估计局部密度
        density = self.estimate_local_density(x_query, 
                                             [d for d, _ in distances])
        
        # 自适应选择K值
        k = self.adaptive_k_selection(density)
        
        # 使用选择的K值进行预测
        k_nearest_indices = [idx for _, idx in distances[:k]]
        k_nearest_labels = y_train[k_nearest_indices]
        
        # 多数投票
        from collections import Counter
        label_counts = Counter(k_nearest_labels)
        return label_counts.most_common(1)[0][0]
```

#### 5.4.3 基于验证集曲线的K值选择

除了交叉验证，还可以观察验证集曲线：

1. **训练集-验证集误差曲线**：
   - 训练误差随K增大而增加
   - 验证误差先减小后增大
   - 最优K在验证误差最小处

2. **学习曲线分析**：
   - 固定K值，观察不同训练集大小的表现
   - 判断KNN是否适合当前问题规模


## 6. 计算优化：从朴素实现到工业级应用

原始KNN的时间复杂度为（n个样本，d维特征），预测时需要计算查询点与所有训练点的距离。对于大数据集，这是不可接受的。

### 6.1 空间划分树结构

#### 6.1.1 KD树（K-Dimensional Tree）

KD树是一种二叉树，用于高效组织k维空间中的点。

```python title="KD树（K-Dimensional Tree）"
import numpy as np

class KDNode:
    """KD树节点"""
    def __init__(self, point=None, left=None, right=None, axis=None):
        self.point = point  # 节点存储的点
        self.left = left    # 左子树
        self.right = right  # 右子树
        self.axis = axis    # 划分轴

class KDTree:
    """KD树实现"""
    
    def __init__(self, points):
        self.root = self.build_tree(points)
        self.n_dim = points.shape[1] if len(points) > 0 else 0
    
    def build_tree(self, points, depth=0):
        """递归构建KD树"""
        if len(points) == 0:
            return None
        
        # 选择划分轴：轮转选择维度
        axis = depth % points.shape[1]
        
        # 按当前轴排序并选择中位数作为划分点
        sorted_points = points[points[:, axis].argsort()]
        median_idx = len(sorted_points) // 2
        
        # 创建节点并递归构建子树
        node = KDNode(
            point=sorted_points[median_idx],
            axis=axis
        )
        
        node.left = self.build_tree(sorted_points[:median_idx], depth + 1)
        node.right = self.build_tree(sorted_points[median_idx + 1:], depth + 1)
        
        return node
    
    def nearest_neighbor(self, query_point):
        """寻找最近邻"""
        best = [None, float('inf')]  # [最近点, 最近距离]
        
        def search(node, depth=0):
            if node is None:
                return
            
            # 计算当前节点距离
            dist = np.linalg.norm(query_point - node.point)
            
            # 更新最优解
            if dist < best[1]:
                best[0] = node.point
                best[1] = dist
            
            # 确定搜索方向
            axis = depth % self.n_dim
            next_branch = None
            opposite_branch = None
            
            if query_point[axis] < node.point[axis]:
                next_branch = node.left
                opposite_branch = node.right
            else:
                next_branch = node.right
                opposite_branch = node.left
            
            # 递归搜索主分支
            search(next_branch, depth + 1)
            
            # 检查是否需要搜索另一分支
            if abs(query_point[axis] - node.point[axis]) < best[1]:
                search(opposite_branch, depth + 1)
        
        search(self.root)
        return best[0], best[1]
    
    def k_nearest_neighbors(self, query_point, k=5):
        """寻找K个最近邻"""
        # 使用最大堆存储K个最近邻
        import heapq
        
        heap = []  # 最大堆，存储(-distance, point)
        
        def search(node, depth=0):
            if node is None:
                return
            
            # 计算距离
            dist = np.linalg.norm(query_point - node.point)
            
            # 维护大小为K的最大堆
            if len(heap) < k:
                heapq.heappush(heap, (-dist, node.point))
            elif dist < -heap[0][0]:  # 当前点比堆顶更近
                heapq.heapreplace(heap, (-dist, node.point))
            
            # 确定搜索方向
            axis = depth % self.n_dim
            next_branch = None
            opposite_branch = None
            
            if query_point[axis] < node.point[axis]:
                next_branch = node.left
                opposite_branch = node.right
            else:
                next_branch = node.right
                opposite_branch = node.left
            
            # 搜索主分支
            search(next_branch, depth + 1)
            
            # 检查是否需要搜索另一分支
            if len(heap) < k or abs(query_point[axis] - node.point[axis]) < -heap[0][0]:
                search(opposite_branch, depth + 1)
        
        search(self.root)
        
        # 提取结果并排序
        neighbors = [(-dist, point) for dist, point in heap]
        neighbors.sort(key=lambda x: x[0])  # 按距离升序排序
        return [(point, -dist) for dist, point in neighbors]

# KD树性能测试
def test_kdtree_performance():
    """比较朴素KNN与KD树的性能"""
    import time
    
    np.random.seed(42)
    
    # 生成测试数据
    n_samples = 10000
    n_dim = 10
    X = np.random.randn(n_samples, n_dim)
    
    # 测试查询点
    query_points = np.random.randn(10, n_dim)
    
    # 测试朴素KNN
    start_time = time.time()
    naive_results = []
    for query in query_points:
        distances = [np.linalg.norm(query - x) for x in X]
        nearest_idx = np.argmin(distances)
        naive_results.append((nearest_idx, distances[nearest_idx]))
    naive_time = time.time() - start_time
    
    # 构建KD树
    start_time = time.time()
    kdtree = KDTree(X)
    build_time = time.time() - start_time
    
    # 测试KD树查询
    start_time = time.time()
    kdtree_results = []
    for query in query_points:
        point, dist = kdtree.nearest_neighbor(query)
        kdtree_results.append((point, dist))
    query_time = time.time() - start_time
    
    print(f"数据集: {n_samples}个样本, {n_dim}维")
    print(f"朴素KNN查询时间: {naive_time:.4f}秒 (10次查询)")
    print(f"KD树构建时间: {build_time:.4f}秒")
    print(f"KD树查询时间: {query_time:.4f}秒 (10次查询)")
    print(f"KD树加速比: {naive_time/query_time:.2f}倍")
    
    # 验证结果一致性
    for i in range(len(query_points)):
        naive_dist = naive_results[i][1]
        kdtree_dist = kdtree_results[i][1]
        if abs(naive_dist - kdtree_dist) > 1e-6:
            print(f"警告: 第{i}个查询结果不一致")
    
    return kdtree, naive_time, query_time

if __name__ == "__main__":
    kdtree, naive_time, kd_time = test_kdtree_performance()
```

#### 6.1.2 Ball Tree

Ball Tree是另一种空间划分树，特别适用于高维数据：

- **核心思想**：将数据点组织在嵌套的超球体中
- **优势**：在高维空间中比KD树更高效
- **适用场景**：维度 > 20 的数据

```python title="Ball Tree"
from sklearn.neighbors import BallTree
import numpy as np

def ball_tree_example():
    """Ball Tree示例"""
    np.random.seed(42)
    X = np.random.randn(1000, 50)  # 50维数据
    
    # 构建Ball Tree
    tree = BallTree(X, leaf_size=40, metric='euclidean')
    
    # 查询
    query_point = np.random.randn(50)
    dist, idx = tree.query([query_point], k=5)
    
    print(f"最近邻索引: {idx[0]}")
    print(f"最近邻距离: {dist[0]}")
    
    # 半径查询：查找指定半径内的所有点
    radius = 3.0
    idx_radius = tree.query_radius([query_point], r=radius)
    print(f"半径{radius}内的点数: {len(idx_radius[0])}")
    
    return tree
```

### 6.2 近似最近邻搜索（ANN）

当精确性可以妥协以换取速度时，使用近似最近邻搜索：

#### 6.2.1 局部敏感哈希（LSH）
核心思想是：将原始空间中的相邻点以较高的概率映射到同一个哈希桶中，而不相邻的点则映射到不同的哈希桶中。这样，我们只需要在查询点所在的哈希桶内进行搜索，就可以找到近似最近邻，从而大大减少搜索时间。

```python title="局部敏感哈希（LSH）"
from sklearn.neighbors import LSHForest
import numpy as np

def lsh_forest_example():
    """LSH Forest近似最近邻搜索"""
    np.random.seed(42)
    n_samples = 100000
    n_features = 100
    
    # 生成大数据集
    X = np.random.randn(n_samples, n_features)
    
    # 构建LSH Forest
    lshf = LSHForest(
        n_estimators=50,      # 哈希表数量
        n_candidates=100,     # 候选点数量
        n_neighbors=10,       # 返回的最近邻数量
        random_state=42
    )
    
    lshf.fit(X)
    
    # 近似最近邻查询
    query = np.random.randn(n_features)
    distances, indices = lshf.kneighbors([query])
    
    print(f"近似最近邻索引: {indices[0][:5]}")
    print(f"近似距离: {distances[0][:5]}")
    
    # 与精确结果比较
    from sklearn.neighbors import NearestNeighbors
    nn = NearestNeighbors(n_neighbors=10)
    nn.fit(X)
    exact_distances, exact_indices = nn.kneighbors([query])
    
    # 计算召回率
    recall = len(set(indices[0]) & set(exact_indices[0])) / len(exact_indices[0])
    print(f"Top-10召回率: {recall:.3f}")
    
    return lshf, recall
```

## 7. 高级优化与改进方法

### 7.1 加权KNN的进阶策略

#### 7.1.1 核函数加权
传统的KNN使用均匀权重（所有邻居同等重要）或反距离权重（距离越近权重越大），但这些都是简单的线性关系。核函数加权引入了非线性平滑权重，更好地反映了"距离越近，影响越大"的直觉。

```python title="核函数加权"
def kernel_weighted_knn(X_train, y_train, x_query, k=5, kernel='gaussian', bandwidth=1.0):
    """核函数加权的KNN"""
    
    # 计算所有距离
    distances = np.linalg.norm(X_train - x_query, axis=1)
    
    # 选择K个最近邻
    k_indices = np.argsort(distances)[:k]
    k_distances = distances[k_indices]
    k_labels = y_train[k_indices]
    
    # 计算核权重
    if kernel == 'gaussian':
        weights = np.exp(-k_distances**2 / (2 * bandwidth**2))
    elif kernel == 'epanechnikov':
        weights = np.maximum(0, 1 - (k_distances / bandwidth)**2)
    elif kernel == 'triangular':
        weights = np.maximum(0, 1 - k_distances / bandwidth)
    elif kernel == 'biweight':
        weights = (1 - (k_distances / bandwidth)**2)**2
    else:  # uniform
        weights = np.ones_like(k_distances)
    
    # 归一化权重
    weights = weights / (np.sum(weights) + 1e-8)
    
    # 加权投票（分类）
    from collections import defaultdict
    weighted_votes = defaultdict(float)
    for label, weight in zip(k_labels, weights):
        weighted_votes[label] += weight
    
    # 返回预测结果
    predicted_label = max(weighted_votes.items(), key=lambda x: x[1])[0]
    
    return predicted_label, weighted_votes
```

#### 7.1.2 自适应带宽选择

带宽参数影响核加权的效果，可以自适应选择：

```python title="自适应带宽选择"
def adaptive_bandwidth_selection(distances, method='silverman'):
    """自适应带宽选择"""
    k = len(distances)
    
    if method == 'silverman':
        # Silverman规则
        std = np.std(distances)
        iqr = np.percentile(distances, 75) - np.percentile(distances, 25)
        sigma = min(std, iqr / 1.34)
        h = 1.06 * sigma * k**(-1/5)
        
    elif method == 'scott':
        # Scott规则
        std = np.std(distances)
        h = 1.06 * std * k**(-1/5)
        
    elif method == 'cross_validation':
        # 简化的交叉验证（实际应用需更复杂）
        h_range = np.linspace(0.1, 2.0, 20)
        best_h = h_range[0]
        best_score = -np.inf
        
        # 这里简化为使用留一法
        for h in h_range:
            scores = []
            for i in range(k):
                # 留一交叉验证
                mask = np.ones(k, dtype=bool)
                mask[i] = False
                
                # 使用带宽h预测第i个点
                # 简化实现，实际需要更完整
                pass
            
            # 选择最优h
            pass
        
        h = best_h
    
    else:  # 固定带宽
        h = 1.0
    
    return max(h, 0.1)  # 确保带宽为正
```

其他的优化与改进方法读者如果感兴趣，可以自行研究（例如距离度量学习，编辑紧邻和数据压缩，集成KNN方法等）。

## 8. 对数据预处理的要求

1. **特征标准化**：必须进行！使用StandardScaler或MinMaxScaler
2. **处理缺失值**：KNN对缺失值敏感，需要填充或删除
3. **类别特征编码**：使用独热编码或目标编码
4. **特征选择**：移除不相关特征，减轻维度灾难

Source: https://petwan.tech/20260117_knn/index.mdx