KNN：K 近邻算法

import numpy as np
from collections import Counter

class SimpleKNN:
    """从零实现 KNN 分类器——理解核心逻辑"""

    def __init__(self, k=3):
        self.k = k
        self.X_train = None
        self.y_train = None

    def fit(self, X, y):
        """KNN 的"训练"：仅仅是记住数据"""
        self.X_train = np.array(X)
        self.y_train = np.array(y)
        print(f"已存储 {len(self.X_train)} 个训练样本")

    def _euclidean_distance(self, x1, x2):
        """计算两个样本之间的欧氏距离"""
        return np.sqrt(np.sum((x1 - x2) ** 2))

    def predict(self, X):
        """对每个新样本，找 K 个最近邻居并投票"""
        predictions = []
        for x in X:
            # 计算到所有训练样本的距离
            distances = [self._euclidean_distance(x, xt)
                         for xt in self.X_train]
            # 取距离最小的 K 个索引
            k_indices = np.argsort(distances)[:self.k]
            # 投票：选出出现次数最多的类别
            k_labels = [self.y_train[i] for i in k_indices]
            most_common = Counter(k_labels).most_common(1)[0][0]
            predictions.append(most_common)
        return np.array(predictions)

from sklearn.datasets import make_blobs
import matplotlib.pyplot as plt

# 生成 3 类二维数据
X, y = make_blobs(n_samples=300, centers=3,
                  cluster_std=1.5, random_state=42)

knn = SimpleKNN(k=5)
knn.fit(X, y)

# 生成网格用于绘制决策边界
x_min, x_max = X[:, 0].min() - 1, X[:, 0].max() + 1
y_min, y_max = X[:, 1].min() - 1, X[:, 1].max() + 1
xx, yy = np.meshgrid(np.linspace(x_min, x_max, 100),
                     np.linspace(y_min, y_max, 100))
grid_points = np.c_[xx.ravel(), yy.ravel()]
Z = knn.predict(grid_points).reshape(xx.shape)

plt.contourf(xx, yy, Z, alpha=0.3, cmap='viridis')
plt.scatter(X[:, 0], X[:, 1], c=y, cmap='viridis', edgecolors='k')
plt.title("KNN 决策边界 (K=5)")
plt.show()

import numpy as np
from scipy.spatial.distance import euclidean, cityblock, cosine

def minkowski_distance(x1, x2, p=2):
    """闵可夫斯基距离：统一框架"""
    return np.sum(np.abs(x1 - x2)  p)  (1/p)

# 测试不同距离度量
x = np.array([1.0, 2.0, 3.0])
y = np.array([4.0, 0.0, 3.0])

print(f"欧氏距离 (p=2):   {minkowski_distance(x, y, 2):.4f}")
print(f"曼哈顿距离 (p=1): {minkowski_distance(x, y, 1):.4f}")
print(f"切比雪夫距离:     {max(np.abs(x - y)):.4f}")

# 用 scipy 验证
print(f"scipy 欧氏:       {euclidean(x, y):.4f}")
print(f"scipy 曼哈顿:     {cityblock(x, y):.4f}")
print(f"余弦距离:         {cosine(x, y):.4f}")

import numpy as np
import matplotlib.pyplot as plt

def visualize_distance_shapes():
    """可视化不同距离度量的单位圆形状"""
    fig, axes = plt.subplots(1, 3, figsize=(12, 4))
    theta = np.linspace(0, 2*np.pi, 200)

    # L1 (曼哈顿)：菱形
    ax = axes[0]
    r = 1.0 / (np.abs(np.cos(theta)) + np.abs(np.sin(theta)))
    ax.plot(r * np.cos(theta), r * np.sin(theta))
    ax.set_title("L1 距离（菱形）")
    ax.set_aspect('equal')

    # L2 (欧氏)：圆形
    ax = axes[1]
    ax.plot(np.cos(theta), np.sin(theta))
    ax.set_title("L2 距离（圆形）")
    ax.set_aspect('equal')

    # L∞ (切比雪夫)：正方形
    ax = axes[2]
    r = 1.0 / np.maximum(np.abs(np.cos(theta)), np.abs(np.sin(theta)))
    ax.plot(r * np.cos(theta), r * np.sin(theta))
    ax.set_title("L∞ 距离（正方形）")
    ax.set_aspect('equal')

    plt.tight_layout()
    plt.show()

