GAN_GRID.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yoU94B49NimG"
      },
      "source": [
        "# Preprocessing"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "kU3ChX6DMh5d"
      },
      "outputs": [],
      "source": [
        "import matplotlib.pyplot as plt\n",
        "import numpy as np\n",
        "import os\n",
        "import pandas as pd\n",
        "import seaborn as sns\n",
        "import time\n",
        "\n",
        "import torch\n",
        "from torch.utils.data import DataLoader\n",
        "import torch.utils.data as data_utils\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5uHnLGRUgLDj"
      },
      "outputs": [],
      "source": [
        "import torch\n",
        "\n",
        "\n",
        "\n",
        "# Device\n",
        "device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
        "\n",
        "# Path of the model (saved/to save)\n",
        "modelFolder = './models/'\n",
        "\n",
        "# When True, retrain the whole model\n",
        "retrain = True\n",
        "\n",
        "\n",
        "# Size of the split\n",
        "trainSize = 0.75\n",
        "valSize = 0.05\n",
        "testSize = 0.20\n",
        "\n",
        "# Specify number of seconds for the window. Default: 16\n",
        "window_size = 16\n",
        "\n",
        "# Model hyper-parameters\n",
        "batch_size = 4\n",
        "learning_rate = 1e-3\n",
        "\n",
        "# Seed for reproducibility\n",
        "seed = 42\n",
        "\n",
        "# Classes to drop in the dataset\n",
        "classes_to_drop=[\n",
        "    'stabf','stab']\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EjXMMgcXgo4G"
      },
      "outputs": [],
      "source": [
        "import numpy as np\n",
        "import os\n",
        "import pandas as pd\n",
        "import random\n",
        "\n",
        "from imblearn.under_sampling import RandomUnderSampler\n",
        "from sklearn import preprocessing\n",
        "from sklearn.metrics import f1_score\n",
        "from torch.utils.data import Dataset\n",
        "\n",
        "import torch\n",
        "import torch.nn as nn\n",
        "\n",
        "\n",
        "\n",
        "def setSeed(seed=seed):\n",
        "    \"\"\"\n",
        "    Setting the seed for reproducibility\n",
        "    \"\"\"\n",
        "    random.seed(seed)\n",
        "    os.environ['PYTHONHASHSEED'] = str(seed)\n",
        "    np.random.seed(seed)\n",
        "    torch.manual_seed(seed)\n",
        "    torch.cuda.manual_seed(seed)\n",
        "    torch.cuda.manual_seed_all(seed)\n",
        "    torch.backends.cudnn.deterministic = True\n",
        "    torch.backends.cudnn.benchmark = True\n",
        "\n",
        "setSeed()\n",
        "\n",
        "def min_max_norm(self,col):\n",
        "    self._data[col]=(self._data[col]-self._data[col].min())/(self._data[col].max()-self._data[col].min())\n",
        "\n",
        "\n",
        "def std_scaler(self,col):\n",
        "    self._data[col]=(self._data[col]-self._data[col].mean())/(self._data[col].std())\n",
        "\n",
        "\n",
        "def f1(test_loader, model):\n",
        "    f1 = 0\n",
        "    with torch.no_grad():\n",
        "        for i, (data, labels) in enumerate(test_loader):\n",
        "            outputs = model(data)\n",
        "            pred = outputs.data.max(1, keepdim=True)[1]\n",
        "            f1 += f1_score(labels, pred, average='macro')\n",
        "    avg_f1 = f1/len(test_loader)\n",
        "    return (avg_f1)\n",
        "\n",
        "\n",
        "class CustomDataset(Dataset):\n",
        "    def __init__(self, file_path='/content/new_dataset.csv', classes_to_drop=classes_to_drop, window_size=window_size, normalize=True, normalize_method='mean_std', auth=False, target=None):\n",
        "\n",
        "        self._window_size=window_size\n",
        "        self._data=pd.read_csv(file_path)\n",
        "\n",
        "        # if auth==True:\n",
        "        #     if target != 'J':\n",
        "        #         self._data = self._data[self._data['stabf'].isin([target, 'J'])]\n",
        "        #     else:\n",
        "        #         self._data = self._data[self._data['stabf'].isin([target, 'I'])]\n",
        "\n",
        "        #     self._data['stabf'] = self._data['stabf'].apply(lambda x: target if x == target else 'Z')\n",
        "        #     self._data['stabf'] = self._data['stabf'].map({target: 1, 'Z': 0}).fillna(0).astype(int)\n",
        "\n",
        "\n",
        "        # # Random Undersampling\n",
        "        # X = self._data.drop('stabf', axis=1)\n",
        "        # y = self._data['stabf']\n",
        "\n",
        "        # # sampler = RandomUnderSampler(sampling_strategy='not minority', random_state=seed)\n",
        "        # # X_resampled, y_resampled = sampler.fit_resample(X, y)\n",
        "\n",
        "        # # X_resampled['Class'] = y_resampled\n",
        "        # self._data = X\n",
        "\n",
        "        # The data is sorted by Class A,B,C the indexes of the dataframe have restarted by ignore index\n",
        "        self._data = self._data.sort_values(by=['stabf'], inplace=False,ignore_index = True)\n",
        "\n",
        "        # class_uniq contains the letters of the drivers A,B and it loops across all of them\n",
        "        for class_uniq in list(self._data['stabf'].unique()):\n",
        "            # Find the total number of elements belonging to a class\n",
        "            tot_number=sum(self._data['stabf']==class_uniq)\n",
        "            # Number of elements to drop so that the class element is divisible by window size\n",
        "            to_drop=tot_number%window_size\n",
        "            # Returns the index of the first element of the class\n",
        "            index_to_start_removing=self._data[self._data['stabf']==class_uniq].index[0]\n",
        "            # Drop element from first element to the element required\n",
        "            self._data.drop(self._data.index[index_to_start_removing:index_to_start_removing+to_drop],inplace=True)\n",
        "\n",
        "\n",
        "        # Resetting index of dataframe after dropping values\n",
        "        self._data = self._data.reset_index()\n",
        "        self._data = self._data.drop(['index'], axis=1)\n",
        "\n",
        "        index_starting_class=[] # This array contains the starting index of each class in the df\n",
        "        for class_uniq in list(self._data['stabf'].unique()):\n",
        "            # Appending the index of first element of each clas\n",
        "            index_starting_class.append(self._data[self._data['stabf']==class_uniq].index[0])\n",
        "\n",
        "        # Create the sequence of indexs of the windows\n",
        "        sequences=[]\n",
        "        for i in range(len(index_starting_class)):\n",
        "            # Check if beginning of next class is there\n",
        "            if i!=len(index_starting_class)-1:\n",
        "                ranges=np.arange(index_starting_class[i], index_starting_class[i+1])\n",
        "            else:\n",
        "                ranges = np.arange(index_starting_class[i], len(self._data))\n",
        "            for j in range(0,len(ranges),int(self._window_size/2)):\n",
        "                if len(ranges[j:j+self._window_size])==16:\n",
        "                    sequences.append(ranges[j:j+self._window_size])\n",
        "        self._sequences=sequences\n",
        "\n",
        "\n",
        "        # Take only the 'Class' which are the actual labels and store it in the labels of self\n",
        "        self._labels=self._data['stabf']\n",
        "        # Dropping columns which have constant measurements because they would return nan in std\n",
        "        self._data.drop(classes_to_drop, inplace=True, axis=1)\n",
        "\n",
        "        # Function to normalize the data either with min_max or mean_std\n",
        "        if normalize and not auth:\n",
        "            for col in self._data.columns:\n",
        "                if normalize_method=='min_max':\n",
        "                    min_max_norm(self,col)\n",
        "                elif normalize_method==\"mean_std\":\n",
        "                    std_scaler(self,col)\n",
        "\n",
        "        # Create the array holding the windowed multidimensional arrays\n",
        "        X=np.empty((len(sequences), self._window_size, len(self._data.columns)))\n",
        "        y=[]\n",
        "\n",
        "        for n_row, sequence in enumerate(sequences):\n",
        "            X[n_row,:,:]=self._data.iloc[sequence]\n",
        "            # The corresponding driver of the sequence is the driver at first sequence\n",
        "            y.append(self._labels[sequence[0]])\n",
        "\n",
        "        assert len(y)==len(X)\n",
        "        # Assign the windowed dataset to the X of self\n",
        "        self._X= X\n",
        "\n",
        "        # Targets is a transformed version of y with drivers are encoded into 0 to 9\n",
        "        targets = preprocessing.LabelEncoder().fit_transform(y)\n",
        "        class_labels = encoder.classes_\n",
        "        for code, label in enumerate(class_labels):\n",
        "          print(f'Code: {code} -> Label: {label}')\n",
        "        targets = torch.as_tensor(targets)  # Just converting it to a pytorch tensor\n",
        "        self._y=targets # Assign it to y of self\n",
        "\n",
        "\n",
        "    def __len__(self):\n",
        "        return len(self._X)\n",
        "\n",
        "\n",
        "    def __getitem__(self, index):\n",
        "        return torch.FloatTensor(self._X[index,:,:]), self._y[index]\n",
        "\n",
        "\n",
        "def evaluate(model, dataloader, criterion):\n",
        "    model.eval()\n",
        "    running_loss = 0.0\n",
        "    running_corrects = 0\n",
        "    y_true = []\n",
        "    y_pred = []\n",
        "\n",
        "    for inputs, labels in dataloader:\n",
        "        inputs = inputs.to(device)\n",
        "        labels = labels.to(device)\n",
        "\n",
        "        inputs = inputs\n",
        "        labels = labels\n",
        "\n",
        "        # Forward pass\n",
        "        with torch.no_grad():\n",
        "            outputs = model(inputs)\n",
        "            loss = criterion(outputs, labels)\n",
        "\n",
        "        _, preds = torch.max(outputs, 1)\n",
        "        running_loss += loss.item() * inputs.size(0)\n",
        "        running_corrects += torch.sum(preds == labels.data)\n",
        "\n",
        "        # Collect predictions and true labels\n",
        "        y_true += labels.data.cpu().numpy().tolist()\n",
        "        y_pred += preds.cpu().numpy().tolist()\n",
        "\n",
        "    # Calculate accuracy and loss\n",
        "    epoch_loss = running_loss / len(dataloader.dataset)\n",
        "    epoch_acc = running_corrects.double() / len(dataloader.dataset)\n",
        "    epoch_f1 = f1_score(y_true, y_pred, average='macro')\n",
        "\n",
        "    return epoch_loss, epoch_acc, epoch_f1\n",
        "\n",
        "\n",
        "def evaluateBinary(model, dataloader, criterion):\n",
        "    model.eval()\n",
        "    running_loss = 0.0\n",
        "    running_corrects = 0\n",
        "    y_true = []\n",
        "    y_pred = []\n",
        "\n",
        "    for inputs, labels in dataloader:\n",
        "        inputs = inputs.to(device)\n",
        "        labels = labels.to(device)\n",
        "\n",
        "        # Forward pass\n",
        "        with torch.no_grad():\n",
        "            outputs = model(inputs)\n",
        "            # loss = criterion(outputs, labels)\n",
        "            loss = criterion(outputs.squeeze(), labels.float())\n",
        "\n",
        "        _, preds = torch.max(outputs, 1)\n",
        "        # preds = (outputs > 0.5).float()\n",
        "        running_loss += loss.item() * inputs.size(0)\n",
        "        running_corrects += torch.sum(preds == labels.data)\n",
        "\n",
        "        # Collect predictions and true labels\n",
        "        y_true += labels.data.cpu().numpy().tolist()\n",
        "        y_pred += preds.cpu().numpy().tolist()\n",
        "\n",
        "    # Calculate accuracy and loss\n",
        "    epoch_loss = running_loss / len(dataloader.dataset)\n",
        "    epoch_acc = running_corrects.double() / len(dataloader.dataset)\n",
        "    epoch_f1 = f1_score(y_true, y_pred, average='macro')\n",
        "\n",
        "    return epoch_loss, epoch_acc, epoch_f1\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "WCTrkBiQnHgG"
      },
      "outputs": [],
      "source": [
        "dataset_path = '/content/smart_grid_stability_augmented.csv'\n",
        "df = pd.read_csv(dataset_path)\n",
        "df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rw0YcgjAPj9_"
      },
      "outputs": [],
      "source": [
        "from sklearn.preprocessing import LabelEncoder\n",
        "\n",
        "encoder = LabelEncoder()\n",
        "df['stabf'] = encoder.fit_transform(df['stabf'])\n",
        "\n",
        "# Retrieve the mapping of numerical codes to original class labels\n",
        "class_labels = encoder.classes_\n",
        "\n",
        "# Display the mapping\n",
        "for code, label in enumerate(class_labels):\n",
        "    print(f'Code: {code} -> Label: {label}')\n",
        "df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OkuqT40NhHQM"
      },
      "outputs": [],
      "source": [
        "df.to_csv('new_dataset.csv', index=False)\n",
        "df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "o723dQMYnDGf"
      },
      "outputs": [],
      "source": [
        "import torch\n",
        "import numpy as np\n",
        "np.random.seed(seed)\n",
        "torch.manual_seed(seed)\n",
        "torch.cuda.manual_seed(seed)\n",
        "torch.cuda.manual_seed_all(seed)\n",
        "torch.backends.cudnn.deterministic = True\n",
        "torch.backends.cudnn.benchmark = True\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yEMcnGxzg2Bu"
      },
      "outputs": [],
      "source": [
        "a = CustomDataset()\n",
        "\n",
        "# Defining sizes\n",
        "train_size = int(trainSize * len(a))\n",
        "val_size = int(valSize * len(a))\n",
        "test_size = len(a)-train_size-val_size\n",
        "\n",
        "train_dataset, val_dataset, test_dataset = torch.utils.data.random_split(\n",
        "    a, [train_size, val_size, test_size])\n",
        "\n",
        "\n",
        "train_loader = torch.utils.data.DataLoader(dataset=train_dataset,\n",
        "                                           batch_size=batch_size,\n",
        "                                           shuffle=False,\n",
        "                                           drop_last=True)\n",
        "\n",
        "validation_loader = torch.utils.data.DataLoader(dataset=val_dataset,\n",
        "                                                batch_size=batch_size,\n",
        "                                                shuffle=False,\n",
        "                                                drop_last=True)\n",
        "\n",
        "test_loader = torch.utils.data.DataLoader(dataset=test_dataset,\n",
        "                                          batch_size=batch_size,\n",
        "                                          shuffle=False,\n",
        "                                          drop_last=True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4j3I39FGNndj"
      },
      "source": [
        "# LSTM model"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "HJpVgWGOJeaG"
      },
      "outputs": [],
      "source": [
        "import torch.optim as optim\n",
        "\n",
        "class RNNBinaryClassification(torch.nn.Module):\n",
        "    def __init__(self, batch_size, window_size, num_features, dropout_rate=0.5):\n",
        "        super(RNNBinaryClassification, self).__init__()\n",
        "        self.rnn1 = torch.nn.LSTM(num_features, 220, batch_first=True, bidirectional=True)\n",
        "        self.dropout = torch.nn.Dropout(p=dropout_rate)\n",
        "        self.fc = torch.nn.Linear(440, 1)  # Output size is 1 for binary classification\n",
        "        self.sigmoid = torch.nn.Sigmoid()  # Sigmoid activation for binary classification\n",
        "\n",
        "    def forward(self, x):\n",
        "        rnn1_out, _ = self.rnn1(x)\n",
        "        rnn1_out = self.dropout(rnn1_out[:, -1, :])\n",
        "        fc_out = self.fc(rnn1_out)\n",
        "        out = self.sigmoid(fc_out)\n",
        "        return out\n",
        "\n",
        "model = RNNBinaryClassification(batch_size, window_size, 12).to(device)\n",
        "criterion = nn.BCELoss()\n",
        "\n",
        "optimizer = optim.Adam(model.parameters(), lr=learning_rate)\n",
        "\n",
        "retrain = False\n",
        "\n",
        "if not os.path.exists('./models/rnn_auth.pt') or retrain:\n",
        "    # Training loop\n",
        "    for epoch in range(10):\n",
        "        model.train()\n",
        "        total_loss = 0.0\n",
        "\n",
        "        for inputs, labels in train_loader:\n",
        "            inputs, labels = inputs.to(device), labels.to(device)\n",
        "\n",
        "            optimizer.zero_grad()\n",
        "            outputs = model(inputs)\n",
        "            loss = criterion(outputs.squeeze(), labels.float())\n",
        "            loss.backward()\n",
        "            optimizer.step()\n",
        "\n",
        "            total_loss += loss.item()\n",
        "\n",
        "        average_loss = total_loss / len(train_loader)\n",
        "\n",
        "        print(f'[💪 EPOCH {epoch+1}/{10}] Loss: {average_loss:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "H6LUv4Fp0eWd"
      },
      "outputs": [],
      "source": [
        "# Calculate accuracy\n",
        "correct_predictions = 0\n",
        "total_samples = 0\n",
        "\n",
        "y_true = []\n",
        "y_pred = []\n",
        "\n",
        "for inputs, labels in test_loader:\n",
        "    inputs, labels = inputs.to(device), labels.to(device)\n",
        "    outputs = model(inputs)\n",
        "    predictions = (outputs > 0.5).float()\n",
        "\n",
        "    for p, l in zip(predictions, labels.float()):\n",
        "        if p == l:\n",
        "            correct_predictions += 1\n",
        "\n",
        "    total_samples += labels.size(0)\n",
        "\n",
        "    y_true.extend(labels.cpu().numpy())\n",
        "    y_pred.extend(predictions.cpu().numpy())\n",
        "\n",
        "acc = correct_predictions/total_samples\n",
        "f1 = f1_score(y_true, y_pred, average='binary')\n",
        "\n",
        "print('[👑 TEST GRU AUTH]\\n')\n",
        "print(f'[🎯 ACCURACY] {acc:.3f}')\n",
        "print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xUtiTWBKNqMb"
      },
      "source": [
        "# GAN-GRID against LSTM"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Qv-QwhKMv4Wy"
      },
      "outputs": [],
      "source": [
        "class Generator(nn.Module):\n",
        "    def __init__(self, batch_size, window_size, num_features,):\n",
        "        super(Generator, self).__init__()\n",
        "        self.batch_size = batch_size\n",
        "        self.num_features = num_features\n",
        "        self.window_size = window_size\n",
        "        self.layer1 = nn.Linear(num_features, 128)\n",
        "        self.layer2 = nn.Linear(128, 256)\n",
        "        self.layer3 = nn.Linear(256, 512)\n",
        "        self.layer4 = nn.Linear(512, batch_size*window_size)\n",
        "        self.layer5 = nn.Linear(batch_size*window_size, num_features)\n",
        "\n",
        "        self.leaky_relu = nn.LeakyReLU(0.2)\n",
        "\n",
        "    def forward(self, x):\n",
        "        x = self.leaky_relu(self.layer1(x))\n",
        "        x = self.leaky_relu(self.layer2(x))\n",
        "        x = self.leaky_relu(self.layer3(x))\n",
        "        x = self.leaky_relu(self.layer4(x))\n",
        "        x = self.layer5(x)\n",
        "        return x"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "y2pw81y319rL"
      },
      "outputs": [],
      "source": [
        "def train_gan(generator, surrogate, label, train_loader, num_epochs=100, lr=0.001, device=torch.device('cpu'), ml=False, num_episodes=150):\n",
        "\n",
        "    losses = []\n",
        "\n",
        "    if not ml:\n",
        "        generator = generator.to(device)\n",
        "        surrogate = surrogate.to(device)\n",
        "\n",
        "        # for model in surrogate.models:\n",
        "        #     model.to(device)\n",
        "        #     model.train()\n",
        "\n",
        "    # Define the loss function and optimizer\n",
        "    binary_cross_entropy_loss = nn.BCEWithLogitsLoss()\n",
        "    generator_optimizer = torch.optim.Adam(generator.parameters(), lr=lr)\n",
        "\n",
        "    # Define the reinforcement learning parameters\n",
        "    max_episode_length = 10\n",
        "    alpha = 0.1\n",
        "    gamma = 0.9\n",
        "\n",
        "    for episode in range(num_episodes):\n",
        "        # Initialize the latent input and the episode reward\n",
        "        latent_input = torch.randn(4, 16, 12).to(device)\n",
        "        episode_reward = 0\n",
        "\n",
        "        for step in range(max_episode_length):\n",
        "            # Generate a sample with the current latent input\n",
        "            fake_input = generator(latent_input)\n",
        "\n",
        "            # Evaluate the sample with the surrogate model\n",
        "            if not ml:\n",
        "                surrogate_output = surrogate(fake_input)\n",
        "            else:\n",
        "                surrogate_output = []\n",
        "                # Looping through each\n",
        "                for group in fake_input:\n",
        "                    # Flatten the group to make it\n",
        "                    flat_group = group.view(-1, group.size(-1)).detach().numpy()\n",
        "                    # Get the probabilities\n",
        "                    probabilities = surrogate.predict_proba(flat_group)\n",
        "                    mean_probabilities = np.mean(probabilities, axis=0)\n",
        "                    # Append the probabilities to the array\n",
        "                    surrogate_output.append(mean_probabilities)\n",
        "                with torch.no_grad():\n",
        "                    surrogate_output = torch.tensor(surrogate_output, requires_grad=True)\n",
        "\n",
        "            predictions = (surrogate_output > 0.5).float()\n",
        "            targets = torch.randint_like(predictions, 0, 2)\n",
        "            reward = (predictions == targets).float().mean().item()\n",
        "            episode_reward += reward\n",
        "\n",
        "            # Update the latent input using reinforcement learning\n",
        "            td_error = reward - episode_reward\n",
        "            latent_input += alpha * td_error * gamma**step * torch.randn_like(latent_input)\n",
        "\n",
        "        # Update the generator using the final latent input of the episode\n",
        "        generator_optimizer.zero_grad()\n",
        "        fake_input = generator(latent_input)\n",
        "\n",
        "        if not ml:\n",
        "            surrogate_output = surrogate(fake_input)\n",
        "        else:\n",
        "            surrogate_output = []\n",
        "            # Looping through\n",
        "            for group in fake_input:\n",
        "                flat_group = group.view(-1, group.size(-1)).detach().numpy()\n",
        "                # Get the probabilities\n",
        "                probabilities = surrogate.predict_proba(flat_group)\n",
        "                mean_probabilities = np.mean(probabilities, axis=0)\n",
        "                # Append the probabilities to the array\n",
        "                surrogate_output.append(mean_probabilities)\n",
        "            with torch.no_grad():\n",
        "                surrogate_output = torch.tensor(surrogate_output, requires_grad=True)\n",
        "\n",
        "        target_labels = targets.view(-1, 1).float()\n",
        "\n",
        "        generator_loss = binary_cross_entropy_loss(surrogate_output, target_labels)\n",
        "\n",
        "        if ml:\n",
        "            generator_optimizer.zero_grad()\n",
        "\n",
        "        generator_loss.backward()\n",
        "        generator_optimizer.step()\n",
        "\n",
        "        losses.append(generator_loss.item())\n",
        "\n",
        "        if episode % 10 == 0:\n",
        "            print(f'[⏭️ EP {episode}/{num_episodes} | D{label}] LOSS: {round(generator_loss.item(), 3)}')\n",
        "\n",
        "    print()\n",
        "\n",
        "    return generator, losses"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oo-s6b5uRAs8"
      },
      "source": [
        "Restart the process if the gennrator did not converge in the first try"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "dHQbv_E5kjU0"
      },
      "outputs": [],
      "source": [
        "lr = 3e-3\n",
        "\n",
        "generators = []\n",
        "losses = []\n",
        "\n",
        "inputs, classes = next(iter(train_loader))\n",
        "\n",
        "for d in range(1):\n",
        "        print(f'[🤖 GENERATORS] Label {d}')\n",
        "\n",
        "        batch_size, window_size, num_features = inputs.shape\n",
        "        generator = Generator(batch_size, window_size, num_features)\n",
        "        surrogate_model = model\n",
        "\n",
        "        generator, loss = train_gan(generator, surrogate_model, train_loader=train_loader, num_epochs=20, lr=lr, label=0, ml=False, num_episodes=100)\n",
        "        print()\n",
        "\n",
        "        generators.append(generator)\n",
        "        losses.append(loss)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q8CkqAieLtM1"
      },
      "outputs": [],
      "source": [
        "results = []\n",
        "\n",
        "threshold = 0.5\n",
        "\n",
        "for i in range(1):\n",
        "    predicted_labels = []\n",
        "    generator = generators[i].to(device)\n",
        "\n",
        "    for batch in test_loader:\n",
        "        input_batch, true_labels = batch[0].to(device), batch[1].to(device)\n",
        "        # Generate data\n",
        "        generated_data = generator(torch.randn(4, 16, 12).to(device))\n",
        "        # generated_data = generated_data * ones_tensor\n",
        "\n",
        "        # Add the result to the ones tensor\n",
        "        final_result =  generated_data\n",
        "\n",
        "        # Get the surrogate outputs for each sample in the generated data\n",
        "        surrogate_outputs = model(final_result)\n",
        "\n",
        "        # Apply the threshold for binary classification\n",
        "        predicted_labels_batch = (surrogate_outputs > threshold).float()\n",
        "\n",
        "        # Append the predicted labels to the lists\n",
        "        predicted_labels.extend(predicted_labels_batch.squeeze().tolist())  # Squeeze the tensor\n",
        "\n",
        "    asr = predicted_labels.count(i) / len(predicted_labels)\n",
        "    results.append(asr)\n",
        "    print(f'ASR: {round(asr, 3)}')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Nto0iW_bNy78"
      },
      "source": [
        "#Random noise attack"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "z8ySntjP-X_0"
      },
      "outputs": [],
      "source": [
        "from sklearn.metrics import accuracy_score\n",
        "class WhiteBoxAttack:\n",
        "    def __init__(self, model, epsilon=0.5, num_samples=50):\n",
        "        self.model = model\n",
        "        self.epsilon = epsilon\n",
        "        self.num_samples = num_samples\n",
        "\n",
        "    def attack(self, data_loader):\n",
        "        self.model.eval()  # Set the model to evaluation mode\n",
        "\n",
        "        for inputs, labels in data_loader:\n",
        "            inputs_adv = inputs.clone().detach().to(device)\n",
        "\n",
        "            for _ in range(self.num_samples):\n",
        "                perturbation = torch.randn_like(inputs_adv) * self.epsilon\n",
        "                inputs_perturbed = inputs_adv + perturbation\n",
        "\n",
        "                with torch.no_grad():\n",
        "                    outputs_perturbed = self.model(inputs_perturbed)\n",
        "                    y_pred_perturbed = torch.round(outputs_perturbed).squeeze().cpu().numpy()\n",
        "                    y_pred_orig = torch.round(self.model(inputs_adv)).squeeze().cpu().numpy()\n",
        "\n",
        "                for i in range(len(inputs_adv)):\n",
        "                    if accuracy_score([labels[i].item()], [y_pred_perturbed[i]]) <= accuracy_score([labels[i].item()], [y_pred_orig[i]]):\n",
        "                        inputs_adv[i] = inputs_perturbed[i]\n",
        "\n",
        "            yield inputs_adv\n",
        "attack = WhiteBoxAttack(model, epsilon=0.5, num_samples=50)\n",
        "\n",
        "# Generate adversarial examples\n",
        "X_test_adv_loader = attack.attack(test_loader)\n",
        "\n",
        "with torch.no_grad():\n",
        "    y_true = []\n",
        "    y_pred_orig = []\n",
        "    y_pred_adv = []\n",
        "    for inputs, labels in test_loader:\n",
        "        X_test_tensor = inputs.to(device)\n",
        "        y_true.extend(labels.cpu().numpy())\n",
        "        X_test_adv = next(X_test_adv_loader)  # Get adversarial examples from the generator\n",
        "        y_pred_orig.extend(torch.round(model(inputs)).squeeze().cpu().numpy())\n",
        "        y_pred_adv.extend(torch.round(model(X_test_adv.to(device))).squeeze().cpu().numpy())\n",
        "\n",
        "accuracy_orig = accuracy_score(y_true, y_pred_orig)\n",
        "accuracy_adv = accuracy_score(y_true, y_pred_adv)\n",
        "\n",
        "print(\"Accuracy on original test examples:\", accuracy_orig)\n",
        "print(\"Accuracy on adversarial test examples:\", accuracy_adv)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MJLD3mW2N6Hr"
      },
      "source": [
        "# FGSM,BIM,PGD"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lNWbJJ1zpDPk"
      },
      "outputs": [],
      "source": [
        "pip install adversarial-robustness-toolbox\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "D50eyzu3ZNog"
      },
      "outputs": [],
      "source": [
        "from art.attacks.evasion import FastGradientMethod, ProjectedGradientDescentPyTorch, ProjectedGradientDescentNumpy, CarliniLInfMethod, CarliniWagnerASR,UniversalPerturbation\n",
        "from art.estimators.classification import PyTorchClassifier\n",
        "from art.attacks.evasion.iterative_method import BasicIterativeMethod\n",
        "from sklearn.metrics import accuracy_score, f1_score\n",
        "\n",
        "classifier = PyTorchClassifier(\n",
        "    model=model,\n",
        "    loss=criterion,\n",
        "    input_shape=(12,),\n",
        "    nb_classes=2,\n",
        "    device_type=\"cpu\"\n",
        "    # device_type=\"gpu\" if torch.cuda.is_available() else \"cpu\"\n",
        ")\n",
        "\n",
        "# Define FGSM, BIM, and CW attacks\n",
        "fgsm_attack = FastGradientMethod(estimator=classifier, eps=0.5)\n",
        "bim_attack = BasicIterativeMethod(estimator=classifier, eps=0.5)\n",
        "pgd_attack = ProjectedGradientDescentNumpy(estimator=classifier, eps=0.5)\n",
        "def evaluate_clean_data(model, test_loader):\n",
        "    model.eval()\n",
        "    correct_predictions = 0\n",
        "    total_samples = 0\n",
        "\n",
        "    y_true = []\n",
        "    y_pred = []\n",
        "\n",
        "    with torch.no_grad():\n",
        "        for inputs, labels in test_loader:\n",
        "            inputs, labels = inputs.to(device), labels.to(device)\n",
        "            outputs = model(inputs)\n",
        "            predictions = (outputs > 0.5).float()\n",
        "\n",
        "            correct_predictions += (predictions == labels).sum().item()\n",
        "            total_samples += labels.size(0)\n",
        "\n",
        "            y_true.extend(labels.cpu().numpy())\n",
        "            y_pred.extend(predictions.cpu().numpy())\n",
        "\n",
        "    acc = accuracy_score(y_true,y_pred)\n",
        "    f1 = f1_score(y_true, y_pred, average='binary')\n",
        "\n",
        "    print('[👑 EVALUATION ON CLEAN DATA]\\n')\n",
        "    print(f'[🎯 ACCURACY] {acc:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')\n",
        "\n",
        "def evaluate_attack(model, test_loader, attack):\n",
        "    model.eval()\n",
        "    y_true = []\n",
        "    y_pred = []\n",
        "\n",
        "    for inputs, labels in test_loader:\n",
        "        inputs, labels = inputs.to(device), labels.to(device)  # Move inputs and labels to devic\n",
        "        adv_inputs = attack.generate(x=inputs.cpu().numpy())\n",
        "        adv_inputs = torch.tensor(adv_inputs).to(device)\n",
        "        outputs = model(adv_inputs)\n",
        "        predictions = (outputs > 0.5).float()\n",
        "\n",
        "        y_true.extend(labels.cpu().numpy())\n",
        "        y_pred.extend(predictions.cpu().numpy())\n",
        "\n",
        "    acc = accuracy_score(y_true, y_pred)\n",
        "    f1 = f1_score(y_true, y_pred, average='binary')\n",
        "\n",
        "    print('[👑 EVALUATION UNDER ATTACK]\\n')\n",
        "    print(f'[🎯 ACCURACY] {acc:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')\n",
        "\n",
        "# Evaluate on clean data\n",
        "evaluate_clean_data(model, test_loader)\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "aaRuqQS_o_GM"
      },
      "outputs": [],
      "source": [
        "# Evaluate under FGSM attack\n",
        "evaluate_attack(model, test_loader, fgsm_attack)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "A6OL8bdlo5YT"
      },
      "outputs": [],
      "source": [
        "# Evaluate under BIM attack\n",
        "evaluate_attack(model, test_loader, bim_attack)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "KUliZSepO2Of"
      },
      "outputs": [],
      "source": [
        "evaluate_attack(model, test_loader, pgd_attack)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q_dazFu52EEM"
      },
      "outputs": [],
      "source": [
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# Given ASR values for two sets\n",
        "asr_values_set1 = [0.999,0.995, 0.981, 0.866, 0.729, 0.563, 0.457, 0.404, 0.387, 0.383, 0.383]\n",
        "asr_values_set2 = [0.999,0.995, 0.972, 0.856, 0.714,0.537, 0.454,  0.397, 0.384, 0.383, 0.382]\n",
        "asr_values_set3 = [0.999,0.995, 0.971, 0.859, 0.714, 0.537, 0.442, 0.397, 0.383, 0.383, 0.382]\n",
        "asr_values_set4 = [0.999,0.996, 0.987, 0.94, 0.884, 0.826, 0.755, 0.67, 0.634, 0.568, 0.497]\n",
        "\n",
        "\n",
        "# Create a list of feature numbers starting from one\n",
        "feature_numbers = [0,0.05,0.1,0.15,0.20,0.25,0.30,0.35,0.40,0.45,0.50]\n",
        "\n",
        "# Plotting the graphs\n",
        "plt.ylim(0, 1.1)\n",
        "plt.plot(feature_numbers, asr_values_set1, marker='o', linestyle='-', color='b', label='fgsm')\n",
        "plt.plot(feature_numbers, asr_values_set2, marker='s', linestyle='-', color='r', label='bim')\n",
        "plt.plot(feature_numbers, asr_values_set3, marker='^', linestyle='-', color='g', label='pgd')\n",
        "plt.plot(feature_numbers, asr_values_set4, marker='^', linestyle='-', color='y', label='random_noise') # Different color for set3\n",
        "# Adding legend\n",
        "plt.legend()\n",
        "\n",
        "plt.xlabel('epsilon')\n",
        "plt.ylabel('Accuracy')\n",
        "plt.title('Accuracy vs epsilon')\n",
        "plt.grid(True)\n",
        "plt.tight_layout()\n",
        "plt.savefig('/content/model_WB_LSTM.pdf', format='pdf')\n",
        "plt.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M_woErAvN_H0"
      },
      "source": [
        "# SHAP for LSTM"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "r7e2q3RYIxIN"
      },
      "outputs": [],
      "source": [
        "!pip install shap"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "t6E-WXDfIhoD"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "import shap\n",
        "import torchvision.transforms as transforms\n",
        "\n",
        "# Function to convert DataLoader to numpy array\n",
        "def loader_to_numpy(loader):\n",
        "    all_data = []\n",
        "    all_labels = []\n",
        "    for data, labels in loader:\n",
        "        all_data.append(data)\n",
        "        all_labels.append(labels)\n",
        "    return torch.cat(all_data).numpy(), torch.cat(all_labels).numpy()\n",
        "\n",
        "# Convert train_loader and test_loader to numpy arrays\n",
        "X_train_array, y_train_array = loader_to_numpy(train_loader)\n",
        "X_test_array, y_test_array = loader_to_numpy(test_loader)\n",
        "\n",
        "# Create a SHAP explainer object\n",
        "explainer = shap.GradientExplainer(model, torch.from_numpy(X_train_array))\n",
        "\n",
        "# Calculate the SHAP values for the NumPy array\n",
        "shap_values = explainer.shap_values(torch.from_numpy(X_test_array))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "qELKCKAuI6iG"
      },
      "outputs": [],
      "source": [
        "\n",
        "\n",
        "# Convert the list of SHAP values to a numpy array\n",
        "shap_values_array = np.array(shap_values)\n",
        "\n",
        "# Calculate the mean across all models\n",
        "# shap_values_mean = np.mean(shap_values_array, axis=0)\n",
        "# Reshape shap_values_mean and X_test_array to remove the dimension with size 12\n",
        "shap_values_mean_reshaped = shap_values_array.reshape(-1, 12)\n",
        "X_test_array_reshaped = X_test_array.reshape(-1, 12)\n",
        "\n",
        "# Plot the summary plot\n",
        "shap.summary_plot(shap_values_mean_reshaped, X_test_array_reshaped, feature_names=df.columns,show=False)\n",
        "\n",
        "plt.savefig('/content/SHAP.jpg', dpi=300)\n",
        "\n",
        "# Display the plot (optional)\n",
        "plt.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zATRVBIsOL7j"
      },
      "source": [
        "# Cumulative distribution"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "DLOKgzwKBGI6"
      },
      "outputs": [],
      "source": [
        "#  Create a new DataFrame containing only rows with value 0 \n",
        "new_df = df[df['stabf'] == 0]\n",
        "\n",
        "# Display the new DataFrame\n",
        "new_df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "M_jdbfS-Bwx4"
      },
      "outputs": [],
      "source": [
        "classes_to_drop=['stabf','stab']\n",
        "new_df.drop(classes_to_drop,inplace=True, axis=1)\n",
        "new_df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4al4vAhk-8Qs"
      },
      "outputs": [],
      "source": [
        "generated_data = generator(torch.randn(1358, 16, 12).to(device))\n",
        "generated_data_numpy = generated_data.detach().cpu().numpy()\n",
        "generated_data_reshaped = generated_data_numpy.reshape(-1, 12)\n",
        "df_generated_data = pd.DataFrame(generated_data_reshaped, columns=new_df.columns)\n",
        "df_generated_data\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EtTmz49v_nqV"
      },
      "outputs": [],
      "source": [
        "# Summary statistics for df1\n",
        "new_df.describe()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yjAagi9ZBjlI"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# Function to calculate CDF\n",
        "def calculate_cdf(data):\n",
        "    sorted_data = np.sort(data)\n",
        "    cdf = np.arange(1, len(sorted_data) + 1) / len(sorted_data)\n",
        "    return sorted_data, cdf\n",
        "columns_to_plot = ['tau1', 'tau2','tau3','tau4','g1','g2','g3','g4']\n",
        "# Plot CDF for each feature\n",
        "plt.figure(figsize=(5, 5))\n",
        "for column in columns_to_plot:\n",
        "    sorted_data, cdf = calculate_cdf(new_df[column])\n",
        "    plt.plot(sorted_data, cdf, label=column)\n",
        "\n",
        "plt.xlabel('Value')\n",
        "plt.ylabel('Cumulative Probability')\n",
        "plt.title('CDF Plot')\n",
        "plt.legend()\n",
        "plt.grid(True)\n",
        "plt.savefig('/content/CDF_realdata.jpg', dpi=300)\n",
        "plt.show()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "nM6jdiQJC82C"
      },
      "outputs": [],
      "source": [
        "df_generated_data.describe()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "863619D9CkdX"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# Function to calculate CDF\n",
        "def calculate_cdf(data):\n",
        "    sorted_data = np.sort(data)\n",
        "    cdf = np.arange(1, len(sorted_data) + 1) / len(sorted_data)\n",
        "    return sorted_data, cdf\n",
        "columns_to_plot = ['tau1', 'tau2','tau3','tau4','g1','g2','g3','g4']\n",
        "# Plot CDF for each feature\n",
        "plt.figure(figsize=(5, 5))\n",
        "for column in columns_to_plot:\n",
        "    sorted_data, cdf = calculate_cdf(df_generated_data[column])\n",
        "    plt.plot(sorted_data, cdf, label=column)\n",
        "\n",
        "plt.xlabel('Value')\n",
        "plt.ylabel('Cumulative Probability')\n",
        "plt.title('CDF Plot')\n",
        "plt.legend()\n",
        "plt.grid(True)\n",
        "plt.savefig('/content/CDF_DL.jpg', dpi=300)\n",
        "plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3_YlOG-nOX5j"
      },
      "source": [
        "# XGBoost Model"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "K_N-fXEgZyxq"
      },
      "outputs": [],
      "source": [
        "import torch\n",
        "import numpy as np\n",
        "np.random.seed(seed)\n",
        "torch.manual_seed(seed)\n",
        "torch.cuda.manual_seed(seed)\n",
        "torch.cuda.manual_seed_all(seed)\n",
        "torch.backends.cudnn.deterministic = True\n",
        "torch.backends.cudnn.benchmark = True\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Bx8l5Jq7jsvr"
      },
      "outputs": [],
      "source": [
        "train_datasets = []\n",
        "validation_datasets = []\n",
        "test_datasets = []\n",
        "\n",
        "train_loaders = []\n",
        "validation_loaders = []\n",
        "test_loaders = []\n",
        "\n",
        "\n",
        "for i in range(1):\n",
        "    print(f'[📚 LOADERS] {i}')\n",
        "    target_label = 0\n",
        "    a = CustomDataset(auth=True, target=target_label)\n",
        "\n",
        "    # Defining sizes\n",
        "    train_size = int(trainSize * len(a))\n",
        "    val_size = int(valSize * len(a))\n",
        "    test_size = len(a)-train_size-val_size\n",
        "\n",
        "    train_dataset, val_dataset, test_dataset = torch.utils.data.random_split(\n",
        "        a, [train_size, val_size, test_size])\n",
        "\n",
        "    train_datasets.append(train_dataset)\n",
        "    validation_datasets.append(val_dataset)\n",
        "    test_datasets.append(test_dataset)\n",
        "\n",
        "\n",
        "    train_loader = torch.utils.data.DataLoader(dataset=train_dataset,\n",
        "                                            batch_size=batch_size,\n",
        "                                            shuffle=False,\n",
        "                                            drop_last=True)\n",
        "\n",
        "    train_loaders.append(train_loader)\n",
        "\n",
        "    validation_loader = torch.utils.data.DataLoader(dataset=val_dataset,\n",
        "                                                    batch_size=batch_size,\n",
        "                                                    shuffle=False,\n",
        "                                                    drop_last=True)\n",
        "\n",
        "    validation_loaders.append(validation_loader)\n",
        "\n",
        "    test_loader = torch.utils.data.DataLoader(dataset=test_dataset,\n",
        "                                            batch_size=batch_size,\n",
        "                                            shuffle=False,\n",
        "                                            drop_last=True)\n",
        "\n",
        "    test_loaders.append(test_loader)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "DVKjDzhZjv1n"
      },
      "outputs": [],
      "source": [
        "classes_to_drop=['stabf','stab']\n",
        "df.drop(classes_to_drop,inplace=True, axis=1)\n",
        "df"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "gRRHAJmikLgs"
      },
      "outputs": [],
      "source": [
        "from xgboost import XGBClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = XGBClassifier(objective='binary:logistic', random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "TsT6QA6bQbY3"
      },
      "outputs": [],
      "source": [
        "import numpy as np\n",
        "\n",
        "# Convert DataFrame to arrays\n",
        "X_train_array = X_train.values\n",
        "X_test_array = X_test.values\n",
        "y_train_array = y_train.values\n",
        "y_test_array = y_test.values\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-fiMex-UOce0"
      },
      "source": [
        "# Random noise Attack"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4iu-nC3O2Que"
      },
      "outputs": [],
      "source": [
        "import numpy as np\n",
        "from sklearn.metrics import accuracy_score\n",
        "from xgboost import XGBClassifier\n",
        "\n",
        "class WhiteBoxAttack:\n",
        "    def __init__(self, model, epsilon=0.05, num_samples=50):\n",
        "        self.model = model\n",
        "        self.epsilon = epsilon\n",
        "        self.num_samples = num_samples\n",
        "\n",
        "    def attack(self, X, y):\n",
        "        y_pred_orig = self.model.predict(X)\n",
        "        accuracy_orig = accuracy_score(y, y_pred_orig)\n",
        "\n",
        "        X_adv = np.copy(X)\n",
        "        for i in range(len(X_adv)):\n",
        "            for _ in range(self.num_samples):\n",
        "                perturbation = np.random.normal(loc=0.0, scale=self.epsilon, size=X_adv[i].shape)\n",
        "                X_perturbed = X_adv[i] + perturbation\n",
        "                y_pred_perturbed = self.model.predict([X_perturbed])[0]\n",
        "                if accuracy_score([y[i]], [y_pred_perturbed]) < accuracy_orig:\n",
        "                    X_adv[i] = X_perturbed\n",
        "                    break\n",
        "\n",
        "        return X_adv\n",
        "\n",
        "# Create and train an XGBoost model\n",
        "xgb_model = XGBClassifier()\n",
        "xgb_model.fit(X_train_array, y_train_array)\n",
        "\n",
        "# Instantiate the WhiteBoxAttack\n",
        "attack = WhiteBoxAttack(model=xgb_model, epsilon=0.05, num_samples=50)\n",
        "\n",
        "# Generate adversarial examples\n",
        "X_test_adv = attack.attack(X_test_array, y_test_array)\n",
        "\n",
        "# Evaluate the performance of the attack\n",
        "y_pred_orig = xgb_model.predict(X_test_array)\n",
        "y_pred_adv = xgb_model.predict(X_test_adv)\n",
        "accuracy_orig = accuracy_score(y_test_array, y_pred_orig)\n",
        "accuracy_adv = accuracy_score(y_test_array, y_pred_adv)\n",
        "\n",
        "print(\"Accuracy on original test examples:\", accuracy_orig)\n",
        "print(\"Accuracy on adversarial test examples:\", accuracy_adv)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8H4FCmtwsO0t"
      },
      "outputs": [],
      "source": [
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# Given ASR values for two sets\n",
        "asr_values_set1 = [0.999,0.797, 0.616, 0.464, 0.335, 0.238, 0.162, 0.11, 0.07, 0.05, 0.03]\n",
        "# asr_values_set2 = [0.999,0.995, 0.972, 0.856, 0.714,0.537, 0.454,  0.397, 0.384, 0.383, 0.382]\n",
        "# asr_values_set3 = [0.999,0.995, 0.971, 0.859, 0.714, 0.537, 0.442, 0.397, 0.383, 0.383, 0.382]\n",
        "# asr_values_set4 = [0.999,0.996, 0.987, 0.94, 0.884, 0.826, 0.755, 0.67, 0.634, 0.568, 0.497]\n",
        "\n",
        "\n",
        "# Create a list of feature numbers starting from one\n",
        "feature_numbers = [0,0.05,0.1,0.15,0.20,0.25,0.30,0.35,0.40,0.45,0.50]\n",
        "\n",
        "# Plotting the graphs\n",
        "plt.ylim(0, 1.1)\n",
        "# plt.plot(feature_numbers, asr_values_set1, marker='o', linestyle='-', color='b', label='fgsm')\n",
        "# plt.plot(feature_numbers, asr_values_set2, marker='s', linestyle='-', color='r', label='bim')\n",
        "# plt.plot(feature_numbers, asr_values_set3, marker='^', linestyle='-', color='g', label='pgd')\n",
        "plt.plot(feature_numbers, asr_values_set1, marker='^', linestyle='-', color='y', label='random_noise') # Different color for set3\n",
        "# Adding legend\n",
        "plt.legend()\n",
        "\n",
        "plt.xlabel('epsilon')\n",
        "plt.ylabel('Accuracy')\n",
        "plt.title('Accuracy vs epsilon')\n",
        "plt.grid(True)\n",
        "plt.tight_layout()\n",
        "plt.savefig('/content/model_WB_XGBoost.pdf', format='pdf')\n",
        "plt.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "N85XUCABOkdL"
      },
      "source": [
        "# SHAP values for XGBoost"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "wzJOZb3bW_G8"
      },
      "outputs": [],
      "source": [
        "import shap\n",
        "import matplotlib.pyplot as plt\n",
        "import matplotlib\n",
        "# Create an explainer object using TreeExplainer\n",
        "explainer = shap.TreeExplainer(model)\n",
        "\n",
        "# Calculate Shapley values for a set of samples (e.g., X_test)\n",
        "shap_values = explainer.shap_values(X_test)\n",
        "\n",
        "# Summary plot\n",
        "shap.summary_plot(shap_values, X_test,feature_names=df.columns,show=False)\n",
        "\n",
        "plt.savefig('/content/SHAP.jpg',dpi=300)\n",
        "\n",
        "plt.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RCy11nhcOsus"
      },
      "source": [
        "# GAN-GRID agaisnt XgBoost"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "auyhwNeH-Mqj"
      },
      "outputs": [],
      "source": [
        "def train_gan(generator, surrogate, label, train_loader, num_epochs=100, lr=0.001, device=torch.device('cpu'), ml=False, num_episodes=150):\n",
        "\n",
        "    losses = []\n",
        "\n",
        "    if not ml:\n",
        "        generator = generator.to(device)\n",
        "        surrogate = surrogate\n",
        "\n",
        "        # for model in surrogate.models:\n",
        "        #     model.to(device)\n",
        "        #     model.train()\n",
        "\n",
        "    # Define the loss function and optimizer\n",
        "    cross_entropy_loss = nn.CrossEntropyLoss()\n",
        "    generator_optimizer = torch.optim.Adam(generator.parameters(), lr=lr)\n",
        "\n",
        "    # Define the reinforcement learning parameters\n",
        "    max_episode_length = 10\n",
        "    alpha = 0.1\n",
        "    gamma = 0.9\n",
        "\n",
        "    for episode in range(num_episodes):\n",
        "        # Initialize the latent input and the episode reward\n",
        "        latent_input = torch.randn(4, 16, 12).to(device)\n",
        "        episode_reward = 0\n",
        "\n",
        "        for step in range(max_episode_length):\n",
        "            # Generate a sample with the current latent input\n",
        "            fake_input = generator(latent_input)\n",
        "\n",
        "            # Evaluate the sample with the surrogate model\n",
        "            if not ml:\n",
        "                surrogate_output = surrogate(fake_input)\n",
        "            else:\n",
        "                surrogate_output = []\n",
        "                # Looping through each group of 12\n",
        "                for group in fake_input:\n",
        "                    # Flatten the group to make it compatible\n",
        "                    flat_group = group.view(-1, group.size(-1)).detach().numpy()\n",
        "                    # Get the probabilities\n",
        "                    probabilities = surrogate.predict_proba(flat_group)\n",
        "                    mean_probabilities = np.mean(probabilities, axis=0)\n",
        "                    # Append the probabilities to the array\n",
        "                    surrogate_output.append(mean_probabilities)\n",
        "                with torch.no_grad():\n",
        "                    surrogate_output = torch.tensor(surrogate_output, requires_grad=True)\n",
        "\n",
        "            predictions = surrogate_output.argmax(dim=1)\n",
        "            targets = torch.randint_like(\n",
        "                predictions, 0, surrogate_output.shape[1])\n",
        "            reward = (predictions == targets).float().mean().item()\n",
        "            episode_reward += reward\n",
        "\n",
        "            # Update the latent input using reinforcement learning\n",
        "            td_error = reward - episode_reward\n",
        "            latent_input += alpha * td_error * \\\n",
        "                gamma**step * torch.randn_like(latent_input)\n",
        "\n",
        "        # Update the generator using the final latent input of the episode\n",
        "        generator_optimizer.zero_grad()\n",
        "        fake_input = generator(latent_input)\n",
        "\n",
        "        if not ml:\n",
        "            surrogate_output = surrogate(fake_input)\n",
        "        else:\n",
        "            surrogate_output = []\n",
        "            # Looping through each group of 16\n",
        "            for group in fake_input:\n",
        "                # Flatten the group to make it compatible\n",
        "                flat_group = group.view(-1, group.size(-1)).detach().numpy()\n",
        "                # Get the probabilities\n",
        "                probabilities = surrogate.predict_proba(flat_group)\n",
        "                mean_probabilities = np.mean(probabilities, axis=0)\n",
        "                # Append the probabilities to the array\n",
        "                surrogate_output.append(mean_probabilities)\n",
        "            with torch.no_grad():\n",
        "                surrogate_output = torch.tensor(surrogate_output, requires_grad=True)\n",
        "\n",
        "\n",
        "        target_labels = torch.full_like(surrogate_output.argmax(dim=1), label)\n",
        "\n",
        "        generator_loss = cross_entropy_loss(surrogate_output, target_labels)\n",
        "\n",
        "        if ml:\n",
        "            generator_optimizer.zero_grad()\n",
        "\n",
        "        generator_loss.backward()\n",
        "        generator_optimizer.step()\n",
        "\n",
        "        losses.append(generator_loss.item())\n",
        "\n",
        "        if episode % 10 == 0:\n",
        "            print(f'[⏭️ EP {episode}/{num_episodes} | D{label}] LOSS: {round(generator_loss.item(), 3)}')\n",
        "\n",
        "    print()\n",
        "\n",
        "    return generator, losses"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Byu7Imb5-Bhm"
      },
      "outputs": [],
      "source": [
        "lr = 3e-3\n",
        "\n",
        "generators = []\n",
        "losses = []\n",
        "\n",
        "inputs, classes = next(iter(train_loader))\n",
        "\n",
        "for d in range(1):\n",
        "        print(f'[🤖 GENERATORS] Label {d}')\n",
        "\n",
        "        batch_size, window_size, num_features = inputs.shape\n",
        "        generator = Generator(batch_size, window_size, num_features)\n",
        "        surrogate_model = model\n",
        "\n",
        "        generator, loss = train_gan(generator, surrogate_model, train_loader=train_loader, num_epochs=20, lr=lr, label=0, ml=True, num_episodes=150)\n",
        "        print()\n",
        "\n",
        "        generators.append(generator)\n",
        "        losses.append(loss)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "i9Ho3jFoR1_d"
      },
      "outputs": [],
      "source": [
        "results = []\n",
        "\n",
        "for d in range(1):\n",
        "    predicted_labels = []\n",
        "    generator = generators[d].to(device)\n",
        "\n",
        "    for batch in test_loaders[d]:\n",
        "        input_batch = batch[0].to(device)\n",
        "\n",
        "        # # Create a tensor of zeros with the same shape as `input_batch`\n",
        "        # zeros_tensor = torch.zeros_like(input_batch).to(device)\n",
        "        # ones_tensor = torch.ones_like(input_batch).to(device)\n",
        "\n",
        "        # # Set the corresponding values to one in `zeros_tensor`\n",
        "        # zeros_tensor[:, :, nonmodifiable_indices] = 1\n",
        "        # ones_tensor[:, :, nonmodifiable_indices] = 0\n",
        "\n",
        "        # Multiply the original tensor with the zeros tensor\n",
        "        # result = input_batch\n",
        "\n",
        "        generated_data = generator(torch.randn(4, 16, 12).to(device))\n",
        "\n",
        "        # generated_data = generated_data.view(-1, generated_data.size(-1))\n",
        "\n",
        "        # Add the result to the ones tensor\n",
        "        final_result = generated_data\n",
        "\n",
        "        # Initialize an array to store probabilities\n",
        "        surrogate_outputs = []\n",
        "        surrogate = model\n",
        "\n",
        "        # Loop through each group of 16\n",
        "        for group in final_result:\n",
        "            # Flatten the group to make it compatible\n",
        "            flat_group = group.view(-1, group.size(-1)).cpu().detach().numpy()\n",
        "\n",
        "            # Get the probabilities\n",
        "            probabilities = surrogate.predict_proba(flat_group)\n",
        "\n",
        "            mean_probabilities = np.mean(probabilities, axis=0)\n",
        "\n",
        "            # Append the probabilities to the array\n",
        "            surrogate_outputs.append(mean_probabilities)\n",
        "\n",
        "        with torch.no_grad():\n",
        "            surrogate_outputs = torch.tensor(surrogate_outputs, requires_grad=True)\n",
        "\n",
        "        # Get the predicted class labels for each sample in the generated data\n",
        "        predicted_labels_batch = torch.argmax(surrogate_outputs, dim=1)\n",
        "\n",
        "        # Append the predicted labels to the lists\n",
        "        predicted_labels.extend(predicted_labels_batch.tolist())\n",
        "\n",
        "    asr = predicted_labels.count(0)/len(predicted_labels)\n",
        "    results.append(asr)\n",
        "    print(f'ASR: {round(asr, 3)}')\n",
        "\n",
        "print(f'\\n[🏆 ASR] {round(np.mean(results), 3)}')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BUmkd97Itrl1"
      },
      "source": [
        "# Cumulative distribution"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7Achu_s2bsVA"
      },
      "outputs": [],
      "source": [
        "generated_data = generator(torch.randn(1358, 16, 12).to(device))\n",
        "generated_data_numpy = generated_data.detach().cpu().numpy()\n",
        "generated_data_reshaped = generated_data_numpy.reshape(-1, 12)\n",
        "df_generated_data = pd.DataFrame(generated_data_reshaped, columns=new_df.columns)\n",
        "df_generated_data\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "31fu-Vxlb7Js"
      },
      "outputs": [],
      "source": [
        "df_generated_data.describe()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c8CA-lNMEikv"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# Function to calculate CDF\n",
        "def calculate_cdf(data):\n",
        "    sorted_data = np.sort(data)\n",
        "    cdf = np.arange(1, len(sorted_data) + 1) / len(sorted_data)\n",
        "    return sorted_data, cdf\n",
        "columns_to_plot = ['tau1', 'tau2','tau3','tau4','g1','g2','g3','g4']\n",
        "# Plot CDF for each feature\n",
        "plt.figure(figsize=(5, 5))\n",
        "for column in columns_to_plot:\n",
        "    sorted_data, cdf = calculate_cdf(df_generated_data[column])\n",
        "    plt.plot(sorted_data, cdf, label=column)\n",
        "\n",
        "plt.xlabel('Value')\n",
        "plt.ylabel('Cumulative Probability')\n",
        "plt.title('CDF Plot')\n",
        "plt.legend()\n",
        "plt.grid(True)\n",
        "plt.savefig('/content/cdf_ML.jpg',dpi=300)\n",
        "plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hwdUj2hROnAr"
      },
      "source": [
        "# Other classifiers"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Ew45tCk2yhZz"
      },
      "outputs": [],
      "source": [
        "from lightgbm import LGBMClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = LGBMClassifier(objective='binary', random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "n2aN6oF7zOk-"
      },
      "outputs": [],
      "source": [
        "from sklearn.tree import DecisionTreeClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = DecisionTreeClassifier(random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ojRDZb4fzir0"
      },
      "outputs": [],
      "source": [
        "from sklearn.ensemble import ExtraTreesClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = ExtraTreesClassifier(n_estimators=100, random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rYT5c6gw0Z2c"
      },
      "outputs": [],
      "source": [
        "from sklearn.neighbors import KNeighborsClassifier\n",
        "\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = KNeighborsClassifier(n_neighbors=5)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ulx5wkxm007H"
      },
      "outputs": [],
      "source": [
        "from sklearn.ensemble import GradientBoostingClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = GradientBoostingClassifier(n_estimators=100, random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "v532-GV31Cf5"
      },
      "outputs": [],
      "source": [
        "from sklearn.ensemble import RandomForestClassifier\n",
        "from sklearn.metrics import accuracy_score, confusion_matrix, classification_report\n",
        "from sklearn.model_selection import train_test_split\n",
        "for d in range(1):\n",
        "    X_train = []\n",
        "    y_train = []\n",
        "    for data in train_datasets[d]:\n",
        "        X_train.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_train.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_train = np.array(X_train)\n",
        "    X_train = [tensor for tensor in X_train]\n",
        "    X_train = np.concatenate(X_train, axis=0)\n",
        "    X_train = pd.DataFrame(X_train, columns=df.columns.tolist())\n",
        "\n",
        "    y_train = np.array(y_train)\n",
        "    y_train = pd.DataFrame(y_train, columns=['stabf'])\n",
        "\n",
        "    model = RandomForestClassifier(n_estimators=100, random_state=42)\n",
        "\n",
        "    # Training\n",
        "    model.fit(X_train, y_train)\n",
        "\n",
        "    X_test = []\n",
        "    y_test = []\n",
        "    for data in test_datasets[d]:\n",
        "        X_test.append(data[0])\n",
        "        for i in range(len(data[0])):\n",
        "            y_test.append(data[1])\n",
        "\n",
        "    # Convert lists to numpy arrays\n",
        "    X_test = np.array(X_test)\n",
        "    X_test = [tensor for tensor in X_test]\n",
        "    X_test = np.concatenate(X_test, axis=0)\n",
        "    X_test = pd.DataFrame(X_test, columns=df.columns.tolist())\n",
        "\n",
        "    y_test = np.array(y_test)\n",
        "    y_test = pd.DataFrame(y_test, columns=['stabf'])\n",
        "\n",
        "    # Testing\n",
        "    y_pred = model.predict(X_test)\n",
        "\n",
        "    accuracy = accuracy_score(y_test, y_pred)\n",
        "    f1 = f1_score(y_test, y_pred, average='weighted')\n",
        "\n",
        "    print(f'[💪 TRAINING {d}] ACC: {accuracy:.3f}')\n",
        "    print(f'[⚖️ F1 SCORE] {f1:.3f}')"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}