Домашняя работа: пишем KNN с нуля¶

В этой работе вам предстоит самостоятельно реализовать:

  1. min-max нормализацию;
  2. поиск ближайших соседей;
  3. голосование соседей;
  4. KNN-классификатор с интерфейсом fit / predict;
  5. визуализацию областей решений для разных значений $k$.

Функции расстояния $L_1$, $L_2$ и $L_\infty$ уже реализованы. Использовать sklearn.neighbors.KNeighborsClassifier нельзя.

Правила¶

  • Можно использовать numpy и matplotlib.
  • Не изменяйте ячейки с тестами assert.
  • Замените raise NotImplementedError своим кодом.
  • Запускайте ячейки по порядку.
  • Работа считается выполненной, если все обязательные assert проходят без ошибок и построены требуемые графики.

Автор: Виктория Кулевцова. CC BY 4.0.

In [ ]:
import numpy as np
import matplotlib.pyplot as plt

plt.rcParams["figure.figsize"] = (8, 5)
plt.rcParams["font.size"] = 12

print("Библиотеки загружены")

0. Готовые функции расстояния¶

Эту часть реализовывать не нужно. KNN будет получать функцию расстояния через параметр metric.

In [ ]:
def l1_distance(a, b):
    """Манхэттенское расстояние L1."""
    a = np.asarray(a, dtype=float)
    b = np.asarray(b, dtype=float)
    return np.sum(np.abs(a - b))


def l2_distance(a, b):
    """Евклидово расстояние L2."""
    a = np.asarray(a, dtype=float)
    b = np.asarray(b, dtype=float)
    return np.sqrt(np.sum((a - b) ** 2))


def linf_distance(a, b):
    """Расстояние Чебышёва L∞."""
    a = np.asarray(a, dtype=float)
    b = np.asarray(b, dtype=float)
    return np.max(np.abs(a - b))


a = np.array([1, 2])
b = np.array([4, 6])

assert np.isclose(l1_distance(a, b), 7)
assert np.isclose(l2_distance(a, b), 5)
assert np.isclose(linf_distance(a, b), 4)

print("Готовые функции расстояния работают корректно")

1. Min-max нормализация¶

Для каждого признака нужно применить формулу

$$x'=\frac{x-x_{\min}}{x_{\max}-x_{\min}}.$$

Реализуйте класс MyMinMaxScaler.

Требования¶

  • fit(X) запоминает минимумы, максимумы и размахи столбцов;
  • transform(X) применяет уже найденные параметры;
  • fit_transform(X) уже реализован;
  • если признак постоянный, после нормализации он должен состоять из нулей;
  • значения новых объектов могут выйти за диапазон $[0,1]$ — обрезать их не нужно;
  • fit должен возвращать self.

Важно: новую точку нельзя нормализовать отдельно от обучающих данных.

In [ ]:
class MyMinMaxScaler:
    def fit(self, X):
        X = np.asarray(X, dtype=float)

        if X.ndim != 2:
            raise ValueError("X должен быть двумерным массивом")
        
        # TODO: сохраните минимумы в self.min_
        # TODO: сохраните максимумы в self.max_
        # TODO: сохраните разницу между максимумами и минимумами в self.range_
        # TODO: верните self
        raise NotImplementedError

    def transform(self, X):
        X = np.asarray(X, dtype=float)

        if not hasattr(self, "min_"):
            raise RuntimeError("Сначала вызовите метод fit")

        if X.ndim != 2:
            raise ValueError("X должен быть двумерным массивом")

        if X.shape[1] != len(self.min_):
            raise ValueError("Количество признаков не совпадает")
            
        # TODO: для нулевых размахов используйте безопасный делитель 1
        # TODO: верните нормализованный массив
        raise NotImplementedError

    def fit_transform(self, X):
        return self.fit(X).transform(X)

Самопроверка нормализации¶

Следующая ячейка должна выполниться без ошибок. Если появляется AssertionError, вернитесь к реализации класса.

In [ ]:
X_scaler_test = np.array([
    [2, 10],
    [2, 20],
    [2, 30],
], dtype=float)

scaler_test = MyMinMaxScaler()
returned_object = scaler_test.fit(X_scaler_test)
X_scaler_test_scaled = scaler_test.transform(X_scaler_test)

expected_scaled = np.array([
    [0.0, 0.0],
    [0.0, 0.5],
    [0.0, 1.0],
])

assert returned_object is scaler_test, "fit должен возвращать self"
assert np.allclose(scaler_test.min_, [2, 10])
assert np.allclose(scaler_test.max_, [2, 30])
assert np.allclose(scaler_test.range_, [0, 20])
assert np.allclose(X_scaler_test_scaled, expected_scaled)

