Python中KNN算法常用代码库及实战使用指南

from sklearn.neighbors import KNeighborsClassifier, KNeighborsRegressor
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
 
# 加载数据集
iris = load_iris()
X, y = iris.data, iris.target
 
# 数据预处理
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
 
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(
    X_scaled, y, test_size=0.3, random_state=42
)
 
# 创建KNN分类器
knn = KNeighborsClassifier(n_neighbors=5, metric='euclidean')
knn.fit(X_train, y_train)
 
# 预测和评估
accuracy = knn.score(X_test, y_test)
print(f"准确率: {accuracy:.4f}")

import numpy as np
from pynndescent import NNDescent
 
# 生成示例数据
data = np.random.randn(10000, 128)
 
# 构建近邻索引
index = NNDescent(
    data,
    n_neighbors=15,
    metric='euclidean',
    n_jobs=-1  # 使用所有CPU核心
)
 
# 查询近邻
query_data = np.random.randn(100, 128)
neighbors, distances = index.query(query_data, k=5)

from annoy import AnnoyIndex
import random
 
# 创建索引
f = 40  # 向量维度
t = AnnoyIndex(f, 'angular')  # 可选: 'angular', 'euclidean', 'manhattan'
 
# 添加数据点
for i in range(1000):
    v = [random.gauss(0, 1) for _ in range(f)]
    t.add_item(i, v)
 
# 构建索引树
t.build(10)  # 10棵树
 
# 保存和加载索引
t.save('test.ann')
u = AnnoyIndex(f, 'angular')
u.load('test.ann')
 
# 查询最近邻
nearest = u.get_nns_by_item(0, 10)  # 获取第0个点的10个最近邻

import faiss
import numpy as np
 
# 准备数据
d = 64  # 维度
nb = 100000  # 数据库大小
nq = 10000  # 查询数量
 
np.random.seed(1234)
xb = np.random.random((nb, d)).astype('float32')
xb[:, 0] += np.arange(nb) / 1000.
xq = np.random.random((nq, d)).astype('float32')
xq[:, 0] += np.arange(nq) / 1000.
 
# 构建索引
index = faiss.IndexFlatL2(d)  # L2距离
index.add(xb)  # 添加向量到索引
 
# 搜索
k = 4  # 查找4个最近邻
D, I = index.search(xq, k)  # D是距离，I是索引
print(f"前5个查询的最近邻索引:\n{I[:5]}")

import pandas as pd
import numpy as np
from sklearn.neighbors import NearestNeighbors
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
 
class MovieRecommender:
    """基于KNN的电影推荐系统"""
    
    def __init__(self, n_neighbors=10):
        self.n_neighbors = n_neighbors
        self.model = None
        self.scaler = StandardScaler()
        self.pca = PCA(n_components=50)
        self.movies_df = None
        self.features_scaled = None
        
    def prepare_data(self, movies_data):
        """准备和预处理数据"""
        self.movies_df = movies_data
        
        # 提取特征（这里使用示例特征）
        features = self.movies_df[[
            'rating', 'runtime', 'year', 
            'action', 'comedy', 'drama', 'thriller'
        ]].fillna(0)
        
        # 标准化特征
        features_scaled = self.scaler.fit_transform(features)
        
        # 降维（可选）
        if features_scaled.shape[1] > 50:
            features_scaled = self.pca.fit_transform(features_scaled)
            
        self.features_scaled = features_scaled
        return features_scaled
    
    def fit(self):
        """训练KNN模型"""
        self.model = NearestNeighbors(
            n_neighbors=self.n_neighbors,
            algorithm='ball_tree',
            metric='cosine'
        )
        self.model.fit(self.features_scaled)
        
    def recommend(self, movie_id, n_recommendations=5):
        """为指定电影推荐相似电影"""
        if movie_id not in self.movies_df.index:
            raise ValueError(f"电影ID {movie_id} 不存在")
            
        # 获取电影的特征向量
        movie_idx = self.movies_df.index.get_loc(movie_id)
        movie_features = self.features_scaled[movie_idx].reshape(1, -1)
        
        # 查找最近邻
        distances, indices = self.model.kneighbors(
            movie_features, 
            n_neighbors=n_recommendations + 1
        )
        
        # 排除自身
        similar_indices = indices[0][1:]
        similar_distances = distances[0][1:]
        
        # 构建推荐结果
        recommendations = []
        for idx, dist in zip(similar_indices, similar_distances):
            movie = self.movies_df.iloc[idx]
            recommendations.append({
                'title': movie['title'],
                'similarity': 1 - dist,  # 转换为相似度
                'rating': movie['rating'],
                'year': movie['year']
            })
            
        return pd.DataFrame(recommendations)
    
    def visualize_recommendations(self, movie_id, n_recommendations=5):
        """可视化推荐结果"""
        recommendations = self.recommend(movie_id, n_recommendations)
        
        # 创建可视化
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
        
        # 相似度条形图
        ax1.barh(recommendations['title'], recommendations['similarity'])
        ax1.set_xlabel('相似度')
        ax1.set_title(f'与《{self.movies_df.loc[movie_id, "title"]}》相似的电影')
        ax1.set_xlim([0, 1])
        
        # 评分散点图
        ax2.scatter(recommendations['year'], recommendations['rating'], 
                   s=recommendations['similarity']*500, alpha=0.6)
        for i, txt in enumerate(recommendations['title']):
            ax2.annotate(txt, (recommendations.iloc[i]['year'], 
                              recommendations.iloc[i]['rating']),
                        fontsize=8, ha='center')
        ax2.set_xlabel('年份')
        ax2.set_ylabel('评分')
        ax2.set_title('推荐电影的年份与评分分布')
        
        plt.tight_layout()
        plt.show()
        
        return recommendations
 
