Python機(jī)器學(xué)習(xí)之手寫KNN算法預(yù)測城市空氣質(zhì)量

更新時(shí)間：2021年12月11日 09:06:40 作者：葉庭云

KNN（K-Nearest Neighbor）最鄰近分類算法是數(shù)據(jù)挖掘分類（classification）技術(shù)中常用算法之一，本文將介紹如何通過KNN算法實(shí)現(xiàn)城市空氣質(zhì)量的預(yù)測，感興趣的同學(xué)可以了解一下

一、KNN算法簡介

KNN（K-Nearest Neighbor）最鄰近分類算法是數(shù)據(jù)挖掘分類（classification）技術(shù)中常用算法之一，其指導(dǎo)思想是"近朱者赤，近墨者黑"，即由你的鄰居來推斷出你的類別。

KNN最鄰近分類算法的實(shí)現(xiàn)原理：為了判斷未知樣本的類別，以所有已知類別的樣本作為參照，計(jì)算未知樣本與所有已知樣本的距離，從中選取與未知樣本距離最近的 K 個(gè)已知樣本，再根據(jù)少數(shù)服從多數(shù)的投票法則（majority-voting），將未知樣本與 K 個(gè)最鄰近樣本中所屬類別占比較多的歸為一類。

KNN算法的核心思想：尋找最近的 k 個(gè)數(shù)據(jù)，推測新數(shù)據(jù)的分類

KNN算法的關(guān)鍵：

1.樣本的所有特征都要做可比較的量化

若是樣本特征中存在非數(shù)值的類型，必須采取方法將其量化為數(shù)值。例如樣本特征中包含顏色，可通過將顏色轉(zhuǎn)換為灰度值來實(shí)現(xiàn)距離計(jì)算。

2.樣本特征要做歸一化處理

樣本有多個(gè)參數(shù)，每一個(gè)參數(shù)都有自己的定義域和取值范圍，他們對距離計(jì)算的影響不一樣，如取值較大的影響力會(huì)蓋過取值較小的參數(shù)。所以樣本參數(shù)必須做一些 scale 處理，最簡單的方式就是所有特征的數(shù)值都采取歸一化處理。

3.需要一個(gè)距離函數(shù)以計(jì)算兩個(gè)樣本之間的距離

通常使用的距離函數(shù)有：歐氏距離、余弦距離、漢明距離、曼哈頓距離等，一般選歐氏距離作為距離度量，但是這是只適用于連續(xù)變量。在文本分類這種非連續(xù)變量情況下，余弦距離可以用來作為度量。通常情況下，如果運(yùn)用一些特殊的算法來計(jì)算度量的話，K近鄰分類精度可顯著提高，如運(yùn)用大邊緣最近鄰法或者近鄰成分分析法。

以計(jì)算二維空間中的A(x1,y1)、B(x2,y2)兩點(diǎn)之間的距離為例，常用的歐氏距離的計(jì)算方法如下圖所示：

確定K的值

K值選的太大易引起欠擬合，太小容易過擬合，需交叉驗(yàn)證確定 K 值。

KNN算法的優(yōu)點(diǎn)：

簡單，易于理解，易于實(shí)現(xiàn)，無需估計(jì)參數(shù)，無需訓(xùn)練；

適合對稀有事件進(jìn)行分類；

特別適合于多分類問題(multi-modal，對象具有多個(gè)類別標(biāo)簽)， KNN比 SVM 的表現(xiàn)要好。

KNN算法的缺點(diǎn)：

KNN算法在分類時(shí)有個(gè)主要的不足是：當(dāng)樣本不平衡時(shí)，如一個(gè)類的樣本容量很大，而其他類樣本容量很小時(shí)，有可能導(dǎo)致當(dāng)輸入一個(gè)新樣本時(shí)，該樣本的 K 個(gè)鄰居中大容量類的樣本占多數(shù)。該算法只計(jì)算最近的鄰居樣本，某一類的樣本數(shù)量很大，那么或者這類樣本并不接近目標(biāo)樣本，或者這類樣本很靠近目標(biāo)樣本。無論怎樣，數(shù)量并不能影響運(yùn)行結(jié)果?？梢圆捎脵?quán)值的方法(和該樣本距離小的鄰居權(quán)值大)來改進(jìn)。

該方法的另一個(gè)不足之處是計(jì)算量較大，因?yàn)閷γ恳粋€(gè)待分類的文本都要計(jì)算它到全體已知樣本的距離，才能求得它的 K 個(gè)最近鄰點(diǎn)。

二、KNN算法實(shí)現(xiàn)思路

要自己動(dòng)手用 Python 實(shí)現(xiàn) KNN 算法，主要有以下三個(gè)步驟：

算距離：給定待分類樣本，計(jì)算它與已分類樣本中的每個(gè)樣本的距離。
找鄰居：圈定與待分類樣本距離最近的 K 個(gè)已分類樣本，作為待分類樣本的近鄰。
做分類：根據(jù)這 K 個(gè)近鄰中的大部分樣本所屬的類別來決定待分類樣本該屬于哪個(gè)分類。

