1. 程式人生 > >機器學習-CrossValidation交叉驗證Python實現

機器學習-CrossValidation交叉驗證Python實現

1.原理

1.1 概念

交叉驗證(Cross-validation)主要用於模型訓練或建模應用中,如分類預測、PCR、PLS迴歸建模等。在給定的樣本空間中,拿出大部分樣本作為訓練集來訓練模型,剩餘的小部分樣本使用剛建立的模型進行預測,並求這小部分樣本的預測誤差或者預測精度,同時記錄它們的加和平均值。這個過程迭代K次,即K折交叉。其中,把每個樣本的預測誤差平方加和,稱為PRESS(predicted Error Sum of Squares)。

1.2 目的

用交叉驗證的目的是為了得到可靠穩定的模型。在分類,建立PC 或PLS模型時,一個很重要的因素是取多少個主成分的問題。用cross validation校驗每個主成分下的PRESS值,選擇PRESS值小的主成分數。或PRESS值不再變小時的主成分數。

常用的精度測試方法主要是交叉驗證,例如10折交叉驗證(10-fold cross validation),將資料集分成十份,輪流將其中9份做訓練1份做驗證,10次的結果的均值作為對演算法精度的估計,一般還需要進行多次10折交叉驗證求均值,例如:10次10折交叉驗證,以求更精確一點。
交叉驗證有時也稱為交叉比對,如:10折交叉比對

1.3 常見的交叉驗證形式

Holdout 驗證

方法:將原始資料隨機分為兩組,一組做為訓練集,一組做為驗證集,利用訓練集訓練分類器,然後利用驗證集驗證模型,記錄最後的分類準確率為此Hold-OutMethod下分類器的效能指標.。Hold-OutMethod相對於K-fold Cross Validation 又稱Double cross-validation ,或相對K-CV稱 2-fold cross-validation(2-CV)

一般來說,Holdout 驗證並非一種交叉驗證,因為資料並沒有交叉使用。 隨機從最初的樣本中選出部分,形成交叉驗證資料,而剩餘的就當做訓練資料。 一般來說,少於原本樣本三分之一的資料被選做驗證資料。

  • 優點:好處的處理簡單,只需隨機把原始資料分為兩組即可
  • 缺點:嚴格意義來說Hold-Out Method並不能算是CV,因為這種方法沒有達到交叉的思想,由於是隨機的將原始資料分組,所以最後驗證集分類準確率的高低與原始資料的分組有很大的關係,所以這種方法得到的結果其實並不具有說服性.(主要原因是 訓練集樣本數太少,通常不足以代表母體樣本的分佈,導致 test 階段辨識率容易出現明顯落差。此外,2-CV 中一分為二的分子集方法的變異度大,往往無法達到「實驗過程必須可以被複制」的要求。)

K-fold cross-validation

K折交叉驗證,初始取樣分割成K個子樣本,一個單獨的子樣本被保留作為驗證模型的資料,其他K-1個樣本用來訓練。交叉驗證重複K次,每個子樣本驗證一次,平均K次的結果或者使用其它結合方式,最終得到一個單一估測。這個方法的優勢在於,同時重複運用隨機產生的子樣本進行訓練和驗證,每次的結果驗證一次,10折交叉驗證是最常用的。

  • 優點:K-CV可以有效的避免過學習以及欠學習狀態的發生,最後得到的結果也比較具有說服性.
  • 缺點:K值選取上

留一驗證

正如名稱所建議, 留一驗證(LOOCV)意指只使用原本樣本中的一項來當做驗證資料, 而剩餘的則留下來當做訓練資料。 這個步驟一直持續到每個樣本都被當做一次驗證資料。 事實上,這等同於 K-fold 交叉驗證是一樣的,其中K為原本樣本個數。 在某些情況下是存在有效率的演演算法,如使用kernel regression 和Tikhonov regularization。

2.深入

使用交叉驗證方法的目的主要有3個:

  • (1)從有限的學習資料中獲取儘可能多的有效資訊;
  • (2)交叉驗證從多個方向開始學習樣本的,可以有效的避免陷入區域性最小值;
  • (3)可以在一定程度上避免過擬合問題。

採用交叉驗證方法時需要將學習資料樣本分為兩部分:訓練資料樣本和驗證資料樣本。並且為了得到更好的學習效果,無論訓練樣本還是驗證樣本都要儘可能參與學習。一般選取10重交叉驗證即可達到好的學習效果。下面在上述原則基礎上設計演算法,主要描述下演算法步驟,如下所示。

Algorithm

Step1:  將學習樣本空間 C 分為大小相等的 K 份  
Step2:  for i = 1 to K :
            取第i份作為測試集
            for j = 1 to K:
                if i != j:
                    將第j份加到訓練集中,作為訓練集的一部分
                end if
            end for
        end for
