SVM学习—SMO算法（Sequential Minimal Optimization）

importmath

importpickle

importrandom

from..math.matriximportMatrix

from..math.text2matriximportText2Matrix

from..nlp.segmenterimportSegmenter

from..common.global_infoimportGlobalInfo

from..common.configurationimportConfiguration

classStandardSMO:

'''Platt'sstandardSMOalgorithmforcsvc.'''

def__init__(self,config,nodeName,loadFromFile=False,C=100,tolerance=0.001):

#storeanumbernearlyzero.

self.accuracy=1E-3

#storepenaltycoefficientofslackvariable.

self.C=C

#storetoleranceofKKTconditions.

self.tolerance=tolerance

#storeisTrainedbydata.

self.istrained=loadFromFile

#storelagrangemultipiers.

self.alpha=[]

#storeweight

self.w=[]

#storethreshold.

self.b=float(0)

#storekii

self.kcache={}

self.istrained=False

#-------------------beginmodelinfo-------------------------------

self.curNode=config.GetChild(nodeName)

self.modelPath=self.curNode.GetChild("model_path").GetValue()

self.logPath=self.curNode.GetChild("log_path").GetValue()

#-------------------endmodelinfo-------------------------------

#-------------------beginkernelinfo-------------------------------

self.curNode=config.GetChild(nodeName)

self.kernelNode=self.curNode.GetChild("kernel");

self.kernelName=self.kernelNode.GetChild("name").GetValue();

#togetparametersfromtoptobutton->fromlefttoright->frominnertoouter.

self.parameters=self.kernelNode.GetChild("parameters").GetValue().split(',');

#-------------------endkernelinfo-------------------------------

if(loadFromFile):

f=open(self.modelPath,"r")

modelStr=pickle.load(f)

[self.alphay,self.sv,self.b,self.w]=pickle.loads(modelStr)

f.close()

self.istrained=True

defDotProduct(self,i1,i2):

'''Togetvector'sdotproductfortraining.'''

dot=float(0)

foriinrange(0,self.trainx.nCol):

dot+=self.trainx.Get(i1,i)*self.trainx.Get(i2,i)

returndot

defKernel(self):

'''Togetkernelfunctionwithconfigurationfortraining.

kernelfunctionincludesRBF,Linearandsoon.'''

ifself.kernelName=='RBF':

returnlambdaxi,yi:math.exp((2*self.DotProduct(xi,yi)-self.DotProduct(xi,xi)-self.DotProduct(yi,yi))/(2*float(self.parameters[0])*float(self.parameters[0])))

elifself.kernelName=='Linear':

returnlambdaxi,yi:self.DotProduct(xi,yi)+float(self.parameters[0])

elifself.kernelName=='Polynomial':

returnlambdaxi,yi:(float(self.parameters[0])*self.DotProduct(xi,yi)+float(self.parameters[1]))**int(self.parameters[2])

defDotVectorProduct(self,v1,v2):

'''Togetvector'sdotproductfortesting.'''

iflen(v1)!=len(v2):

print'Thedimensionsoftwovectorshouldequal'

return0.0

dot=float(0)

foriinrange(0,len(v1)):

dot+=v1[i]*v2[i]

returndot

defKernelVector(self,v1,v2):

'''Togetkernelfunctionfortesting.'''

ifself.kernelName=='RBF':

returnmath.exp((2*self.DotVectorProduct(v1,v2)-self.DotVectorProduct(v1,v1)-self.DotVectorProduct(v2,v2))/(2*float(self.parameters[0])*float(self.parameters[0])))

elifself.kernelName=='Linear':

returnself.DotVectorProduct(v1,v2)+float(self.parameters[0])

elifself.kernelName=='Polynomial':

return(float(self.parameters[0])*self.DotVectorProduct(v1,v2)+float(self.parameters[1]))**int(self.parameters[2])

defF(self,i1):

'''Tocalculateoutputofansample.

returnoutput.'''

ifself.kernelName=='Linear':

dot=0

foriinrange(0,self.trainx.nCol):

dot+=self.w[i]*self.trainx.Get(i1,i);

returndot+self.b

K=self.Kernel()

final=0.0

foriinrange(0,len(self.alpha)):

ifself.alpha[i]>0:

key1='%s%s%s'%(str(i1),'-',str(i))

