對於線性不可分的數據集,可以利用核函數(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)

