PyTorch Logistic Regression ~ MLP model

README.md

PyTorch Logistic Regression ~ MLP model

1. logistic_regression.py - Using torch.nn module, analysing sklearn DIGITS dataset
2. logistic_regression_low.py - NOT using torch.nn module, analysing sklearn DIGITS dataset
3. mnist_mlp.py

logistic_regression.py

# -*- coding: utf-8 -*-
#
#   logistic_regression.py
#       date. 8/22/2017
#

import numpy as np
import torch
from torch.autograd import Variable
import torch.nn as nn
import torch.nn.functional as F

from sklearn.datasets import load_digits
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, confusion_matrix

# CPU演算とGPU演算を切り換えるスイッチ．GPU演算では，CPU-GPU間のメモリ・コピーが行われる．
if torch.cuda.is_available():
    GPU_sw = True
else:
    GPU_sw = False
print('GPU_sw = ', GPU_sw)


def load_data():
    digits = load_digits()
    y = digits.target
    n_samples = len(digits.images)
    X = digits.images.reshape((n_samples, -1))

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=0)

    return X_train, X_test, y_train, y_test


class Net(nn.Module):
    # torch.nn modeling
    def __init__(self, n_feature, n_class):
        super(Net, self).__init__()
        self.fc = nn.Linear(n_feature, n_class)

    def forward(self, x):
        y = self.fc(x)

        return y


def train_feed(X_train, y_train, idx, batch_size):
    # Training phase におけるデータ供給
    dtype = torch.FloatTensor
    n_samples = X_train.shape[0]
    if idx == 0:
        perm = np.random.permutation(n_samples)
        X_train = X_train[perm]
        y_train = y_train[perm]
    start = idx
    end = start + batch_size
    if end > n_samples:
        perm = np.random.permutation(n_samples)
        X_train = X_train[perm]
        y_train = y_train[perm]
        start = 0
        end = start + batch_size

    batch_x = np.asarray(X_train[start:end], dtype=np.float32)
    batch_x = batch_x / 16.0    # image のスケーリング
    batch_y = np.asarray(y_train[start:end], dtype=np.int64)
    idx = end

    if GPU_sw:
        var_x = Variable(torch.from_numpy(batch_x).cuda())
        var_y = Variable(torch.from_numpy(batch_y).cuda())
    else:
        var_x = Variable(torch.from_numpy(batch_x))
        var_y = Variable(torch.from_numpy(batch_y))

    return var_x, var_y, idx


if __name__ == '__main__':
    # Load Data
    X_train, X_test, y_train, y_test = load_data()
    n_feature = 64
    n_class = 10
    batch_size = 100

    # define network
    if GPU_sw:
        net = Net(n_feature, n_class).cuda()
    else:
        net = Net(n_feature, n_class)

    loss_fn = torch.nn.CrossEntropyLoss()    # 損失関数の定義
    optimizer = torch.optim.SGD(net.parameters(), 
                                lr=0.003, momentum=0.9)  # オプティマイザ

    print('Training...')
    train_index = 0
    for t in range(10000):
        batch_x, batch_y, train_index = train_feed(X_train, y_train, 
                                                   train_index, batch_size)
        y_pred = net.forward(batch_x)
        loss = loss_fn(y_pred, batch_y)

        if t % 1000 == 0:
            print('{:>5d}: loss = {:>10.3f}'.format(t, loss.data[0]))
        
        # zero the gradient buffers, 勾配gradを初期化（ゼロ化）する．
        optimizer.zero_grad()

        # Backward pass: 誤差の逆伝搬を行って，パラメータの変化量を算出する．
        loss.backward()

        # パラメータ更新
        optimizer.step()


    # Test process
    X_test = np.asarray(X_test, dtype=np.float32) / 16.0      # imageのスケーリング
    if GPU_sw:
        var_x_te = Variable(torch.from_numpy(X_test).cuda())
        y_pred_te = net.forward(var_x_te)
        y_pred_proba = y_pred_te.data.cpu().numpy()
    else:
        var_x_te = Variable(torch.from_numpy(X_test))
        y_pred_te = net.forward(var_x_te)
        y_pred_proba = y_pred_te.data.numpy()
    y_pred_ = np.argmax(y_pred_proba, axis=1)

    # テスト結果の評価
    confmat = confusion_matrix(y_test, y_pred_)
    print('\nconfusion matrix:')
    print(confmat)
    accu = accuracy_score(y_test, y_pred_)
    print('\naccyracy = {:>.4f}'.format(accu))

