Feature_Engineering_Group.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Import Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "from pyspark.ml.feature import StringIndexer, OneHotEncoder\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Load Data and Convert to Spark Data Frame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# load data as dataframe\n",
    "loan_df = spark.read.csv(\"loan.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "loan_df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Create response variable and features\n",
    "### 3.1 Remove some columns based on EDA results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "loan_df2=loan_df.select(\n",
    "    'loan_status', # response variable\n",
    "    'grade', # dummy\n",
    "    'sub_grade', # dummy\n",
    "    'emp_length',\n",
    "    'home_ownership', # dummy    \n",
    "    'purpose',  # dummy\n",
    "    'verification_status', # dummy\n",
    "    loan_df.loan_amnt.cast(\"float\"),\n",
    "    loan_df.funded_amnt.cast(\"float\"),\n",
    "    loan_df.funded_amnt_inv.cast(\"float\"),\n",
    "    loan_df.int_rate.cast(\"float\"),\n",
    "    loan_df.installment.cast(\"float\"),\n",
    "    loan_df.annual_inc.cast(\"float\"),\n",
    "    loan_df.dti.cast(\"float\"),\n",
    "    loan_df.delinq_2yrs.cast(\"integer\"),  \n",
    "    loan_df.inq_last_6mths.cast(\"integer\"),\n",
    "    loan_df.open_acc.cast(\"integer\"), \n",
    "    loan_df.pub_rec.cast(\"integer\"),\n",
    "    loan_df.revol_bal.cast(\"float\"), \n",
    "    loan_df.revol_util.cast(\"float\"),\n",
    "    loan_df.total_acc.cast(\"integer\"),\n",
    "    loan_df.acc_now_delinq.cast(\"integer\"),\n",
    "    loan_df.tot_coll_amt.cast(\"float\"),\n",
    "    loan_df.tot_cur_bal.cast(\"float\"),\n",
    "    loan_df.total_rev_hi_lim.cast(\"integer\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.2 Create response variable and remove rows with no valid response variable "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def whetherpaid(x):\n",
    "    if x in ['Default', 'Charged Off', 'Does not meet the credit policy. Status:Charged Off']:\n",
    "        return 1\n",
    "    elif x in ['Does not meet the credit policy. Status:Fully Paid', 'Fully Paid']:\n",
    "        return 0\n",
    "    else:\n",
    "        return -1\n",
    "    \n",
    "paidflag = udf(lambda x: whetherpaid(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "loan_df3 = loan_df2.withColumn('paid_flag', paidflag('loan_status').\n",
    "                               cast(\"integer\")).where(\"paid_flag != -1\").drop('loan_status')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- grade: string (nullable = true)\n",
      " |-- sub_grade: string (nullable = true)\n",
      " |-- emp_length: string (nullable = true)\n",
      " |-- home_ownership: string (nullable = true)\n",
      " |-- purpose: string (nullable = true)\n",
      " |-- verification_status: string (nullable = true)\n",
      " |-- loan_amnt: float (nullable = true)\n",
      " |-- funded_amnt: float (nullable = true)\n",
      " |-- funded_amnt_inv: float (nullable = true)\n",
      " |-- int_rate: float (nullable = true)\n",
      " |-- installment: float (nullable = true)\n",
      " |-- annual_inc: float (nullable = true)\n",
      " |-- dti: float (nullable = true)\n",
      " |-- delinq_2yrs: integer (nullable = true)\n",
      " |-- inq_last_6mths: integer (nullable = true)\n",
      " |-- open_acc: integer (nullable = true)\n",
      " |-- pub_rec: integer (nullable = true)\n",
      " |-- revol_bal: float (nullable = true)\n",
      " |-- revol_util: float (nullable = true)\n",
      " |-- total_acc: integer (nullable = true)\n",
      " |-- acc_now_delinq: integer (nullable = true)\n",
      " |-- tot_coll_amt: float (nullable = true)\n",
      " |-- tot_cur_bal: float (nullable = true)\n",
      " |-- total_rev_hi_lim: integer (nullable = true)\n",
      " |-- paid_flag: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "loan_df3.printSchema()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.3 Add more features\n",
    "#### 3.3.1 Create a numeric feature for \"emp_length\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import re\n",
    "def convert_to_int(s):\n",
    "    s = re.sub('\\\\D', '', s)  #remove any non-digital character\n",
    "    #\\d matches any digital, #\\D matches any non-digital\n",
    "    try:\n",
    "        return s\n",
    "    except ValueError:\n",
    "        return None\n",
    "\n",
    "emp_to_num = udf(convert_to_int)\n",
    "loan_df4 = loan_df3.withColumn('emp_len',emp_to_num('emp_length').cast('integer')).drop('emp_length')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.3.2 Regroup home_ownership"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def home_ownership_func(x):\n",
    "    if x in ['ANY','OTHER','NONE']:\n",
    "        return 'Other'\n",
    "    else: \n",
    "        return x\n",
    "\n",
    "home_ownership = udf(home_ownership_func)  #fixed a function\n",
    "loan_df5 = loan_df4.withColumn('home_ownership',home_ownership('home_ownership'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------+\n",
      "|home_ownership| count|\n",
      "+--------------+------+\n",
      "|           OWN| 22282|\n",
      "|         Other|   228|\n",
      "|          RENT|107831|\n",
      "|      MORTGAGE|126598|\n",
      "+--------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "loan_df5.select('home_ownership').groupBy('home_ownership').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.3.3 Add loan_inc_ratio and installment_inc_ratio as new features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def calculate_ratio(a, b):\n",
    "    try:\n",
    "        return a/float(b)\n",
    "    except TypeError:\n",
    "        return None\n",
    "    except ZeroDivisionError:\n",
    "        return None\n",
    "\n",
    "def calculate_monthly_ratio(a, b):\n",
    "    try:\n",
    "        b = float(b) / 12\n",
    "        return calculate_ratio(a, b)\n",
    "    except TypeError:\n",
    "        return None\n",
    "\n",
    "# User-defined functions to compute ratios\n",
    "ratio = udf(calculate_ratio) # compute loan amount / annual income\n",
    "monthly_ratio = udf(calculate_monthly_ratio) # compute installment / monthly income"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "loan_df6 = loan_df5.withColumn('loan_inc_ratio',ratio('loan_amnt','annual_inc').cast('float'))\n",
    "loan_df6 = loan_df6.withColumn('instal_inc_ratio',monthly_ratio('installment','annual_inc').cast('float'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.3.6 Create dummy variables\n",
    "* Use StringIndexer to encode a string column of labels to a column of label indices, and most frequent label gets index 0.\n",
    "\n",
    "* Use OneHotEncoder to convert indices into dummy vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "purpose converted\n",
      "grade converted\n",
      "sub_grade converted\n",
      "verification_status converted\n",
      "home_ownership converted\n"
     ]
    }
   ],
   "source": [
    "convert_list = [      \n",
    "               'purpose',\n",
    "               'grade',\n",
    "               'sub_grade',\n",
    "               'verification_status',\n",
    "               'home_ownership',\n",
    "                ]\n",
    "\n",
    "for item in convert_list:\n",
    "    indexer = StringIndexer(inputCol=item, outputCol=item + 'Index')    \n",
    "    loan_df6 = indexer.fit(loan_df6).transform(loan_df6).drop(item)\n",
    "    onehotenc = OneHotEncoder(inputCol=item + 'Index', outputCol=item + \"-onehot\", dropLast=False)\n",
    "    loan_df6 = onehotenc.transform(loan_df6).drop(item +'Index')\n",
    "    print item + ' converted'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "loan_df6 = loan_df6.fillna(0.0, ['tot_coll_amt','tot_cur_bal', 'total_rev_hi_lim'])\n",
    "loan_df6 = loan_df6.dropna()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "loan_df_final = loan_df6.select(\n",
    "    'paid_flag',\n",
    "    'loan_amnt',\n",
    "    'funded_amnt',\n",
    "    'funded_amnt_inv',\n",
    "    'int_rate',\n",
    "    'installment',\n",
    "    'annual_inc',\n",
    "    'dti',\n",
    "    'delinq_2yrs',\n",
    "    'inq_last_6mths',\n",
    "    'open_acc',\n",
    "    'pub_rec',\n",
    "    'revol_bal',\n",
    "    'revol_util',\n",
    "    'total_acc',\n",
    "    'acc_now_delinq',\n",
    "    'tot_coll_amt',\n",
    "    'tot_cur_bal',\n",
    "    'total_rev_hi_lim',\n",
    "    'emp_len',\n",
    "    'loan_inc_ratio',\n",
    "    'instal_inc_ratio',\n",
    "    'purpose-onehot',\n",
    "    'grade-onehot',\n",
    "    'sub_grade-onehot',\n",
    "    'verification_status-onehot',\n",
    "    'home_ownership-onehot'\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Modeling"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# import ML packages\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.mllib.evaluation import MulticlassMetrics\n",
    "from pyspark.ml.linalg import Vectors\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder, TrainValidationSplit\n",
    "from pyspark.ml.feature import StringIndexer, VectorIndexer, VectorAssembler"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1 Splitting and Formatting Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# create the final dataset\n",
    "loan_df_final = loan_df_final.withColumnRenamed('paid_flag','label')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "va = VectorAssembler(outputCol='features',inputCols=loan_df_final.columns[1:])\n",
    "data = va.transform(loan_df_final).select('features','label')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Creating training and test sets by random split 0.8,0.2\n",
    "data_sets=data.randomSplit([0.8,0.2])\n",
    "\n",
    "data_train=data_sets[0].cache()\n",
    "data_test=data_sets[1].cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "training dataset fraud propotition: 0.180456353232\n",
      "test dataset fraud propotition: 0.181495671917\n"
     ]
    }
   ],
   "source": [
    "print 'training dataset fraud propotition:', float(data_train.filter('label=1').count())/float((data_train.filter('label=1').count() + data_train.filter('label=0').count()))\n",
    "print 'test dataset fraud propotition:', float(data_test.filter('label=1').count())/float((data_test.filter('label=1').count() + data_test.filter('label=0').count()))\n",
    "\n",
    "data_train_paid = data_train.filter(data_train.label == 0)\n",
    "data_train_not_paid = data_train.filter(data_train.label == 1)\n",
    "\n",
    "num_samples_train = data_train.count()\n",
    "half_num_samples_train = num_samples_train / 2\n",
    "\n",
    "data_train_paid = data_train_paid.sample(True, float(half_num_samples_train) / float(data_train_paid.count())) \n",
    "data_train_not_paid = data_train_not_paid.sample(True, float(half_num_samples_train) / float(data_train_not_paid.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "197091"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# union two random sample sets together, paid & unpaid\n",
    "data_train = data_train_paid.union(data_train_not_paid)  \n",
    "\n",
    "# total observations remain close enough to the original dataset\n",
    "data_train.count() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2 Random Forest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Fit random forest\n",
    "rf = RandomForestClassifier()\n",
    "evaluator_rf = BinaryClassificationEvaluator()\n",
    "cv = CrossValidator().setEstimator(rf).setEvaluator(evaluator_rf).setNumFolds(5)\n",
    "paramGrid = ParamGridBuilder().addGrid(rf.numTrees,[31]).addGrid(rf.maxDepth,[5,15]).build()\n",
    "\n",
    "cv.setEstimatorParamMaps(paramGrid)\n",
    "cvmodel_rf = cv.fit(data_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Make predictions\n",
    "predictions_rf = cvmodel_rf.bestModel.transform(data_test)\n",
    "prediction_list_rf = predictions_rf.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create a list of tuples as (prediction, label)\n",
    "predictionAndLabels_rf = []\n",
    "for item in prediction_list_rf:\n",
    "    predictionAndLabels_rf.append((float(item[4]), float(item[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "areaUnderROC:0.698704209361\n"
     ]
    }
   ],
   "source": [
    "score=evaluator_rf.evaluate(predictions_rf)\n",
    "print evaluator_rf.getMetricName()+\":\"+str(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[ 25856.  14520.]\n",
      " [  3168.   5785.]]\n",
      "0.284905195765\n",
      "0.646152127778\n",
      "0.641427963267\n"
     ]
    }
   ],
   "source": [
    "metrics_rf = MulticlassMetrics(sc.parallelize(predictionAndLabels_rf))\n",
    "print metrics_rf.confusionMatrix().toArray()\n",
    "print metrics_rf.precision(1)\n",
    "print metrics_rf.recall(1)\n",
    "print metrics_rf.accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.ml.classification.RandomForestClassificationModel"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rfmodel=cvmodel_rf.bestModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(3, 0.17016971772952169),\n",
       " (19, 0.070145635212076132),\n",
       " (288, 0.066363043067363459),\n",
       " (20, 0.06182702391140641),\n",
       " (6, 0.050874038220614787),\n",
       " (12, 0.041545190442791008),\n",
       " (5, 0.038451809983694456),\n",
       " (286, 0.028610698537262409),\n",
       " (16, 0.023086853530975979),\n",
       " (4, 0.021965160807738617),\n",
       " (13, 0.021719786779805569),\n",
       " (291, 0.021054837479358976),\n",
       " (11, 0.020758402764765083),\n",
       " (1, 0.020715153617940719),\n",
       " (0, 0.020619211375286568)]"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "importance_list = list(enumerate(rfmodel.featureImportances))\n",
    "importance_list.sort(key=lambda tup: -tup[1])\n",
    "top_importances = importance_list[:15]\n",
    "top_importances"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3 Logistic Regression"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lr = LogisticRegression()\n",
    "evaluator = BinaryClassificationEvaluator()\n",
    "\n",
    "cv = CrossValidator().setEstimator(lr).setEvaluator(evaluator).setNumFolds(5)\n",
    "paramGrid = ParamGridBuilder().addGrid(lr.maxIter,[100]).addGrid(lr.regParam,[0.1,0.01,0.001,1]).build()\n",
    "\n",
    "cv.setEstimatorParamMaps(paramGrid)\n",
    "cvmodel = cv.fit(data_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#cvmodel.bestModel.coefficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "areaUnderROC:0.706364985953\n"
     ]
    }
   ],
   "source": [
    "predictions = cvmodel.bestModel.transform(data_test)\n",
    "score = evaluator.evaluate(cvmodel.bestModel.transform(data_test))\n",
    "print evaluator.getMetricName()+\":\"+str(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "prediction_list = predictions.collect()\n",
    "predictionAndLabels = []\n",
    "for item in prediction_list:\n",
    "    predictionAndLabels.append((float(item[4]), float(item[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[ 25876.  14500.]\n",
      " [  3066.   5887.]]\n",
      "0.288762446657\n",
      "0.657544956998\n",
      "0.64390115348\n"
     ]
    }
   ],
   "source": [
    "metrics = MulticlassMetrics(sc.parallelize(predictionAndLabels))\n",
    "print metrics.confusionMatrix().toArray()\n",
    "print metrics.precision(1)\n",
    "print metrics.recall(1)\n",
    "print metrics.accuracy"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}