今回は「Python 機械学習プログラミング：第2章」のコードを実装しました。

第2章：分類問題 単純な機械学習アルゴリズムの訓練

この章では初期の機械学習アルゴリズムのうち、パーセプトロンとADALINEの説明とPythonで再現するコードが記載されてます。

パーセプトロンとADALINEについて詳しく知りたい方は下記の記事を参照して下さい。

それではコードを記載します。

パーセプトロンの実装コード

import numpy as np

class Perceptron(object) :
    """ パーセプトロンの分類機
    
    パラメータ
    ----------------
    eta : float
            学習率（0.0より大きく1.0以下の値）
    n_iter : int
            訓練データの訓練回数
    random_state : int
            重みを初期化するための乱数シード
    
    属性
    ----------------
    w_ : 1次元配列
            適合後の重み
    errors_ : リスト
            各エポックでの誤分類（更新）の数
    """
    def __init__(self, eta=0.01, n_iter=50, random_state=1) :
        self.eta = eta
        self.n_iter = n_iter
        self.random_state = random_state
        
    def fit(self, X, y) :
        """訓練データに適合させる
        
        パラメータ
        ----------------
        X : （配列のようなデータ構造）, shape = [n_examples, n_features]
              訓練データ
              n_examples
        y : 配列のようなデータ構造, shape = [n_examples]
             目的関数
        
        戻り値
        ----------------
        self : object
        
        """
        rgen = np.random.RandomState(self.random_state)
        self.w_ = rgen.normal(loc=0.0, scale=0.01, size=1 + X.shape[1])
        self.errors_ = []
        
        for _ in range(self.n_iter) : # 訓練回数分までデータを反復
            errors = 0
            for xi, target in zip(X, y) : # 各訓練データで重みを更新
                # 重み w1 ... ,wmの更新
                update = self.eta * (target - self.predict(xi))
                self.w_[1:] += update * xi
                self.w_[0] += update
                # 重みの更新が0でない場合は誤分類としてカウント
                errors += int(update != 0.0)
            # 反復ごとの誤差を格納
            self.errors_.append(errors)
        return self
    
    def net_input(self, X) :
        """総入力を計算"""
        return np.dot(X, self.w_[1:]) + self.w_[0]
    def predict(self, X) :
        """1ステップ後のクラスラベルを返す"""
        return np.where(self.net_input(X) >= 0.0, 1, -1)

IrisデータセットをDataFrameオブジェクトに直接読み込む

import os
import pandas as pd
s = os.path.join('https://archive.ics.uci.edu', 'ml', 'machine-learning-databases', 'iris', 'iris.data')

print('URL', s)
df = pd.read_csv(s, header=None, encoding='utf-8')
df.tail()

Iris-setisaの50枚の花とIris-versicolorの50枚の花に対応する先頭の100このクラスラベルを抽出する。これらのクラスラベルを2つの整数のクラスラベル1と-1に変換する。このクラスラベルはベクトルyに代入されており、pandasライブラリのDateFrameオブジェクトのvaluesメソッドによっる戻り値のNumPy表現となっている。同じ容量で100この訓練データの特徴量の列から1列目（がく片の長さ）と3列目（花びらの長さ）を抽出し、それらのデータを特徴量行列Xに代入する。2次元散布図を使ってこれらのデータを可視化してみよう。

import matplotlib.pyplot as plt
import numpy as np

# 1-100行目の目的関数を抽出
y = df.iloc[0:100, 4].values

# Iris-setosaを-1、Iris-versicolorを1に変換
y = np.where(y == 'Iris-setosa', -1, 1)

# 1-100行目の1,3列目の抽出
X = df.iloc[0:100, [0,2]].values

# 品種setosaのプロット（赤の○）
plt.scatter(X[:50, 0], X[:50,1], color='red', marker='o', label='setosa')

# 品種versicolorのプロット（赤の×）
plt.scatter(X[51:100, 0], X[51:100, 1], color='blue', marker='x', label='versicolor')

# 軸ラベルの設定
plt.xlabel('sepal length [cm]')
plt.ylabel('petal length [cm]')

# 凡例の設定（左下に配置）
plt.legend(loc='upper left')

# 図の表示
plt.show()

Irisデータに含まれてる品種の訓練データの分布状況と、2つの特徴量軸（cm単位の花びらの長さ、がく片の長さ）を表している。次にアルゴリズムを収束し、Irisデータセットの2つの品種を分ける決定境界が検出されたかチェックするために、各エポックの誤分類の個数もプロットする。横軸をエポック数、縦軸を誤分類の個数とする。