三、KNN算法預(yù)測城市空氣質(zhì)量

1. 獲取數(shù)據(jù)

數(shù)據(jù)來源：http://www.tianqihoubao.com/aqi/chengdu-201901.html

對于這種 Table 表格型數(shù)據(jù)，可以直接用 pandas 的 read_html() 大法，將數(shù)據(jù)保存到csv，也就不用再寫爬蟲去解析網(wǎng)頁和提取數(shù)據(jù)了。

# -*- coding: UTF-8 -*-
"""
@File    ：spider.py
@Author  ：葉庭云
@CSDN    ：https://yetingyun.blog.csdn.net/
@http://www.tianqihoubao.com/aqi/beijing-201901.html
"""
import pandas as pd
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s')

for page in range(1, 13):   # 12個(gè)月
    if page < 10:
        url = f'http://www.tianqihoubao.com/aqi/guangzhou-20190{page}.html'
        df = pd.read_html(url, encoding='gbk')[0]
        if page == 1:
            df.to_csv('2019年廣州空氣質(zhì)量數(shù)據(jù).csv', mode='a+', index=False, header=False)
        else:
            df.iloc[1:,::].to_csv('2019年廣州空氣質(zhì)量數(shù)據(jù).csv', mode='a+', index=False, header=False)
    else:
        url = f'http://www.tianqihoubao.com/aqi/guangzhou-2019{page}.html'
        df = pd.read_html(url, encoding='gbk')[0]
        df.iloc[1:,::].to_csv('2019年廣州空氣質(zhì)量數(shù)據(jù).csv', mode='a+', index=False, header=False)

    logging.info(f'{page}月空氣質(zhì)量數(shù)據(jù)下載完成！')

多爬取幾個(gè)城市 2019 年歷史空氣質(zhì)量數(shù)據(jù)保存到本地

2. 生成測試集和訓(xùn)練集

import pandas as pd

# 將2019年成都空氣質(zhì)量數(shù)據(jù)作為測試集
df = pd.read_csv('2019年成都空氣質(zhì)量數(shù)據(jù).csv')
# 取質(zhì)量等級  AQI指數(shù)  當(dāng)天AQI排名  PM2.5 。。。8列數(shù)據(jù)
# SettingWithCopyWarning: A value is trying to be set on a copy of a slice from a DataFrame 解決方法
df1 = df[['AQI指數(shù)', '當(dāng)天AQI排名', 'PM2.5', 'PM10', 'So2', 'No2', 'Co', 'O3']].copy()

air_quality = []
# print(df['質(zhì)量等級'].value_counts())
# 質(zhì)量等級列數(shù)據(jù)為字符串  轉(zhuǎn)為為標(biāo)簽  便于判斷預(yù)測
for i in df['質(zhì)量等級']:
    if i == "優(yōu)":
        air_quality.append('1')
    elif i == "良":
        air_quality.append('2')
    elif i == "輕度污染":
        air_quality.append('3')
    elif i == "中度污染":
        air_quality.append('4')
    elif i == "重度污染":
        air_quality.append('5')
	elif i == "嚴(yán)重污染":
        air_quality.append('6')

print(air_quality)
df1['空氣質(zhì)量'] = air_quality

# 將數(shù)據(jù)寫入test.txt
# print(df1.values, type(df1.values)) # <class 'numpy.ndarray'>
with open('test.txt', 'w') as f:
    for x in df1.values:
        print(x)
        s = ','.join([str(i) for i in x])
        # print(s, type(s))
        f.write(s + '\n')

import pandas as pd

# 自定義其他幾個(gè)城市空氣質(zhì)量數(shù)據(jù)作為訓(xùn)練集
df = pd.read_csv('2019年天津空氣質(zhì)量數(shù)據(jù).csv', encoding='utf-8')
# 取質(zhì)量等級  AQI指數(shù)  當(dāng)天AQI排名  PM2.5 。。。8列數(shù)據(jù)
# SettingWithCopyWarning: A value is trying to be set on a copy of a slice from a DataFrame 解決方法
df1 = df[['AQI指數(shù)', '當(dāng)天AQI排名', 'PM2.5', 'PM10', 'So2', 'No2', 'Co', 'O3']].copy()

air_quality = []
# print(df['質(zhì)量等級'].value_counts())
# 質(zhì)量等級列數(shù)據(jù)為字符串  轉(zhuǎn)為為數(shù)字標(biāo)識
for i in df['質(zhì)量等級']:
    if i == "優(yōu)":
        air_quality.append('1')
    elif i == "良":
        air_quality.append('2')
    elif i == "輕度污染":
        air_quality.append('3')
    elif i == "中度污染":
        air_quality.append('4')
    elif i == "重度污染":
        air_quality.append('5')
    elif i == "嚴(yán)重污染":
        air_quality.append('6')

print(air_quality)
df1['空氣質(zhì)量'] = air_quality

# 將數(shù)據(jù)寫入追加寫入到train.txt
# print(df1.values, type(df1.values)) # <class 'numpy.ndarray'>
with open('train.txt', 'a+') as f:
    for x in df1.values:
        print(x)
        s = ','.join([str(i) for i in x])
        # print(s, type(s))
        f.write(s + '\n')

