Augmentation/Heart.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Heart experiments\n",
    "\n",
    "This is a Jupytor notebook that reproduces some of the results shown in paper \"Collinear datasets augmentation using Procrustes validation sets\" (in a more simple form).  \n",
    "\n",
    "First we load packages and define ANN discrimination model for Heart experiments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "from torch.utils.data import DataLoader, TensorDataset\n",
    "from prcv.methods import pcvpls\n",
    "from pandas import read_csv\n",
    "\n",
    "# ANN learning parameters\n",
    "N_EPOCH = 300    # number of epochs\n",
    "LR = 0.000001    # learning rate\n",
    "BATCH_SIZE = 10  # batch size\n",
    "\n",
    "device = 'cuda' if torch.cuda.is_available() else 'cpu'\n",
    "\n",
    "class ANNModel(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(ANNModel, self).__init__()\n",
    "        self.fc1 = nn.Linear(in_features=17, out_features=34)\n",
    "        self.fc20 = nn.Linear(in_features=34, out_features=68)\n",
    "        self.fc21 = nn.Linear(in_features=68, out_features=68)\n",
    "        self.fc22 = nn.Linear(in_features=68, out_features=68)\n",
    "        self.fc3 = nn.Linear(in_features=68, out_features=34)\n",
    "        self.fc4 = nn.Linear(in_features=34, out_features=1)\n",
    "        self.fc5 = nn.Sigmoid()\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.fc1(x)\n",
    "        x = nn.functional.relu(x)\n",
    "        x = self.fc20(x)\n",
    "        x = nn.functional.relu(x)\n",
    "        x = self.fc21(x)\n",
    "        x = nn.functional.relu(x)\n",
    "        x = self.fc22(x)\n",
    "        x = nn.functional.relu(x)\n",
    "        x = self.fc3(x)\n",
    "        x = nn.functional.relu(x)\n",
    "        x = self.fc4(x)\n",
    "        x = self.fc5(x)\n",
    "        return x\n",
    "\n",
    "# method for making predictions (returns predicted response values, and accuracy)\n",
    "def predict(model, X, Y):\n",
    "    X_tensor = torch.from_numpy(X).float().to(device)\n",
    "    Yp_tensor = model(X_tensor)\n",
    "    Yp = Yp_tensor.detach().cpu().numpy()\n",
    "    TP = (Yp[:, 0] > 0.5) == (Y[:, 0] > 0.5)\n",
    "    Acc = sum(TP) / Y.shape[0]\n",
    "    return Yp, Acc\n",
    "\n",
    "# method for counting all model parameters\n",
    "def count_parameters(model):\n",
    "    return sum(p.numel() for p in model.parameters() if p.requires_grad)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "14757"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# initialize model and show number of parameters\n",
    "model = ANNModel()\n",
    "count_parameters(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Function which generates `n` PV-sets using PLS based algorithm with `ncomp` components and random splits into `nseg` segments and then train the ANN model based on the augmented data. If `n = 0` model will be trained using the original training set, without augmentation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def doit(Xc, Yc, n, ncomp, nseg):\n",
    "\n",
    "    # initially augmented training set is the same as the original one\n",
    "    Xc_aug = Xc\n",
    "    Yc_aug = Yc\n",
    "\n",
    "    if (n > 0):\n",
    "        # loop for generation of additional data points using PCV for PLS and adding them\n",
    "        # to the training set\n",
    "        cv = {'type': 'rand', 'nseg': nseg}\n",
    "        for i in range(n):\n",
    "            Xpv, D = pcvpls(Xc, Yc, ncomp = ncomp, center = True, scale = False, cv = cv)\n",
    "            Xc_aug = np.concatenate((Xc_aug, Xpv), axis = 0)\n",
    "            Yc_aug = np.concatenate((Yc_aug, Yc), axis = 0)\n",
    "\n",
    "    # turn training set to Torch tensors\n",
    "    Xc_tensor = torch.from_numpy(Xc_aug).float().to(device)\n",
    "    Yc_tensor = torch.from_numpy(Yc_aug).float().to(device)\n",
    "\n",
    "    # create Torch dataset and data loader with given batch size\n",
    "    cal = TensorDataset(Xc_tensor, Yc_tensor)\n",
    "    cal_loader = DataLoader(cal, batch_size = BATCH_SIZE, shuffle = True)\n",
    "\n",
    "    # learning using given number of epochs and learning rate\n",
    "    model = ANNModel().to(device)\n",
    "    criterion = nn.BCELoss()\n",
    "    optimizer = torch.optim.Adam(model.parameters(), lr = LR)\n",
    "    losses = np.zeros(N_EPOCH, dtype = np.float64)\n",
    "    for epoch in range(N_EPOCH):\n",
    "        for i, (inputs, targets) in enumerate(cal_loader):\n",
    "            # zero the gradients\n",
    "            optimizer.zero_grad()\n",
    "\n",
    "            # forward pass\n",
    "            outputs = model(inputs)\n",
    "            loss = criterion(outputs, targets)\n",
    "            losses[epoch] += loss\n",
    "\n",
    "            # backward pass and optimization\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "\n",
    "    return model, losses"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load, split and preprocess dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load data from CSV file (it already contains dummy variables)\n",
    "d = read_csv('Heart.csv', delimiter=\",\", decimal=\".\")\n",
    "n = d.shape[0]\n",
    "\n",
    "# split to X and Y\n",
    "Y = d.iloc[:, 17:18].to_numpy()\n",
    "X = d.iloc[:, 0:17].to_numpy()\n",
    "\n",
    "# split to members and strangers\n",
    "ind0 = Y[:,0] == 0\n",
    "ind1 = Y[:,0] == 1\n",
    "X0 = X[ind0, ]\n",
    "Y0 = Y[ind0, ]\n",
    "X1 = X[ind1, ]\n",
    "Y1 = Y[ind1, ]\n",
    "\n",
    "n0 = X0.shape[0]\n",
    "n1 = X1.shape[0]\n",
    "\n",
    "# compute size of calibration and test set based on 75/25 split\n",
    "n0_cal = round(n0 * 0.75)\n",
    "n0_tst = n0 - n0_cal\n",
    "n1_cal = round(n1 * 0.75)\n",
    "n0_tst = n1 - n1_cal"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we repeat the training/test procedure 10 times, save all Accuracies (this will take some time)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 0.4520547945205479 0.863013698630137\n"
     ]
    }
   ],
   "source": [
    "n = 10\n",
    "ACC1_v = np.zeros(n)\n",
    "ACC2_v = np.zeros(n)\n",
    "\n",
    "for i in range(n):\n",
    "\n",
    "    # split data to training set and calibration set in each class\n",
    "    ind0_all = list(range(n0))\n",
    "    random.shuffle(ind0_all)\n",
    "    ind0_cal = ind0_all[0:n0_cal]\n",
    "    ind0_tst = ind0_all[n0_cal:n0]\n",
    "\n",
    "    ind1_all = list(range(n1))\n",
    "    random.shuffle(ind1_all)\n",
    "    ind1_cal = ind1_all[0:n1_cal]\n",
    "    ind1_tst = ind1_all[n1_cal:n1]\n",
    "\n",
    "    X0c = X0[ind0_cal, ]\n",
    "    Y0c = Y0[ind0_cal, ]\n",
    "    X0t = X0[ind0_tst, ]\n",
    "    Y0t = Y0[ind0_tst, ]\n",
    "\n",
    "    X1c = X1[ind1_cal, ]\n",
    "    Y1c = Y1[ind1_cal, ]\n",
    "    X1t = X1[ind1_tst, ]\n",
    "    Y1t = Y1[ind1_tst, ]\n",
    "\n",
    "    # combine the classes together to form calibration set\n",
    "    Xc = np.vstack((X0c, X1c))\n",
    "    Yc = np.vstack((Y0c, Y1c))\n",
    "\n",
    "    # combine the classes together to form test set\n",
    "    Xt = np.vstack((X0t, X1t))\n",
    "    Yt = np.vstack((Y0t, Y1t))\n",
    "\n",
    "    # compute mean and std for the calibration set\n",
    "    mXc = Xc.mean(axis=0)\n",
    "    sXc = Xc.std(axis=0)\n",
    "\n",
    "    # autoscale both sets using the computed values\n",
    "    Xc = (Xc - mXc) / sXc\n",
    "    Xt = (Xt - mXc) / sXc\n",
    "\n",
    "    # train the models. Model 1 is based on original training set. Model 2 is based on training\n",
    "    # set augmented with 10 PV-sets computed using 10 latent variables and cross-validation\n",
    "    # splits with 10 segments.\n",
    "    model1, losses1 = doit(Xc, Yc,  0,  0,  0)\n",
    "    model2, losses2 = doit(Xc, Yc, 10, 10, 10)\n",
    "\n",
    "    # make predictions for the test set\n",
    "    Yp1, ACC1_v[i] = predict(model1, Xt, Yt)\n",
    "    Yp2, ACC2_v[i] = predict(model2, Xt, Yt)\n",
    "    print(i, ACC1_v[i], ACC2_v[i])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Compare the results visually."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "p = plt.boxplot(np.column_stack((ACC1_v, ACC2_v)), labels = [\"No augmentation\", \"Augmented\"])"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}