# パーせプロトンのオブジェクト生成（インスタンス化）
ppn = Perceptron(eta=0.01, n_iter=10)

# 訓練データへのモデルの適合
ppn.fit(X, y)

# エポックと誤分類の関係を表す折れ線グラフをプロット
plt.plot(range(1, len(ppn.errors_) + 1), ppn.errors_, marker='o')

# 軸ラベルの設定
plt.xlabel('Epochs')
plt.ylabel('Number of update')

# 図の表示
plt.show()

6回目のエポックの後、パーセプトロンは収束して訓練データを完璧に分類できるようになってる。2次元のデータセットの決定境界を可視化するために簡単で便利な関数を実装。

from matplotlib.colors import ListedColormap

def plot_decision_regions(X, y, classifier, resolution=0.02) :
    
    # マーカーとカラーマップの準備
    markers = ('s', 'x', 'o', '^', 'v')
    colors = ('red' ,'blue', 'lightgreen', 'gray', 'cyan')
    cmap = ListedColormap(colors[:len(np.unique(y))])
    
    # 決定領域のプロット
    x1_min, x1_max = X[:, 0].min() - 1, X[:, 0].max() +1
    x2_min, x2_max = X[:, 1].min() - 1, X[:, 1].max() +1
    # グリッドポイントの生成
    xx1, xx2 = np.meshgrid(np.arange(x1_min, x1_max, resolution),
                                              np.arange(x2_min, x2_max, resolution))
    # 各特徴量を1次元配列に変換して予測を実行
    Z = classifier.predict(np.array([xx1.ravel(), xx2.ravel()]).T)
    # 予測結果を元のグリッドポイントのデータサイズに変換
    Z = Z.reshape(xx1.shape)
    # グリッドポイントの等高線のプロット
    plt.contourf(xx1, xx2, Z, alpha=0.3, cmap=cmap)
    # 軸の範囲の設定
    plt.xlim(xx1.min(), xx1.max())
    plt.ylim(xx2.min(), xx2.max())
    
    # クラスごとのに訓練データをプロット
    for idx, cl in enumerate(np.unique(y)):
        plt.scatter(x=X[y == cl, 0],
                            y=X[y == cl, 1],
                            alpha=0.8,
                            c=colors[idx],
                            marker=markers[idx],
                            label=cl,
                            edgecolors='black')

予測されたクラスごとに決定領域を逸れざれ異なる色にマッピングする。

# 決定領域のプロット
plot_decision_regions(X, y, classifier=ppn)

# 軸のラベルの設定
plt.xlabel('sepal length [cm]')
plt.ylabel('petal length [cm]')

# 凡例の設定（左下に配置）
plt.legend(loc='upper left')

# 図の表示
plt.show()

ADALINEをPythonで実装する

class AdalineGD(object):
    """ADAptive LInear NEuron 分類機
    
    パラメータ
    -----------------
    eta : float
        学習率(0.0より大きく1.0以下の値)
    n_iter : int
        訓練データの訓練回数
    random_state : int
        重みを初期化するための乱数シード
        
    属性
    -----------------
    w_：1次元配列
        適合後の重み
    cost_：リスト
        各エポックでの残差平方和残すと関数
    """
    
    def __init__(self, eta=0.01, n_iter=50, random_state=1):
        self.eta = eta
        self.n_iter = n_iter
        self.random_state = random_state
    
    def fit(self, X, y):
        """訓練データに適合させる
        
        パラメータ
        -----------------
        X：（配列のようなデータ構造）, shape = [n_examples, n_features]
            訓練データ
            n_examplesは訓練データの個数、n_featuresは特徴量の個数
        y：（配列のようなデータ構造）, shape = [n_examples]
            目的変数
            
        戻り値
        -----------------
        self：object
        """
        rgen = np.random.RandomState(self.random_state)
        self.w_ = rgen.normal(loc=0.0, scale=0.01, size=1 + X.shape[1])
        self.cost_ = []
        
        for i in range(self.n_iter): # 訓練回数分まで訓練データを反復
            net_input = self.net_input(X)
            # activationメソッドは単なる恒等関数であるため
            # このコードでは何の効果もないことに注意。代わりに
            # 直接　'output = self.net_input(X)'と記述することもできた。
            # activationメソッドの目的はより概念的な者である。
            # つまり、(もちほど記述する)ロジスティック回帰の場合は
            # ロジスティック回帰の分類機の実装するためにシグモイド関数に変更することもできる。
            output = self.activation(net_input)
            # 誤差の計算
            errors = (y - output)
            # w1...wmの計算
            self.w_[1:] += self.eta * X.T.dot(errors)
            # w0の更新
            self.w_[0] += self.eta * errors.sum()
            # コスト関数の計算
            cost = (errors**2).sum() / 2.0
            # コストの格納
            self.cost_.append(cost)
        return self
    
    def net_input(self, X):
        """総入力を計算"""
        return np.dot(X, self.w_[1:]) + self.w_[0]
    
    def activation(self, X):
        """線型活性化関数の出力を計算"""
        return X
    
    def predict(self, X):
        """1ステップ後のクラスラベルを返す"""
        return np.where(self.activation(self.net_input(X)) >= 0.0, 1, -1)