logistic_regression_low.py

# -*- coding: utf-8 -*-
#
#   logistic_regression_low.py
#       date. 8/22/2017
#

import numpy as np
import torch
from torch.autograd import Variable

from sklearn.datasets import load_digits
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, confusion_matrix

# CPU演算とGPU演算を切り換えるスイッチ．GPU演算では，CPU-GPU間のメモリ・コピーが行われる．
if torch.cuda.is_available():
    GPU_sw = True
else:
    GPU_sw = False
print('GPU_sw = ', GPU_sw)


def load_data():
    digits = load_digits()
    y = digits.target
    n_samples = len(digits.images)
    X = digits.images.reshape((n_samples, -1))

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=0)

    return X_train, X_test, y_train, y_test


class Model:
    # torch.nn APIを用いないモデル構築
    def __init__(self, n_feature, n_class):
        if GPU_sw:
            dtype = torch.cuda.FloatTensor
        else:
            dtype = torch.FloatTensor

        self.W = Variable((torch.randn(n_feature, n_class) * 0.01).type(dtype), 
                           requires_grad=True)
        self.b = Variable(torch.zeros(n_class).type(dtype), requires_grad=True)

    def forward(self, x):
        """
          calculate network forwarding prop. 
          return logits before activation (LogSoftMax)
        """
        y = torch.mm(x, self.W) + self.b    # using "Broadcasting" for "b"

        return y

    @property
    def params(self):
        return [self.W, self.b]


def train_feed(X_train, y_train, idx, batch_size):
    # Training phase におけるデータ供給
    dtype = torch.FloatTensor
    n_samples = X_train.shape[0]
    if idx == 0:
        perm = np.random.permutation(n_samples)
        X_train = X_train[perm]
        y_train = y_train[perm]
    start = idx
    end = start + batch_size
    if end > n_samples:
        perm = np.random.permutation(n_samples)
        X_train = X_train[perm]
        y_train = y_train[perm]
        start = 0
        end = start + batch_size

    batch_x = np.asarray(X_train[start:end], dtype=np.float32)
    batch_x = batch_x / 16.0    # image のスケーリング
    batch_y = np.asarray(y_train[start:end], dtype=np.int64)
    idx = end

    if GPU_sw:
        var_x = Variable(torch.from_numpy(batch_x).cuda())
        var_y = Variable(torch.from_numpy(batch_y).cuda())
    else:
        var_x = Variable(torch.from_numpy(batch_x))
        var_y = Variable(torch.from_numpy(batch_y))

    return var_x, var_y, idx


if __name__ == '__main__':
    # Load Data
    X_train, X_test, y_train, y_test = load_data()
    n_feature = 64
    n_class = 10
    batch_size = 100

    # define network
    model = Model(n_feature, n_class)

    loss_fn = torch.nn.CrossEntropyLoss()    # 損失関数の定義
    optimizer = torch.optim.SGD(model.params, 
                                lr=0.003, momentum=0.9)  # オプティマイザ

    print('Training...')
    train_index = 0
    for t in range(10000):
        batch_x, batch_y, train_index = train_feed(X_train, y_train, 
                                                   train_index, batch_size)
        y_pred = model.forward(batch_x)
        loss = loss_fn(y_pred, batch_y)

        if t % 1000 == 0:
            print('{:>5d}: loss = {:>10.3f}'.format(t, loss.data[0]))

        # 勾配gradを初期化（ゼロ化）
        optimizer.zero_grad()

        # Backward pass: 誤差の逆伝搬を行って，パラメータの変化量を算出する．
        loss.backward()

        # パラメータ更新
        optimizer.step()


    # Test process
    X_test = np.asarray(X_test, dtype=np.float32) / 16.0      # imageのスケーリング
    if GPU_sw:
        var_x_te = Variable(torch.from_numpy(X_test).cuda())
        y_pred_te = model.forward(var_x_te)
        y_pred_proba = y_pred_te.data.cpu().numpy()
    else:
        var_x_te = Variable(torch.from_numpy(X_test))
        y_pred_te = model.forward(var_x_te)
        y_pred_proba = y_pred_te.data.numpy()
    y_pred_ = np.argmax(y_pred_proba, axis=1)

    # テスト結果の評価
    confmat = confusion_matrix(y_test, y_pred_)
    print('\nconfusion matrix:')
    print(confmat)
    accu = accuracy_score(y_test, y_pred_)
    print('\naccyracy = {:>.4f}'.format(accu))

