[1,2,5 Layer] Deep Neural Network implementation NOTE: Use you own vector data [Read Comment to select number of layers]

dnn_ravyg.py

#!/usr/bin/env python
# -*- coding: utf-8 -*- 
# Parts of this code have be taken from Dr.Calix's ITS520 Class at Purdue University.
# Co-Authored: Ravish Gupta
##########################################################
import csv
import warnings
import matplotlib.pyplot as plt
import tensorflow as tf
import numpy as np
from numpy import genfromtxt
import matplotlib.pyplot as plt

# Preprocessor.
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import LabelEncoder
# Performace Evaluator.
from sklearn.metrics import accuracy_score
from sklearn.metrics import precision_score
from sklearn.metrics import recall_score,f1_score
from sklearn.metrics import confusion_matrix
from sklearn.metrics import roc_auc_score
from sklearn.metrics import classification_report
# Loose warnings.
warnings.filterwarnings("ignore", category=DeprecationWarning) 

def convertOneHot_data2(data):
    y=np.array([int(i) for i in data])
    rows = len(y)
    columns = y.max()+1
    a = np.zeros(shape = (rows,columns))
    for i,j in enumerate(y):
        a[i][j]=1
    return (a)
X_train = []
y_train = []
X_test = []
y_test = []

# Balanced Set.
# Put your training and testing data here.
# NOTE: These training and testing data file are not attached with this gist please use your own data.
# My training data file is names 8770_train.csv where 8770 is number of records same can be infered to testing data.
traindata= list(csv.reader(open ('tweets_data/8770_train.csv', 'rU'), delimiter =',')) # Train Data
testdata = list(csv.reader(open ('tweets_data/821_test.csv', 'rU'), delimiter =',')) # Test Data

# Adjest input dimensions for you training and testing data here.
for data in traindata:
   # training data.
   # I am only using columns 4 to 22 of my training and testing data.
   X_train.append(data[4:22])
   # training label
   # Column 23 containg my lables (annotated class)
   y_train.append(data[23])

for data in testdata:
   # Testing data. 
   X_test.append(data[4:22])
   # Testing label
   y_test.append(data[23])

le = LabelEncoder()
y_train = le.fit_transform(y_train)
y_test = le.fit_transform(y_test)

accuracy_score_list = []
precision_score_list =[]
def print_stats_metrics(y_test, y_pred):
    # Accuracy.
    print('Accuracy: %.3f' % accuracy_score(y_test, y_pred))
    accuracy_score_list.append(accuracy_score(y_test,y_pred))
    # Precision.
    precision_score_list.append(precision_score(y_test, y_pred))
    print('Presicion: %.3f' % precision_score(y_test, y_pred))
    # F1 Score.
    print ("f1_score %.3f" % f1_score(y_true, y_pred))
    # Recall Score.
    print ("recall_score %.3f" % recall_score(y_true, y_pred))
    # AUC
    # store the predicted probabilities for class 1
    auc_val = roc_auc_score(y_pred, y_true)
    print("roc_auc_score %.3f" % auc_val)
    # Confusion Matrix.
    cm = confusion_matrix(y_test, y_pred)
    print("Confusion matrix:")
    print(cm)

    # Classification report.
    target_names = ['Class0', 'Class1']
    print(classification_report(y_true, y_pred, target_names=target_names))

def plot_metric_per_epoch():
    x_epochs = []
    y_epochs = []
    for i, val in enumerate(accuracy_score_list):
        x_epochs.append(i)
        y_epochs.append(val)
    plt.scatter(x_epochs, y_epochs, s=50, c='lightgreen', marker='s', label='score')
    plt.xlabel('epochs')
    plt.ylabel('score')
    plt.title('Score per epoch')
    plt.grid()
    plt.show()

def layer(input, weight_shape, bias_shape):
    weight_stddev = (2.0/weight_shape[0])**0.5
    # reason we get different results.
    w_init = tf.random_normal_initializer(stddev=weight_stddev)
    bias_init = tf.constant_initializer(value=0)
    # Initializing it differently. 
    W = tf.get_variable("W", weight_shape, initializer=w_init)
    b = tf.get_variable("b", bias_shape, initializer=bias_init)
    return tf.nn.relu(tf.matmul(input, W) + b)

#Deep Neural Net - 5 hidden layers.
def inference_DeepNet5layers(x_tf, A, B):
    # First Hidder Layer 1
    with tf.variable_scope("hidden_1"):
        hidden_1 = layer(x_tf, [A, 64],[64])
    # Inner Layer 2    
    with tf.variable_scope("hidden_2"):
        hidden_2 = layer(hidden_1, [64, 32],[32])
    # Inner Layer 3
    with tf.variable_scope("hidden_3"):
        hidden_3 = layer(hidden_2, [32, 16],[16])
    # Inner Layer 4
    with tf.variable_scope("hidden_4"):
        hidden_4 = layer(hidden_3, [16, 8],[8])
    # Inner Layer 5    
    with tf.variable_scope("hidden_5"):
        hidden_5 = layer(hidden_4, [8, 4],[4])    
    # Final output Layer 
    with tf.variable_scope("output"):
        output = layer(hidden_5, [4, B], [B])
    return output