3. 實(shí)現(xiàn)KNN算法

讀取數(shù)據(jù)集

def read_dataset(filename1, filename2, trainingSet, testSet):
      with open(filename1, 'r') as csvfile:
          lines = csv.reader(csvfile)  # 讀取所有的行
          dataset1 = list(lines)       # 轉(zhuǎn)化成列表
          for x in range(len(dataset1)):  # 每一行數(shù)據(jù)
              for y in range(8):
                  dataset1[x][y] = float(dataset1[x][y])   # 8個(gè)參數(shù)轉(zhuǎn)換為浮點(diǎn)數(shù)
              testSet.append(dataset1[x])    # 生成測試集

      with open(filename2, 'r') as csvfile:
          lines = csv.reader(csvfile)  # 讀取所有的行
          dataset2 = list(lines)       # 轉(zhuǎn)化成列表
          for x in range(len(dataset2)):   # 每一行數(shù)據(jù)
              for y in range(8):
                  dataset2[x][y] = float(dataset2[x][y])  # 8個(gè)參數(shù)轉(zhuǎn)換為浮點(diǎn)數(shù)
              trainingSet.append(dataset2[x])  # 生成訓(xùn)練集

計(jì)算歐氏距離

def calculateDistance(testdata, traindata, length):  # 計(jì)算距離
      distance = 0  # length表示維度 數(shù)據(jù)共有幾維
      for x in range(length):
          distance += pow((int(testdata[x]) - int(traindata[x])), 2)
      return round(math.sqrt(distance), 3)    # 保留3位小數(shù)

找 K 個(gè)相鄰最近的鄰居

def getNeighbors(self, trainingSet, test_instance, k):  # 返回最近的k個(gè)邊距
      distances = []
      length = len(test_instance)
      # 對訓(xùn)練集的每一個(gè)數(shù)計(jì)算其到測試集的實(shí)際距離
      for x in range(len(trainingSet)):
          dist = self.calculateDistance(test_instance, trainingSet[x], length)
          print('訓(xùn)練集：{} --- 距離：{}'.format(trainingSet[x], dist))
          distances.append((trainingSet[x], dist))
      distances.sort(key=operator.itemgetter(1))  # 按距離從小到大排列
      # print(distances)
      neighbors = []
      # 排序完成后取距離最小的前k個(gè)
      for x in range(k):
          neighbors.append(distances[x][0])
      print(neighbors)
      return neighbors

計(jì)算比例最大的分類

def getResponse(neighbors):   # 根據(jù)少數(shù)服從多數(shù)，決定歸類到哪一類
      class_votes = {}
      for x in range(len(neighbors)):
          response = neighbors[x][-1]  # 統(tǒng)計(jì)每一個(gè)分類的多少  空氣質(zhì)量的數(shù)字標(biāo)識
          if response in class_votes:
              class_votes[response] += 1
          else:
              class_votes[response] = 1
      print(class_votes.items())
      sortedVotes = sorted(class_votes.items(), key=operator.itemgetter(1), reverse=True)  # 按分類大小排序  降序
      return sortedVotes[0][0]    # 分類最大的  少數(shù)服從多數(shù)   為預(yù)測結(jié)果

預(yù)測準(zhǔn)確率計(jì)算

def getAccuracy(test_set, predictions):
      correct = 0
      for x in range(len(test_set)):
          # predictions預(yù)測的與testset實(shí)際的比對  計(jì)算預(yù)測的準(zhǔn)確率
          if test_set[x][-1] == predictions[x]:
              correct += 1
          else:
              # 查看錯(cuò)誤預(yù)測
              print(test_set[x], predictions[x])

      print('有{}個(gè)預(yù)測正確，共有{}個(gè)測試數(shù)據(jù)'.format(correct, len(test_set)))
      return (correct / (len(test_set))) * 100.0

run函數(shù)調(diào)用

# -*- coding: UTF-8 -*-
"""
@Author  ：葉庭云
@公眾號  ：修煉Python
@CSDN    ：https://yetingyun.blog.csdn.net/
"""
def run(self):
    training_set = []    # 訓(xùn)練集
    test_set = []        # 測試集
    self.read_dataset('./train_4/test.txt', './train_4/train.txt', training_set, test_set)  # 數(shù)據(jù)劃分
    print('Train set: ' + str(len(training_set)))
    print('Test set: ' + str(len(test_set)))
    # generate predictions
    predictions = []
    k = 7  # 取最近的6個(gè)數(shù)據(jù)
    for x in range(len(test_set)):  # 對所有的測試集進(jìn)行測試
        neighbors = self.getNeighbors(training_set, test_set[x], k)  # 找到8個(gè)最近的鄰居
        result = self.getResponse(neighbors)  # 找這7個(gè)鄰居歸類到哪一類
        predictions.append(result)

    accuracy = self.getAccuracy(test_set, predictions)
    print('預(yù)測準(zhǔn)確度為:  {:.2f}%'.format(accuracy))   # 保留2位小數(shù)

運(yùn)行效果如下：