Step3:  for i in (K-1訓練集):
            訓練第i個訓練集,得到一個分類模型
            使用該模型在第N個數據集上測試,計算並儲存模型評估指標
        end for
Step4:  計算模型的平均效能
Step5:  用這K個模型在最終驗證集的分類準確率平均值作為此K-CV下分類器的效能指標.

3.實現

3.1 scikit-learn交叉驗證

使用方法:

首先載入資料集

>>> import numpy as np
>>> from sklearn import cross_validation
>>> from sklearn import datasets
>>> from sklearn import svm
>>> iris = datasets.load_iris()
>>> iris.data.shape, iris.target.shape
((150, 4), (150,))

通過上面程式碼,資料集特徵和類標籤分別為iris.data, iris.target,接著進行交叉驗證

>>> X_train, X_test, y_train, y_test = cross_validation.train_test_split(
...     iris.data, iris.target, test_size=0.4, random_state=0)
>>> X_train.shape, y_train.shape
((90, 4), (90,))
>>> X_test.shape, y_test.shape
((60, 4), (60,))
>>> clf = svm.SVC(kernel='linear', C=1).fit(X_train, y_train)
>>> clf.score(X_test, y_test)                           
0.96...

上面的clf是分類器,可以自己替換,比如我可以使用RandomForest

clf = RandomForestClassifier(n_estimators=400)

一個比較有用的函式是train_test_split。功能是從樣本中隨機的按比例選取train data和test data。形式為

X_train, X_test, y_train, y_test = cross_validation.train_test_split(train_data,train_target, test_size=0.4, random_state=0)

test_size是樣本佔比。如果是整數的話就是樣本的數量。random_state是隨機數的種子。

3.2 抽樣與CV結合

由於我跑的實驗,資料是非均衡資料,不能直接套用,所以這裡自己寫了一個交叉驗證的程式碼,僅供參考,如有問題,歡迎交流。

首先有一個自適應的資料載入函式,主要用於載入本地文字資料,同時文字每行資料以”\t”隔開,最後一列為類標號,資料樣例如下:

A1001   708 K   -4  -3  6   2   -13 0   2   -4  -4  -10 -9  1
A1002   709 L   -4  -4  -1  -2  -11 -1  0   -12 -7  -5  -1  -1
A1003   710 G   0   -6  -2  -6  -8  -4  -6  -6  -9  -4  0   -1
A1004   711 R   0   0   1   -3  -10 -1  -3  -4  -6  -9  -6  1

說明:前面三個不是特徵,所以在載入資料集的時候,特徵部分起始位置修改了下,loadDataSet函式如下:

def loadDataSet(fileName):
    fr = open(fileName)
    dataMat = []; labelMat = []
    for eachline in fr:
        lineArr = []
        curLine = eachline.strip().split('\t') #remove '\n'
        for i in range(3, len(curLine)-1):
            lineArr.append(float(curLine[i])) #get all feature from inpurfile
        dataMat.append(lineArr)
        labelMat.append(int(curLine[-1])) #last one is class lable
    fr.close()
    return dataMat,labelMat

返回的dataMat為純特徵矩陣,labelMat為類別標號。

下面的splitDataSet用來切分資料集,如果是十折交叉,則split_size取10,filename為整個資料集檔案,outdir則是切分的資料集的存放路徑。

def splitDataSet(fileName, split_size,outdir):
    if not os.path.exists(outdir): #if not outdir,makrdir
        os.makedirs(outdir)
    fr = open(fileName,'r')#open fileName to read
    num_line = 0
    onefile = fr.readlines()
    num_line = len(onefile)        
    arr = np.arange(num_line) #get a seq and set len=numLine
    np.random.shuffle(arr) #generate a random seq from arr
    list_all = arr.tolist()
    each_size = (num_line+1) / split_size #size of each split sets
    split_all = []; each_split = []
    count_num = 0; count_split = 0  #count_num 統計每次遍歷的當前個數
                                    #count_split 統計切分次數
    for i in range(len(list_all)): #遍歷整個數字序列
        each_split.append(onefile[int(list_all[i])].strip()) 
        count_num += 1
        if count_num == each_size:
            count_split += 1 
            array_ = np.array(each_split)
            np.savetxt(outdir + "/split_" + str(count_split) + '.txt',\
                        array_,fmt="%s", delimiter='\t')  #輸出每一份資料
            split_all.append(each_split) #將每一份資料加入到一個list中
            each_split = []
            count_num = 0
    return split_all

underSample(datafile)方法為抽樣函式,強正負樣本比例固定為1:1,返回的是一個正負樣本比例均等的資料集合。

