對於線性不可分的數據集,可以利用核函數(kernel)將數據轉換成易於分類器理解的形式。
如下圖,如果在x軸和y軸構成的坐標系中插入直線進行分類的話, 不能得到理想的結果,或許我們可以對圓中的數據進行某種形式的轉換,從而得到某些新的變量來表示數據。在這種表示情況下,我們就更容易得到大於0或者小於0的測試結果。在這個例子中,我們將數據從一個特征空間轉換到另一個特征空間,在新的空間下,我們可以很容易利用已有的工具對數據進行處理,將這個過程稱之為從一個特征空間到另一個特征空間的映射。在通常情況下,這種映射會將低維特征空間映射到高維空間。
這種從某個特征空間到另一個特征空間的映射是通過核函數來。
SVM優化中一個特別好的地方就是,所有的運算都可以寫成內積(inner product)的形式。向量的內積指的就是兩個向量相乘,之后得到單個標量或者數值。我們可以把內積運算替換成核函數,而並不必做簡化處理。將內積替換成核函數的方法被稱之為核技巧(kernel trick)或者核“變電”(kernel substation)。
徑向基核函數
徑向基核函數是SVM中常用的一個核函數。徑向基函數是一個采用向量作為自變量的函數,能夠基於向量距離運算輸出一個標量。
'''#######******************************** 以下是有核函數的版本 '''#######******************************** class optStruct: def __init__(self,dataMatIn, classLabels, C, toler, kTup): # Initialize the structure with the parameters self.X = dataMatIn self.labelMat = classLabels self.C = C self.tol = toler self.m = shape(dataMatIn)[0] self.alphas = mat(zeros((self.m,1))) self.b = 0 self.eCache = mat(zeros((self.m,2))) #first column is valid flag self.K = mat(zeros((self.m,self.m))) for i in range(self.m): self.K[:,i] = kernelTrans(self.X, self.X[i,:], kTup) def calcEk(oS, k): #計算誤差 fXk = float(multiply(oS.alphas,oS.labelMat).T*oS.K[:,k] + oS.b) Ek = fXk - float(oS.labelMat[k]) return Ek def selectJ(i, oS, Ei): #用於選擇第2個循環(內循環)的alpha值,內循環中的啟發式方法 maxK = -1; maxDeltaE = 0; Ej = 0 oS.eCache[i] = [1,Ei] #set valid #choose the alpha that gives the maximum delta E validEcacheList = nonzero(oS.eCache[:,0].A)[0] if (len(validEcacheList)) > 1: for k in validEcacheList: #loop through valid Ecache values and find the one that maximizes delta E if k == i: continue #跳過本身 Ek = calcEk(oS, k) deltaE = abs(Ei - Ek) if (deltaE > maxDeltaE): #選取具有最大步長的j maxK = k; maxDeltaE = deltaE; Ej = Ek return maxK, Ej else: #in this case (first time around) we don't have any valid eCache values j = selectJrand(i, oS.m) Ej = calcEk(oS, j) return j, Ej def updateEk(oS, k): #alpha改變后,更新緩存 Ek = calcEk(oS, k) oS.eCache[k] = [1,Ek] #內部循環的代碼和簡版的SMO代碼很相似 def innerL(i, oS): Ei = calcEk(oS, i) #判斷每一個alpha是否被優化過,如果誤差很大,就對該alpha值進行優化,toler是容錯率 if ((oS.labelMat[i]*Ei < -oS.tol) and (oS.alphas[i] < oS.C)) or ((oS.labelMat[i]*Ei > oS.tol) and (oS.alphas[i] > 0)): j,Ej = selectJ(i, oS, Ei) #使用啟發式方法選取第2個alpha,選取使得誤差最大的alpha alphaIold = oS.alphas[i].copy(); alphaJold = oS.alphas[j].copy(); #保證alpha在0與C之間 if (oS.labelMat[i] != oS.labelMat[j]): #當y1和y2異號,計算alpha的取值范圍 L = max(0, oS.alphas[j] - oS.alphas[i]) H = min(oS.C, oS.C + oS.alphas[j] - oS.alphas[i]) else: #當y1和y2同號,計算alpha的取值范圍 L = max(0, oS.alphas[j] + oS.alphas[i] - oS.C) H = min(oS.C, oS.alphas[j] + oS.alphas[i]) if L==H: print "L==H"; return 0 #eta是alpha[j]的最優修改量,eta=K11+K22-2*K12,也是f(x)的二階導數,K表示核函數 eta = 2.0 * oS.K[i,j] - oS.K[i,i] - oS.K[j,j] #changed for kernel #如果二階導數-eta <= 0,說明一階導數沒有最小值,就不做任何改變,本次循環結束直接運行下一次for循環 if eta >= 0: print "eta>=0"; return 0 oS.alphas[j] -= oS.labelMat[j]*(Ei - Ej)/eta #利用公式更新alpha[j],alpha2new=alpha2-yj(Ei-Ej)/eta oS.alphas[j] = clipAlpha(oS.alphas[j],H,L) #判斷alpha的范圍是否在0和C之間 updateEk(oS, j) #在alpha改變的時候更新Ecache print "j=",j print oS.alphas.A[j] #如果alphas[j]沒有調整,就忽略下面語句,本次循環結束直接運行下一次for循環 if (abs(oS.alphas[j] - alphaJold) < 0.00001): print "j not moving enough"; return 0 oS.alphas[i] += oS.labelMat[j]*oS.labelMat[i]*(alphaJold - oS.alphas[j])#update i by the same amount as j updateEk(oS, i) #在alpha改變的時候更新Ecache print "i=",i print oS.alphas.A[i] #已經計算出了alpha,接下來根據模型的公式計算b b1 = oS.b - Ei- oS.labelMat[i]*(oS.alphas[i]-alphaIold)*oS.K[i,i] - oS.labelMat[j]*(oS.alphas[j]-alphaJold)*oS.K[i,j] b2 = oS.b - Ej- oS.labelMat[i]*(oS.alphas[i]-alphaIold)*oS.K[i,j]- oS.labelMat[j]*(oS.alphas[j]-alphaJold)*oS.K[j,j] #根據公式確定偏移量b,理論上可選取任意支持向量來求解,但是現實任務中通常使用所有支持向量求解的平均值,這樣更加魯棒 if (0 < oS.alphas[i]) and (oS.C > oS.alphas[i]): oS.b = b1 elif (0 < oS.alphas[j]) and (oS.C > oS.alphas[j]): oS.b = b2 else: oS.b = (b1 + b2)/2.0 return 1 #如果有任意一對alpha發生改變,返回1 else: return 0 #完整版Platt SMO的外循環 def smoP(dataMatIn, classLabels, C, toler, maxIter,kTup=('lin', 0)): oS = optStruct(mat(dataMatIn),mat(classLabels).transpose(),C,toler, kTup) iter = 0 entireSet = True; alphaPairsChanged = 0 while (iter < maxIter) and ((alphaPairsChanged > 0) or (entireSet)): #有alpha改變同時遍歷次數小於最大次數,或者需要遍歷整個集合 alphaPairsChanged = 0 #首先進行完整遍歷,過程和簡化版的SMO一樣 if entireSet: for i in range(oS.m): alphaPairsChanged += innerL(i,oS) #i是第1個alpha的下標 print "完整遍歷, 迭代次數: %d i:%d, 成對改變的次數 %d" % (iter,i,alphaPairsChanged) iter += 1 #非邊界遍歷,挑選其中alpha值在0和C之間非邊界alpha進行優化 else: nonBoundIs = nonzero((oS.alphas.A > 0) * (oS.alphas.A < C))[0] #然后挑選其中值在0和C之間的非邊界alpha進行遍歷 for i in nonBoundIs: alphaPairsChanged += innerL(i,oS) print "非邊界, 迭代次數: %d i:%d, 成對改變的次數 %d" % (iter,i,alphaPairsChanged) iter += 1 #如果這次是完整遍歷的話,下次不用進行完整遍歷 if entireSet: entireSet = False #終止完整循環 elif (alphaPairsChanged == 0): entireSet = True #如果alpha的改變數量為0的話,再次遍歷所有的集合一次 print "iteration number: %d" % iter return oS.b,oS.alphas def calcWs(alphas,dataArr,classLabels): #計算模型的參數w,即alpha*y*x轉置的累加 X = mat(dataArr); labelMat = mat(classLabels).transpose() m,n = shape(X) w = zeros((n,1)) for i in range(m): w += multiply(alphas[i]*labelMat[i],X[i,:].T) return w def testRbf(k1=1.3): dataArr,labelArr = loadDataSet('testSetRBF.txt') b,alphas = smoP(dataArr, labelArr, 200, 0.0001, 10000, ('rbf', k1)) #C=200 important datMat=mat(dataArr); labelMat = mat(labelArr).transpose() svInd=nonzero(alphas.A>0)[0] sVs=datMat[svInd] #get matrix of only support vectors labelSV = labelMat[svInd]; print "there are %d Support Vectors" % shape(sVs)[0] m,n = shape(datMat) errorCount = 0 for i in range(m): kernelEval = kernelTrans(sVs,datMat[i,:],('rbf', k1)) predict=kernelEval.T * multiply(labelSV,alphas[svInd]) + b if sign(predict)!=sign(labelArr[i]): errorCount += 1 print "the training error rate is: %f" % (float(errorCount)/m) dataArr,labelArr = loadDataSet('testSetRBF2.txt') errorCount = 0 datMat=mat(dataArr); labelMat = mat(labelArr).transpose() m,n = shape(datMat) for i in range(m): kernelEval = kernelTrans(sVs,datMat[i,:],('rbf', k1)) predict=kernelEval.T * multiply(labelSV,alphas[svInd]) + b if sign(predict)!=sign(labelArr[i]): errorCount += 1 print "the test error rate is: %f" % (float(errorCount)/m) def img2vector(filename): returnVect = zeros((1,1024)) fr = open(filename) for i in range(32): lineStr = fr.readline() for j in range(32): returnVect[0,32*i+j] = int(lineStr[j]) return returnVect def loadImages(dirName): from os import listdir hwLabels = [] trainingFileList = listdir(dirName) #load the training set m = len(trainingFileList) trainingMat = zeros((m,1024)) for i in range(m): fileNameStr = trainingFileList[i] fileStr = fileNameStr.split('.')[0] #take off .txt classNumStr = int(fileStr.split('_')[0]) if classNumStr == 9: hwLabels.append(-1) else: hwLabels.append(1) trainingMat[i,:] = img2vector('%s/%s' % (dirName, fileNameStr)) return trainingMat, hwLabels def testDigits(kTup=('rbf', 10)): dataArr,labelArr = loadImages('trainingDigits') b,alphas = smoP(dataArr, labelArr, 200, 0.0001, 10000, kTup) datMat=mat(dataArr); labelMat = mat(labelArr).transpose() svInd=nonzero(alphas.A>0)[0] sVs=datMat[svInd] labelSV = labelMat[svInd]; print "there are %d Support Vectors" % shape(sVs)[0] m,n = shape(datMat) errorCount = 0 for i in range(m): kernelEval = kernelTrans(sVs,datMat[i,:],kTup) predict=kernelEval.T * multiply(labelSV,alphas[svInd]) + b if sign(predict)!=sign(labelArr[i]): errorCount += 1 print "the training error rate is: %f" % (float(errorCount)/m) dataArr,labelArr = loadImages('testDigits') errorCount = 0 datMat=mat(dataArr); labelMat = mat(labelArr).transpose() m,n = shape(datMat) for i in range(m): kernelEval = kernelTrans(sVs,datMat[i,:],kTup) predict=kernelEval.T * multiply(labelSV,alphas[svInd]) + b if sign(predict)!=sign(labelArr[i]): errorCount += 1 print "the test error rate is: %f" % (float(errorCount)/m)