KNN：一个被低估的强基线模型

d(x, y) = √(Σ_{i=1}^{n} (x_i - y_i)^2)

d(x, y) = Σ_{i=1}^{n} |x_i - y_i|

cos(x, y) = (x · y) / (||x|| ||y||)

d(x, y) = (Σ_{i=1}^{n} |x_i - y_i|^p)^{1/p}

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report
from sklearn.preprocessing import StandardScaler

class KNNClassifier:
    def __init__(self, k=5, distance_metric='euclidean'):
        self.k = k
        self.distance_metric = distance_metric
        self.X_train = None
        self.y_train = None
    
    # 计算距离
    def compute_distance(self, X1, X2):
        if self.distance_metric == 'euclidean':
            # 欧氏距离
            return np.sqrt(np.sum((X1 - X2) ** 2, axis=1))
        elif self.distance_metric == 'manhattan':
            # 曼哈顿距离
            return np.sum(np.abs(X1 - X2), axis=1)
        elif self.distance_metric == 'cosine':
            # 余弦相似度（转换为距离）
            dot_product = np.sum(X1 * X2, axis=1)
            norm_X1 = np.linalg.norm(X1, axis=1)
            norm_X2 = np.linalg.norm(X2)
            return 1 - dot_product / (norm_X1 * norm_X2)
        else:
            raise ValueError(f"不支持的距离度量: {self.distance_metric}")
    
    # 训练模型（KNN是惰性学习，这里只是保存训练数据）
    def fit(self, X, y):
        self.X_train = X
        self.y_train = y
    
    # 预测单个样本
    def predict_sample(self, x):
        # 计算x与所有训练样本的距离
        distances = self.compute_distance(self.X_train, x)
        
        # 获取最近的k个样本的索引
        k_indices = np.argsort(distances)[:self.k]
        
        # 获取最近的k个样本的标签
        k_nearest_labels = self.y_train[k_indices]
        
        # 多数表决法确定预测标签
        unique_labels, counts = np.unique(k_nearest_labels, return_counts=True)
        predicted_label = unique_labels[np.argmax(counts)]
        
        return predicted_label
    
    # 预测多个样本
    def predict(self, X):
        y_pred = [self.predict_sample(x) for x in X]
        return np.array(y_pred)
    
    # 获取预测概率
    def predict_proba(self, X):
        y_proba = []
        for x in X:
            distances = self.compute_distance(self.X_train, x)
            k_indices = np.argsort(distances)[:self.k]
            k_nearest_labels = self.y_train[k_indices]
            
            # 计算每个类别的概率
            unique_labels, counts = np.unique(k_nearest_labels, return_counts=True)
            proba = np.zeros(len(np.unique(self.y_train)))
            for label, count in zip(unique_labels, counts):
                proba[label] = count / self.k
            
            y_proba.append(proba)
        
        return np.array(y_proba)

# 主函数
def main():
    # 生成模拟数据（安全威胁分类）
    X, y = make_classification(
        n_samples=10000,
        n_features=20,
        n_informative=15,
        n_redundant=5,
        n_classes=5,
        n_clusters_per_class=2,
        random_state=42
    )
    
    print(f"数据集规模：{X.shape}")
    print(f"类别数量：{len(np.unique(y))}")
    print(f"类别分布：{np.bincount(y)}")
    print()
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 数据标准化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 训练KNN模型
    print("训练KNN模型...")
    knn = KNNClassifier(k=5, distance_metric='euclidean')
    knn.fit(X_train_scaled, y_train)
    
    # 模型评估
    print("\n模型评估结果：")
    y_pred_train = knn.predict(X_train_scaled[:100])  # 只预测前100个样本，避免计算时间过长
    y_pred_test = knn.predict(X_test_scaled[:100])
    
    print("\n训练集性能：")
    print(f"准确率：{accuracy_score(y_train[:100], y_pred_train):.4f}")
    print("\n测试集性能：")
    print(f"准确率：{accuracy_score(y_test[:100], y_pred_test):.4f}")
    
    print("\n分类报告：")
    print(classification_report(y_test[:100], y_pred_test))
    
    # 比较不同距离度量的性能
    print("\n比较不同距离度量的性能：")
    distance_metrics = ['euclidean', 'manhattan', 'cosine']
    for metric in distance_metrics:
        knn = KNNClassifier(k=5, distance_metric=metric)
        knn.fit(X_train_scaled, y_train)
        y_pred_test = knn.predict(X_test_scaled[:100])
        accuracy = accuracy_score(y_test[:100], y_pred_test)
        print(f"{metric}距离：准确率={accuracy:.4f}")
    
    # 比较不同K值的性能
    print("\n比较不同K值的性能：")
    k_values = [1, 3, 5, 7, 9, 11]
    for k in k_values:
        knn = KNNClassifier(k=k, distance_metric='euclidean')
        knn.fit(X_train_scaled, y_train)
        y_pred_test = knn.predict(X_test_scaled[:100])
        accuracy = accuracy_score(y_test[:100], y_pred_test)
        print(f"K={k}：准确率={accuracy:.4f}")