key2='%s%s%s'%(str(i),'-',str(i1))

ifself.kcache.has_key(key1):

k=self.kcache[key1]

elifself.kcache.has_key(key2):

k=self.kcache[key2]

else:

k=K(i1,i)

self.kcache[key1]=k

final+=self.alpha[i]*self.trainy[i]*k

final+=self.b

returnfinal

defexamineExample(self,i1):

'''Tofindthefirstlagrangemultipliers.

thenfindthesecondlagrangemultipliers.'''

y1=self.trainy[i1]

alpha1=self.alpha[i1]

E1=self.F(i1)-y1

kkt=y1*E1

if(kkt>self.toleranceandkkt>0)or(kkt<-self.toleranceandkkt<self.C):#notabidebyKKTconditions

ifself.FindMaxNonbound(i1,E1):

return1

elifself.FindRandomNonbound(i1):

return1

elifself.FindRandom(i1):

return1

return0

defFindMaxNonbound(self,i1,E1):

'''Tofindsecondlagrangemultipliersfromnon-bound.

conditionismaximum|E1-E2|ofnon-boundlagrangemultipliers.'''

i2=-1

maxe1e2=None

foriinrange(0,len(self.alpha)):

ifself.alpha[i]>0andself.alpha[i]<self.C:

E2=self.F(i)-self.trainy[i]

tmp=math.fabs(E1-E2)

ifmaxe1e2==Noneormaxe1e2<tmp:

maxe1e2=tmp

i2=i

ifi2>=0andself.StepOnebyOne(i1,i2):

return1

return0

defFindRandomNonbound(self,i1):

'''Tofindsecondlagrangemultipliersfromnon-bound.

conditionisrandomofnon-boundlagrangemultipliers.'''

k=random.randint(0,len(self.alpha)-1)

foriinrange(0,len(self.alpha)):

i2=(i+k)%len(self.alpha)

ifself.alpha[i2]>0andself.alpha[i2]<self.Candself.StepOnebyOne(i1,i2):

return1

return0

defFindRandom(self,i1):

'''Tofindsecondlagrangemultipliersfromall.

conditionisrandomoneofalllagrangemultipliers.'''

k=random.randint(0,len(self.alpha)-1)

foriinrange(0,len(self.alpha)):

i2=(i+k)%len(self.alpha)

ifself.StepOnebyOne(i1,i2):

return1

return0

defW(self,alpha1new,alpha2newclipped,i1,i2,E1,E2,k11,k22,k12):

'''TocalculateWvalue.'''

K=self.Kernel()

alpha1=self.alpha[i1]

alpha2=self.alpha[i2]

y1=self.trainy[i1]

y2=self.trainy[i2]

s=y1*y2

w1=alpha1new*(y1*(self.b-E1)+alpha1*k11+s*alpha2*k12)

w1+=alpha2newclipped*(y2*(self.b-E2)+alpha2*k22+s*alpha1*k12)

w1=w1-k11*alpha1new*alpha1new/2-k22*alpha2newclipped*alpha2newclipped/2-s*k12*alpha1new*alpha2newclipped

returnw1

defStepOnebyOne(self,i1,i2):

'''Tosolvetwolagrangemultipliersproblem.

thealgorithmcanreferencetheblog.'''

ifi1==i2:

return0

#togetkernelfunction.

K=self.Kernel()

alpha1=self.alpha[i1]

alpha2=self.alpha[i2]

alpha1new=-1.0

alpha2new=-1.0

alpha2newclipped=-1.0

y1=self.trainy[i1]

y2=self.trainy[i2]

s=y1*y2

key11='%s%s%s'%(str(i1),'-',str(i1))

key22='%s%s%s'%(str(i2),'-',str(i2))

key12='%s%s%s'%(str(i1),'-',str(i2))

key21='%s%s%s'%(str(i2),'-',str(i1))

ifself.kcache.has_key(key11):

k11=self.kcache[key11]

else:

k11=K(i1,i1)

self.kcache[key11]=k11

ifself.kcache.has_key(key22):

k22=self.kcache[key22]

else:

k22=K(i2,i2)

self.kcache[key22]=k22

ifself.kcache.has_key(key12):

k12=self.kcache[key12]

elifself.kcache.has_key(key21):