visualize_distance_shapes()

from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import cross_val_score
import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification

# 生成数据
X, y = make_classification(n_samples=500, n_features=10,
                           n_informative=5, random_state=42)

# 交叉验证选择最优 K
k_range = range(1, 31)
cv_scores = []
for k in k_range:
    knn = KNeighborsClassifier(n_neighbors=k)
    scores = cross_val_score(knn, X, y, cv=5, scoring='accuracy')
    cv_scores.append(scores.mean())

# 绘制 K-准确率曲线
plt.figure(figsize=(8, 5))
plt.plot(k_range, cv_scores, marker='o', linewidth=2)
best_k = k_range[np.argmax(cv_scores)]
plt.axvline(x=best_k, color='r', linestyle='--',
            label=f'最佳 K={best_k}')
plt.xlabel('K 值')
plt.ylabel('交叉验证准确率')
plt.title('K 值选择：交叉验证')
plt.legend()
plt.grid(alpha=0.3)
plt.show()

print(f"最佳 K 值: {best_k}")
print(f"最高准确率: {max(cv_scores):.4f}")

import numpy as np

def empirical_k_rule(n_samples, n_classes=2):
    """经验法则估算 K 值范围"""
    k_sqrt = int(np.sqrt(n_samples))
    # 调整为奇数
    if k_sqrt % 2 == 0:
        k_sqrt += 1
    k_min = max(1, k_sqrt - 2)
    k_max = k_sqrt + 4
    print(f"样本数: {n_samples}")
    print(f"建议 K 范围: [{k_min}, {k_max}]")
    print(f"起点 K={k_sqrt}")
    return k_min, k_max, k_sqrt

# 不同数据规模下的经验 K 值
for n in [100, 500, 1000, 5000]:
    empirical_k_rule(n)
    print("-" * 30)

import numpy as np
from collections import defaultdict

class WeightedKNN:
    """加权 KNN 分类器——距离越近，权重越大"""

    def __init__(self, k=5, weight_type='distance'):
        self.k = k
        self.weight_type = weight_type  # 'uniform', 'distance', 'gaussian'
        self.X_train = None
        self.y_train = None

    def fit(self, X, y):
        self.X_train = np.array(X)
        self.y_train = np.array(y)

    def _get_weight(self, distances):
        """根据距离计算权重"""
        if self.weight_type == 'uniform':
            return np.ones_like(distances)
        elif self.weight_type == 'distance':
            epsilon = 1e-8
            return 1.0 / (distances + epsilon)
        elif self.weight_type == 'gaussian':
            sigma = np.mean(distances)
            return np.exp(-(distances  2) / (2 * sigma  2 + 1e-8))

    def predict(self, X):
        predictions = []
        for x in X:
            dists = np.sqrt(np.sum((self.X_train - x) ** 2, axis=1))
            k_idx = np.argsort(dists)[:self.k]
            k_dists = dists[k_idx]
            k_labels = self.y_train[k_idx]
            weights = self._get_weight(k_dists)

            # 加权投票
            class_votes = defaultdict(float)
            for label, weight in zip(k_labels, weights):
                class_votes[label] += weight
            predictions.append(max(class_votes, key=class_votes.get))
        return np.array(predictions)

from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.neighbors import KNeighborsClassifier
from sklearn.metrics import accuracy_score

# 生成类别不平衡数据
X, y = make_classification(n_samples=1000, n_features=5,
                           n_informative=3, weights=[0.85, 0.15],
                           random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42, stratify=y)

