diff --git a/experiments/decision_tree/basic_dTree.py b/experiments/decision_tree/basic_dTree.py new file mode 100644 index 0000000..08d950e --- /dev/null +++ b/experiments/decision_tree/basic_dTree.py @@ -0,0 +1,110 @@ +import numpy as np + +''' +Purpose:provide basic implementation of decision tree. +''' + +class Leaf: + def __init__(self, y): + self.value = self._majority_class(y) + + def _majority_class(self, y): + from collections import Counter + return Counter(y).most_common(1)[0][0] + +class Node: + def __init__(self, best_feature, best_threshold, left_child, right_child): + self.best_feature = best_feature + self.best_threshold = best_threshold + self.left_child = left_child + self.right_child = right_child + +class DecisionTree: + def __init__(self): + self.depth = 0 + + def build_tree(self, X, y, depth, max_depth): + if depth >= max_depth or len(set(y)) == 1: + return Leaf(y) + + best_feature, best_threshold = self.find_best_split(X, y) + + if best_feature is None: + return Leaf(y) + + left_data, right_data, y_left, y_right = self.split(X, y, best_feature, best_threshold) + + left_child = self.build_tree(left_data, y_left, depth + 1, max_depth) + right_child = self.build_tree(right_data, y_right, depth + 1, max_depth) + + return Node(best_feature, best_threshold, left_child, right_child) + + def split(self, X, y, b_f, b_th): + left_data = [] + right_data = [] + + y_left = [] + y_right = [] + + for i in range(len(X)): + if X[i][b_f] <= b_th: + left_data.append(X[i]) + y_left.append(y[i]) + else: + right_data.append(X[i]) + y_right.append(y[i]) + + return left_data, right_data, y_left, y_right + + def calculate_gini(self, labels): + _, counts = np.unique(labels, return_counts=True) + probs = counts / counts.sum() + return 1 - np.sum(probs ** 2) + + def find_best_split(self, X, y): + n_samples, n_features = len(X), len(X[0]) + gini_parent = self.calculate_gini(y) + + best_IG = 0 + best_feature = None + best_threshold = None + + X = np.array(X) # ensure numpy + y = np.array(y) + + for f in range(n_features): + values = np.unique(X[:, f]) + thresholds = (values[:-1] + values[1:]) / 2 + + for t in thresholds: + left_idx = X[:, f] < t + right_idx = X[:, f] >= t + + if len(y[left_idx]) == 0 or len(y[right_idx]) == 0: + continue + + gini_left = self.calculate_gini(y[left_idx]) + gini_right = self.calculate_gini(y[right_idx]) + + curr_IG = gini_parent - ( + (len(y[left_idx]) / n_samples) * gini_left + + (len(y[right_idx]) / n_samples) * gini_right + ) + + if curr_IG > best_IG: + best_IG = curr_IG + best_feature = f + best_threshold = t + + return best_feature, best_threshold + + def predict_sample(self, x, node): + # if leaf → return prediction + if isinstance(node, Leaf): + return node.value + + # otherwise follow the split + if x[node.best_feature] <= node.best_threshold: + return self.predict_sample(x, node.left_child) + else: + return self.predict_sample(x, node.right_child) \ No newline at end of file diff --git a/experiments/decision_tree/run_basic_dtree.py b/experiments/decision_tree/run_basic_dtree.py new file mode 100644 index 0000000..98bffcf --- /dev/null +++ b/experiments/decision_tree/run_basic_dtree.py @@ -0,0 +1,25 @@ +from experiments.decision_tree.basic_dTree import DecisionTree + +''' +Purpose: Test the basic implementation of decision tree +''' + +X1 = [[1], [2], [3], [10], [11], [12]] +y1 = [0, 0, 0, 1, 1, 1] + +X2 = [ + [1, 1], [2, 1], [1, 2], + [8, 8], [9, 8], [8, 9] +] +y2 = [0, 0, 0, 1, 1, 1] + +X3 = [[1], [2], [3], [4], [5]] +y3 = [0, 0, 1, 0, 1] + +model = DecisionTree() + +node = model.build_tree(X1, y1, 0, 3) + +pred = model.predict_sample([[6]], node) + +print(pred) \ No newline at end of file diff --git a/glassboxml/models/decision_tree.py b/glassboxml/models/decision_tree.py new file mode 100644 index 0000000..dc9c80f --- /dev/null +++ b/glassboxml/models/decision_tree.py @@ -0,0 +1,106 @@ +import numpy as np + +class Leaf: + def __init__(self, y): + self.value = self._majority_class(y) + + def _majority_class(self, y): + from collections import Counter + return Counter(y).most_common(1)[0][0] + + +class Node: + def __init__(self, best_feature, best_threshold, left_child, right_child): + self.best_feature = best_feature + self.best_threshold = best_threshold + self.left_child = left_child + self.right_child = right_child + +class DecisionTree: + def __init__(self, max_depth): + self.max_depth = max_depth + self.root = None + + def fit(self, X, y): + X = np.array(X) + y = np.array(y) + + self.root = self.build_tree(X,y, 0) + + def build_tree(self, X, y, depth): + # if we either reach maximum depth or all of the elements left are of the same class, then return the leaf + if depth >= self.max_depth or len(set(y))==1: + return Leaf(y) + + best_feature, best_threshold = self.find_best_split(X, y) + + left_data, right_data, y_left, y_right = self.split(X, y, best_feature, best_threshold) + + left_child = self.build_tree(left_data, y_left, depth+1) + right_child = self.build_tree(right_data, y_right, depth+1) + + return Node(best_feature, best_threshold, left_child, right_child) + + def split(self, X, y, b_f, b_th): + left_data = [] + right_data = [] + + y_left = [] + y_right = [] + + for i in range(len(X)): + if X[i][b_f] <= b_th: + left_data.append(X[i]) + y_left.append(y[i]) + else: + right_data.append(X[i]) + y_right.append(y[i]) + + return left_data, right_data, y_left, y_right + + def calculate_gini(self, y): + _, counts = np.unique(y, return_counts=True) + probs = counts / counts.sum() + return 1 - np.sum(probs ** 2) + + def find_best_split(self, X, y): + n_samples, n_features = X.shape + best_IG = 0 + best_feature = 0 + best_threshold = 0 + + gini_parent = self.calculate_gini(y) + + for f in range(n_features): + values = np.unique(X[:, f]) + thresholds = (values[:-1] + values[1:])/2 + for t in thresholds: + left_idx = X[:,f] < t + right_idx = X[:,f] > t + + if (len(y[left_idx]) == 0 or len(y[right_idx]) == 0): + continue + + gini_left = self.calculate_gini(y[left_idx]) + gini_right = self.calculate_gini(y[right_idx]) + + curr_IG = gini_parent - ( + (len(y[left_idx]) / n_samples) * gini_left + + (len(y[right_idx]) / n_samples) * gini_right + ) + + if curr_IG > best_IG: + best_IG = curr_IG + best_feature = f + best_threshold = t + + return best_feature, best_threshold + + def predict_sample(self, x, node): + if isinstance(node, Leaf): + return node.value + + if x[node.best_feature] <= node.best_threshold: + return self.predict_sample(x, node.left_child) + else: + return self.predict_sample(x, node.right_child) \ No newline at end of file diff --git a/tests/models/test_decision_tree.py b/tests/models/test_decision_tree.py new file mode 100644 index 0000000..465f938 --- /dev/null +++ b/tests/models/test_decision_tree.py @@ -0,0 +1,52 @@ +import numpy as np +from glassboxml.models.decision_tree import DecisionTree, Leaf, Node + +def test_tree_perfect_split(): + X = [[1], [2], [3], [10], [11], [12]] + y = [0, 0, 0, 1, 1, 1] + + model = DecisionTree(max_depth=2) + model.fit(X, y) + + preds = [model.predict_sample(x, model.root) for x in X] + + assert preds == y, f"Predictions {preds} do not match labels {y}" + +def test_pure_node_returns_leaf(): + X = [[1], [2], [3]] + y = [1, 1, 1] + + model = DecisionTree(max_depth=3) + model.fit(X, y) + + assert isinstance(model.root, Leaf) + assert model.root.value == 1 + +def test_max_depth_limits_growth(): + X = [[1], [2], [3], [10], [11], [12]] + y = [0, 0, 0, 1, 1, 1] + + model = DecisionTree(max_depth=0) + model.fit(X, y) + + assert isinstance(model.root, Leaf) + +def test_gini_impurity(): + model = DecisionTree(max_depth=1) + + y = np.array([0, 0, 1, 1]) + gini = model.calculate_gini(y) + + assert np.isclose(gini, 0.5), f"Gini {gini} should be 0.5" + +def test_predict_consistency(): + X = [[1], [2], [3], [10], [11], [12]] + y = [0, 0, 0, 1, 1, 1] + + model = DecisionTree(max_depth=2) + model.fit(X, y) + + preds1 = [model.predict_sample(x, model.root) for x in X] + preds2 = [model.predict_sample(x, model.root) for x in X] + + assert preds1 == preds2 \ No newline at end of file