问题描述

图1 15-17
图2 18-20

程序实现

# 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

图3 15结果

16

图4 16结果1
图5 16结果2

17

图6 17结果1
图7 17结果2

对比16与17的结果：

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

16.1

图8 16.1结果1
图9 16.1结果2

17.1

图10 17.1结果1
图11 17.1结果2

对比16.1与17.1的结果：

16.1中步长1.0，17.1中步长0.5，可见步长对更新次数有影响；
再看16 vs 16.1、17 vs 17.1，前者初始值[0,0,0,0,0]，后者初始值[1,0,0,0,0]，可见初始值对更新次数有影响。

18

图12 18结果1
图13 18结果2

19

图14 19结果1
图15 19结果2

对比18与19的结果：

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

20

图16 20结果1
图17 20结果2

18.1

图18 18.1结果1
图19 18.1结果2

20.1

图20 20.1结果1
图21 20.1结果2

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

结论与PLA处类似，
W初始值、更新步长对分类器表现有影响。