# 对比三种加权策略
for w in ['uniform', 'distance']:
    knn = KNeighborsClassifier(n_neighbors=7, weights=w)
    knn.fit(X_train, y_train)
    y_pred = knn.predict(X_test)
    acc = accuracy_score(y_test, y_pred)
    print(f"weights={w:>10s} | 准确率: {acc:.4f}")

# 查看不同 K 值下的加权效果
print(" 不同 K 值的加权效果:")
print(f"{'K':>3s} | {'uniform':>8s} | {'distance':>8s}")
print("-" * 28)
for k in [1, 3, 5, 7, 9, 11, 15]:
    for w in ['uniform', 'distance']:
        knn = KNeighborsClassifier(n_neighbors=k, weights=w)
        knn.fit(X_train, y_train)
        acc = accuracy_score(y_test, knn.predict(X_test))
        if w == 'uniform':
            print(f"{k:3d} | {acc:.4f}", end="")
        else:
            print(f" | {acc:.4f}")

from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import train_test_split
from sklearn.datasets import load_wine
from sklearn.metrics import accuracy_score
import numpy as np

# 加载数据（特征尺度差异很大）
wine = load_wine()
X, y = wine.data, wine.target
print(f"原始特征范围:")
for i, name in enumerate(wine.feature_names[:4]):
    print(f"  {name:>20s}: [{X[:, i].min():.1f}, {X[:, i].max():.1f}]")

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 对比标准化前后的效果
for scaler_name, scaler in [("无标准化", None),
                             ("StandardScaler", StandardScaler()),
                             ("MinMaxScaler", MinMaxScaler())]:
    if scaler:
        X_train_s = scaler.fit_transform(X_train)
        X_test_s = scaler.transform(X_test)
    else:
        X_train_s, X_test_s = X_train, X_test

    knn = KNeighborsClassifier(n_neighbors=5)
    knn.fit(X_train_s, y_train)
    acc = accuracy_score(y_test, knn.predict(X_test_s))
    print(f"{scaler_name:>20s}: 准确率 = {acc:.4f}")

import numpy as np
import matplotlib.pyplot as plt

def demonstrate_scaling_impact():
    """可视化标准化对 KNN 决策边界的影响"""
    from sklearn.datasets import make_classification
    from sklearn.preprocessing import StandardScaler

    # 创建尺度差异巨大的数据
    np.random.seed(42)
    X1 = np.random.randn(100, 2)
    X1[:, 0] *= 100  # 特征 1 尺度大
    X1[:, 1] *= 1    # 特征 2 尺度小
    y1 = np.zeros(100)

    X2 = np.random.randn(100, 2) + np.array([300, 1])
    X2[:, 0] *= 100
    X2[:, 1] *= 1
    y2 = np.ones(100)

    X = np.vstack([X1, X2])
    y = np.concatenate([y1, y2])

    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    fig, axes = plt.subplots(1, 2, figsize=(12, 5))

    for ax, data, title in [
        (axes[0], X, "未标准化"),
        (axes[1], X_scaled, "标准化后")
    ]:
        ax.scatter(data[:100, 0], data[:100, 1],
                   c='blue', label='类别 0', alpha=0.7)
        ax.scatter(data[100:, 0], data[100:, 1],
                   c='red', label='类别 1', alpha=0.7)
        ax.set_title(title)
        ax.legend()
        ax.set_xlabel("特征 1")
        ax.set_ylabel("特征 2")

    plt.tight_layout()
    plt.show()

demonstrate_scaling_impact()

from sklearn.datasets import make_moons, make_circles
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import matplotlib.pyplot as plt
import numpy as np

# 在非线性数据上对比多种分类器
datasets = {
    "moons": make_moons(n_samples=500, noise=0.3, random_state=42),
    "circles": make_circles(n_samples=500, noise=0.15, random_state=42),
}

classifiers = {
    "KNN (k=7)": KNeighborsClassifier(n_neighbors=7),
    "决策树": DecisionTreeClassifier(max_depth=5, random_state=42),
    "SVM (RBF)": SVC(kernel='rbf', gamma='scale', random_state=42),
    "逻辑回归": LogisticRegression(max_iter=1000, random_state=42),
}

