PRML - Lab 1: KNN 算法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def _distance(p1: np.ndarray, p2: np.ndarray, mode: int = 2) -> float:
    '''
    Get the distance between two points.

    :param `p1`: the first point
    :param `p2`: the second point
    :param `mode`: which exponent to use when calculating distance,
                   using `2` by default for Euclidean distance
    '''

    assert p1.shape == p2.shape, (
        '_distance: dimensions not match for '
        f'{p1.shape} and {p2.shape}'
    )
    return np.linalg.norm(p1 - p2, ord=mode)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def _get_k_nearest_neighbors(
    self, k: int, base_p: np.ndarray, dataset: int = Dataset.TRAIN_SET
) -> List[int]:
    '''
    Get k nearest neighbors of a point from dataset. Each point (except the
    base point) is denoted by its index in the dataset.

    :param `base_p`: the base point
    :param `dataset`: which dataset to use
    '''

    if dataset == Dataset.TRAIN_SET:
        data = self.train_data
    elif dataset == Dataset.DEV_SET:
        data = self.dev_data
    else:
        data = self.test_data

    # Use a min heap of size k to get the k nearest neighbors
    heap: List[Tuple[float, np.ndarray]] = []
    for p_i in range(data.shape[0]):
        dist: float = _distance(base_p, data[p_i])
        if (len(heap) < k):
            heappush(heap, (-dist, p_i))
        else:
            heappushpop(heap, (-dist, p_i))

    # Return the indices of the k points in the dataset
    return [item[1] for item in heap]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def _get_most_common_label(
    self, labels_i: List[int], dataset: int = Dataset.TRAIN_SET
) -> int:
    '''
    Get the most common label in given labels. Each label is denoted by
    its data point's index in the dataset.

    :param `labels_i`: the indices of given labels
    :param `dataset`: which dataset to use
    '''

    if dataset == Dataset.TRAIN_SET:
        all_labels = self.train_label
    else:
        all_labels = self.dev_label

    labels: List[int] = [all_labels[i] for i in labels_i]
    return max(set(labels), key=labels.count)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

def fit(self, train_data: np.ndarray, train_label: np.ndarray) -> None:
    '''
    Train the model using a training set with labels.

    :param `train_data`: training set
    :param `train_label`: provided labels for data in training set
    '''

    # Shuffle the dataset with labels
    assert train_data.shape[0] == train_label.shape[0], (
        'fit: data size not match for '
        f'{train_data.shape[0]} and {train_label.shape[0]}'
    )
    shuffled_i = np.random.permutation(train_data.shape[0])
    shuffled_data: np.ndarray = train_data[shuffled_i]
    shuffled_label: np.ndarray = train_label[shuffled_i]

    # Separate training set and development set (for validation)
    train_ratio: float = 0.75
    train_size: int = floor(shuffled_data.shape[0] * train_ratio)
    self.train_data = shuffled_data[:train_size]
    self.train_label = shuffled_label[:train_size]
    self.dev_data = shuffled_data[train_size:]
    self.dev_label = shuffled_label[train_size:]

    print('=== Training ===')

    # Compare the predicted and expected results, calculate the accuracy
    # for each parameter k, and find out the best k for prediction.
    k_threshold: int = train_size if train_size < 20 else 20
    accuracy_table: List[float] = [0.0]
    max_accuracy: float = 0.0

    for k in range(1, k_threshold):
        predicted_labels: List[int] = []
        for p in self.dev_data:
            k_nearest_neighbors: List[int] = self._get_k_nearest_neighbors(
                k, p, Dataset.TRAIN_SET
            )
            predicted_label: int = self._get_most_common_label(
                k_nearest_neighbors, Dataset.TRAIN_SET
            )
            predicted_labels.append(predicted_label)
        prediction: np.ndarray = np.array(predicted_labels)

        accuracy: float = np.mean(np.equal(prediction, self.dev_label))
        accuracy_table.append(accuracy)
        print(f'k = {k}, train_acc = {accuracy * 100} %')
        if accuracy > max_accuracy:
            max_accuracy, self.k = accuracy, k

    print(f'best k = {self.k}\n')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

