Decision Tree, Il Modello
Parte (2/3)
Potrebbe interessarti:
Il progetto Python, che comprende gli script mostrati in questo post, e quelli relativi alla creazione dei grafici Matplotlib e Plotly, è disponibile assieme al corso Introduzione al Machine Learning.
Implementazione del Decision Tree
Dopo aver intuito l’algoritmo che sta alla base di un Decision Tree, e dopo aver visto come calcolare l’impurità del modello tramite la funzione di costo Entropy o Gini Impurity, vediamo ora come implementare tale classificatore “da zero”. Di seguito, alcune considerazioni sulle convenzioni dei nomi applicate ai metodi delle classi che andremo a sviluppare:
Python e i Modificatori di Accesso ai Metodi
Python è un linguaggio tipizzato dinamicamente e interpretato,
quindi l’applicazione rigorosa dei modificatori di accesso
in fase di compilazione è meno coerente e rigorosa rispetto a quella di linguaggi come C++, Java o C#.
Python non usa parole chiave come public, protected o private, ma segue le seguenti convenzioni:
- metodi public: tutti i metodi che non iniziano per underscore, ad esempio:
predict. - metodi protected: tutti i metodi che cominciano con un singolo carattere underscore, ad esempio:
_impurity. - metodi private: tutti i metodi che cominciano con due caratteri underscore, come
__build.
La Classe Node
Come prima cosa, definiamo la classe Node, che verrà usata per costruire la struttura ad albero del classificatore:
from __future__ import annotations
from abc import ABC, abstractmethod
import numpy as np
from numpy import ndarray
from scipy import stats
class Node(object):
"""
A Decision Tree node object.
"""
def __init__(self, parent: Node, data: ndarray) -> None:
"""
Node constructor.
Parameters
----------
parent : Node
the parent node, or None for the root node
data : ndarray
the dataset of shape (n_examples, n_features + 1) associated to this node.
The dataset is a matrix with n+1 columns, where the first n columns are
the features vectors, while the last column is the target label vector.
"""
self.parent: Node = parent # parent node
self.depth: int = 1 # deep level or recursion level
self.feature_idx: int # the index of the feature used by the splitting process
self.split: float # the feature value used for the split
self.data = data # the dataset associated to this node
self.left: Node = None # left child node
self.right: Node = None # right child node
# for a classifier, the class label that appears most frequently in the input dataset.
# For a regressor, the mean of the values associated to the leaf node
self.leaf_value = None
if parent is not None:
self.depth = parent.depth + 1
La Superclasse DecisionTree
Ora definiamo la superclasse astratta DecisionTree, che useremo per implementare
il classificatore DecisionTreeClassifier e, più avanti, il regressore DecisionTreeRegressor.
A livello di interfaccia pubblica, DecisionTree fornisce l’implementazione del metodo fit per eseguire il training del modello,
e del metodo predict per fare predizioni su dati di input, successivamente alla fase di training.
Le sottoclassi di DecisionTree dovranno fornire l’implementazione per i metodi protetti _impurity e _leaf_value:
# In Python, abstract base classes require the ABC module ("abstract base class")
# is passed as an argument
class DecisionTree(ABC):
"""
Decision Tree abstract base class that defines functionalities
that are common to both the classification and regression use cases.
Subclasses should provide an implementation for _impurity and _leaf_value methods.
"""
def __init__(self, max_depth: int = None, min_samples_split: int = 2) -> None:
"""
Constructor.
Parameters
----------
max_depth : int
maximum depth the tree can grow.
min_samples_split : int
minimum number of samples required to split a node. Default value is 2.
"""
self.tree = None
self.max_depth = max_depth
self.min_samples_split = min_samples_split
def fit(self, X: np.array, y: np.array) -> DecisionTree:
"""
Train the Decision Tree model.
Parameters
----------
X : ndarray
the training dataset with shape (n_examples, n_features).
y : ndarray
target class label values for the training dataset, with shape (n_examples, ).
"""
# concatenate the dataset with the target class labels (as the latest column vector)
data = np.concatenate((X, y.reshape(-1, 1)), axis=1)
# set the root node of the tree
self.tree = Node(None, data)
# build the tree
self.__build(self.tree)
return self
def predict(self, X: np.array) -> np.array:
"""
Predict the class labels of the input examples.
Parameters
----------
X : ndarray
the dataset with shape (n_examples, n_features)
containing the examples used for the prediction.
Returns
-------
ndarray
an array with shape (n_examples, ) with the predicted class label value
for each dataset's example.
"""
# for each dataset example, traverse the tree to get the predicted value,
# that is, the class label associated to the leaf node
predict = [self.__traverse(self.tree, x)
for x in X]
# return a ndarray with the predictions
return np.array(predict)
@abstractmethod
def _impurity(self, data: np.array) -> None:
"""
Compute the impurity for the input dataset,
that is, the reduction in entropy or surprise of that dataset.
Subclasses should provide an implementation that computes the impurity
using a loss function like Entropy or Gini Impurity.
Parameters
----------
data : ndarray
the training dataset with shape (n_examples, n_features + 1).
The first n columns are the features vectors.
For a DecisionTreeClassifier, the last column is the target labels vector.
For a DecisionTreeRegressor, the last column is the target values vector.
Returns
-------
float
the impurity value for the input dataset.
A value of 0 indicates the dataset is pure,
while a number greater than 0 indicates a certain grade of impurity.
"""
pass
@abstractmethod
def _leaf_value(self, data: np.array) -> None:
"""
Get the Leaf Value associated to the input dataset.
For a Classification Tree, the Leaf Value is the Class Label
that appears most frequently in the dataset.
For a Regression Tree, the Leaf Value is the Mean of the Label values of the dataset.
Parameters
----------
data : ndarray
the training dataset with shape (n_examples, n_features + 1).
The first n columns are the features vectors,
while the last column is the target label vector.
Returns
-------
float
the Leaf Value associated to the input dataset.
"""
pass
La Costruzione dell’Albero
Il metodo ricorsivo __build viene usato per costruire la struttura dell’albero.
Se il nodo passato al metodo non è un nodo foglia, e sussistono le giuste condizioni per la suddivisione
del dataset associato, il codice chiama il metodo __split_data, che genera due nuovi subset, sinistro e destro.
Questi vengono successivamente associati a due nuovi nodi, figli del nodo padre corrente.
Per ognuno di essi viene nuovamente chiamato il metodo __build,
per continuare il processo di costruzione dell’albero, e di suddivisione del dataset corrente.
La ricorsione su un determinato ramo termina quando il subset corrente non può più essere suddiviso,
oppure quando esso è puro, o si è raggiunto il massimo valore di profondità dell’albero:
def __build(self, node: Node) -> None:
"""
Build the Decision Tree.
Starting from the root node, find the decision rule (a feature / split combination)
that best splits the initial multi-class dataset into two sub-datasets with fewer classes,
minimizing the related impurity value.
Then, associate the newly created subsets to a new left and right child node,
and for each child node, apply the same recursive process.
When the split generates a pure dataset (a dataset that only contain examples
of the same class), or when the process reaches the maximum recursion level
allowed for the tree, stop the recursion for that branch, and generate a leaf node
that contains the classification target we are looking for
(the class label that appears most frequently in the leaf node's dataset).
Parameters
----------
node : Node
the root node on which to grow the tree.
"""
n_samples = node.data.shape[0]
can_be_split = (n_samples >= self.min_samples_split)
y = node.data[:, -1]
impure_dataset = np.unique(y).size > 1
max_depth_not_reached = (self.max_depth is None) or (node.depth <= self.max_depth)
# root or internal node
if can_be_split and impure_dataset and max_depth_not_reached:
data_left, data_right = self.__split_data(node)
# stop recursion if the split generates two empty subsets
if (data_left is None or data_left.shape[0] == 0) and
(data_right is None or data_right.shape[0] == 0):
node.leaf_value = self._leaf_value(node.data)
return
# define two new child nodes and apply the same recursive process
# until the split operation generates a pure dataset,
# or until the process reaches the maximum recursion level allowed for this Decision Tree
if data_left.shape[0] > 0:
node.left = Node(node, data_left)
self.__build(node.left)
if data_right.shape[0] > 0:
node.right = Node(node, data_right)
self.__build(node.right)
# leaf node
else:
node.leaf_value = self._leaf_value(node.data)
return
La Suddivisione del Dataset
Il metodo __split_data ha il compito di trovare la regola decisionale migliore,
quella che permette di suddividere il dataset del nodo di input in due subset
aventi un numero minore di classi, rispetto al dataset padre.
Questo processo incrementa la purezza dei subset figli, e quindi incrementa l’Information Gain del modello.
In relazione al calcolo del valore di impurità, rispetto a quanto visto nel capitolo sulla
Massimizzazione dell’Information Gain,
__split_data utilizza un paio di semplificazioni:
i valori di tutte le feature vengono considerati numerici. A livello di dataset di input, i valori delle feature booleane (“True”, “False”) devono essere convertiti nei rispettivi valori numerici ($1$, $0$).
non si tiene conto del valore di impurità del dataset associato al nodo padre, ma solo di quello calcolato sui due subset associati ai nodi figli.
Il metodo usa un doppio ciclo for che considera tutte le feature, e per ogni singola feature,
calcola tutti i valori mediani ad essa associati.
Questi valori vengono poi usati per dividere il dataset padre in due, e calcolare il valore di impurità totale.
Le variabili locali node_feature_idx e node_split tengono traccia della combinazione (feature, valore di split)
che minimizza l’impurità totale per il nodo corrente.
Alla fine del ciclo, il nodo di input conterrà le informazioni relative alla feature prescelta,
e al valore di suddivisione del dataset associato:
def __split_data(self, node):
"""
Find the decision rule (a feature / split combination)
that best splits the input node's dataset into two sub-datasets with fewer classes,
and return them.
Parameters
----------
node : Node
the node having the dataset to split.
Returns
-------
node_data_left : ndarray
the left dataset that includes the examples whose feature value is lower or equal
to the split value.
node_data_right : ndarray
the right dataset that includes the examples whose feature value is higher
than the split value.
"""
node_imp = None
node_feature_idx = None
node_split = None
node_data_left = None
node_data_right = None
n_samples = node.data.shape[0]
n_features = node.data.shape[1] - 1
# iterate through the features
for feature_idx in range(n_features):
# get the current feature vector, and compute the related median values
feature_vector = node.data[:, feature_idx]
split_values = self.__median_values(feature_vector)
# iterate through the unique median values of the current feature vector
for split in split_values:
# use the current (feature, split) combination to define a decision rule
# that splits current node's dataset into two sub-datasets.
# The left dataset includes examples whose feature value is lower or equal
# to the split value. The right dataset includes examples having that feature value
# higher than the split value
data_left = node.data[feature_vector <= split]
data_right = node.data[feature_vector > split]
# calculate the total impurity generated by the current split,
# as the weighted sum of the impurity of the two sub-dataset
w_left = data_left.shape[0] / n_samples
w_right = data_right.shape[0] / n_samples
total_imp = w_left * self._impurity(data_left) + w_right * self._impurity(data_right)
# save the (feature, split) combination
# that minimizes the total impurity value for the current node
if (node_imp is None) or (total_imp < node_imp):
node_imp = total_imp
node_feature_idx = feature_idx
node_split = split
node_data_left = data_left # feature <= split
node_data_right = data_right # feature > split
# set the feature and split data for the current node
node.feature_idx = node_feature_idx
node.split = node_split
return node_data_left, node_data_right
def __median_values(self, X):
"""Compute the unique values for the input data, and sort them in ascending order.
Then, for each pair of (x_i, x_i+1) values, compute the related median value."""
X_u = np.unique(X)
count = len(X_u) - 1
values = np.zeros(count)
for i in range(count):
median = np.median((X_u[i], X_u[i + 1]))
values[i] = median
return values
L’Attraversamento dell’Albero
Il metodo privato __traverse, richiamato da predict, è un metodo ricorsivo
che riceve in input una matrice di esempi, e per ogni esempio $x^{(i)}$, esegue l’attraversamento dell’albero.
L’algoritmo parte dal nodo root, e valuta i dati di $x^{(i)}$ rispetto alla regola decisionale associata al nodo attuale.
Ad esempio, se la regola è “Age $\gt$ 15”, e il vettore feature contiene “Age = 7”,
l’espressione “Age $\gt$ 15” ritornerà False, e l’algoritmo seguirà il nodo figlio sinistro.
Altrimenti, seguirà quello destro.
La ricorsione continua in questo modo fino al raggiungimento del nodo foglia,
contenente il valore di predizione della label di classe per l’osservazione corrente:
def __traverse(self, node: Node, x: np.array) -> int | float:
"""
Pass the input example through the tree,
starting at the root node, and move to either the left or right child node,
depending on how the example's features satisfy the decision rules associated
to each internal node. We navigate through the tree in this manner
until a leaf node is reached. At this point, the class label associated
to the leaf node is provided as the prediction.
Parameters
----------
node : Node
a tree node.
x : ndarray
the example used to traverse the tree, with shape (n_features, ).
Returns
-------
int
the predicted value, that is, the class label associated to the leaf node.
"""
# root / internal node
if node.leaf_value is None:
# apply the decision rule associated to the current node
feature = x[node.feature_idx]
go_left = (feature <= node.split)
# go left or right?
if (go_left):
return self.__traverse(node.left, x)
else:
return self.__traverse(node.right, x)
# leaf node
else:
return node.leaf_value
La Classe DecisionTreeClassifier
Per ultima, implementiamo la classe DecisionTreeClassifier, che estende DecisionTree,
ed implementa i metodi _impurity e _leaf_value.
In particolare, l’implementazione di _leaf_value per il classificatore
ritorna la label della classe che appare più frequentemente nel dataset di input:
class DecisionTreeClassifier(DecisionTree):
"""
Decision Tree Classifier.
Inherits from DecisionTree and provide an implementation
for _impurity and _leaf_value methods.
"""
def __init__(self, max_depth: int = None, min_samples_split: int = 2, criterion: str = "gini"):
"""
Initializer
Parameters
----------
max_depth : int
maximum depth the tree can grow.
min_samples_split : int
minimum number of samples required to split a node. Default value is 2.
criterion : str
the name of the loss function to use during training.
Supported values are "gini" and "entropy".
Default value is "gini".
"""
DecisionTree.__init__(self, max_depth, min_samples_split)
self.criterion = criterion
def _impurity(self, data: np.array) -> float:
# use the selected loss function to calculate the node impurity
if self.criterion == 'gini':
return self.__gini(data)
if self.criterion == 'entropy':
return self.__entropy(data)
def _leaf_value(self, data: np.array) -> int:
"""
Get the Class Label that appears most frequently in the input dataset.
Parameters
----------
data : ndarray
the training dataset with shape (n_examples, n_features + 1).
The first n columns are the features vectors, while the last column
is the target label vector.
Returns
-------
int
the class label that appears most frequently in the input dataset.
"""
# get the target label's column vector
y = data[:, -1]
# get the class that appears most frequently in the target label's column vector
y_mode = stats.mode(y, keepdims=False)[0]
return int(y_mode)
def __gini(self, data: np.array) -> float:
"""
Compute the Gini Impurity loss function for the input dataset.
Parameters
----------
data : ndarray
the training dataset with shape (n_examples, n_features + 1).
The first n columns are the features vectors,
while the last column is the target label vector.
Returns
-------
float
the Gini Impurity value for the input dataset.
"""
gini = 1
y = data[:, -1]
# iterate through the unique classes
for c in np.unique(y):
# Gini(n) = ∑(c=1,C){ p(c|n) (1 − p(c|n)) }
# = 1 - ∑(c=1,C){ p(c|n)^2 }
#
# where p(c|n) is the proportion of the examples that belong to class c
# for a particular node n
data_c = data[y == c]
p = data_c.shape[0] / data.shape[0]
gini -= p**2
return gini
def __entropy(self, data: np.array) -> float:
"""
Compute the Shannon's Entropy loss function for the input dataset.
Parameters
----------
data : ndarray
the training dataset with shape (n_examples, n_features + 1).
The first n columns are the features vectors,
while the last column is the target label vector.
Returns
-------
float
the Entropy value for the input dataset.
"""
entropy = 0
y = data[:, -1]
# iterate through the unique classes
for c in np.unique(y):
# Entropy(n) = -∑(c=1,C){ p(c|n) * log(p(c|n)) }
#
# where p(c|n) is the proportion of the examples that belong to class c
# for a particular node n
data_c = data[y == c]
p = data_c.shape[0] / data.shape[0]
entropy -= p * np.log2(p)
return entropy