random_forest.ipynb

{
 "metadata": {
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.2-final"
  },
  "orig_nbformat": 2,
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.9.2 64-bit ('.venv')",
   "metadata": {
    "interpreter": {
     "hash": "0a11edd73418ac0304368ea2890697e5d158a4fa8829350d01bf2337fc71853d"
    }
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2,
 "cells": [
  {
   "source": [
    "# Random Forest"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import Optional\n",
    "from dataclasses import dataclass\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd"
   ]
  },
  {
   "source": [
    "## Some Helper Functions"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 186,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "0.9182958340544896\n0.4591479170272448\n(array([[ 3, 10],\n       [ 1, 22],\n       [ 2, 28]]), array([[ 5, 32],\n       [ 4, 32]]), array([1, 1, 0]), array([0, 1]))\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "(array([[  6.   , 148.   ,  72.   , ...,  33.6  ,   0.627,  50.   ],\n",
       "        [  1.   ,  85.   ,  66.   , ...,  26.6  ,   0.351,  31.   ],\n",
       "        [  8.   , 183.   ,  64.   , ...,  23.3  ,   0.672,  32.   ],\n",
       "        ...,\n",
       "        [  5.   , 121.   ,  72.   , ...,  26.2  ,   0.245,  30.   ],\n",
       "        [  1.   , 126.   ,  60.   , ...,  30.1  ,   0.349,  47.   ],\n",
       "        [  1.   ,  93.   ,  70.   , ...,  30.4  ,   0.315,  23.   ]]),\n",
       " array([1, 0, 1, 0, 1, 0, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1, 1, 1, 0, 1, 0, 0,\n",
       "        1, 1, 1, 1, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 1,\n",
       "        0, 1, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 1, 0,\n",
       "        1, 0, 0, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0,\n",
       "        1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1,\n",
       "        1, 1, 0, 0, 1, 1, 1, 0, 0, 0, 1, 0, 0, 0, 1, 1, 0, 0, 1, 1, 1, 1,\n",
       "        1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0,\n",
       "        1, 1, 0, 0, 0, 1, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 0, 0, 1,\n",
       "        0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 0, 0, 1, 1, 0, 1, 0, 1,\n",
       "        1, 1, 0, 0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 1, 1, 1, 1, 0, 1, 1,\n",
       "        1, 1, 0, 0, 0, 0, 0, 1, 0, 0, 1, 1, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0,\n",
       "        1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 0, 1, 0, 0,\n",
       "        1, 0, 1, 0, 0, 1, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 1, 1, 0,\n",
       "        0, 1, 0, 0, 0, 1, 1, 1, 0, 0, 1, 0, 1, 0, 1, 1, 0, 1, 0, 0, 1, 0,\n",
       "        1, 1, 0, 0, 1, 0, 1, 0, 0, 1, 0, 1, 0, 1, 1, 1, 0, 0, 1, 0, 1, 0,\n",
       "        0, 0, 1, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0,\n",
       "        0, 0, 0, 1, 1, 1, 0, 1, 1, 0, 0, 1, 0, 0, 1, 0, 0, 1, 1, 0, 0, 0,\n",
       "        0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 1, 0, 0, 1, 0,\n",
       "        0, 1, 0, 1, 1, 0, 1, 0, 1, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 1,\n",
       "        0, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0,\n",
       "        1, 0, 0, 1, 1, 1, 0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 0, 0, 1, 0, 0, 0,\n",
       "        0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0,\n",
       "        1, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0,\n",
       "        1, 0, 0, 0, 1, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0,\n",
       "        0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 1, 1, 1, 0, 0, 1, 1, 0, 0, 0,\n",
       "        0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0,\n",
       "        0, 0, 0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0,\n",
       "        0, 1, 0, 0, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0,\n",
       "        0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0,\n",
       "        1, 0, 0, 0, 1, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1,\n",
       "        0, 1, 1, 1, 1, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 1,\n",
       "        0, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0,\n",
       "        0, 0, 1, 0, 1, 1, 0, 0, 1, 0, 0, 1, 1, 0, 0, 1, 0, 0, 1, 0, 0, 0,\n",
       "        0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 1, 0, 0, 1, 0,\n",
       "        1, 1, 1, 0, 0, 1, 1, 1, 0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 1, 0],\n",
       "       dtype=int64))"
      ]
     },
     "metadata": {},
     "execution_count": 186
    }
   ],
   "source": [
    "def compute_entropy(labels: np.ndarray) -> float:\n",
    "    \"\"\"Computes the entropy of a given set of labels (assumed to be either 0 or 1).\"\"\"\n",
    "\n",
    "    entropy: float = 0.0\n",
    "\n",
    "    # there is zero entropy in an empty set of labels\n",
    "    if labels.shape[0] == 0:\n",
    "        return 0.0\n",
    "\n",
    "    # calculate ratio of true to false labels\n",
    "    true_rate = labels.sum() / labels.shape[0]\n",
    "    false_rate = 1 - true_rate\n",
    "\n",
    "    # return zero entropy if either of these rates are 0\n",
    "    if true_rate == 0 or false_rate == 0:\n",
    "        return 0.0\n",
    "\n",
    "    return -1 * false_rate * np.log2(false_rate) - true_rate * np.log2(true_rate)\n",
    "\n",
    "def compute_information_gain(old_labels: np.ndarray, new_labels_left: np.ndarray, new_labels_right: np.ndarray) -> float:\n",
    "    \"\"\"Computes the information gained from splitting up the labels in old_labels to the two new sets.\"\"\"\n",
    "\n",
    "    information_gain: float = 0.0\n",
    "\n",
    "    # compute old entropy\n",
    "    old_entropy = compute_entropy(old_labels)\n",
    "\n",
    "    # compute new entropy\n",
    "    left_weight = new_labels_left.shape[0] / old_labels.shape[0]\n",
    "    right_weight = new_labels_right.shape[0] / old_labels.shape[0]\n",
    "    new_entropy = (left_weight * compute_entropy(new_labels_left)) + (right_weight * compute_entropy(new_labels_right))\n",
    "\n",
    "    return old_entropy - new_entropy\n",
    "\n",
    "def split_on_attribute(X: np.ndarray, y: np.ndarray, split_attribute_index: int, split_value: float) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:\n",
    "    \"\"\"Splits the data in X and y according to the chosen split attribute index (a column in X) and\n",
    "        the defined split value.\"\"\"\n",
    "    \n",
    "    X_left = []\n",
    "    X_right = []\n",
    "    y_left = []\n",
    "    y_right = []\n",
    "\n",
    "    for i, feature_vector in enumerate(X):\n",
    "        if feature_vector[split_attribute_index] <= split_value:\n",
    "            X_left.append(feature_vector)\n",
    "            y_left.append(y[i])\n",
    "        else:\n",
    "            X_right.append(feature_vector)\n",
    "            y_right.append(y[i])\n",
    "    \n",
    "    return np.array(X_left), np.array(X_right), np.array(y_left), np.array(y_right)\n",
    "\n",
    "def find_best_split(X: np.ndarray, y: np.ndarray) -> tuple[int, float, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:\n",
    "    \"\"\"Finds the best way to split up the current dataset for the highest information gain.\"\"\"\n",
    "\n",
    "    N = X.shape[0]\n",
    "    d = X.shape[1]\n",
    "\n",
    "    highest_information_gain = np.NINF\n",
    "    attribute_index = 0\n",
    "    split_value = 0\n",
    "    X_left = np.array([])\n",
    "    X_right = np.array([])\n",
    "    y_left = np.array([])\n",
    "    y_right = np.array([])\n",
    "\n",
    "    # loop through a random choice of attributes\n",
    "    for new_attribute_index in np.random.choice(np.arange(d), int(d**0.5), replace=False).tolist():\n",
    "\n",
    "        # first calculate the value to split on, which will be the mean in this example\n",
    "        new_split_value = X[:, attribute_index].sum() / N\n",
    "\n",
    "        # build partitions\n",
    "        X_left_tmp, X_right_tmp, y_left_tmp, y_right_tmp = split_on_attribute(X, y, new_attribute_index, new_split_value)\n",
    "\n",
    "        # then find the information gain for this split\n",
    "        new_info_gain = compute_information_gain(y, y_left_tmp, y_right_tmp)\n",
    "\n",
    "        # if this is higher than any previous gain, save off the result\n",
    "        if new_info_gain > highest_information_gain:\n",
    "            highest_information_gain = new_info_gain\n",
    "            attribute_index = new_attribute_index\n",
    "            split_value = new_split_value\n",
    "            X_left = X_left_tmp\n",
    "            X_right = X_right_tmp\n",
    "            y_left = y_left_tmp\n",
    "            y_right = y_right_tmp\n",
    "\n",
    "    return attribute_index, split_value, X_left, X_right, y_left, y_right\n",
    "\n",
    "def is_leaf(y: np.ndarray) -> bool:\n",
    "    \"\"\"Determines if the labels array is for a leaf node.\"\"\"\n",
    "\n",
    "    return np.all(y == y[0]).item()\n",
    "\n",
    "def get_mode(y: np.ndarray) -> int:\n",
    "    \"\"\"Finds the mode of the labels.\"\"\"\n",
    "\n",
    "    # could have also used -> 1 if a.sum() > len(a) // 2 else 0\n",
    "    return max(set(y.tolist()), key=y.tolist().count)\n",
    "\n",
    "def load_data() -> tuple[np.ndarray, np.ndarray]:\n",
    "    df = pd.read_csv(\"./data/pima-diabetes.csv\", names=[\"A\", \"B\", \"C\", \"D\", \"E\", \"F\", \"G\", \"H\", \"label\"], header=None)\n",
    "    X = df[[\"A\", \"B\", \"C\", \"D\", \"E\", \"F\", \"G\", \"H\"]].to_numpy()\n",
    "    y = df[[\"label\"]].to_numpy().reshape(len(df))\n",
    "    return X, y\n",
    "\n",
    "\n",
    "\n",
    "print(compute_entropy(np.array([0,0,0,1,1,1,1,1,1])))\n",
    "print(compute_information_gain(np.array([0,0,0,1,1,1]), np.array([0,0]), np.array([0,1,1,1])))\n",
    "print(split_on_attribute(np.array([[3, 10],[1,22],[2,28],[5,32],[4,32]]), np.array([1,1,0,0,1]), 0, 3))\n",
    "load_data()"
   ]
  },
  {
   "source": [
    "## Define a Decision Tree Class"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass()\n",
    "class Node:\n",
    "    split_attribute_index: Optional[int] = None\n",
    "    split_value: Optional[float] = None\n",
    "    left_branch: Optional[\"Node\"] = None\n",
    "    right_branch: Optional[\"Node\"] = None\n",
    "    leaf_label: Optional[int] = None\n",
    "\n",
    "    def is_leaf_node(self) -> bool:\n",
    "        return self.leaf_label != None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 195,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Correctly classified 520/768 for an accuracy of 0.6770833333333334\n"
     ]
    }
   ],
   "source": [
    "class DecisionTree:\n",
    "    def __init__(self, max_depth: int) -> None:\n",
    "        self.__root_node: Node = Node()\n",
    "        \n",
    "        if max_depth > 0:\n",
    "            self.__max_depth = max_depth\n",
    "        else:\n",
    "            raise Exception(\"tree must have a max depth of greater than zero\")\n",
    "    \n",
    "    def learn(self, X: np.ndarray, y: np.ndarray, current_node: Node = None, depth: int = 1) -> None:\n",
    "        \"\"\"Construct the tree with recursive nodes.\"\"\"\n",
    "\n",
    "        if depth == 1:\n",
    "            # this is the first iteration and there is no parent node, so operate on the root node variable\n",
    "\n",
    "            # find best split in data\n",
    "            split_attribute_index, split_value, X_left, X_right, y_left, y_right = find_best_split(X, y)\n",
    "            # print(X_left.shape, X_right.shape, y_left.shape, y_right.shape)\n",
    "\n",
    "            # assign the split data to the root node\n",
    "            self.__root_node.split_attribute_index = split_attribute_index\n",
    "            self.__root_node.split_value = split_value\n",
    "\n",
    "            # build the left side of the tree recursively\n",
    "            self.__root_node.left_branch = Node()\n",
    "            self.learn(X_left, y_left, self.__root_node.left_branch, depth + 1)\n",
    "\n",
    "            # build the right side of the tree recursively\n",
    "            self.__root_node.right_branch = Node()\n",
    "            self.learn(X_right, y_right, self.__root_node.right_branch, depth + 1)\n",
    "\n",
    "        elif depth < self.__max_depth:\n",
    "            # continue building the tree and populating the current node information\n",
    "\n",
    "            # first check if the labels presented are for a leaf node\n",
    "            if is_leaf(y):\n",
    "                current_node.leaf_label = y[0]\n",
    "            else:\n",
    "                # split the data again\n",
    "                split_attribute_index, split_value, X_left, X_right, y_left, y_right = find_best_split(X, y)\n",
    "\n",
    "                # determine if we have resulted in any empty set of labels on either side\n",
    "                if len(y_left) == 0 or len(y_right) == 0:\n",
    "                    # if this is the case for either of the sides, we know that we don't want to hand off\n",
    "                    #   another recursive call with an empty dataset. In this case, make it a leaf node\n",
    "                    #   with the mode of the labels found on the populated side.\n",
    "                    current_node.leaf_label = get_mode(y)\n",
    "                else:\n",
    "                    # assign split data to the current node\n",
    "                    current_node.split_attribute_index = split_attribute_index\n",
    "                    current_node.split_value = split_value\n",
    "\n",
    "                    # continue building the left side of the tree\n",
    "                    current_node.left_branch = Node()\n",
    "                    self.learn(X_left, y_left, current_node.left_branch, depth + 1)\n",
    "\n",
    "                    # continue building the right side of the tree\n",
    "                    current_node.right_branch = Node()\n",
    "                    self.learn(X_right, y_right, current_node.right_branch, depth + 1)\n",
    "\n",
    "        else:\n",
    "            # depth limit has been reached, do not split anymore and create a leaf node\n",
    "            current_node.leaf_label = get_mode(y)\n",
    "\n",
    "    def classify(self, record: np.ndarray) -> int:\n",
    "        \"\"\"Uses the tree to classify a feature vector.\"\"\"\n",
    "\n",
    "        node = self.__root_node\n",
    "\n",
    "        while not node.is_leaf_node():\n",
    "            # we are still traversing the tree, determine which side to go down\n",
    "            if record[node.split_attribute_index] <= node.split_value:\n",
    "                # go down the left side of the tree\n",
    "                node = node.left_branch\n",
    "            else:\n",
    "                # go down the right side of the tree\n",
    "                node = node.right_branch\n",
    "        \n",
    "        return node.leaf_label\n",
    "\n",
    "\n",
    "X, y = load_data()\n",
    "tree = DecisionTree(10)\n",
    "tree.learn(X, y)\n",
    "\n",
    "correct = 0\n",
    "for i in range(len(X)):\n",
    "    if tree.classify(X[i]) == y[i]:\n",
    "        correct += 1\n",
    "\n",
    "print(f\"Correctly classified {correct}/{len(y)} for an accuracy of {correct/len(y)}\")"
   ]
  },
  {
   "source": [
    "## Grow a Random Forest"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 178,
   "metadata": {},
   "outputs": [],
   "source": [
    "class RandomForest:\n",
    "    def __init__(self, size: int, tree_depth: int) -> None:\n",
    "        self.__size = size\n",
    "        self.__trees = [DecisionTree(tree_depth) for _ in range(self.__size)]\n",
    "        self.__bootstrap_datasets = []\n",
    "        self.__bootstrap_labels = []\n",
    "\n",
    "    def __run_bootstrapping(self, X: np.ndarray, y: np.ndarray, size) -> tuple[np.ndarray, np.ndarray]:\n",
    "        indicies = np.random.randint(len(X), size=size)\n",
    "        return X[indicies], y[indicies]\n",
    "    \n",
    "    def train(self, X: np.ndarray, y: np.ndarray) -> None:\n",
    "        for i in range(self.__size):\n",
    "            # build bootstrap dataset for this tree\n",
    "            X_boot, y_boot = self.__run_bootstrapping(X, y, len(X))\n",
    "            # print(len(X_boot), \" \", len(y_boot))\n",
    "            # print(y_boot)\n",
    "            self.__bootstrap_datasets.append(X_boot)\n",
    "            self.__bootstrap_labels.append(y_boot)\n",
    "\n",
    "            # train the tree\n",
    "            self.__trees[i].learn(X_boot, y_boot)\n",
    "    \n",
    "    def voting(self, X: np.ndarray) -> np.ndarray:\n",
    "        y = []\n",
    "\n",
    "        for record in X:\n",
    "            votes = []\n",
    "\n",
    "            for i, dataset in enumerate(self.__bootstrap_datasets):\n",
    "                if record not in dataset:\n",
    "                    # record vote for out-of-bag tree\n",
    "                    votes.append(self.__trees[i].classify(record))\n",
    "\n",
    "            counts = np.bincount(votes)\n",
    "\n",
    "            if len(counts) == 0:\n",
    "                # this is when record was found in all bootstrap datasets, classify on all trees instead\n",
    "                for tree in self.__trees:\n",
    "                    votes.append(tree.classify(record))\n",
    "            \n",
    "                # reassign counts to this new bincount\n",
    "                counts = np.bincount(votes)\n",
    "\n",
    "            y.append(np.argmax(counts))\n",
    "        \n",
    "        return np.array(y)\n"
   ]
  },
  {
   "source": [
    "## Train the Forest"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 179,
   "metadata": {},
   "outputs": [],
   "source": [
    "np.random.seed(98238923)\n",
    "FOREST_SIZE = 30\n",
    "TREE_DEPTH = 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 197,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Accuracy: 0.651\n"
     ]
    }
   ],
   "source": [
    "X, y = load_data()\n",
    "\n",
    "forest = RandomForest(FOREST_SIZE, TREE_DEPTH)\n",
    "forest.train(X, y)\n",
    "predictions = forest.voting(X)\n",
    "accuracy = np.sum(predictions == y) / len(y)\n",
    "\n",
    "print(f\"Accuracy: {round(accuracy, 4)}\")\n"
   ]
  }
 ]
}