# Для нового объекта используем старые минимумы и максимумы.
new_object_scaled = scaler_test.transform([[2, 40]])
assert np.allclose(new_object_scaled, [[0.0, 1.5]])

# fit_transform должен давать тот же результат, что fit + transform.
fit_transform_scaler = MyMinMaxScaler()
assert np.allclose(
    fit_transform_scaler.fit_transform(X_scaler_test),
    expected_scaled,
)

# Объекты вне обучающего диапазона не обрезаются до [0, 1].
outside_objects_scaled = scaler_test.transform([
    [2, 0],
    [2, 40],
])
assert np.allclose(
    outside_objects_scaled,
    [[0.0, -0.5], [0.0, 1.5]],
)

# Если все значения признака одинаковы, после преобразования получаются нули.
constant_scaler = MyMinMaxScaler()
constant_scaled = constant_scaler.fit_transform([
    [5, -2],
    [5, -2],
])
assert np.allclose(constant_scaled, np.zeros((2, 2)))

print("Все тесты MyMinMaxScaler пройдены")

2. Собственный KNN¶

Реализуйте классификатор MyKNNClassifier.

Что должен делать каждый метод¶

  • fit(X, y) — запомнить обучающие данные;
  • _distances(query) — найти расстояние от query до каждого обучающего объекта;
  • kneighbors(query) — вернуть индексы и расстояния $k$ ближайших соседей;
  • predict_one(query) — провести голосование для одной точки;
  • predict(X) — получить предсказания для нескольких точек.

Правило при ничьей¶

Если несколько классов набрали одинаковое максимальное число голосов, выбирайте класс с меньшим номером. Например, при голосах [0, 1] ответом будет 0.

Порядок соседей¶

Соседи сортируются сначала по расстоянию, а при одинаковом расстоянии — по индексу объекта. Для этого можно использовать np.lexsort.

In [ ]:
class MyKNNClassifier:
    def __init__(self, k=3, metric=l2_distance):
        if not isinstance(k, int) or k <= 0:
            raise ValueError("k должно быть положительным целым числом")

        self.k = k
        self.metric = metric

    def fit(self, X, y):
        X = np.asarray(X, dtype=float)
        y = np.asarray(y)

        if X.ndim != 2:
            raise ValueError("X должен быть двумерным массивом")

        if y.ndim != 1:
            raise ValueError("y должен быть одномерным массивом")

        if len(X) != len(y):
            raise ValueError(
                "Количество объектов в X и y не совпадает"
            )

        # TODO: проверьте, что k не больше числа обучающих объектов
        # TODO: сохраните X в self.X_train_
        # TODO: сохраните y в self.y_train_
        # TODO: верните self
        raise NotImplementedError

    def _distances(self, query):
        if not hasattr(self, "X_train_"):
            raise RuntimeError("Сначала вызовите метод fit")
        query = np.asarray(query, dtype=float)

        # TODO: посчитайте расстояние от query до каждой строки self.X_train_
        # Результат должен быть одномерным numpy-массивом длины n_train.
        if query.ndim != 1:
            raise ValueError(
                "query должен быть одномерным массивом"
            )

        if len(query) != self.X_train_.shape[1]:
            raise ValueError(
                "Количество признаков query не совпадает с X"
            )
        raise NotImplementedError

    def kneighbors(self, query):
        # TODO: получите все расстояния
        # TODO: отсортируйте объекты по паре (расстояние, индекс)
        # TODO: оставьте первые self.k объектов
        # Верните (neighbor_indices, neighbor_distances).
        raise NotImplementedError

    def predict_one(self, query):
        # TODO: найдите соседей
        # TODO: получите их классы
        # TODO: посчитайте количество голосов каждого класса
        # TODO: верните класс-победитель
        # Подсказка: np.unique(..., return_counts=True).
        raise NotImplementedError

    def predict(self, X):
        X = np.asarray(X, dtype=float)
        if X.ndim != 2:
            raise ValueError("X должен быть двумерным массивом")
        return np.array([self.predict_one(point) for point in X])

Самопроверка KNN: расстояния и соседи¶

In [ ]:
X_knn_test = np.array([
    [0, 0],
    [0, 2],
    [3, 0],
    [3, 2],
], dtype=float)

y_knn_test = np.array([0, 0, 1, 1])
query_knn_test = np.array([0.2, 0.3])

knn_test = MyKNNClassifier(k=3, metric=l2_distance)
returned_object = knn_test.fit(X_knn_test, y_knn_test)