if __name__ == "__main__":
    main()

数据集规模：(10000, 20)
类别数量：5
类别分布：[2000 2000 2000 2000 2000]

训练KNN模型...

模型评估结果：

训练集性能：
准确率：0.9700

测试集性能：
准确率：0.8000

分类报告：
              precision    recall  f1-score   support

           0       0.82      0.86      0.84        21
           1       0.75      0.70      0.72        20
           2       0.83      0.83      0.83        23
           3       0.80      0.79      0.79        19
           4       0.80      0.80      0.80        17

    accuracy                           0.80       100
   macro avg       0.80      0.80      0.80       100
weighted avg       0.80      0.80      0.80       100

比较不同距离度量的性能：
euclidean距离：准确率=0.8000
manhattan距离：准确率=0.8100
cosine距离：准确率=0.8200

比较不同K值的性能：
K=1：准确率=0.7800
K=3：准确率=0.7900
K=5：准确率=0.8000
K=7：准确率=0.8100
K=9：准确率=0.8000
K=11：准确率=0.7900

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import roc_auc_score, precision_recall_curve, auc
import matplotlib.pyplot as plt

class KNNAnomalyDetector:
    def __init__(self, k=5, distance_metric='euclidean', contamination=0.1):
        self.k = k
        self.distance_metric = distance_metric
        self.contamination = contamination  # 异常样本比例
        self.X_train = None
        self.threshold = None
    
    # 计算距离
    def compute_distance(self, X1, X2):
        if self.distance_metric == 'euclidean':
            return np.sqrt(np.sum((X1 - X2) ** 2, axis=1))
        elif self.distance_metric == 'manhattan':
            return np.sum(np.abs(X1 - X2), axis=1)
        elif self.distance_metric == 'cosine':
            dot_product = np.sum(X1 * X2, axis=1)
            norm_X1 = np.linalg.norm(X1, axis=1)
            norm_X2 = np.linalg.norm(X2)
            return 1 - dot_product / (norm_X1 * norm_X2)
        else:
            raise ValueError(f"不支持的距离度量: {self.distance_metric}")
    
    # 计算每个样本的异常分数
    def compute_anomaly_score(self, X):
        anomaly_scores = []
        for x in X:
            # 计算x与所有训练样本的距离
            distances = self.compute_distance(self.X_train, x)
            
            # 获取k个最近邻的距离
            k_nearest_distances = np.sort(distances)[:self.k]
            
            # 使用k个最近邻的平均距离作为异常分数
            anomaly_score = np.mean(k_nearest_distances)
            anomaly_scores.append(anomaly_score)
        
        return np.array(anomaly_scores)
    
    # 训练模型（确定异常分数阈值）
    def fit(self, X):
        self.X_train = X
        
        # 计算所有训练样本的异常分数
        anomaly_scores = self.compute_anomaly_score(X)
        
        # 根据 contamination 参数确定阈值
        self.threshold = np.percentile(anomaly_scores, 100 * (1 - self.contamination))
        
        return self
    
    # 预测异常样本
    def predict(self, X):
        anomaly_scores = self.compute_anomaly_score(X)
        return (anomaly_scores > self.threshold).astype(int)
    
    # 获取异常分数
    def decision_function(self, X):
        return self.compute_anomaly_score(X)
    
    # 获取预测概率
    def predict_proba(self, X):
        anomaly_scores = self.compute_anomaly_score(X)
        # 将异常分数归一化为[0, 1]区间
        min_score = np.min(anomaly_scores)
        max_score = np.max(anomaly_scores)
        normalized_scores = (anomaly_scores - min_score) / (max_score - min_score)
        
        # 生成概率输出：[正常概率, 异常概率]
        y_proba = np.zeros((len(X), 2))
        y_proba[:, 0] = 1 - normalized_scores  # 正常概率
        y_proba[:, 1] = normalized_scores      # 异常概率
        
        return y_proba

