ift7025-projet/Code/DecisionTree.py

# -*- coding: utf-8 -*-
"""
Vous allez definir une classe pour chaque algorithme que vous allez développer,
votre classe doit contenit au moins les 3 methodes definies ici bas, 
    * train     : pour entrainer le modèle sur l'ensemble d'entrainement
    * predict     : pour prédire la classe d'un exemple donné
    * test         : pour tester sur l'ensemble de test
vous pouvez rajouter d'autres méthodes qui peuvent vous etre utiles, mais moi
je vais avoir besoin de tester les méthodes train, predict et test de votre code.
"""

import numpy as np
import metrics
import time

# le nom de votre classe
# NeuralNet pour le modèle Réseaux de Neurones
# DecisionTree le modèle des arbres de decision

class DecisionTree: #nom de la class à changer

    def __init__(self, attribute_type, **kwargs):
        """
        c'est un Initializer. 
        Vous pouvez passer d'autre paramètres au besoin,
        c'est à vous d'utiliser vos propres notations
        """
        self.tree=[]
        self.attribute_type=attribute_type
        
    def plurality_value(self,train_labels):
        values,_,counts = np.unique(train_labels,return_index=True, return_counts=True)
        return values[np.argmax(counts)]
        
    def importance(self,train, train_labels, attribute):
        #Entropie Avant
        _,_,counts = np.unique(train_labels,return_index=True, return_counts=True)
        total_count = sum(counts)
        entropie_total = -sum(counts/total_count * np.log2(counts/total_count))
        #print("Entropie Total:"+str(entropie_total))
        
        if (self.attribute_type=="continuous"):
            #Trouver split
            attribute_sort_order = np.argsort(train[:,attribute])
            sorted_labels = train_labels[attribute_sort_order]
            lags = np.hstack((np.array([False]),sorted_labels[:-1] != sorted_labels[1:]))
            lags2 = np.hstack((np.array([False]),lags))[:-1]
            potential_splits = (
                    0.5*train[attribute_sort_order,attribute][lags]+
                    0.5*train[attribute_sort_order,attribute][lags2]
                    )
            if (len(potential_splits)==0):
                potential_splits = np.array([np.median(train[attribute_sort_order,attribute])])
            #print("Potential Split:"+str(potential_splits))
            split_gain = []
            for v in potential_splits:
                split_labels_1 = train_labels[train[:,attribute] <= v]
                split_labels_2 = train_labels[train[:,attribute] > v]
                _,_,counts1 = np.unique(split_labels_1,return_index=True, return_counts=True)
                total_count1 = sum(counts1)
                entropie_total1 = -sum(counts1/total_count1 * np.log2(counts1/total_count1))
                _,_,counts2 = np.unique(split_labels_2,return_index=True, return_counts=True)
                total_count2 = sum(counts2)
                entropie_total2 = -sum(counts2/total_count2 * np.log2(counts2/total_count2))
                split_gain.append(entropie_total-(total_count1/total_count*entropie_total1+total_count2/total_count*entropie_total2))
            #Valeur unique attribut
            #print("Split Gain:"+str(split_gain))
            best_split = potential_splits[np.argmax(split_gain)]
            best_gain = max(split_gain)
            return best_gain,best_split
        if (self.attribute_type=="discrete"):
            _,counts1 = np.unique(train[:,attribute], return_counts=True)
            total_count1 = sum(counts1)
            entropie_total1 = -sum(counts1/total_count1 * np.log2(counts1/total_count1))
            gain = entropie_total - entropie_total1
            return gain,None
        
    
    def decision_tree_learning(self,train, train_labels, attributes, parent_examples):
        n_examples,n_attributes = train.shape
        classes_uniques = np.unique(train_labels)
        # la feuille est vide
        if (n_examples == 0):
            l1 = []
            l1.append(("Feuille",self.plurality_value(parent_examples)))
            return list(l1)
        # tous les exemples ont la même classe
        elif len(classes_uniques)==1:
            l1 = []
            l1.append(("Feuille",classes_uniques[0]))
            return l1
        # la liste d'attributs est vides
        elif (sum(attributes)==0):
            l1 = []
            l1.append(("Feuille",self.plurality_value(train_labels)))
            return l1
        else:
            # Calcul du gain
            attr = np.where(attributes==1)[0]
            a_gain = []
            a_split = []
            for a in attr:
                gain, split = self.importance(train, train_labels, a)
                a_gain.append(gain)
                a_split.append(split)
                
            # Calcul du meilleur attribut
            
            pos_gain_max = np.argmax(a_gain)
            a_max = attr[pos_gain_max]
            a_max_split = a_split[pos_gain_max]
            attributes[a_max]=0
            
            # Nouvel arbre
            
            tree = []
            
            # pour chaque valeur de l'attribut, faire un sous-arbre
            if (self.attribute_type=="continuous"):
                for v in [True,False]:
                    #print("Nouvelle branche: l'attribut "+str(a_max)+"<="+str(a_max_split)+" est: "+str(v))
                    train_pos = np.where((train[:,a_max] <= a_max_split) == v)
                    subtree = self.decision_tree_learning(train[train_pos],train_labels[train_pos],attributes,train_labels)
                    tree.append(("Branche",a_max,a_max_split,v,subtree))
            if (self.attribute_type=="discrete"):
                for v in np.unique(train[:,a_max]):
                    #print("Nouvelle branche: l'attribut "+str(a_max)+" est: "+str(v))
                    train_pos = np.where(train[:,a_max] == v)
                    subtree = self.decision_tree_learning(train[train_pos],train_labels[train_pos],attributes,train_labels)
                    tree.append(("Branche",a_max,v,subtree))
        return tree
                
        
    def train(self, train, train_labels,verbose=True): #vous pouvez rajouter d'autres attribus au besoin
        """
        c'est la méthode qui va entrainer votre modèle,
        train est une matrice de taille nxm, avec 
        n : le nombre d'exemple d'entrainement dans le dataset
        m : le mobre d'attribus (le nombre de caractéristiques)
        
        train_labels : est une matrice de taille nx1
        verbose: afficher les métriques calculées
        
        vous pouvez rajouter d'autres arguments, il suffit juste de
        les expliquer en commentaire
        
        
        
        ------------
        
        """
        
        n,m = train.shape
        # Vecteur booléen qui définit quels attributs n'ont pas été utilisés
        attributes = np.ones(m)
        
        self.tree = self.decision_tree_learning(train, train_labels, attributes, None)
        
        return self.test(train, train_labels,verbose)
        
        
    def extract_tree(self,myTree,exemple):
        for b in myTree:
            # On a atteint la feuille
            if b[0] == 'Feuille':
                return b[1]
            # On est dans une branche, on teste le split
            if self.attribute_type=="continuous":
                if ((exemple[b[1]] <= b[2]) == b[3]):
                    return self.extract_tree(b[4],exemple)
            if self.attribute_type=="discrete":
                if (exemple[b[1]] == b[2]):
                    return self.extract_tree(b[3],exemple)
        return None
            
        
    def predict(self, exemple, label):
        """
        Prédire la classe d'un exemple donné en entrée
        exemple est de taille 1xm
        
        si la valeur retournée est la meme que la veleur dans label
        alors l'exemple est bien classifié, si non c'est une missclassification

        """
        
        return self.extract_tree(self.tree,exemple)
        
    def test(self, test, test_labels, verbose=True):
        """
        c'est la méthode qui va tester votre modèle sur les données de test
        l'argument test est une matrice de taille nxm, avec 
        n : le nombre d'exemple de test dans le dataset
        m : le mobre d'attribus (le nombre de caractéristiques)
        verbose: afficher les métriques calculées
        
        test_labels : est une matrice taille nx1
        
        vous pouvez rajouter d'autres arguments, il suffit juste de
        les expliquer en commentaire
        
        Faites le test sur les données de test, et afficher :
        - la matrice de confision (confusion matrix)
        - l'accuracy (ou le taux d'erreur)
        
        Bien entendu ces tests doivent etre faits sur les données de test seulement
        
        """
        start_time = time.time()
        prediction_test = [self.predict(exemple,label) for exemple,label in zip(test,test_labels)]
        cm = metrics.confusion_matrix(test_labels,prediction_test)
        accuracy, precision, recall = metrics.prediction_metrics(cm,test_labels,prediction_test)
        compute_time = time.time() - start_time
        if (verbose):
            metrics.print_prediction_metrics(cm,accuracy,precision,recall,compute_time)

        return cm,accuracy,precision,recall,compute_time
    
    
    # Vous pouvez rajouter d'autres méthodes et fonctions,
    # il suffit juste de les commenter.