Python实现---用KNN解决回归问题KNN用于回归问题实验数据集python代码实现实验结果

125 阅读 0 评论 83 点赞

我是靠谱客的博主活力马里奥，这篇文章主要介绍Python实现---用KNN解决回归问题KNN用于回归问题实验数据集python代码实现实验结果，现在分享给大家，希望可以做个参考。

KNN用于回归问题

KNN不仅可以用于分类问题，也可以用回归问题。
主要算法思路就是：使用KNN计算某个数据点的预测值时，模型从训练数据集中选择离该数据点最近的k个数据点，并且把这些数据的y值取均值，把求出的这个均值作为新数据点的预测值。

实验数据集

data.csv:
数据是csv格式的，第一行是标签。
在这里插入图片描述

python代码实现

以下代码在dataTransform函数里改一下数据集，可以直接在pycharm中运行

复制代码

import numpy as np
import pandas as pd
import matplotlib as mpl
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler, StandardScaler

# KNN核心算法
def classify(inX, dataSet, y_train, k):
    m, n = dataSet.shape  # shape（m, n）测试集中有m个个体和n个特征
    # 计算测试数据到每个点的欧式距离
    distances = []
    for i in range(m):
        sum = 0
        for j in range(n):
            sum += (inX[j] - dataSet[i][j]) ** 2
        distances.append(sum ** 0.5)
    sortDist = sorted(distances)  # 得到的是按照distance排序好的
    # 求k个最近的值的平均值
    sum = 0  
    for i in range(k):
        sum += y_train[distances.index(sortDist[i])]
    return sum/k

# 用于归一化
x_scaler = MinMaxScaler(feature_range=(-1, 1))
y_scaler = MinMaxScaler(feature_range=(-1, 1))

# 数据的读取和归一化
def dataTransform():
    df = pd.read_csv('data.csv', encoding='utf-8')
    x = df[['AVGACURATE', 'AVGSPEED', 'AVGHEADDIST', 'LIGHTVEH', 'LONGVEH', 'HUGEVEH', 'MIDVEH', 'MINIVEH']]
    y = df['AVGFLOW']
    y = y.values.reshape(-1, 1)  # 在sklearn中，所有的数据都应该是二维矩阵,所以需要使用.reshape(1,-1)进行转换
    # 对数据进行最大最小值归一化
    x = x_scaler.fit_transform(x)
    y = y_scaler.fit_transform(y)
    # 训练集
    x_train = x[0:1000, :]  # 二维
    y_train = y[0:1000]
    # 测试集
    x_test = x[1001:1292, :]
    y_test = y[1001:1292]
    return x_train, y_train, x_test, y_test

# 测试算法
def Test():
    x_train, y_train, x_test, y_test = dataTransform()
    predict = []  # 记录预测值
    err = 0
    for i in range(len(x_test)):  # 对每一个测试数据
        predict.append(classify(x_test[i], x_train, y_train, 5))  # 返回平均值
        # print(predict[i], y_test[i])
        err += np.square(y_test[i]-predict[i])  # 计算误差和
    mse_err = err/len(x_test)
    print("the total mse error is: ", mse_err)
    predict = np.array(predict)  # 转成array
    draw(predict, y_test)

# 画图函数
def draw(predict, y_test):
    # 先转化为实际值
    predict = predict.reshape(-1, 1)
    predict = y_scaler.inverse_transform(predict)
    y_test = y_scaler.inverse_transform(y_test)
    # 解决中文无法显示的问题
    plt.rcParams['font.sans-serif'] = [u'SimHei']
    plt.rcParams['axes.unicode_minus'] = False
    plt.figure(figsize=(8, 6))
    plt.plot(predict, label='pred')
    plt.plot(y_test, label='actual')
    plt.title('车流预测_测试集', )
    plt.legend()
    plt.show()

if __name__ == '__main__':
    Test()

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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import numpy as np
import pandas as pd
import matplotlib as mpl
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler, StandardScaler


# KNN核心算法
def classify(inX, dataSet, y_train, k):
    m, n = dataSet.shape  # shape（m, n）测试集中有m个个体和n个特征
    # 计算测试数据到每个点的欧式距离
    distances = []
    for i in range(m):
        sum = 0
        for j in range(n):
            sum += (inX[j] - dataSet[i][j]) ** 2
        distances.append(sum ** 0.5)
    sortDist = sorted(distances)  # 得到的是按照distance排序好的
    # 求k个最近的值的平均值
    sum = 0  
    for i in range(k):
        sum += y_train[distances.index(sortDist[i])]
    return sum/k


# 用于归一化
x_scaler = MinMaxScaler(feature_range=(-1, 1))
y_scaler = MinMaxScaler(feature_range=(-1, 1))


# 数据的读取和归一化
def dataTransform():
    df = pd.read_csv('data.csv', encoding='utf-8')
    x = df[['AVGACURATE', 'AVGSPEED', 'AVGHEADDIST', 'LIGHTVEH', 'LONGVEH', 'HUGEVEH', 'MIDVEH', 'MINIVEH']]
    y = df['AVGFLOW']
    y = y.values.reshape(-1, 1)  # 在sklearn中，所有的数据都应该是二维矩阵,所以需要使用.reshape(1,-1)进行转换
    # 对数据进行最大最小值归一化
    x = x_scaler.fit_transform(x)
    y = y_scaler.fit_transform(y)
    # 训练集
    x_train = x[0:1000, :]  # 二维
    y_train = y[0:1000]
    # 测试集
    x_test = x[1001:1292, :]
    y_test = y[1001:1292]
    return x_train, y_train, x_test, y_test


# 测试算法
def Test():
    x_train, y_train, x_test, y_test = dataTransform()
    predict = []  # 记录预测值
    err = 0
    for i in range(len(x_test)):  # 对每一个测试数据
        predict.append(classify(x_test[i], x_train, y_train, 5))  # 返回平均值
        # print(predict[i], y_test[i])
        err += np.square(y_test[i]-predict[i])  # 计算误差和
    mse_err = err/len(x_test)
    print("the total mse error is: ", mse_err)
    predict = np.array(predict)  # 转成array
    draw(predict, y_test)


# 画图函数
def draw(predict, y_test):
    # 先转化为实际值
    predict = predict.reshape(-1, 1)
    predict = y_scaler.inverse_transform(predict)
    y_test = y_scaler.inverse_transform(y_test)
    # 解决中文无法显示的问题
    plt.rcParams['font.sans-serif'] = [u'SimHei']
    plt.rcParams['axes.unicode_minus'] = False
    plt.figure(figsize=(8, 6))
    plt.plot(predict, label='pred')
    plt.plot(y_test, label='actual')
    plt.title('车流预测_测试集', )
    plt.legend()
    plt.show()


if __name__ == '__main__':
    Test()