def underSample(datafile): #只針對一個數據集的下采樣
    dataMat,labelMat = loadDataSet(datafile) #載入資料
    pos_num = 0; pos_indexs = []; neg_indexs = []   
    for i in range(len(labelMat)):#統計正負樣本的下標    
        if labelMat[i] == 1:
            pos_num +=1
            pos_indexs.append(i)
            continue
        neg_indexs.append(i)
    np.random.shuffle(neg_indexs)
    neg_indexs = neg_indexs[0:pos_num]
    fr = open(datafile, 'r')
    onefile = fr.readlines()
    outfile = []
    for i in range(pos_num):
        pos_line = onefile[pos_indexs[i]]    
        outfile.append(pos_line)
        neg_line= onefile[neg_indexs[i]]      
        outfile.append(neg_line)
    return outfile #輸出單個數據集取樣結果

下面的generateDataset(datadir,outdir)方法是從切分的資料集中留出一份作為測試集(無需抽樣),對其餘的進行抽樣然後合併為一個作為訓練集,程式碼如下:

def generateDataset(datadir,outdir): #從切分的資料集中,對其中九份抽樣匯成一個,\
    #剩餘一個做為測試集,將最後的結果按照訓練集和測試集輸出到outdir中
    if not os.path.exists(outdir): #if not outdir,makrdir
        os.makedirs(outdir)
    listfile = os.listdir(datadir)
    train_all = []; test_all = [];cross_now = 0
    for eachfile1 in listfile:
        train_sets = []; test_sets = []; 
        cross_now += 1 #記錄當前的交叉次數
        for eachfile2 in listfile:
            if eachfile2 != eachfile1:#對其餘九份欠抽樣構成訓練集
                one_sample = underSample(datadir + '/' + eachfile2)
                for i in range(len(one_sample)):
                    train_sets.append(one_sample[i])
        #將訓練集和測試集檔案單獨儲存起來
        with open(outdir +"/test_"+str(cross_now)+".datasets",'w') as fw_test:
            with open(datadir + '/' + eachfile1, 'r') as fr_testsets:
                for each_testline in fr_testsets:                
                    test_sets.append(each_testline) 
            for oneline_test in test_sets:
                fw_test.write(oneline_test) #輸出測試集
            test_all.append(test_sets)#儲存訓練集
        with open(outdir+"/train_"+str(cross_now)+".datasets",'w') as fw_train:
            for oneline_train in train_sets:   
                oneline_train = oneline_train
                fw_train.write(oneline_train)#輸出訓練集
            train_all.append(train_sets)#儲存訓練集
    return train_all,test_all

因為需要評估交叉驗證,所以我寫了一個performance方法根據真實類標籤紙和預測值來計算SN和SP,當然如果需要其他的評估標準,繼續新增即可。

def performance(labelArr, predictArr):#類標籤為int型別
    #labelArr[i] is actual value,predictArr[i] is predict value
    TP = 0.; TN = 0.; FP = 0.; FN = 0.   
    for i in range(len(labelArr)):
        if labelArr[i] == 1 and predictArr[i] == 1:
            TP += 1.
        if labelArr[i] == 1 and predictArr[i] == -1:
            FN += 1.
        if labelArr[i] == -1 and predictArr[i] == 1:
            FP += 1.
        if labelArr[i] == -1 and predictArr[i] == -1:
            TN += 1.
    SN = TP/(TP + FN) #Sensitivity = TP/P  and P = TP + FN 
    SP = TN/(FP + TN) #Specificity = TN/N  and N = TN + FP
    #MCC = (TP*TN-FP*FN)/math.sqrt((TP+FP)*(TP+FN)*(TN+FP)*(TN+FN))
    return SN,SP

classifier(clf,train_X, train_y, test_X, test_y)方法是交叉驗證中每次用的分類器訓練過程以及測試過程,裡面使用的分類器是scikit-learn自帶的。該方法會將一些訓練結果寫入到檔案中並儲存到本地,同時在最後會返回ACC,SP,SN。

def classifier(clf,train_X, train_y, test_X, test_y):#X:訓練特徵,y:訓練標號
    # train with randomForest 
    print " training begin..."
    clf = clf.fit(train_X,train_y)
    print " training end."
    #==========================================================================
    # test randomForestClassifier with testsets
    print " test begin."
    predict_ = clf.predict(test_X) #return type is float64
    proba = clf.predict_proba(test_X) #return type is float64
    score_ = clf.score(test_X,test_y)
    print " test end."
    #==========================================================================
    # Modeal Evaluation
    ACC = accuracy_score(test_y, predict_)
    SN,SP = performance(test_y, predict_)
    MCC = matthews_corrcoef(test_y, predict_)
    #AUC = roc_auc_score(test_labelMat, proba)
    #==========================================================================
    #save output 
    eval_output = []
    eval_output.append(ACC);eval_output.append(SN)  #eval_output.append(AUC)
    eval_output.append(SP);eval_output.append(MCC)
    eval_output.append(score_)
    eval_output = np.array(eval_output,dtype=float)
    np.savetxt("proba.data",proba,fmt="%f",delimiter="\t")
    np.savetxt("test_y.data",test_y,fmt="%f",delimiter="\t")
    np.savetxt("predict.data",predict_,fmt="%f",delimiter="\t") 
    np.savetxt("eval_output.data",eval_output,fmt="%f",delimiter="\t")
    print "Wrote results to output.data...EOF..."
    return ACC,SN,SP