# 使用示例
if __name__ == "__main__":
    # 创建示例数据
    movies_data = pd.DataFrame({
        'title': ['The Matrix', 'Inception', 'Interstellar', 'The Dark Knight',
                 'Pulp Fiction', 'Fight Club', 'The Shawshank Redemption',
                 'The Godfather', 'Forrest Gump', 'The Avengers'],
        'rating': [8.7, 8.8, 8.6, 9.0, 8.9, 8.8, 9.3, 9.2, 8.8, 8.0],
        'runtime': [136, 148, 169, 152, 154, 139, 142, 175, 142, 143],
        'year': [1999, 2010, 2014, 2008, 1994, 1999, 1994, 1972, 1994, 2012],
        'action': [1, 1, 0, 1, 0, 1, 0, 0, 0, 1],
        'comedy': [0, 0, 0, 0, 1, 0, 0, 0, 1, 0],
        'drama': [0, 1, 1, 1, 1, 1, 1, 1, 1, 0],
        'thriller': [1, 1, 1, 1, 1, 1, 0, 1, 0, 0]
    })
    
    # 初始化推荐系统
    recommender = MovieRecommender(n_neighbors=5)
    recommender.prepare_data(movies_data)
    recommender.fit()
    
    # 获取推荐
    recommendations = recommender.recommend(0, n_recommendations=3)
    print("推荐结果：")
    print(recommendations)

from sklearn.preprocessing import MinMaxScaler, RobustScaler
from sklearn.feature_selection import SelectKBest, f_classif
 
def optimize_features(X, y, n_features=20):
    """特征优化流水线"""
    # 使用RobustScaler处理异常值
    scaler = RobustScaler()
    X_scaled = scaler.fit_transform(X)
    
    # 特征选择
    selector = SelectKBest(f_classif, k=n_features)
    X_selected = selector.fit_transform(X_scaled, y)
    
    return X_selected, selector.get_support()

from sklearn.neighbors import DistanceMetric
 
# 不同距离度量的比较
metrics = ['euclidean', 'manhattan', 'chebyshev', 'minkowski']
results = {}
 
for metric in metrics:
    knn = KNeighborsClassifier(n_neighbors=5, metric=metric)
    knn.fit(X_train, y_train)
    score = knn.score(X_test, y_test)
    results[metric] = score
    
print("不同距离度量的性能：")
for metric, score in results.items():
    print(f"{metric}: {score:.4f}")

from sklearn.model_selection import GridSearchCV
 
def optimize_k(X_train, y_train, k_range=(1, 30)):
    """使用网格搜索优化K值"""
    param_grid = {
        'n_neighbors': list(range(k_range[0], k_range[1], 2)),
        'weights': ['uniform', 'distance'],
        'metric': ['euclidean', 'manhattan']
    }
    
    knn = KNeighborsClassifier()
    grid_search = GridSearchCV(
        knn, param_grid, 
        cv=5, 
        scoring='accuracy',
        n_jobs=-1
    )
    
    grid_search.fit(X_train, y_train)
    
    print(f"最佳参数: {grid_search.best_params_}")
    print(f"最佳得分: {grid_search.best_score_:.4f}")
    
    return grid_search.best_estimator_