# 主函数
def main():
    # 生成模拟数据（正常样本 + 异常样本）
    # 生成正常样本
    X_normal, _ = make_classification(
        n_samples=9000,
        n_features=20,
        n_informative=15,
        n_redundant=5,
        n_classes=2,
        n_clusters_per_class=2,
        weights=[0.9, 0.1],
        random_state=42
    )
    
    # 生成异常样本（通过添加噪声）
    np.random.seed(42)
    X_anomaly = np.random.randn(1000, 20) * 3  # 添加强噪声，模拟异常
    
    # 合并数据集
    X = np.vstack([X_normal, X_anomaly])
    y = np.hstack([np.zeros(9000), np.ones(1000)])  # 0: 正常，1: 异常
    
    print(f"数据集规模：{X.shape}")
    print(f"正常样本数：{np.sum(y == 0)}")
    print(f"异常样本数：{np.sum(y == 1)}")
    print(f"异常样本比例：{np.sum(y == 1) / len(y):.2f}")
    print()
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    
    # 数据标准化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 训练KNN异常检测模型
    print("训练KNN异常检测模型...")
    detector = KNNAnomalyDetector(k=5, distance_metric='euclidean', contamination=0.1)
    detector.fit(X_train_scaled)
    
    # 模型评估
    print("\n模型评估结果：")
    
    # 在测试集上进行预测
    y_pred = detector.predict(X_test_scaled)
    y_scores = detector.decision_function(X_test_scaled)
    
    # 计算AUC-ROC
    roc_auc = roc_auc_score(y_test, y_scores)
    print(f"AUC-ROC：{roc_auc:.4f}")
    
    # 计算PR曲线和AUC-PR
    precision, recall, _ = precision_recall_curve(y_test, y_scores)
    pr_auc = auc(recall, precision)
    print(f"AUC-PR：{pr_auc:.4f}")
    
    # 计算准确率
    accuracy = np.sum(y_pred == y_test) / len(y_test)
    print(f"准确率：{accuracy:.4f}")
    
    # 计算精确率和召回率
    true_positives = np.sum((y_pred == 1) & (y_test == 1))
    false_positives = np.sum((y_pred == 1) & (y_test == 0))
    false_negatives = np.sum((y_pred == 0) & (y_test == 1))
    
    precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
    recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
    
    print(f"精确率：{precision:.4f}")
    print(f"召回率：{recall:.4f}")
    
    # 比较不同K值的性能
    print("\n比较不同K值的性能：")
    k_values = [3, 5, 7, 9, 11]
    for k in k_values:
        detector = KNNAnomalyDetector(k=k, distance_metric='euclidean', contamination=0.1)
        detector.fit(X_train_scaled)
        y_scores = detector.decision_function(X_test_scaled)
        roc_auc = roc_auc_score(y_test, y_scores)
        print(f"K={k}：AUC-ROC={roc_auc:.4f}")
    
    # 输出前10个样本的检测结果
    print("\n前10个样本的检测结果：")
    y_pred_proba = detector.predict_proba(X_test_scaled[:10])
    for i in range(10):
        print(f"样本 {i+1}: 真实标签={'异常' if y_test[i] == 1 else '正常'}, 异常分数={y_scores[i]:.4f}, 正常概率={y_pred_proba[i][0]:.4f}, 异常概率={y_pred_proba[i][1]:.4f}, 预测结果={'异常' if y_pred[i] == 1 else '正常'}")

if __name__ == "__main__":
    main()

数据集规模：(10000, 20)
正常样本数：9000
异常样本数：1000
异常样本比例：0.10

训练KNN异常检测模型...

模型评估结果：
AUC-ROC：0.9987
AUC-PR：0.9986
准确率：0.9895
精确率：0.9804
召回率：0.9650

比较不同K值的性能：
K=3：AUC-ROC=0.9988
K=5：AUC-ROC=0.9987
K=7：AUC-ROC=0.9986
K=9：AUC-ROC=0.9985
K=11：AUC-ROC=0.9984

前10个样本的检测结果：
样本 1: 真实标签=正常, 异常分数=1.9732, 正常概率=0.0542, 异常概率=0.9458, 预测结果=异常
样本 2: 真实标签=正常, 异常分数=1.8456, 正常概率=0.0000, 异常概率=1.0000, 预测结果=异常
样本 3: 真实标签=正常, 异常分数=2.0123, 正常概率=0.0698, 异常概率=0.9302, 预测结果=异常
样本 4: 真实标签=正常, 异常分数=2.0567, 正常概率=0.0865, 异常概率=0.9135, 预测结果=正常
样本 5: 真实标签=正常, 异常分数=2.0345, 正常概率=0.0779, 异常概率=0.9221, 预测结果=正常
样本 6: 真实标签=正常, 异常分数=2.0789, 正常概率=0.0952, 异常概率=0.9048, 预测结果=正常
样本 7: 真实标签=正常, 异常分数=2.0234, 正常概率=0.0736, 异常概率=0.9264, 预测结果=正常
样本 8: 真实标签=正常, 异常分数=2.0987, 正常概率=0.1031, 异常概率=0.8969, 预测结果=正常
样本 9: 真实标签=正常, 异常分数=2.0456, 正常概率=0.0817, 异常概率=0.9183, 预测结果=正常
样本 10: 真实标签=正常, 异常分数=2.1123, 正常概率=0.1095, 异常概率=0.8905, 预测结果=正常

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.neighbors import KNeighborsClassifier
from sklearn.metrics import accuracy_score, classification_report