assert returned_object is knn_test, "fit должен возвращать self"
assert np.allclose(knn_test.X_train_, X_knn_test)
assert np.array_equal(knn_test.y_train_, y_knn_test)

actual_distances = knn_test._distances(query_knn_test)
expected_distances = np.array([
    np.sqrt(0.13),
    np.sqrt(2.93),
    np.sqrt(7.93),
    np.sqrt(10.73),
])

assert actual_distances.shape == (4,)
assert np.allclose(actual_distances, expected_distances)

neighbor_indices, neighbor_distances = knn_test.kneighbors(query_knn_test)
assert np.array_equal(neighbor_indices, [0, 1, 2])
assert np.allclose(neighbor_distances, expected_distances[[0, 1, 2]])

# При равных расстояниях меньший индекс должен идти первым.
equidistant_model = MyKNNClassifier(k=2, metric=l2_distance)
equidistant_model.fit(
    np.array([[1, 0], [-1, 0], [0, 2]], dtype=float),
    np.array([0, 1, 1]),
)
equal_indices, equal_distances = equidistant_model.kneighbors([0, 0])
assert np.array_equal(equal_indices, [0, 1])
assert np.allclose(equal_distances, [1.0, 1.0])

print("Расстояния и поиск соседей работают корректно")

Самопроверка KNN: предсказания и ничья¶

In [ ]:
assert knn_test.predict_one([0.2, 0.3]) == 0
assert knn_test.predict_one([2.9, 1.5]) == 1

batch_prediction = knn_test.predict([
    [0.1, 0.1],
    [2.9, 1.5],
])
assert np.array_equal(batch_prediction, [0, 1])

# Специальный тест на ничью: один голос за 0 и один за 1.
tie_model = MyKNNClassifier(k=2, metric=l2_distance)
tie_model.fit(
    np.array([[0, 0], [2, 0]], dtype=float),
    np.array([1, 0]),
)
assert tie_model.predict_one([1, 0]) == 0

# Проверяем, что можно передать другую метрику.
l1_model = MyKNNClassifier(k=1, metric=l1_distance)
l1_model.fit(X_knn_test, y_knn_test)
assert l1_model.predict_one([2.8, 0.2]) == 1

assert batch_prediction.shape == (2,)

# k не может быть больше количества обучающих объектов.
too_many_neighbors = MyKNNClassifier(k=len(X_knn_test) + 1)
try:
    too_many_neighbors.fit(X_knn_test, y_knn_test)
    raise AssertionError("При k > n_train ожидалась ошибка ValueError")
except ValueError:
    pass

print("Все обязательные тесты MyKNNClassifier пройдены")

3. Датасет «Маги и воины»¶

Теперь применим собственный KNN к небольшому датасету.

Признаки:

  1. здоровье;
  2. магическая сила.

Классы:

  • 0 — воин;
  • 1 — маг.

Точка [760, 3] — редкий маг, похожий по признакам на воинов. Благодаря ей будет хорошо видно отличие маленького $k$ от большого.

In [ ]:
X_train = np.array([
    [900, 2],
    [850, 3],
    [780, 2],
    [820, 4],
    [740, 3],
    [880, 5],
    [350, 9],
    [420, 8],
    [300, 10],
    [450, 9],
    [500, 7],
    [380, 6],
    [760, 3],  # редкий маг среди воинов
], dtype=float)

y_train = np.array([
    0, 0, 0, 0, 0, 0,
    1, 1, 1, 1, 1, 1,
    1,
])

X_test = np.array([
    [800, 3],
    [410, 9],
    [690, 8],
    [840, 4],
    [520, 8],
    [730, 2],
], dtype=float)

y_test = np.array([0, 1, 1, 0, 1, 0])
query = np.array([700, 9], dtype=float)

class_names = {0: "Воин", 1: "Маг"}
class_colors = {0: "tomato", 1: "royalblue"}

print("Обучающих объектов:", len(X_train))
print("Тестовых объектов:", len(X_test))

3.1. Обучение без нормализации¶

В KNN обучение почти целиком состоит в запоминании обучающих объектов. Тем не менее мы используем привычный метод fit.

In [ ]:
knn_raw = MyKNNClassifier(k=3, metric=l2_distance)
knn_raw.fit(X_train, y_train)

raw_query_prediction = knn_raw.predict(query.reshape(1, -1))[0]
raw_test_prediction = knn_raw.predict(X_test)
raw_accuracy = np.mean(raw_test_prediction == y_test)