for name, (X, y) in datasets.items():
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=42)
    print(f" === {name.upper()} 数据集 ===")
    for clf_name, clf in classifiers.items():
        clf.fit(X_train, y_train)
        acc = accuracy_score(y_test, clf.predict(X_test))
        print(f"  {clf_name:>15s}: {acc:.4f}")

import numpy as np
import time
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import make_classification

# 训练速度和预测速度对比
print(f"{'样本数':>8s} | {'KNN训练':>8s} | {'KNN预测':>8s} | {'树训练':>8s} | {'树预测':>8s}")
print("-" * 60)

for n in [1000, 5000, 10000, 50000]:
    X, y = make_classification(n_samples=n, n_features=10,
                               random_state=42)

    # KNN
    knn = KNeighborsClassifier(n_neighbors=5)
    t0 = time.time()
    knn.fit(X, y)
    t_train = time.time() - t0
    t0 = time.time()
    knn.predict(X[:100])
    t_predict = time.time() - t0

    # 决策树
    tree = DecisionTreeClassifier(random_state=42)
    t0 = time.time()
    tree.fit(X, y)
    t_train_tree = time.time() - t0
    t0 = time.time()
    tree.predict(X[:100])
    t_predict_tree = time.time() - t0

    print(f"{n:8d} | {t_train:8.4f}s | {t_predict:8.4f}s | "
          f"{t_train_tree:8.4f}s | {t_predict_tree:8.4f}s")

from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.datasets import load_breast_cancer
from sklearn.metrics import classification_report
import time

# 加载乳腺癌数据集
data = load_breast_cancer()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.2, random_state=42, stratify=data.target)

# 使用 Pipeline 确保标准化在交叉验证内部执行
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('knn', KNeighborsClassifier())
])

# 网格搜索最优超参数
param_grid = {
    'knn__n_neighbors': [3, 5, 7, 9, 11, 15],
    'knn__weights': ['uniform', 'distance'],
    'knn__metric': ['euclidean', 'manhattan'],
    'knn__algorithm': ['auto', 'kd_tree', 'ball_tree'],
}

grid_search = GridSearchCV(pipeline, 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}")
print(f" 测试集报告:")
print(classification_report(y_test, grid_search.predict(X_test),
                            target_names=data.target_names))

from sklearn.neighbors import KDTree, BallTree
import numpy as np
import time

# 手动演示 KD 树加速原理
np.random.seed(42)
X = np.random.randn(10000, 5)
query = np.random.randn(1, 5)

# 方法 1：暴力搜索（Brute Force）
t0 = time.time()
dists_brute = np.sqrt(np.sum((X - query) ** 2, axis=1))
k_indices_brute = np.argsort(dists_brute)[:5]
t_brute = time.time() - t0

# 方法 2：KD 树
t0 = time.time()
kdt = KDTree(X, leaf_size=30, metric='euclidean')
dists_kd, indices_kd = kdt.query(query, k=5)
t_kd = time.time() - t0

# 方法 3：Ball 树
t0 = time.time()
bt = BallTree(X, leaf_size=30, metric='euclidean')
dists_bt, indices_bt = bt.query(query, k=5)
t_bt = time.time() - t0

print(f"查询时间对比 (10000 样本, 5 维):")
print(f"  暴力搜索: {t_brute*1000:.2f} ms")
print(f"  KD 树:    {t_kd*1000:.2f} ms  ({t_brute/t_kd:.1f}x 加速)")
print(f"  Ball 树:  {t_bt*1000:.2f} ms  ({t_brute/t_bt:.1f}x 加速)")

# 验证结果一致性
print(f" KD 树最近邻索引: {indices_kd[0]}")
print(f"暴力搜索最近邻:    {k_indices_brute}")
print(f"结果一致: {set(indices_kd[0]) == set(k_indices_brute)}"
)