原文地址：http://www.jianshu.com/p/5b4a64874650

问题描述

程序实现

# coding: utf-8

import numpy as np

import matplotlib.pyplot as plt

import time

def read_data(dataFile):

    with open(dataFile, 'r') as file:

        data_list = []

        for line in file.readlines():

            line = line.strip().split()

            # add x0=1.0

            data_list.append([1.0] + [float(l) for l in line])

        num_data = len(data_list)

        data_array = np.array(data_list)

        return (num_data,data_array)

def sign(n):

    if(n>0):

        return 1

    else:

        return -1

# define PLA class

class PLA(object):

    def __init__(self,num_data,data_array,training_epochs=2000,ita=1.0,qID=15):

        self.num_data=num_data

        self.data_array=data_array

        self.training_epochs=training_epochs

        self.ita=ita

        self.qID=qID

    def train(self,w=np.zeros([5])):

        self.update_counts_list=[]

        self.last_error_id_list=[]

        for k in range(self.training_epochs):

            if self.training_epochs==1:

                id_array=np.array([m for m in range(self.num_data)])

            else:

                np.random.seed(k)

                id_array = np.random.permutation([m for m in range(self.num_data)])

            update_counts = 0

            total_counts = 0

            self.w=np.array(w)

            id = 0

            error_point_id = -1

            while (total_counts <= self.num_data):

                g = 0

                g += np.dot(self.w, self.data_array[id_array[id]][:5])

                if sign(g) == self.data_array[id_array[id]][5]:

                    total_counts += 1

                else:

                    self.w += self.ita*self.data_array[id_array[id]][5] * self.data_array[id_array[id]][:5]

                    error_point_id = id_array[id]

                    update_counts += 1

                    total_counts = 0

                id += 1

                id = id % self.num_data

            self.update_counts_list.append(update_counts)

            self.last_error_id_list.append(error_point_id)

        return

    def show_results(self):

        print("\n",self.qID,"...")

        print("training:")

        if self.training_epochs==1:

            print("the number of updates: ", self.update_counts_list[0])

            print("the final error point id: ", self.last_error_id_list[0])

            print("-----------------------")

            return

        else:

            print("the list of update counts: ",self.update_counts_list)

            print("the list of last error point id: ",self.last_error_id_list)

            print("the average number of updates:", sum(self.update_counts_list) / self.training_epochs)

            print("-----------------------")

            plt.figure()

            plt.hist(self.update_counts_list)

            plt.xlabel("the number of updates")

            plt.ylabel("frequency")

            plt.title(self.qID)

            plt.savefig("%s_train.png"%self.qID)

            return

def total_error_counts(w,data_array,num_data):

    total_error_counts=0

    for i in range(num_data):

        if sign(np.dot(w, data_array[i][:5])) != data_array[i][5]:

            total_error_counts+=1

    return total_error_counts

# define PA class

class PA(PLA):

    def __init__(self,num_data,data_array,num_test,test_array,

                 training_epochs=2000,given_updates=50,ita=1.0,pla_flag=False,qID=18):

        PLA.__init__(self,num_data,data_array,training_epochs,ita,qID)

        self.num_test=num_test

        self.test_array=test_array

        self.given_updates=given_updates

        self.pla_flag=pla_flag

    def train_and_test(self,w=np.zeros([5])):

        self.last_error_id_list=[]

        self.test_error_rate_list=[]

        for k in range(self.training_epochs):

            # train

            if self.training_epochs==1:

                id_array=np.array([m for m in range(self.num_data)])

            else:

                np.random.seed(k)

                id_array = np.random.permutation([m for m in range(self.num_data)])

            update_counts = 0

            id = 0

            self.pocket_w = np.array(w) # create a copy of w and give it to self.w

            w=np.array(w)

            error_point_id = -1

            while (update_counts <= self.given_updates):

                g = 0

                g += np.dot(w, self.data_array[id_array[id]][:5])

                if sign(g) != self.data_array[id_array[id]][5]:

                    w += self.ita*self.data_array[id_array[id]][5] * self.data_array[id_array[id]][:5]

                    if(self.pla_flag or (total_error_counts(w,self.data_array,self.num_data)<total_error_counts(self.pocket_w,self.data_array,self.num_data))):

                        self.pocket_w=np.array(w)

                    error_point_id = id_array[id]

                    update_counts += 1

                id += 1

                id = id % self.num_data

            self.last_error_id_list.append(error_point_id)

            # test

            self.test_error_rate_list.append(total_error_counts(self.pocket_w, self.test_array, self.num_test)

                                             / self.num_test)

        return

    def show_results(self):

        print("\n",self.qID,"...")

        print("training:")

        if self.training_epochs==1:

            print("the final error point id: ", self.last_error_id_list[0])

        else:

            print("the list of last error point id: ",self.last_error_id_list)

        print("testing:")

        print("the average error rate on test set: ",np.sum(self.test_error_rate_list)/self.training_epochs)

        print("-----------------------")

        plt.figure()

        plt.hist(self.test_error_rate_list)

        plt.xlabel("error rate")

        plt.ylabel("frequency")

        plt.title(self.qID)

        plt.savefig("%s_test.png"%self.qID)

        return