mnist_mlp.py

# -*- coding: utf-8 -*-
#
#   mnist_mlp.py
#       date. 8/24/2017
#       library: torch v0.2.0
#

import numpy as np
import torch
from torch.autograd import Variable
import torch.nn as nn
import torch.nn.functional as F
from torchvision import datasets as dset
from torchvision import transforms
from sklearn.metrics import accuracy_score, confusion_matrix

# CPU演算とGPU演算を切り換えるスイッチ．GPU演算では，CPU-GPU間のメモリ・コピーが行われる．
GPU_sw = torch.cuda.is_available()
print('GPU_sw = ', GPU_sw)

def mk_data_loader(dirn, batch_size, train=True, gpu_sw=False):
    """
      MNIST data loader 関数
    """
    kwargs = {'num_workers': 1, 'pin_memory': True} if gpu_sw else {}
    data_loader = torch.utils.data.DataLoader(
        dset.MNIST(dirn, train=train, download=False,
                   transform=transforms.Compose([
                       transforms.ToTensor(),
                        # torchvision.transforms.Normalize(mean, std)
                       transforms.Normalize((0.1307,), (0.3081,))
                   ])),
        batch_size=batch_size, shuffle=True, **kwargs)

    return data_loader

class Net(nn.Module):
    """
      torch.nn modeling - 3層 MLP model
    """
    def __init__(self, n_feature, n_class, n_hidden1=512, n_hidden2=256):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(n_feature, n_hidden1)
        self.fc2 = nn.Linear(n_hidden1, n_hidden2)
        self.fc3 = nn.Linear(n_hidden2, n_class)
    def forward(self, x):
        net = F.relu(self.fc1(x))
        net = F.relu(self.fc2(net))
        y = self.fc3(net)

        return y

if __name__ == '__main__':
    # Data Loader
    batch_size = 100
    train_loader = mk_data_loader('../MNIST_data', batch_size,
                                  train=True, gpu_sw=GPU_sw)
    test_loader = mk_data_loader('../MNIST_data', batch_size,
                                 train=False, gpu_sw=GPU_sw)
    # define network
    n_feature = 784
    n_class = 10
    if GPU_sw:
        net = Net(n_feature, n_class).cuda()
    else:
        net = Net(n_feature, n_class)

    loss_fn = torch.nn.CrossEntropyLoss()    # 損失関数の定義
    optimizer = torch.optim.SGD(net.parameters(), 
                                lr=0.003, momentum=0.9)  # オプティマイザ

    # Train プロセス
    print('Training...')
    n_epochs = 10
    for epoch in range(1, n_epochs+1):
        for i, (data, target) in enumerate(train_loader):
            data = data.resize_([batch_size, n_feature])
            if GPU_sw:
                data, target = data.cuda(), target.cuda()
            data, target = Variable(data), Variable(target)

            y_pred = net.forward(data)
            loss = loss_fn(y_pred, target)

            if i % 100 == 0:
                print('epoch {:>3d}:{:>5d}: loss = {:>10.3f}'.format(
                        epoch, i, loss.data[0]))
            
            # zero the gradient buffers, 勾配gradを初期化（ゼロ化）する．
            optimizer.zero_grad()
            # Backward pass: 誤差の逆伝搬を行って，パラメータの変化量を算出する．
            loss.backward()
            # パラメータ更新
            optimizer.step()

    # Test プロセス
    y_pred = []
    y_target = []
    for data, target in test_loader:
        data = data.resize_([batch_size, n_feature])
        if GPU_sw:
            data, target = data.cuda(), target.cuda()
        data, target = Variable(data), Variable(target)
        y_pred_ = net.forward(data)

        if GPU_sw:
            y_pred.extend(y_pred_.data.cpu().numpy())
            y_target.extend(target.data.cpu().numpy())
        else:
            y_pred.extend(y_pred_.data.numpy())
            y_target.extend(target.data.numpy())

    y_pred_am = np.argmax(np.asarray(y_pred), axis=1)
    y_target = np.asarray(y_target)

    # テスト結果の評価
    confmat = confusion_matrix(y_target, y_pred_am)
    print('\nconfusion matrix:')
    print(confmat)
    accu = accuracy_score(y_target, y_pred_am)
    print('\naccyracy = {:>.4f}\n'.format(accu))