k12=self.kcache[key21]

else:

k12=K(i1,i2)

self.kcache[key12]=k12

eta=k11+k22-2*k12

E1=self.F(i1)-y1

E2=self.F(i2)-y2

#tocalucatebound.

L=0.0

H=0.0

ify1*y2==-1:

gamma=alpha2-alpha1

ifgamma>0:

L=gamma

H=self.C

else:

L=0

H=self.C+gamma

ify1*y2==1:

gamma=alpha2+alpha1

ifgamma-self.C>0:

L=gamma-self.C

H=self.C

else:

L=0

H=gamma

ifH==L:

return0

#------------------------begintomovelagrangemultipliers.----------------------------

if-eta<0:

#tocalculateapha2'snewvalue

alpha2new=alpha2+y2*(E1-E2)/eta

ifalpha2new<L:

alpha2newclipped=L

elifalpha2new>H:

alpha2newclipped=H

else:

alpha2newclipped=alpha2new

else:

w1=self.W(alpha1+s*(alpha2-L),L,i1,i2,E1,E2,k11,k22,k12)

w2=self.W(alpha1+s*(alpha2-H),H,i1,i2,E1,E2,k11,k22,k12)

ifw1-w2>self.accuracy:

alpha2newclipped=L

elifw2-w1>self.accuracy:

alpha2newclipped=H

else:

alpha2newclipped=alpha2

ifmath.fabs(alpha2newclipped-alpha2)<self.accuracy*(alpha2newclipped+alpha2+self.accuracy):

return0

alpha1new=alpha1+s*(alpha2-alpha2newclipped)

ifalpha1new<0:

alpha2newclipped+=s*alpha1new

alpha1new=0

elifalpha1new>self.C:

alpha2newclipped+=s*(alpha1new-self.C)

alpha1new=self.C

#------------------------endtomovelagrangemultipliers.----------------------------

ifalpha1new>0andalpha1new<self.C:

self.b+=(alpha1-alpha1new)*y1*k11+(alpha2-alpha2newclipped)*y2*k12-E1

elifalpha2newclipped>0andalpha2newclipped<self.C:

self.b+=(alpha1-alpha1new)*y1*k12+(alpha2-alpha2newclipped)*y2*k22-E2

else:

b1=(alpha1-alpha1new)*y1*k11+(alpha2-alpha2newclipped)*y2*k12-E1+self.b

b2=(alpha1-alpha1new)*y1*k12+(alpha2-alpha2newclipped)*y2*k22-E2+self.b

self.b=(b1+b2)/2

ifself.kernelName=='Linear':

forjinrange(0,self.trainx.nCol):

self.w[j]+=(alpha1new-alpha1)*y1*self.trainx.Get(i1,j)+(alpha2newclipped-alpha2)*y2*self.trainx.Get(i2,j)

self.alpha[i1]=alpha1new

self.alpha[i2]=alpha2newclipped

print'a',i1,'=',alpha1new,'a',i2,'=',alpha2newclipped

return1

defTrain(self,trainx,trainy):

'''Totrainsamples.

self.trainxistrainingmatrixandself.trainyisclassifyinglabel'''

self.trainx=trainx

self.trainy=trainy

iflen(self.trainy)!=self.trainx.nRow:

print"ERROR!,x.nRowshould==len(y)"

return0

numChanged=0;

examineAll=1;

#toinitializealllagrangemultipierswithzero.

foriinrange(0,self.trainx.nRow):

self.alpha.append(0.0)

#toinitializewwithzero.

forjinrange(0,self.trainx.nCol):

self.w.append(float(0))

whilenumChanged>0orexamineAll:

numChanged=0

print'numChanged=',numChanged

ifexamineAll:

forkinrange(0,self.trainx.nRow):

numChanged+=self.examineExample(k);#firsttimeoralloflagrangemultipiersareabidebyKKTconditionsthenexaminallsamples.

else:

forkinrange(0,self.trainx.nRow):

ifself.alpha[k]!=0andself.alpha[k]!=self.C:

numChanged+=self.examineExample(k);#toexaminallnon-boundlagrangemultipliers

if(examineAll==1):

examineAll=0

elif(numChanged==0):

examineAll=1

else:

#storesupportvectormachine.

self.alphay=[]

self.index=[]

