KNN：距离、特征缩放和邻居数量

from math import dist

query = (3.0, 4.0)
samples = [((1.0, 4.0), "A"), ((5.0, 5.0), "B"), ((2.0, 1.0), "A")]
ranked = sorted((dist(query, point), label) for point, label in samples)
print(ranked)

from sklearn.model_selection import cross_val_score
from sklearn.neighbors import KNeighborsClassifier
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

for neighbors in (1, 3, 7, 15, 31):
    candidate = make_pipeline(
        StandardScaler(), KNeighborsClassifier(n_neighbors=neighbors)
    )
    score = cross_val_score(candidate, X_train, y_train, cv=5).mean()
    print(neighbors, round(score, 3))

import numpy as np
from sklearn.neighbors import KNeighborsClassifier
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

rng = np.random.default_rng(12)
size = rng.normal(40, 12, 500)
queue = rng.normal(5, 2, 500)
noise_feature = rng.normal(20_000, 4_000, 500)
y = ((size + queue * 4 + rng.normal(0, 8, 500)) > 62).astype(int)
X = np.column_stack([size, queue, noise_feature])
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, stratify=y, random_state=42,
)

plain = KNeighborsClassifier(n_neighbors=7).fit(X_train, y_train)
scaled = make_pipeline(
    StandardScaler(), KNeighborsClassifier(n_neighbors=7)
).fit(X_train, y_train)
print("plain:", round(accuracy_score(y_test, plain.predict(X_test)), 3))
print("scaled:", round(accuracy_score(y_test, scaled.predict(X_test)), 3))

from sklearn.model_selection import cross_validate

for neighbors in (3, 7, 15, 31):
    candidate = make_pipeline(
        StandardScaler(),
        KNeighborsClassifier(n_neighbors=neighbors, weights="distance"),
    )
    result = cross_validate(candidate, X_train, y_train, cv=5, scoring="f1")
    print(
        neighbors,
        "mean=", round(result["test_score"].mean(), 3),
        "std=", round(result["test_score"].std(), 3),
    )

from time import perf_counter

for query_count in (1, 50, 200):
    started = perf_counter()
    scaled.predict(X_test[:query_count])
    elapsed_ms = (perf_counter() - started) * 1000
    print(query_count, round(elapsed_ms, 3), "ms")