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

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

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

一、KNN算法簡介

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

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

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

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

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

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

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

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

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

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

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

確定K的值

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

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

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

適合對稀有事件進行分類；

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

KNN算法的缺點：

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

該方法的另一個不足之處是計算量較大，因為對每一個待分類的文本都要計算它到全體已知樣本的距離，才能求得它的 K 個最近鄰點。

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

要自己動手用 Python 實現(xiàn) KNN 算法，主要有以下三個步驟：

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

三、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個月
    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ù)下載完成！')

多爬取幾個城市 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ù)  當天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ù)', '當天AQI排名', 'PM2.5', 'PM10', 'So2', 'No2', 'Co', 'O3']].copy()

air_quality = []
# print(df['質(zhì)量等級'].value_counts())
# 質(zhì)量等級列數(shù)據(jù)為字符串  轉(zhuǎn)為為標簽  便于判斷預(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 == "嚴重污染":
        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

# 自定義其他幾個城市空氣質(zhì)量數(shù)據(jù)作為訓(xùn)練集
df = pd.read_csv('2019年天津空氣質(zhì)量數(shù)據(jù).csv', encoding='utf-8')
# 取質(zhì)量等級  AQI指數(shù)  當天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ù)', '當天AQI排名', 'PM2.5', 'PM10', 'So2', 'No2', 'Co', 'O3']].copy()

air_quality = []
# print(df['質(zhì)量等級'].value_counts())
# 質(zhì)量等級列數(shù)據(jù)為字符串  轉(zhuǎn)為為數(shù)字標識
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 == "嚴重污染":
        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. 實現(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個參數(shù)轉(zhuǎ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個參數(shù)轉(zhuǎn)換為浮點數(shù)
              trainingSet.append(dataset2[x])  # 生成訓(xùn)練集

計算歐氏距離

def calculateDistance(testdata, traindata, length):  # 計算距離
      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 個相鄰最近的鄰居

def getNeighbors(self, trainingSet, test_instance, k):  # 返回最近的k個邊距
      distances = []
      length = len(test_instance)
      # 對訓(xùn)練集的每一個數(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個
      for x in range(k):
          neighbors.append(distances[x][0])
      print(neighbors)
      return neighbors

計算比例最大的分類

def getResponse(neighbors):   # 根據(jù)少數(shù)服從多數(shù)，決定歸類到哪一類
      class_votes = {}
      for x in range(len(neighbors)):
          response = neighbors[x][-1]  # 統(tǒng)計每一個分類的多少  空氣質(zhì)量的數(shù)字標識
          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ù)測準確率計算

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

      print('有{}個預(yù)測正確，共有{}個測試數(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個數(shù)據(jù)
    for x in range(len(test_set)):  # 對所有的測試集進行測試
        neighbors = self.getNeighbors(training_set, test_set[x], k)  # 找到8個最近的鄰居
        result = self.getResponse(neighbors)  # 找這7個鄰居歸類到哪一類
        predictions.append(result)

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

運行效果如下：