KNN算法笔记

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

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))

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]

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()

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

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

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

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)  # 确保带宽为正