Tutoriel de Theano (http://deeplearning.net/tutorial/contents.html) Sur la base de, j'ai implémenté une régression logistique avec les données MNIST de Kaggle.
from __future__ import print_function
__docformat__ = 'restructedtext en'
import six.moves.cPickle as pickle
import timeit
import numpy
import theano
import theano.tensor as T
import numpy as np
rng = numpy.random.RandomState(8000)
#Définition des variables partagées
#Les données créées par cette expression sont
#-------------------------------------------------------------------------
#Classe de régression logistique
#-------------------------------------------------------------------------
class LogisticRegression(object):
"""Multi-class Logistic Regression Clasｓ
"""
def __init__(self, input, n_in, n_out):
#MODEL
#Représentation de variable partagée
#À propos de W
self.W = theano.shared(
value=numpy.zeros(
(n_in, n_out),
dtype=theano.config.floatX
),
name='W',
borrow=True
)
#borrow détermine si l'entité de données définie dans l'espace Python est également partagée par des variables partagées.
#À propos de b
self.b = theano.shared(
value=numpy.zeros(
(n_out,),
dtype=theano.config.floatX
),
name='b',
borrow=True
)
#Calcul de la fonction softmax (calcul de la probabilité d'attribution à la classe)
#ｐ(y＝i|x,W,b)
self.p_y_given_x = T.nnet.softmax(T.dot(input, self.W) + self.b)
#Prendre argmax et afficher la valeur prédite
#ypred
self.y_pred = T.argmax(self.p_y_given_x, axis=1)
#.Stocké dans paramsn
self.params = [self.W, self.b]
#.Stocker en entrée
self.input = input
#Fonction de perte
def negative_log_likelihood(self, y):
#Définition de la fonction de perte
#self.p_y_given_x est la fonction softmax (sortie yn)
#T.arange(y.shape[0])Est la ligne et la colonne de Numpy[]
#-Ln(p(t｜w)).PRML4.Type 90
return -T.mean(T.log(self.p_y_given_x)[T.arange(y.shape[0]), y])
#Taux d'erreur
def errors(self, y):
#Vérification des dimensions de la sortie
if y.ndim != self.y_pred.ndim:
raise TypeError(
'y should have the same shape as self.y_pred',
('y', y.type, 'y_pred', self.y_pred.type)
)
#Vérifiez si le type de données est correct
if y.dtype.startswith('int'):
#neq sorties pas égales la réponse de sortie est correcte et probabilité correcte
#Premier argument: Argument avec l'image de probabilité de sortie maximale
#Deuxième argument: étiquette correcte
return T.mean(T.neq(self.y_pred, y))
else:
raise NotImplementedError()
#Méthode de descente de gradient stochastique en mini-lots
def sgd_optimization_mnist(learning_rate=0.10, n_epochs=100,
dataset='mnist.pkl.gz',
batch_size=500):
dtype = np.float32
data = np.loadtxt("../input/train.csv", dtype=dtype,
delimiter=',', skiprows=1)
test_data = np.loadtxt("../input/test.csv", dtype=dtype,
delimiter=',', skiprows=1)
test_datasets=[test_data]
print (data.shape)
labels = data[:,0]
data = data[:, 1:]
index = T.lscalar() #index de minibatch
NUM_TRAIN = len(data)
NUM_TEST = len(test_data)
if NUM_TRAIN % batch_size != 0:
whole = (NUM_TRAIN // batch_size) * batch_size
data = data[:whole]
NUM_TRAIN = len(data)
#tri aléatoire
indices = rng.permutation(NUM_TRAIN)
data, labels = data[indices, :], labels[indices]
# batch_la taille est de 500, (480, 20)Et entraînez-vous avec 96% de l'ensemble de données et validez avec le reste
is_train = numpy.array( ([0]* (batch_size - 20) + [1] * 20) * (NUM_TRAIN // batch_size))
test_indices = rng.permutation(NUM_TEST)
test_data, test_labels = test_data[test_indices, :], labels[test_indices]
is_test = numpy.array( ([0]* (batch_size - 20) + [1] * 20) * (NUM_TEST // batch_size))
# trai_set,valid_set,test_Préparation du set
train_set_x, train_set_y = numpy.array(data[is_train==0]), labels[is_train==0]
valid_set_x, valid_set_y = numpy.array(data[is_train==1]), labels[is_train==1]
test_set_x,test_set_y = numpy.array(data[is_test==0]), labels[is_test==0]
#calcul de minibatch
n_train_batches = len(train_set_y) // batch_size
n_valid_batches = len(valid_set_y) // batch_size
n_test_batches = len(test_set_y) // batch_size
##############
#Construire un modèle#
##############
print('... building the model')
#scalaire de type long
index = T.lscalar()
#type de matrice
x = T.matrix('x')
#vecteur de type int
y = T.ivector('y')
#Construire une classe de régression logistique
#Les données MNIST sont 28X28
classifier = LogisticRegression(input=x, n_in=28 * 28, n_out=10)
#la fonction de coût est la probabilité logarithmique négative
cost = classifier.negative_log_likelihood(y)
#Mini lot type theano.Résumer avec la fonction
train_set_x = theano.shared(numpy.asarray(train_set_x, dtype=theano.config.floatX))
train_set_y = T.cast(theano.shared(numpy.asarray(train_set_y, dtype=theano.config.floatX)), 'int32')
test_set_x = theano.shared(numpy.asarray(test_set_x, dtype=theano.config.floatX))
test_set_y = T.cast(theano.shared(numpy.asarray(test_set_y, dtype=theano.config.floatX)), 'int32')
valid_set_x = theano.shared(numpy.asarray(valid_set_x, dtype=theano.config.floatX))
valid_set_y = T.cast(theano.shared(numpy.asarray(valid_set_y, dtype=theano.config.floatX)), 'int32')
#valider le modèle
validate_model = theano.function(
inputs=[index],
outputs=classifier.errors(y),
givens={
x: valid_set_x[index * batch_size: (index + 1) * batch_size],
y: valid_set_y[index * batch_size: (index + 1) * batch_size]
}
)
#modèle de test
test_model = theano.function(
inputs=[index],
outputs=classifier.errors(y),
givens={
x: valid_set_x[index * batch_size:(index + 1) * batch_size],
y: valid_set_y[index * batch_size:(index + 1) * batch_size]
}
)
#Différencier le coût par W
g_W = T.grad(cost=cost, wrt=classifier.W)
#Différencier le coût par b
g_b = T.grad(cost=cost, wrt=classifier.b)
#la fonction de coût est la probabilité logarithmique négative
#Paramètres (W,b) Mettre à jour
updates = [(classifier.W, classifier.W - learning_rate * g_W),
(classifier.b, classifier.b - learning_rate * g_b)]
#modèle de train
#Paramètres définis dans les mises à jour (W,b) Mettre à jour
train_model = theano.function(
inputs=[index],
outputs=cost,
updates=updates,
givens={
x: train_set_x[index * batch_size: (index + 1) * batch_size],
y: train_set_y[index * batch_size: (index + 1) * batch_size]
}
)
#################
#Formation modèle#
#################
print('... training the model')
#Nombre maximum de BOUCLES
patience = 50000
#S'il n'augmente pas de plus de 2
patience_increase = 5
#Limite supérieure des notes
improvement_threshold = 0.995
#Fréquence des réunions de patience. Dans ce cas, vérifiez à chaque époque
validation_frequency = min(n_train_batches, patience // 2)
best_validation_loss = numpy.inf
test_score = 0.
start_time = timeit.default_timer()
#Le début de LOOP
done_looping = False
epoch = 0
minibatch_index = 1
while (epoch < n_epochs) and (not done_looping):
epoch = epoch + 1
for minibatch_index in range(n_train_batches):
minibatch_avg_cost = train_model(minibatch_index)
#Calcul de l'itération
iter = (epoch - 1) * n_train_batches + minibatch_index
if (iter + 1) % validation_frequency == 0:
# zero-Une perte calculée par Validation
validation_losses = [validate_model(i)
for i in range(n_valid_batches)]
this_validation_loss = numpy.mean(validation_losses)
print(
'epoch %i, minibatch %i/%i, validation error %f %%' %
(
epoch,
minibatch_index + 1,
n_train_batches,
this_validation_loss * 100.
)
)
#Si la perte de validation est la plus faible
if this_validation_loss < best_validation_loss:
if this_validation_loss < best_validation_loss * \
improvement_threshold:
patience = max(patience, iter * patience_increase)
best_validation_loss = this_validation_loss
#Test avec ensemble de test
test_losses = [test_model(i)
for it in range(len(test_datasets))]
test_score = numpy.mean(test_losses)
print(
(
' epoch %i, minibatch %i/%i, test error of'
' best model %f %%'
) %
(
epoch,
minibatch_index + 1,
n_train_batches,
test_score * 100.
)
)
#Enregistrer le meilleur modèle
with open('best_model.pkl', 'wb') as f:
pickle.dump(classifier, f)
if patience <= iter:
done_looping = True
break
end_time = timeit.default_timer()
print(
(
'Optimization complete with best validation score of %f %%,'
'with test performance %f %%'
)
% (best_validation_loss * 100., test_score * 100.)
)
print('The code run for %d epochs, with %f epochs/sec' % (
epoch, 1. * epoch / (end_time - start_time)))
def predict():
"""
An example of how to load a trained model and use it
to predict labels.
"""
#Appeler le modèle entraîné
classifier = pickle.load(open('best_model.pkl'))
#prédire le modèle
predict_model = theano.function(
inputs=[classifier.input],
outputs=classifier.y_pred)
#Appeler l'ensemble de test
test_data = np.loadtxt("../input/test.csv", dtype=dtype,
delimiter=',', skiprows=1)
#prédire
predicted_values = predict_model(test_data)
print("Predicted values for the first 10 examples in test set:")
print(predicted_values)
np.savetxt('Submit_TheanoLR3.csv', np.c_[range(1, len(test_data) + 1), predicted_values], delimiter=',', comments = '', header = 'ImageId,Label', fmt='%d')
if __name__ == '__main__':
sgd_optimization_mnist()
Le résultat est 0,90757. Il était de 0,90900 dans scikit-learn, et cela ressemble à ceci.
Recommended Posts