foriinrange(0,len(self.alpha)):

ifself.alpha[i]>0:

self.index.append(i)

self.alphay.append(self.alpha[i]*self.trainy[i])

self.sv=[[0forjinrange(self.trainx.nCol)]foriinrange(len(self.index))]

foriinrange(0,len(self.index)):

forjinrange(0,self.trainx.nCol):

self.sv[i][j]=self.trainx.Get(self.index[i],j)

#dumpmodelpath

f=open(self.modelPath,"w")

modelStr=pickle.dumps([self.alphay,self.sv,self.b,self.w],1)

pickle.dump(modelStr,f)

f.close()

self.istrained=True

defTest(self,testx,testy):

'''Totestsamples.

self.testxistrainingmatrixandself.testyisclassifyinglabel'''

#checkparameter

if(notself.istrained):

print"Error!,nottrained!"

returnFalse

if(testx.nRow!=len(testy)):

print"Error!testx.nRowshould==len(testy)"

returnFalse

self.trainx=testx

self.trainy=testy

correct=0.0

foriinrange(0,self.trainx.nRow):

fxi=0.0

rowvector=[self.trainx.Get(i,k)forkinrange(0,self.trainx.nCol)]

ifself.kernelName=='Linear':

fxi+=self.KernelVector(self.w,rowvector)+self.b

else:

forjinrange(0,len(self.alphay)):

fxi+=self.alphay[j]*self.KernelVector(self.sv[j],rowvector)

fxi+=self.b

iffxi*self.trainy[i]>=0:

correct+=1

print'outputis',fxi,'labelis',self.trainy[i]

print'acu=',correct/len(self.trainy)

importsys,os

importmath

sys.path.append(os.path.join(os.getcwd(),'../'))

frompymining.math.matriximportMatrix

frompymining.math.text2matriximportText2Matrix

frompymining.nlp.segmenterimportSegmenter

frompymining.common.global_infoimportGlobalInfo

frompymining.common.configurationimportConfiguration

frompymining.preprocessor.chisquare_filterimportChiSquareFilter

frompymining.classifier.standard_smo_csvcimportStandardSMO

if__name__=="__main__":

config=Configuration.FromFile("conf/test.xml")

GlobalInfo.Init(config,"__global__")

txt2mat=Text2Matrix(config,"__matrix__")

[trainx,trainy]=txt2mat.CreateTrainMatrix("data/train-csvc.txt")

chiFilter=ChiSquareFilter(config,"__filter__")

chiFilter.TrainFilter(trainx,trainy)

[trainx,trainy]=chiFilter.MatrixFilter(trainx,trainy)

nbModel=StandardSMO(config,"standard_smo_csvc")

nbModel.Train(trainx,trainy)

[trainx,trainy]=txt2mat.CreatePredictMatrix("data/train-csvc-test.txt")

[trainx,trainy]=chiFilter.MatrixFilter(trainx,trainy)

nbModel=StandardSMO(config,"standard_smo_csvc",True)

nbModel.Test(trainx,trainy)

<config>

<__segmenter__>

<main_dict>dict/dict.main</main_dict>

</__segmenter__>

<__matrix__>

</__matrix__>

<__global__>

<term_to_id>mining/term_to_id</term_to_id>

<id_to_term>mining/id_to_term</id_to_term>

<id_to_doc_count>mining/id_to_doc_count</id_to_doc_count>

<class_to_doc_count>mining/class_to_doc_count</class_to_doc_count>

<id_to_idf>mining/id_to_idf</id_to_idf>

</__global__>

<__filter__>

<rate>0.9</rate>

<method>max</method>

<log_path>mining/filter.log</log_path>

<model_path>mining/filter.model</model_path>

</__filter__>

<naive_bayes>

<model_path>mining/naive_bayes.model</model_path>

<log_path>mining/naive_bayes.log</log_path>

</naive_bayes>

<twc_naive_bayes>

<model_path>mining/naive_bayes.model</model_path>

<log_path>mining/naive_bayes.log</log_path>

</twc_naive_bayes>

<standard_smo_csvc>

<model_path>mining/standard_smo.model</model_path>

<log_path>mining/standard_smo.log</log_path>

<kernel>

<name>RBF</name>

<parameters>10</parameters>

</kernel>

</standard_smo_csvc>

</config>