def predict(self, test_data: np.ndarray) -> np.ndarray:
    '''
    Predict the label of a point using our model.

    :param `test_data`: testing set
    '''

    self.test_data = test_data

    print('=== Predicting ===')

    predicted_labels: List[int] = []
    for p in self.test_data:
        k_nearest_neighbors: List[int] = self._get_k_nearest_neighbors(
            self.k, p, Dataset.TRAIN_SET
        )
        predicted_label: int = self._get_most_common_label(
            k_nearest_neighbors, Dataset.TRAIN_SET
        )
        predicted_labels.append(predicted_label)
    prediction: np.ndarray = np.array(predicted_labels)
    return prediction

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

def generate() -> None:
    '''
    Generate datasets using different parameters, and save to a file for
    further use.
    '''

    class Parameter(NamedTuple):
        mean: Tuple[int, int]
        cov: List[List[float]]
        size: int
        label: int

    def _generate_with_parameters(param: Parameter) -> np.ndarray:
        '''
        Generate a dataset using given parameters.

        :param `param`: a tuple of `mean`, `cov`, `size`
            `mean`: the mean of the dataset
            `cov`: the coefficient of variation (COV) of the dataset
            `size`: the number of points in the dataset
        '''

        return np.random.multivariate_normal(
            param.mean,
            param.cov,
            param.size,
        )

    parameters: List[Parameter] = [
        Parameter(
            mean=(1, 2),
            cov=[[73, 0], [0, 22]],
            size=800,
            label=0,
        ),
        Parameter(
            mean=(16, -5),
            cov=[[21.2, 0], [0, 32.1]],
            size=200,
            label=1,
        ),
        Parameter(
            mean=(10, 22),
            cov=[[10, 5], [5, 10]],
            size=1000,
            label=2,
        ),
    ]

    data: List[np.ndarray] = [
        _generate_with_parameters(param) for param in parameters
    ]

    indices: np.ndarray = np.arange(2000)
    np.random.shuffle(indices)
    all_data: np.ndarray = np.concatenate(data)
    all_label = np.concatenate([
        np.ones(param.size, int) * param.label for param in parameters
    ])
    shuffled_data: np.ndarray = all_data[indices]
    shuffled_label: np.ndarray = all_label[indices]

    train_data: np.ndarray = shuffled_data[:1600]
    train_label: np.ndarray = shuffled_label[:1600]
    test_data: np.ndarray = shuffled_data[1600:]
    test_label: np.ndarray = shuffled_label[1600:]
    np.save('data.npy', (
        (train_data, train_label),
        (test_data, test_label),
    ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def display(data: np.ndarray, label: np.ndarray, name: str) -> None:
    '''
    Visualize dataset with labels using `matplotlib.pyplot`.

    :param `data`: dataset
    :param `label`: labels for data in the dataset
    :param `name`: file name when saving to file
    '''

    datasets_with_label: List[List[np.ndarray]] = [[], [], []]
    for i in range(data.shape[0]):
        datasets_with_label[label[i]].append(data[i])

    for dataset_with_label in datasets_with_label:
        dataset_with_label_: np.ndarray = np.array(dataset_with_label)
        plt.scatter(dataset_with_label_[:, 0], dataset_with_label_[:, 1])

    plt.savefig(f'img/{name}')
    plt.show()

1
2
3
4
5

# 训练模型及预测
python ./source.py g

# 展示数据集
python ./source.py d

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

=== Training ===
k = 1, train_acc = 67.5 %
k = 2, train_acc = 67.75 %
k = 3, train_acc = 70.5 %
k = 4, train_acc = 72.5 %
k = 5, train_acc = 72.75 %
k = 6, train_acc = 73.25 %
k = 7, train_acc = 74.25 %
k = 8, train_acc = 74.0 %
k = 9, train_acc = 75.75 %
k = 10, train_acc = 75.5 %
k = 11, train_acc = 75.75 %
k = 12, train_acc = 75.5 %
k = 13, train_acc = 75.75 %
k = 14, train_acc = 75.5 %
k = 15, train_acc = 75.5 %
k = 16, train_acc = 74.5 %
k = 17, train_acc = 75.0 %
k = 18, train_acc = 75.0 %
k = 19, train_acc = 75.25 %
best k = 9

=== Predicting ===
k = 9, predict_acc = 71.5 %

1
2
3

mean = (1, 2)
cov = [[73, 0], [0, 22]]
size = 800

1
2
3

mean = (16, -5)
cov = [[21.2, 0], [0, 32.1]]
size = 200

1
2
3

mean = (10, 22)
cov = [[10, 5], [5, 10]]
size = 1000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

k = 1, train_acc = 95.75 %
k = 2, train_acc = 95.75 %
k = 3, train_acc = 97.25 %
k = 4, train_acc = 96.25 %
k = 5, train_acc = 96.5 %
k = 6, train_acc = 96.5 %
k = 7, train_acc = 96.75 %
k = 8, train_acc = 96.75 %
k = 9, train_acc = 96.75 %
k = 10, train_acc = 96.5 %
k = 11, train_acc = 96.5 %
k = 12, train_acc = 96.5 %
k = 13, train_acc = 96.75 %
k = 14, train_acc = 97.0 %
k = 15, train_acc = 96.75 %
k = 16, train_acc = 96.75 %
k = 17, train_acc = 96.75 %
k = 18, train_acc = 96.75 %
k = 19, train_acc = 97.0 %

1

k = 3, predict_acc = 96.0 %

1
2
3

mean = (-5, 2)
cov = [[73, 0], [0, 22]]
size = 800

1
2
3

mean = (30, -10)
cov = [[21.2, 0], [0, 32.1]]
size = 200

1
2
3

mean = (20, 40)
cov = [[10, 5], [5, 10]]
size = 1000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

k = 1, train_acc = 100.0 %
k = 2, train_acc = 100.0 %
k = 3, train_acc = 100.0 %
k = 4, train_acc = 100.0 %
k = 5, train_acc = 100.0 %
k = 6, train_acc = 100.0 %
k = 7, train_acc = 100.0 %
k = 8, train_acc = 100.0 %
k = 9, train_acc = 100.0 %
k = 10, train_acc = 100.0 %
k = 11, train_acc = 100.0 %
k = 12, train_acc = 100.0 %
k = 13, train_acc = 100.0 %
k = 14, train_acc = 100.0 %
k = 15, train_acc = 100.0 %
k = 16, train_acc = 100.0 %
k = 17, train_acc = 100.0 %
k = 18, train_acc = 100.0 %
k = 19, train_acc = 100.0 %

1

k = 1, predict_acc = 100.0 %

1
2
3

mean = (1, 2)
cov = [[73, 0], [0, 22]]
size = 800

1
2
3

mean = (3, -2)
cov = [[21.2, 0], [0, 32.1]]
size = 200

1
2
3

mean = (-5, 4)
cov = [[10, 5], [5, 10]]
size = 1000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

k = 1, train_acc = 65.0 %
k = 2, train_acc = 65.75 %
k = 3, train_acc = 69.25 %
k = 4, train_acc = 68.25 %
k = 5, train_acc = 72.5 %
k = 6, train_acc = 71.5 %
k = 7, train_acc = 73.75 %
k = 8, train_acc = 75.0 %
k = 9, train_acc = 76.0 %
k = 10, train_acc = 75.75 %
k = 11, train_acc = 76.0 %
k = 12, train_acc = 74.5 %
k = 13, train_acc = 75.25 %
k = 14, train_acc = 75.0 %
k = 15, train_acc = 74.75 %
k = 16, train_acc = 75.5 %
k = 17, train_acc = 75.0 %
k = 18, train_acc = 75.0 %
k = 19, train_acc = 74.5 %

1

k = 9, predict_acc = 76.0 %