print("Ответ для нового персонажа:", class_names[raw_query_prediction])
print("Предсказания на тесте:     ", raw_test_prediction)
print("Правильные ответы:         ", y_test)
print(f"Accuracy: {raw_accuracy:.3f}")

assert raw_query_prediction == 0, "Без нормализации ожидается класс «Воин»"

assert np.array_equal(raw_test_prediction, [0, 1, 0, 0, 1, 0])
assert np.isclose(raw_accuracy, 5 / 6)

3.2. Обучение с нормализацией¶

Обратите внимание на порядок действий:

  1. scaler.fit вызывается только на X_train;
  2. обучающие и тестовые данные преобразуются одними параметрами;
  3. KNN обучается на нормализованной обучающей выборке.
In [ ]:
scaler = MyMinMaxScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
query_scaled = scaler.transform(query.reshape(1, -1))

knn_scaled = MyKNNClassifier(k=3, metric=l2_distance)
knn_scaled.fit(X_train_scaled, y_train)

scaled_query_prediction = knn_scaled.predict(query_scaled)[0]
scaled_test_prediction = knn_scaled.predict(X_test_scaled)
scaled_accuracy = np.mean(scaled_test_prediction == y_test)

print("Без нормализации:", class_names[raw_query_prediction])
print("С нормализацией: ", class_names[scaled_query_prediction])
print(f"Accuracy без нормализации: {raw_accuracy:.3f}")
print(f"Accuracy с нормализацией:   {scaled_accuracy:.3f}")

assert scaled_query_prediction == 1, "После нормализации ожидается класс «Маг»"

assert np.array_equal(scaled_test_prediction, [0, 1, 1, 0, 1, 0])
assert np.isclose(scaled_accuracy, 1.0)

Письменный ответ¶

Ответьте в 2–4 предложениях:

  1. Почему предсказание для query изменилось после нормализации?
  2. Какой признак доминировал до нормализации?
  3. Почему нельзя отдельно вызывать fit_transform для X_test?

Ваш ответ: …

4. Визуализация¶

Функция ниже строит области решений вашей модели. Она работает в исходных координатах, но перед предсказанием при необходимости нормализует точки сетки.

Разберитесь, что делают np.meshgrid, reshape и contourf. Саму вспомогательную функцию изменять не нужно.

In [ ]:
def plot_decision_regions(
    ax,
    model,
    X_original,
    y,
    scaler=None,
    title="",
):
    x_min, x_max = X_original[:, 0].min() - 80, X_original[:, 0].max() + 80
    y_min, y_max = X_original[:, 1].min() - 1, X_original[:, 1].max() + 1

    xx, yy = np.meshgrid(
        np.linspace(x_min, x_max, 140),
        np.linspace(y_min, y_max, 100),
    )
    grid_original = np.column_stack([xx.ravel(), yy.ravel()])

    if scaler is None:
        grid_for_model = grid_original
    else:
        grid_for_model = scaler.transform(grid_original)

    grid_prediction = model.predict(grid_for_model).reshape(xx.shape)

    ax.contourf(
        xx, yy, grid_prediction,
        levels=[-0.5, 0.5, 1.5],
        colors=["mistyrose", "lightblue"],
        alpha=0.8,
    )

    for class_id in np.unique(y):
        mask = y == class_id
        ax.scatter(
            X_original[mask, 0],
            X_original[mask, 1],
            color=class_colors[class_id],
            edgecolor="black",
            s=75,
            label=class_names[class_id],
        )

    ax.set_xlabel("Здоровье")
    ax.set_ylabel("Магическая сила")
    ax.set_title(title)
    ax.grid(alpha=0.15)

4.1. До и после нормализации¶

Постройте две области решений для $k=3$ и $L_2$. Добавьте на каждый график новую точку query в виде звезды.

In [ ]:
fig, axes = plt.subplots(1, 2, figsize=(15, 5), sharex=True, sharey=True)

plot_decision_regions(
    axes[0],
    knn_raw,
    X_train,
    y_train,
    scaler=None,
    title="Без нормализации, k=3",
)

plot_decision_regions(
    axes[1],
    knn_scaled,
    X_train,
    y_train,
    scaler=scaler,
    title="С min-max нормализацией, k=3",
)

# TODO: добавьте query на оба графика через axes[i].scatter(...).
# Используйте marker='*', color='gold', edgecolor='black', s=300.

for ax in axes:
    ax.legend()

plt.tight_layout()
plt.savefig("knn_normalization_comparison.png", dpi=150, bbox_inches="tight")
plt.show()