2つの学習率を用いて、エポック数に対するコストをプロット。

# 描画領域を1行2列に分割
fig, ax = plt.subplots(nrows=1, ncols=2, figsize=(10, 4))
# 勾配降下法によるADALINEの学習(学習率 eta=0.01)
adal = AdalineGD(n_iter=10, eta=0.01).fit(X, y)
# エポック数とコストの関係を表す折れ線グラフのプロット（縦軸のコストは常用対数）
ax[0].plot(range(1, len(adal.cost_)+1), np.log10(adal.cost_), marker='o')
# 軸ラベルの設定
ax[0].set_xlabel('Epochs')
ax[0].set_ylabel('log(Sum-squared-errors)')
# タイトルの設定
ax[0].set_title('Adline - Learning rate 0.01')
# 勾配降下法によるADALINEの学習（学習率　eta=0.0001）
ada2 = AdalineGD(n_iter=10, eta=0.0001).fit(X, y)
# エポック数とコストの関係を表す折れ線グラフ
ax[1].plot(range(1, len(ada2.cost_)+1), ada2.cost_, marker='o')
# 軸ラベルの設定
ax[1].set_xlabel('Epochs')
ax[1].set_ylabel('Sum-squared-errors')
# タイトルの設定
ax[1].set_title('Adaline - Learning - rate 0.0001')
# 図の表示
plt.show()

標準化を実現

# データのコピー
X_std = np.copy(X)
# 各列の標準化
X_std[:, 0] = (X[:, 0] - X[:, 0].mean()) / X[:, 0].std()
X_std[:, 1] = (X[:, 1] - X[:, 1].mean()) / X[:, 1].std()

# 勾配降下法によるADALINEの学習（標準化後、学習率 eta=0.01）
ada_gd = AdalineGD(n_iter=15, eta=0.01)
# モデルの適合
ada_gd.fit(X_std, y)
# 境界領域のプロット
plot_decision_regions(X_std, y, classifier=ada_gd)
# タイトルの設定
plt.title('Adaline - Gradiend Descent')
# 軸ラベルの設定
plt.xlabel('sepal length [standardized]')
plt.ylabel('petal length [standardized]')
# 凡例の設定（左上に配置）
plt.legend(loc='upper left')
# 図の表示
plt.tight_layout()
plt.show()
# エポック数とコストの関係を表す折れ線グラフのプロット
plt.plot(range(1, len(ada_gd.cost_) + 1), ada_gd.cost_, marker='o')
# 軸ラベルの設定
plt.xlabel('Epochs')
plt.ylabel('Sum-squred-error')
# 図の表示
plt.tight_layout()
plt.show()

確率的勾配降下方に基づいて重みを更新する学習アルゴリズム。

from numpy.random import seed

