Python如何手寫KNN算法預(yù)測城市空氣質(zhì)量

本篇內(nèi)容介紹了“Python如何手寫KNN算法預(yù)測城市空氣質(zhì)量”的有關(guān)知識，在實(shí)際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

一、KNN算法簡介

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

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

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

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

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

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

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

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

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

通常使用的距離函數(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，對象具有多個類別標(biāo)簽)， KNN比 SVM 的表現(xiàn)要好。

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

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

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

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

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

• 算距離：給定待分類樣本，計(jì)算它與已分類樣本中的每個樣本的距離。

• 找鄰居：圈定與待分類樣本距離最近的 K 個已分類樣本，作為待分類樣本的近鄰。

• 做分類：根據(jù)這 K 個近鄰中的大部分樣本所屬的類別來決定待分類樣本該屬于哪個分類。

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

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

對于這種 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ù)  當(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

# 自定義其他幾個城市空氣質(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個參數(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個參數(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 個相鄰最近的鄰居

def getNeighbors(self, trainingSet, test_instance, k):  # 返回最近的k個邊距
      distances = []
      length = len(test_instance)
      # 對訓(xùn)練集的每一個數(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個
      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ì)每一個分類的多少  空氣質(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:
              # 查看錯誤預(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)):  # 對所有的測試集進(jìn)行測試
        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ù)測準(zhǔn)確度為:  {:.2f}%'.format(accuracy))   # 保留2位小數(shù)

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

可以通過增加訓(xùn)練集城市空氣質(zhì)量數(shù)據(jù)量，調(diào)節(jié)找鄰居的數(shù)量k，提高預(yù)測準(zhǔn)確率。

“Python如何手寫KNN算法預(yù)測城市空氣質(zhì)量”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！