from sklearn.neighbors import LocalOutlierFactor
import numpy as np
 
class KNNAnomalyDetector:
    """基于KNN的异常检测器"""
    
    def __init__(self, n_neighbors=20, contamination=0.1):
        self.n_neighbors = n_neighbors
        self.contamination = contamination
        self.lof = LocalOutlierFactor(
            n_neighbors=n_neighbors,
            contamination=contamination,
            novelty=True
        )
        
    def fit(self, X_train):
        """训练异常检测模型"""
        self.lof.fit(X_train)
        return self
        
    def detect_anomalies(self, X_test):
        """检测异常样本"""
        # 预测（-1表示异常，1表示正常）
        predictions = self.lof.predict(X_test)
        
        # 计算异常分数
        anomaly_scores = self.lof.score_samples(X_test)
        
        # 构建结果DataFrame
        results = pd.DataFrame({
            'is_anomaly': predictions == -1,
            'anomaly_score': -anomaly_scores,  # 转换为正值，越大越异常
            'data_index': range(len(X_test))
        })
        
        return results.sort_values('anomaly_score', ascending=False)
    
    def visualize_anomalies(self, X_test, feature_names=None):
        """可视化异常检测结果"""
        results = self.detect_anomalies(X_test)
        
        # 使用PCA降维到2D进行可视化
        pca = PCA(n_components=2)
        X_2d = pca.fit_transform(X_test)
        
        plt.figure(figsize=(10, 6))
        
        # 绘制正常点
        normal_mask = ~results['is_anomaly']
        plt.scatter(X_2d[normal_mask, 0], X_2d[normal_mask, 1], 
                   c='blue', label='正常', alpha=0.6)
        
        # 绘制异常点
        anomaly_mask = results['is_anomaly'].values
        plt.scatter(X_2d[anomaly_mask, 0], X_2d[anomaly_mask, 1], 
                   c='red', label='异常', s=100, marker='x')
        
        plt.xlabel(f'主成分1 ({pca.explained_variance_ratio_[0]:.2%}方差)')
        plt.ylabel(f'主成分2 ({pca.explained_variance_ratio_[1]:.2%}方差)')
        plt.title('KNN异常检测结果')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.show()
        
        return results
 
# 使用示例
if __name__ == "__main__":
    # 生成包含异常的数据
    np.random.seed(42)
    n_samples = 300
    n_outliers = 30
    
    # 正常数据
    X_normal = 0.3 * np.random.randn(n_samples, 2)
    
    # 异常数据
    X_outliers = np.random.uniform(low=-4, high=4, size=(n_outliers, 2))
    
    # 合并数据
    X = np.vstack([X_normal, X_outliers])
    
    # 创建检测器
    detector = KNNAnomalyDetector(n_neighbors=20, contamination=0.1)
    detector.fit(X_normal)  # 只用正常数据训练
    
    # 检测异常
    anomalies = detector.visualize_anomalies(X)
    print(f"\n检测到 {anomalies['is_anomaly'].sum()} 个异常点")
    print("\n异常分数最高的5个样本：")
    print(anomalies.head())

# Trae IDE会自动识别并建议优化的代码模式
# 原始代码
distances = []
for point in test_points:
    dist = np.sqrt(np.sum((train_point - point) ** 2))
    distances.append(dist)
 
# Trae IDE建议的优化版本
distances = np.linalg.norm(train_points - test_points[:, np.newaxis], axis=2)

import unittest
from sklearn.datasets import make_classification
 
class TestKNNImplementation(unittest.TestCase):
    """KNN实现的单元测试"""
    
    def setUp(self):
        """准备测试数据"""
        self.X, self.y = make_classification(
            n_samples=100, 
            n_features=20, 
            n_informative=15,
            n_redundant=5,
            random_state=42
        )
        
    def test_knn_accuracy(self):
        """测试KNN分类准确率"""
        knn = KNeighborsClassifier(n_neighbors=5)
        knn.fit(self.X[:80], self.y[:80])
        score = knn.score(self.X[80:], self.y[80:])
        self.assertGreater(score, 0.7, "准确率应大于70%")
        
    def test_different_k_values(self):
        """测试不同K值的影响"""
        k_values = [3, 5, 7, 9]
        scores = []
        
        for k in k_values:
            knn = KNeighborsClassifier(n_neighbors=k)
            knn.fit(self.X[:80], self.y[:80])
            score = knn.score(self.X[80:], self.y[80:])
            scores.append(score)
            
        # 确保所有K值都能产生合理的结果
        self.assertTrue(all(s > 0.5 for s in scores))
        