下面的mean_fun用於求列表list中數值的平均值,主要是求ACC_mean,SP_mean,SN_mean,用來評估模型好壞。

def mean_fun(onelist):
    count = 0
    for i in onelist:
        count += i
    return float(count/len(onelist))

交叉驗證程式碼

def crossValidation(clf, clfname, curdir,train_all, test_all):
    os.chdir(curdir)
    #構造出純資料型樣本集
    cur_path = curdir
    ACCs = [];SNs = []; SPs =[]
    for i in range(len(train_all)):
        os.chdir(cur_path)
        train_data = train_all[i];train_X = [];train_y = []
        test_data = test_all[i];test_X = [];test_y = []
        for eachline_train in train_data:
            one_train = eachline_train.split('\t') 
            one_train_format = []
            for index in range(3,len(one_train)-1):
                one_train_format.append(float(one_train[index]))
            train_X.append(one_train_format)
            train_y.append(int(one_train[-1].strip()))
        for eachline_test in test_data:
            one_test = eachline_test.split('\t')
            one_test_format = []
            for index in range(3,len(one_test)-1):
                one_test_format.append(float(one_test[index]))
            test_X.append(one_test_format)
            test_y.append(int(one_test[-1].strip()))
        #======================================================================
        #classifier start here
        if not os.path.exists(clfname):#使用的分類器
            os.mkdir(clfname)
        out_path = clfname + "/" + clfname + "_00" + str(i)#計算結果資料夾
        if not os.path.exists(out_path):
            os.mkdir(out_path)
        os.chdir(out_path)
        ACC, SN, SP = classifier(clf, train_X, train_y, test_X, test_y)
        ACCs.append(ACC);SNs.append(SN);SPs.append(SP)
        #======================================================================
    ACC_mean = mean_fun(ACCs)
    SN_mean = mean_fun(SNs)
    SP_mean = mean_fun(SPs)
    #==========================================================================
    #output experiment result
    os.chdir("../")
    os.system("echo `date` '" + str(clf) + "' >> log.out")
    os.system("echo ACC_mean=" + str(ACC_mean) + " >> log.out")
    os.system("echo SN_mean=" + str(SN_mean) + " >> log.out")
    os.system("echo SP_mean=" + str(SP_mean) + " >> log.out")
    return ACC_mean, SN_mean, SP_mean

測試:

if __name__ == '__main__':
    os.chdir("your workhome") #你的資料存放目錄
    datadir = "split10_1" #切分後的檔案輸出目錄
    splitDataSet('datasets',10,datadir)#將資料集datasets切為十個儲存到datadir目錄中
    #==========================================================================
    outdir = "sample_data1" #抽樣的資料集存放目錄
    train_all,test_all = generateDataset(datadir,outdir) #抽樣後返回訓練集和測試集
    print "generateDataset end and cross validation start"
    #==========================================================================
    #分類器部分
    from sklearn.ensemble import RandomForestClassifier
    clf = RandomForestClassifier(n_estimators=500) #使用隨機森林分類器來訓練
    clfname = "RF_1"    #==========================================================================
    curdir = "experimentdir" #工作目錄
    #clf:分類器,clfname:分類器名稱,curdir:當前路徑,train_all:訓練集,test_all:測試集
    ACC_mean, SN_mean, SP_mean = crossValidation(clf, clfname, curdir,train_all,test_all)
    print ACC_mean,SN_mean,SP_mean  #將ACC均值,SP均值,SN均值都輸出到控制檯

上面的程式碼主要用於抽樣後的十倍交叉驗證,該怎麼設定引數,還得具體分析。

總之,交叉驗證在一定程度上能夠避免陷入區域性最小值。一般實際操作中使用的是十折交叉驗證,單具體情況還得具體分析,並沒有一個統一的標準固定十倍交叉的引數或者是演算法的選擇以及演算法引數的選擇。不同的資料使用不同的演算法往往會的得到不同的最優分類器。So,just try it!Happy coding!

本欄目Machine Learning持續更新中,歡迎關注CSDN部落格:dream_angel_z