preparing_data/pgd.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Adversarial samples generated and saved to pgd_train.csv and pgd_test.csv\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from sklearn.model_selection import train_test_split\n",
    "from torch.autograd import Variable\n",
    "\n",
    "# Load the dataset\n",
    "df = pd.read_csv('/Users/aryankargwal/adverscredit/models/data/credit.csv')\n",
    "\n",
    "# Split the dataset into features and target\n",
    "X = df.drop('Class', axis=1).values\n",
    "y = df['Class'].values\n",
    "\n",
    "# Split the dataset into training and testing sets\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "\n",
    "# Convert data to torch tensors\n",
    "X_train = torch.tensor(X_train, dtype=torch.float32)\n",
    "y_train = torch.tensor(y_train, dtype=torch.float32)\n",
    "X_test = torch.tensor(X_test, dtype=torch.float32)\n",
    "y_test = torch.tensor(y_test, dtype=torch.float32)\n",
    "\n",
    "# Define a simple neural network model\n",
    "class SimpleNN(nn.Module):\n",
    "    def __init__(self, input_dim):\n",
    "        super(SimpleNN, self).__init__()\n",
    "        self.fc1 = nn.Linear(input_dim, 64)\n",
    "        self.fc2 = nn.Linear(64, 64)\n",
    "        self.fc3 = nn.Linear(64, 1)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        x = torch.relu(self.fc1(x))\n",
    "        x = torch.relu(self.fc2(x))\n",
    "        x = torch.sigmoid(self.fc3(x))\n",
    "        return x\n",
    "\n",
    "# Initialize the model, loss function, and optimizer\n",
    "model = SimpleNN(X_train.shape[1])\n",
    "criterion = nn.BCELoss()\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# Train the model\n",
    "num_epochs = 10\n",
    "for epoch in range(num_epochs):\n",
    "    model.train()\n",
    "    optimizer.zero_grad()\n",
    "    outputs = model(X_train)\n",
    "    loss = criterion(outputs.squeeze(), y_train)\n",
    "    loss.backward()\n",
    "    optimizer.step()\n",
    "\n",
    "# PGD attack function\n",
    "def pgd_attack(model, X, y, epsilon=0.01, alpha=0.001, num_iter=10):\n",
    "    X_adv = X.clone().detach().requires_grad_(True)\n",
    "    for _ in range(num_iter):\n",
    "        model.zero_grad()\n",
    "        outputs = model(X_adv)\n",
    "        loss = criterion(outputs.squeeze(), y)\n",
    "        loss.backward()\n",
    "        with torch.no_grad():\n",
    "            X_adv += alpha * X_adv.grad.sign()\n",
    "            eta = torch.clamp(X_adv - X, min=-epsilon, max=epsilon)\n",
    "            X_adv = torch.clamp(X + eta, min=0, max=1).detach_()\n",
    "            X_adv.requires_grad_(True)\n",
    "    return X_adv\n",
    "\n",
    "# Generate adversarial samples\n",
    "X_train_adv = pgd_attack(model, X_train, y_train)\n",
    "X_test_adv = pgd_attack(model, X_test, y_test)\n",
    "\n",
    "# Combine adversarial and original samples\n",
    "def scatter_adversarial_samples(X, X_adv, y):\n",
    "    X_combined = X.clone().detach().cpu().numpy()\n",
    "    y_combined = y.clone().detach().cpu().numpy()\n",
    "    X_adv_np = X_adv.clone().detach().cpu().numpy()\n",
    "    \n",
    "    num_samples = len(y)\n",
    "    num_adv_samples = len(y) // 10  # Adding adversarial samples equal to 10% of original samples\n",
    "    indices = np.random.choice(num_samples, num_adv_samples, replace=False)\n",
    "    \n",
    "    X_combined[indices] = X_adv_np[indices]\n",
    "    \n",
    "    return X_combined, y_combined\n",
    "\n",
    "X_train_combined, y_train_combined = scatter_adversarial_samples(X_train, X_train_adv, y_train)\n",
    "X_test_combined, y_test_combined = scatter_adversarial_samples(X_test, X_test_adv, y_test)\n",
    "\n",
    "# Save the new datasets with adversarial samples\n",
    "train_combined_df = pd.DataFrame(X_train_combined, columns=df.columns[:-1])\n",
    "train_combined_df['Class'] = y_train_combined\n",
    "train_combined_df.to_csv('pgd_train.csv', index=False)\n",
    "\n",
    "test_combined_df = pd.DataFrame(X_test_combined, columns=df.columns[:-1])\n",
    "test_combined_df['Class'] = y_test_combined\n",
    "test_combined_df.to_csv('pgd_test.csv', index=False)\n",
    "\n",
    "print(\"Adversarial samples generated and saved to pgd_train.csv and pgd_test.csv\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.11.0 ('base')",
   "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.11.0"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "40d3a090f54c6569ab1632332b64b2c03c39dcf918b08424e98f38b5ae0af88f"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}