# Проверяем, что новая точка действительно добавлена на оба графика.
for ax in axes:
    plotted_offsets = [
        collection.get_offsets()
        for collection in ax.collections
        if hasattr(collection, "get_offsets")
    ]
    point_is_plotted = any(
        np.any(np.all(np.isclose(offsets, query), axis=1))
        for offsets in plotted_offsets
        if len(offsets) > 0
    )
    assert point_is_plotted, "Добавьте query на оба графика"

4.2. Как влияет значение $k$¶

Постройте четыре модели на нормализованных данных:

$$k \in \{1, 3, 5, 9\}.$$

Для каждой модели нарисуйте область решений. Весь необходимый код обучения моделей нужно написать в следующей ячейке самостоятельно.

In [ ]:
k_values = [1, 3, 5, 9]
fig, axes = plt.subplots(2, 2, figsize=(14, 10), sharex=True, sharey=True)

for ax, k in zip(axes.ravel(), k_values):
    # TODO: создайте MyKNNClassifier с текущим k и метрикой L2
    # TODO: обучите его на X_train_scaled и y_train
    # TODO: вызовите plot_decision_regions; не забудьте передать scaler
    pass

handles, labels = axes[0, 0].get_legend_handles_labels()
fig.legend(handles, labels, loc="upper center", ncol=2)
fig.suptitle("Области решений собственного KNN", fontsize=16, y=1.01)
plt.tight_layout()
plt.savefig("knn_different_k.png", dpi=150, bbox_inches="tight")
plt.show()

# На каждом из четырёх подграфиков должны появиться области и обучающие точки.
for ax in axes.ravel():
    assert ax.get_xlabel() == "Здоровье"
    assert ax.get_ylabel() == "Магическая сила"
    assert len(ax.collections) >= 3, "Постройте область решений для каждого k"

Вывод по графикам¶

Ответьте в 3–5 предложениях:

  1. При каком $k$ граница получилась наиболее сложной?
  2. Что произошло вокруг редкого мага [760, 3] при $k=1$?
  3. Как менялась граница при увеличении $k$?
  4. Какое значение $k$ вы выбрали бы для этих данных и почему?

Ваш ответ: …

5. Бонус: метрика Минковского¶

Метрики $L_1$ и $L_2$ — частные случаи расстояния Минковского:

$$d_p(a,b)=\left(\sum_i |a_i-b_i|^p\right)^{1/p}.$$

Реализуйте функцию minkowski_distance. Параметр $p$ должен быть не меньше 1.

In [ ]:
def minkowski_distance(a, b, p=3):
    # TODO: проверьте, что p >= 1
    # TODO: реализуйте формулу расстояния Минковского
    raise NotImplementedError
In [ ]:
a = np.array([1, 2])
b = np.array([4, 6])

assert np.isclose(minkowski_distance(a, b, p=1), l1_distance(a, b))
assert np.isclose(minkowski_distance(a, b, p=2), l2_distance(a, b))
assert np.isclose(minkowski_distance(a, b, p=3), 91 ** (1 / 3))

try:
    minkowski_distance(a, b, p=0.5)
    raise AssertionError("При p < 1 ожидалась ошибка ValueError")
except ValueError:
    pass

assert np.isclose(minkowski_distance(a, a, p=3), 0.0)
assert np.isclose(
    minkowski_distance([-1, 2], [2, -2], p=1),
    7.0,
)

print("Бонусные тесты пройдены")

Использование новой метрики в KNN¶

Поскольку MyKNNClassifier принимает любую функцию двух аргументов, метрику с $p=3$ можно передать через lambda.

In [ ]:
minkowski_p3 = lambda a, b: minkowski_distance(a, b, p=3)

knn_minkowski = MyKNNClassifier(k=3, metric=minkowski_p3)
knn_minkowski.fit(X_train_scaled, y_train)

minkowski_prediction = knn_minkowski.predict(query_scaled)[0]
print("Предсказание с метрикой Минковского p=3:", class_names[minkowski_prediction])

Финальный чек-лист¶

Перед сдачей проверьте:

  • все обязательные ячейки с assert выполняются без ошибок;
  • в обязательных частях не используется sklearn;
  • MyMinMaxScaler обучается только на X_train;
  • реализованы все методы MyKNNClassifier;
  • на графике сравнения новая точка отмечена звездой;
  • построены четыре области решений для разных $k$;
  • даны письменные ответы про нормализацию и влияние $k$;
  • ноутбук сохранён после выполнения всех ячеек;
  • вместе с ноутбуком приложены изображения knn_normalization_comparison.png и knn_different_k.png.