class AdalineSGD(object):
    """ADAptive LInesr Neuron 分類器
    
    パラメータ
    -------------
    eta：float
        学習率（0.0より大きく1.0以下の値）
    n_iter : int
        訓練データの訓練回数
    shuffle : bool(デフォルト：True)
        Trueの場合は、循環を回避するためにエポックごとに訓練データをシャッフル
    random_state : int
        重みを初期化するための乱数シード
        
    属性
    -------------
    w_ : 1次元配列
        適合後の重み
    cost_ : リスト
        各エポックで全ての訓練データの平均を求める誤差平方和コスト関数
    
    """
    
    def __init__(self, eta=0.01, n_iter=10, shuffle=True, random_state=None):
        # 学習率の初期化
        self.eta = eta
        # 訓練回数の初期化
        self.n_iter = n_iter
        # 重みの初期化フラグはFalseに設定
        self.w_initialized = False
        # 各エポックで訓練データをシャッフルするかどうかのフラグを初期化
        self.shuffle = shuffle
        # 乱数シードを設定
        self.random_state = random_state
    
    def fit(self, X, y):
        """訓練データに適合させる
        
        パラメータ
        --------------
        X : (配列のようなデータ構造)　、shape = [n_examples, n_features]
            n_examplesは訓練データの個数、n_featureは特徴量の個数
        y : 配列のようなデータ構造、shape=[n_examples]
            目的変数
        
        戻り値
        --------------
        self : object
        
        """
        # 重みベクトルの生成
        self._initialize_weights(X.shape[1])
        # コストを格納するリストの生成
        self.cost_ = []
        # 訓練回数分まで訓練データを反復
        for i in range(self.n_iter):
            # 指定された場合は訓練データをシャッフル
            if self.shuffle:
                X, y = self._shuffle(X, y)
                # 各訓練データのコストを格納するリストの生成
                cost = []
                # 各訓練データに対する計算
                for xi, target in zip(X, y):
                    # 特徴量xiと目的変数yを用いた重みの更新とコストの計算
                    cost.append(self._update_weights(xi, target))
                # 訓練データの平均コストの計算
                avg_cost = sum(cost) / len(y)
                # 平均コストを格納
                self.cost_.append(avg_cost)
        return self
    
    def partial_fit(self, X, y):
        """重みを初期化することなく訓練データに適合させる"""
        # 初期化されてない場合は初期化を実行
        if not self.w_initialized:
            self._initialize_weights(X.shape[1])
        # 目的変数yの要素数が2以上の場合は各訓練データの特徴量xiと目的変数targetで重みを更新
        if y.ravel().shape[0] > 1:
            for xi, target in zip(X, y):
                self._update_weights(xi, target)
        # 目的変数yの要素数が1の場合は訓練データ全体の特徴量Xと目的変数yで重みを更新
        else:
            self._update_weights(X, y)
        return self
    
    def _shuffle(self, X, y):
        """訓練データをシャッフル"""
        r = self.rgen.permutation(len(y))
        return X[r], y[r]
    
    def _initialize_weights(self, m):
        """重みを小さな乱数に初期化"""
        self.rgen= np.random.RandomState(self.random_state)
        self.w_ = self.rgen.normal(loc=0.0, scale=0.01, size=1 + m)
        self.w_initialized = True
        
    def _update_weights(self, xi, target):
        """ADALINEの学習規則を用いて重みを更新"""
        # 活性化関数の出力の計算
        output = self.activation(self.net_input(xi))
        # 誤差の計算
        error = (target - output)
        # 重みの更新
        self.w_[1:] += self.eta * xi.dot(error)
        # 重みの更新
        self.w_[0] += self.eta * error
        # コストの計算
        cost = 0.5 * error ** 2
        return cost
    
    def net_input(self, X):
        """総入力を計算"""
        return np.dot(X, self.w_[1:]) + self.w_[0]
    
    def activation(self, X):
        return X
    
    def predict(self, X):
        """1ステップ後のクラスラベルを返す"""
        return np.where(self.activation(self.net_input(X)) >= 0.0, 1, -1)


# 確率的勾配降下法によるADALINEの学習
ada_sgd = AdalineSGD(n_iter=15, eta=0.01, random_state=1)
# モデルへの適合
ada_sgd.fit(X_std, y)
# 境界領域のプロット
plot_decision_regions(X_std, y, classifier=ada_sgd)
# タイトルの設定
plt.title('Adaline - Stochastic Gradient Descent')
# 軸ラベルの設定
plt.xlabel('sepal length [standardized]')
plt.ylabel('patel length [standardized]')
# plotの表示
plt.show()
# エポックとコストの折れ線グラフのプロット
plt.plot(range(1, len(ada_sgd.cost_) + 1), ada_sgd.cost_, marker='o')
# 軸ラベルの設定
plt.xlabel('Epochs')
plt.ylabel('Average Cost')
# プロットの表示
plt.tight_layout(),
plt.show()

コードをコピペしたような記事ですが、パーセプトロンとADALINEのアルゴリズムをちゃんと理解できるようにしないとな〜と思ってます。