if __name__=="__main__":

    num_data,data_array=read_data("hw1_15_train.dat")

    # 15

    pla_15=PLA(num_data,data_array,training_epochs=1,qID=15)

    pla_15.train()

    pla_15.show_results()

    # 16

    pla_16=PLA(num_data,data_array,qID=16)

    pla_16.train()

    pla_16.show_results()

    # 17

    pla_17=PLA(num_data,data_array,ita=0.5,qID=17)

    pla_17.train()

    pla_17.show_results()

    # 16 else 1

    pla_16_1=PLA(num_data,data_array,qID=161)

    pla_16_1.train(w=np.array([1.0,0,0,0,0]))

    pla_16_1.show_results()

    # 17 else 1

    pla_17_1=PLA(num_data,data_array,ita=0.5,qID=171)

    pla_17_1.train(w=np.array([1.0,0,0,0,0]))

    pla_17_1.show_results()

    num_data,data_array=read_data("hw1_18_train.dat")

    num_test,test_array=read_data("hw1_18_test.dat")

    # 18

    pa_18=PA(num_data,data_array,num_test,test_array,qID=18)

    start = time.time()

    pa_18.train_and_test()

    end=time.time()

    print("total time for train and test of 18: ",end-start," s")

    pa_18.show_results()

    # 19

    pa_19=PA(num_data,data_array,num_test,test_array,pla_flag=True,qID=19)

    start=time.time()

    pa_19.train_and_test()

    end=time.time()

    print("total time for train and test of 19: ",end-start," s")

    pa_19.show_results()

    # 20

    pa_20=PA(num_data,data_array,num_test,test_array,given_updates=100,qID=20)

    pa_20.train_and_test()

    pa_20.show_results()

    # 18 else 1

    pa_18_1=PA(num_data,data_array,num_test,test_array,qID=181)

    pa_18_1.train_and_test(w=np.array([1.0,0,0,0,0]))

    pa_18_1.show_results()

    # 20 else 1

    pa_20_1=PA(num_data,data_array,num_test,test_array,given_updates=100,qID=201)

    pa_20_1.train_and_test(w=np.array([1.0,0,0,0,0]))

    pa_20_1.show_results()

运行结果及分析

15

16

17

对比16与17的结果：

16中步长1.0，17中步长0.5，看似步长对更新次数无影响？

16.1

17.1

对比16.1与17.1的结果：

16.1中步长1.0，17.1中步长0.5，可见步长对更新次数有影响；

再看16 vs 16.1、17 vs 17.1，前者\(W\)初始值[0,0,0,0,0]，后者\(W\)初始值[1,0,0,0,0]，可见\(W\)初始值对更新次数有影响。

18

19

对比18与19的结果：

可见PA(18)速度明显慢于PLA(19)；但在数据线性不可分的情况下，PA表现比PLA好。

20

18.1

20.1

分别对比18与20、18.1与20.1、18与18.1、20与20.1的结果：

结论与PLA处类似，

W初始值、更新步长对分类器表现有影响。