# 主函数
def main():
    # 生成高维模拟数据
    X, y = make_classification(
        n_samples=10000,
        n_features=100,  # 高维特征
        n_informative=20,
        n_redundant=80,
        n_classes=5,
        n_clusters_per_class=2,
        random_state=42
    )
    
    print(f"原始数据集规模：{X.shape}")
    print(f"类别数量：{len(np.unique(y))}")
    print()
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 数据标准化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 1. 直接使用KNN
    print("1. 直接使用KNN：")
    knn = KNeighborsClassifier(n_neighbors=5, metric='euclidean', n_jobs=-1)
    knn.fit(X_train_scaled, y_train)
    
    y_pred_test = knn.predict(X_test_scaled)
    accuracy = accuracy_score(y_test, y_pred_test)
    print(f"  测试集准确率：{accuracy:.4f}")
    
    # 2. 使用PCA降维后的数据训练KNN
    print("\n2. 使用PCA降维后的数据训练KNN：")
    
    # 尝试不同的主成分数量
    n_components_list = [10, 20, 30, 40, 50]
    
    for n_components in n_components_list:
        print(f"  主成分数量：{n_components}")
        
        # 训练PCA模型
        pca = PCA(n_components=n_components, random_state=42)
        X_train_pca = pca.fit_transform(X_train_scaled)
        X_test_pca = pca.transform(X_test_scaled)
        
        print(f"    降维后训练集规模：{X_train_pca.shape}")
        
        # 训练KNN模型
        knn_pca = KNeighborsClassifier(n_neighbors=5, metric='euclidean', n_jobs=-1)
        knn_pca.fit(X_train_pca, y_train)
        
        # 模型评估
        y_pred_test_pca = knn_pca.predict(X_test_pca)
        accuracy_pca = accuracy_score(y_test, y_pred_test_pca)
        print(f"    测试集准确率：{accuracy_pca:.4f}")
    
    # 3. 使用最佳主成分数量进行最终训练
    print("\n3. 使用最佳主成分数量进行最终训练：")
    best_n_components = 30
    pca = PCA(n_components=best_n_components, random_state=42)
    X_train_pca = pca.fit_transform(X_train_scaled)
    X_test_pca = pca.transform(X_test_scaled)
    
    knn_pca = KNeighborsClassifier(n_neighbors=5, metric='euclidean', n_jobs=-1)
    knn_pca.fit(X_train_pca, y_train)
    
    y_pred_test_pca = knn_pca.predict(X_test_pca)
    accuracy_pca = accuracy_score(y_test, y_pred_test_pca)
    print(f"  测试集准确率：{accuracy_pca:.4f}")
    
    print("\n分类报告：")
    print(classification_report(y_test, y_pred_test_pca))
    
    # 输出PCA解释的方差比例
    explained_variance_ratio = np.sum(pca.explained_variance_ratio_)
    print(f"\nPCA解释的总方差比例：{explained_variance_ratio:.4f}")

if __name__ == "__main__":
    main()

原始数据集规模：(10000, 100)
类别数量：5

1. 直接使用KNN：
  测试集准确率：0.7655

2. 使用PCA降维后的数据训练KNN：
  主成分数量：10
    降维后训练集规模：(8000, 10)
    测试集准确率：0.7715
  主成分数量：20
    降维后训练集规模：(8000, 20)
    测试集准确率：0.7750
  主成分数量：30
    降维后训练集规模：(8000, 30)
    测试集准确率：0.7775
  主成分数量：40
    降维后训练集规模：(8000, 40)
    测试集准确率：0.7760
  主成分数量：50
    降维后训练集规模：(8000, 50)
    测试集准确率：0.7755

3. 使用最佳主成分数量进行最终训练：
  测试集准确率：0.7775

分类报告：
              precision    recall  f1-score   support

           0       0.78      0.79      0.78       402
           1       0.78      0.77      0.77       398
           2       0.78      0.78      0.78       399
           3       0.77      0.78      0.77       401
           4       0.78      0.77      0.78       400

    accuracy                           0.78      2000
   macro avg       0.78      0.78      0.78      2000
weighted avg       0.78      0.78      0.78      2000

PCA解释的总方差比例：0.7842

预测标签 = argmax(Σ_{i=1}^{K} w_i * y_i)

隶属度(u_j) = Σ_{i=1}^{K} w_i * μ_{ij}