#Deep Neural Net - 4 hidden layers.
def inference_DeepNet4layers(x_tf, A, B):
    # First Hidder Layer 1
    with tf.variable_scope("hidden_1"):
        hidden_1 = layer(x_tf, [A, 20],[20])
    # Inner Layer 2    
    with tf.variable_scope("hidden_2"):
        hidden_2 = layer(hidden_1, [20, 16],[16])
    # Inner Layer 3
    with tf.variable_scope("hidden_3"):
        hidden_3 = layer(hidden_2, [16, 8],[8])
    # Inner Layer 4
    with tf.variable_scope("hidden_4"):
        hidden_4 = layer(hidden_3, [8, 4],[4])  
    # Final output Layer 
    with tf.variable_scope("output"):
        output = layer(hidden_4, [4, B], [B])
    return output

#Deep Neural Net - 2 hidden layers.
def inference_DeepNet2layers(x_tf, A, B):
    # First Hidder Layer 1
    with tf.variable_scope("hidden_1"):
        hidden_1 = layer(x_tf, [A, 7],[7])
    # Inner Layer 2        
    with tf.variable_scope("hidden_2"):
        hidden_2 = layer(hidden_1, [7, 3],[3])    
    # Final output Layer 
    with tf.variable_scope("output"):
        output = layer(hidden_2, [3, B], [B])
    return output

#Deep Neural Net - 1 hidden layers.
def inference_DeepNet1layers(x_tf, A, B):
    # First Hidder Layer 1
    with tf.variable_scope("hidden_1"):
        hidden_1 = layer(x_tf, [A, 6],[6])   
    # Final output Layer 
    with tf.variable_scope("output"):
        output = layer(hidden_1, [6, B], [B])
    return output


# Loss (cost) Functions for 2 Layered Net.
def loss_DeepNet2layers(output, y_tf):
    xentropy = tf.nn.softmax_cross_entropy_with_logits(output, y_tf)
    loss = tf.reduce_mean(xentropy) 
    return loss


# Defines the network architecture.
# Simple logistic regression.
def inference(x_tf, A, B):
    W = tf.Variable(tf.zeros([A,B]))
    b = tf.Variable(tf.zeros([B]))
    output = tf.nn.softmax(tf.matmul(x_tf, W) + b)
    return output
   

def loss(output, y_tf):
    dot_product = y_tf * tf.log(output)
    xentropy = -tf.reduce_sum(dot_product, reduction_indices=1)#remove indices?
    loss = tf.reduce_mean(xentropy) #remove this line?
    return loss
    
def training(cost):
    optimizer = tf.train.GradientDescentOptimizer(0.001)
    train_op = optimizer.minimize(cost)
    return train_op

## Add accuracy checking nodes.
def evaluate(output, y):
    correct_prediction = tf.equal(tf.argmax(output,1), tf.argmax(y,1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
    return accuracy

sc= StandardScaler()
sc.fit(X_train)
X_train = sc.transform(X_train)
X_test = sc.transform(X_test)

y_train_onehot=convertOneHot_data2(y_train)
y_test_onehot=convertOneHot_data2(y_test)


A = X_train.shape[1]         # number of features
B = y_train_onehot.shape[1]   # number of classes


x_tf=tf.placeholder(tf.float32, [None, A])  # features
y_tf=tf.placeholder(tf.float32,[None, B])   ##correct label for x sample, this y_tf has to be onehot encoded

# output = inference_DeepNet5layers(x_tf, A, B) ## for deep NN with 5 hidden layers
# output = inference_DeepNet4layers(x_tf, A, B) ## for deep NN with 4 hidden layers
# output = inference_DeepNet2layers(x_tf, A, B) ## for deep NN with 2 hidden layers
output = inference_DeepNet1layers(x_tf, A, B) ## for deep NN with 1 hidden layers

cost = loss_DeepNet2layers(output, y_tf)
train_op = training(cost)
eval_op = evaluate(output, y_tf)

##################################################################
# Initialize and run
init = tf.initialize_all_variables()
sess = tf.Session()
sess.run(init)
n_epochs = 1000
batch_size = 200
number_of_samples_train_test = X_train.shape[0]
num_batches = int(number_of_samples_train_test/batch_size)
y_p_metrics = tf.argmax(output,1)

for i in range(n_epochs):
    for batch_n in range(num_batches):
        sta = batch_n*batch_size
        end = sta + batch_size
        sess.run(train_op, feed_dict={x_tf:X_train[sta:end,:], y_tf:y_train_onehot[sta:end,:]})
    print("iteration %d " % i)
    print "***********************************************************"
    result, y_result_matrics = sess.run([eval_op, y_p_metrics], feed_dict={x_tf:X_test, y_tf:y_test_onehot})
    y_true = np.argmax(y_test_onehot, 1)
    print_stats_metrics(y_true, y_result_matrics)
    # print precision_score_list
    print "Run {} , {}".format(i, result)
    #print("W: %f" % sess.run(W))

plot_metric_per_epoch()
print "<<<<<<Processing done!>>>>>>"

To select number of layers comment/uncomment:

# output = inference_DeepNet5layers(x_tf, A, B) ## for deep NN with 5 hidden layers
# output = inference_DeepNet4layers(x_tf, A, B) ## for deep NN with 4 hidden layers
# output = inference_DeepNet2layers(x_tf, A, B) ## for deep NN with 2 hidden layers
output = inference_DeepNet1layers(x_tf, A, B) ## for deep NN with 1 hidden layers