if __name__ == '__main__':
    unittest.main()

import time
import numpy as np
from sklearn.neighbors import KNeighborsClassifier
from sklearn.datasets import make_classification
import pandas as pd
 
class KNNBenchmark:
    """KNN算法性能基准测试"""
    
    def __init__(self):
        self.results = []
        
    def benchmark_sklearn(self, X_train, y_train, X_test):
        """测试scikit-learn实现"""
        start_time = time.time()
        
        knn = KNeighborsClassifier(n_neighbors=5, algorithm='ball_tree')
        knn.fit(X_train, y_train)
        predictions = knn.predict(X_test)
        
        elapsed_time = time.time() - start_time
        
        return {
            'library': 'scikit-learn',
            'time': elapsed_time,
            'predictions': predictions
        }
    
    def benchmark_faiss(self, X_train, X_test, k=5):
        """测试FAISS实现"""
        import faiss
        
        start_time = time.time()
        
        # 构建索引
        index = faiss.IndexFlatL2(X_train.shape[1])
        index.add(X_train.astype('float32'))
        
        # 搜索
        D, I = index.search(X_test.astype('float32'), k)
        
        elapsed_time = time.time() - start_time
        
        return {
            'library': 'FAISS',
            'time': elapsed_time,
            'neighbors': I
        }
    
    def run_benchmark(self, n_samples=10000, n_features=50, n_test=1000):
        """运行完整基准测试"""
        print(f"生成数据集: {n_samples} 样本, {n_features} 特征")
        
        X, y = make_classification(
            n_samples=n_samples,
            n_features=n_features,
            n_informative=int(n_features * 0.8),
            random_state=42
        )
        
        X_train, y_train = X[:-n_test], y[:-n_test]
        X_test = X[-n_test:]
        
        # 运行各种实现的基准测试
        results = []
        
        # scikit-learn
        sklearn_result = self.benchmark_sklearn(X_train, y_train, X_test)
        results.append(sklearn_result)
        
        # FAISS
        faiss_result = self.benchmark_faiss(X_train, X_test)
        results.append(faiss_result)
        
        # 汇总结果
        df_results = pd.DataFrame(results)[['library', 'time']]
        df_results['samples_per_second'] = n_test / df_results['time']
        
        print("\n基准测试结果：")
        print(df_results.to_string(index=False))
        
        return df_results
 
# 运行基准测试
if __name__ == "__main__":
    benchmark = KNNBenchmark()
    results = benchmark.run_benchmark(
        n_samples=50000,
        n_features=100,
        n_test=5000
    )

def auto_tune_knn(X_train, y_train, X_val, y_val):
    """自动调优KNN参数"""
    best_score = 0
    best_params = {}
    
    # K值范围
    k_range = range(3, min(30, len(X_train) // 10), 2)
    
    # 权重选项
    weights = ['uniform', 'distance']
    
    # 算法选项
    algorithms = ['ball_tree', 'kd_tree', 'brute']
    
    for k in k_range:
        for weight in weights:
            for algorithm in algorithms:
                try:
                    knn = KNeighborsClassifier(
                        n_neighbors=k,
                        weights=weight,
                        algorithm=algorithm
                    )
                    knn.fit(X_train, y_train)
                    score = knn.score(X_val, y_val)
                    
                    if score > best_score:
                        best_score = score
                        best_params = {
                            'n_neighbors': k,
                            'weights': weight,
                            'algorithm': algorithm
                        }
                except:
                    continue
    
    print(f"最佳参数: {best_params}")
    print(f"验证集得分: {best_score:.4f}")
    
    return best_params

def memory_efficient_knn(X_train, batch_size=1000):
    """内存高效的KNN实现"""
    from sklearn.neighbors import NearestNeighbors
    import gc
    
    # 使用较小的数据类型
    if X_train.dtype == np.float64:
        X_train = X_train.astype(np.float32)
    
    # 分批构建索引
    nn = NearestNeighbors(n_neighbors=5, algorithm='ball_tree')
    
    for i in range(0, len(X_train), batch_size):
        batch = X_train[i:i+batch_size]
        if i == 0:
            nn.fit(batch)
        else:
            # 增量更新（注意：sklearn不直接支持，这里仅作示例）
            pass
    
    # 强制垃圾回收
    gc.collect()
    
    return nn