trabalho_big_data_deploy.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "displayHTML('<div style=\"text-align:center\"><img src =\"https://github.com/romulomadu/PEDS/blob/master/algebra/tarefas/logos.png?raw=true\" /></div>')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "import random\n",
    "import numpy as np\n",
    "import scipy\n",
    "import pandas as pd\n",
    "import math\n",
    "import sklearn\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.feature_extraction.text import TfidfVectorizer\n",
    "from sklearn.metrics.pairwise import cosine_similarity\n",
    "from scipy.sparse.linalg import svds\n",
    "from collections import OrderedDict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# File location and type\n",
    "file_location = \"/FileStore/tables/Churn_Modelling.csv\"\n",
    "file_type = \"csv\"\n",
    "\n",
    "file_prods = \"\"\n",
    "\n",
    "# CSV options\n",
    "infer_schema = \"true\"\n",
    "first_row_is_header = \"true\"\n",
    "delimiter = \",\"\n",
    "\n",
    "# The applied options are for CSV files. For other file types, these will be ignored.\n",
    "df_ = spark.read.format(file_type) \\\n",
    "  .option(\"inferSchema\", infer_schema) \\\n",
    "  .option(\"header\", first_row_is_header) \\\n",
    "  .option(\"sep\", delimiter) \\\n",
    "  .load(file_location)\n",
    "\n",
    "#df_.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prod_df = df_.toPandas()\n",
    "prod_df['Poup'] = prod_df.EstimatedSalary > 40000\n",
    "prod_df['Prev'] = (prod_df.EstimatedSalary > 40000) & (prod_df.Age >30)\n",
    "prod_df['Stocks'] = (prod_df.EstimatedSalary > 40000) & (prod_df.Age >30) & (prod_df.CreditScore > 700)\n",
    "prod_df['PIC'] = ((prod_df.Age < 35) | (prod_df.Age > 60)) & (prod_df.EstimatedSalary < 40000)\n",
    "prod_df['CDB'] = prod_df.Age > 50\n",
    "prod_df['TD'] = prod_df.EstimatedSalary > 30000\n",
    "prod_df['Micro'] = (prod_df.EstimatedSalary < 30000) & (prod_df.Gender == 'Female')\n",
    "prod_df['CrCard'] = (prod_df.EstimatedSalary > 30000) & (prod_df.HasCrCard == 0)\n",
    "prod_df['Univ'] = (prod_df.Age <= 23)\n",
    "#prod_df['PortSal'] = (prod_df.Age > 23)\n",
    "prods = ['Poup', 'Prev', 'Stocks', 'PIC', 'CDB', 'TD', 'Micro', 'CrCard', 'Univ']#, 'PortSal'] \n",
    "prod_df = prod_df[['CustomerId', 'Balance'] + prods]\n",
    "prod_df['balProd'] = prod_df[prods].sum(axis=1).max()/prod_df[prods].sum(axis=1)\n",
    "prod_df.loc[prod_df['balProd']==np.inf, 'balProd']=0\n",
    "prod_df_old = prod_df.copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# A little randomness\n",
    "def randomness(col, bias=0.5):\n",
    "  rand_list = [(random.randint(0,10)/10.) - bias for i in range(10000)]\n",
    "  return prod_df[[col]].add(pd.Series(rand_list), axis=0) >= 1\n",
    "  \n",
    "prod_df['PIC'] = randomness('PIC', -0.3)\n",
    "prod_df['CDB'] = randomness('CDB', 0.3)\n",
    "prod_df['TD'] = randomness('TD', 0.6)\n",
    "\n",
    "for prod in prods:\n",
    "  prod_df[prod] = randomness(prod)\n",
    "\n",
    "# If balance is zero, there are no products\n",
    "for col in prods:\n",
    "  prod_df[col] = (prod_df[col]) & (prod_df.Balance > 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "melt_prod_df = pd.melt(prod_df, ['CustomerId', 'Balance', 'balProd'])\n",
    "melt_prod_df= melt_prod_df[melt_prod_df['value']].drop('value', axis=1)\n",
    "#prod_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "numeric_features = [t[0] for t in df_.dtypes if t[1] == 'int']\n",
    "numeric_data = df_.select(numeric_features).toPandas()\n",
    "n = len(numeric_data.columns)\n",
    "df = df_.select(\n",
    " 'CreditScore',\n",
    " 'Geography',\n",
    " 'Gender',\n",
    " 'Age',\n",
    " 'Tenure',\n",
    " 'Balance',\n",
    " 'NumOfProducts',\n",
    " 'HasCrCard',\n",
    " 'IsActiveMember',\n",
    " 'EstimatedSalary',\n",
    " 'Exited')\n",
    "cols = df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler\n",
    "categoricalColumns = ['Geography', 'Gender', 'HasCrCard', 'IsActiveMember']#, 'Exited']\n",
    "stages = []\n",
    "for categoricalCol in categoricalColumns:\n",
    "    stringIndexer = StringIndexer(inputCol = categoricalCol, outputCol = categoricalCol + 'Index')\n",
    "    encoder = OneHotEncoderEstimator(inputCols=[stringIndexer.getOutputCol()], outputCols=[categoricalCol + \"classVec\"])\n",
    "    stages += [stringIndexer, encoder]\n",
    "label_stringIdx = StringIndexer(inputCol = 'Exited', outputCol = 'label')\n",
    "stages += [label_stringIdx]\n",
    "numericCols = ['CreditScore',\n",
    " 'Age',\n",
    " 'Tenure',\n",
    " 'Balance',\n",
    " 'NumOfProducts',\n",
    " 'EstimatedSalary']\n",
    "assemblerInputs = [c + \"classVec\" for c in categoricalColumns] + numericCols\n",
    "assembler = VectorAssembler(inputCols=assemblerInputs, outputCol=\"features\")\n",
    "stages += [assembler]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cols = ['CreditScore',\n",
    " 'Geography',\n",
    " 'Gender',\n",
    " 'Age',\n",
    " 'Tenure',\n",
    " 'Balance',\n",
    " 'NumOfProducts',\n",
    " 'HasCrCard',\n",
    " 'IsActiveMember',\n",
    " 'EstimatedSalary']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "pipeline = Pipeline(stages = stages)\n",
    "pipelineModel = pipeline.fit(df)\n",
    "df = pipelineModel.transform(df)\n",
    "selectedCols = ['label', 'features'] + cols\n",
    "df = df.select(selectedCols)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = df.randomSplit([0.7, 0.3], seed = 2018)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "lr = LogisticRegression(featuresCol = 'features', labelCol = 'label', maxIter=10)\n",
    "lrModel = lr.fit(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "plt.clf()\n",
    "import numpy as np\n",
    "beta = np.sort(lrModel.coefficients)\n",
    "plt.plot(beta)\n",
    "plt.ylabel('Beta Coefficients')\n",
    "#display(plt.show())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.clf()\n",
    "trainingSummary = lrModel.summary\n",
    "roc = trainingSummary.roc.toPandas()\n",
    "plt.plot(roc['FPR'],roc['TPR'])\n",
    "plt.ylabel('False Positive Rate')\n",
    "plt.xlabel('True Positive Rate')\n",
    "plt.title('ROC Curve')\n",
    "#display(plt.show())\n",
    "print('Training set areaUnderROC: ' + str(trainingSummary.areaUnderROC))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions = lrModel.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "dt = DecisionTreeClassifier(featuresCol = 'features', labelCol = 'label', maxDepth = 3)\n",
    "dtModel = dt.fit(train)\n",
    "predictions = dtModel.transform(test)\n",
    "#predictions.select('age', 'job', 'label', 'rawPrediction', 'prediction', 'probability').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "rf = RandomForestClassifier(featuresCol = 'features', labelCol = 'label')\n",
    "rfModel = rf.fit(train)\n",
    "predictions = rfModel.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import GBTClassifier\n",
    "gbt = GBTClassifier(maxIter=10)\n",
    "gbtModel = gbt.fit(train)\n",
    "predictions = gbtModel.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dbutils.widgets.text(name='CustId_input', defaultValue='15634602', label='CustId')\n",
    "custid_input= int(getArgument(\"CustId_input\"))\n",
    "plt.clf()\n",
    "input_df = df_.filter('CustomerId == {}'.format(custid_input))\n",
    "input_df = input_df.drop('CustomerId', 'RowNumber', 'Surname')\n",
    "input_df = pipelineModel.transform(input_df)\n",
    "#selectedCols = ['label', 'features'] + cols\n",
    "#input_df2 = input_df.select(selectedCols)\n",
    "pred_input = rfModel.transform(input_df)\n",
    "prob_exit_input = round((pred_input.select('probability').toPandas()['probability'].loc[0][1])*100,2)\n",
    "strategy = int(prob_exit_input/20) + 1\n",
    "if strategy == 1:\n",
    "  message = 'Não oferecer taxas diferenciadas'\n",
    "elif strategy > 1:\n",
    "  message = 'Ofereça taxas diferenciadas de nível {}'.format(strategy)\n",
    "\n",
    "displayHTML(\"\"\"<center><h1 style=\"font-family:verdana;\">O cliente tem {}% de chances de sair do Banco</h1>\n",
    "            <h3>{}</h3>\n",
    "            </center>\n",
    "            \"\"\".format(prob_exit_input, message))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "custid_input= int(getArgument(\"CustId_input\"))\n",
    "pred_tot = gbtModel.transform(train)\n",
    "probs = pred_tot.select('probability').toPandas()['probability']\n",
    "probs = [probs.iloc[i][1] for i in range(len(probs))]\n",
    "plt.clf()\n",
    "fig, ax = plt.subplots()\n",
    "plt.title('Probabilidade de saida dos clientes de acordo com o modelo')\n",
    "hist_data = ax.hist(probs,bins=8)\n",
    "x_hist = (hist_data[1]> prob_exit_input/100)\n",
    "y_hist = hist_data[0]\n",
    "plt.ylim([0,y_hist.max()*1.15])\n",
    "height = y_hist[np.where(x_hist==False)[0][-1]]\n",
    "ax.scatter(round(prob_exit_input/100,2), height*1.1, c='r', s=80)\n",
    "display(plt.show())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Recommender System\n",
    "### Collaborative Filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "interactions_train_df, interactions_test_df = train_test_split(melt_prod_df, \n",
    "                                   test_size=0.20,\n",
    "                                   random_state=42)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "interactions_full_df = melt_prod_df\n",
    "interactions_full_indexed_df = melt_prod_df.set_index('CustomerId')\n",
    "interactions_train_indexed_df = interactions_train_df.set_index('CustomerId')\n",
    "interactions_test_indexed_df = interactions_test_df.set_index('CustomerId')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_items_interacted(person_id, interactions_df):\n",
    "    # Get the user's data and merge in the movie information.\n",
    "    interacted_items = interactions_df.loc[person_id]['variable']\n",
    "    return set(interacted_items if type(interacted_items) == pd.Series else [interacted_items])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Creating a sparse pivot table with users in rows and items in columns\n",
    "users_items_pivot_matrix_df = interactions_full_df.pivot(index='CustomerId', \n",
    "                                                          columns='variable', \n",
    "                                                          values='balProd').fillna(0)\n",
    "\n",
    "users_items_pivot_matrix = users_items_pivot_matrix_df.as_matrix()\n",
    "users_ids = list(users_items_pivot_matrix_df.index)\n",
    "#The number of factors to factor the user-item matrix.\n",
    "NUMBER_OF_FACTORS_MF = 5\n",
    "#Performs matrix factorization of the original user item matrix\n",
    "U, sigma, Vt = svds(users_items_pivot_matrix, k = NUMBER_OF_FACTORS_MF)\n",
    "sigma = np.diag(sigma)\n",
    "all_user_predicted_ratings = np.dot(np.dot(U, sigma), Vt)\n",
    "#Converting the reconstructed matrix back to a Pandas dataframe\n",
    "cf_preds_df = pd.DataFrame(all_user_predicted_ratings, columns = users_items_pivot_matrix_df.columns, index=users_ids).transpose()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class CFRecommender:\n",
    "    \n",
    "    MODEL_NAME = 'Collaborative Filtering'\n",
    "    \n",
    "    def __init__(self, cf_predictions_df, items_df=None):\n",
    "        self.cf_predictions_df = cf_predictions_df\n",
    "        self.items_df = items_df\n",
    "        \n",
    "    def get_model_name(self):\n",
    "        return self.MODEL_NAME\n",
    "        \n",
    "    def recommend_items(self, user_id, items_to_ignore=[], topn=10, verbose=False):\n",
    "        # Get and sort the user's predictions\n",
    "        sorted_user_predictions = self.cf_predictions_df[user_id].sort_values(ascending=False) \\\n",
    "                                    .reset_index().rename(columns={user_id: 'recStrength'})\n",
    "\n",
    "        # Recommend the highest predicted rating movies that the user hasn't seen yet.\n",
    "        recommendations_df = sorted_user_predictions[~sorted_user_predictions['variable'].isin(items_to_ignore)] \\\n",
    "                               .sort_values('recStrength', ascending = False) \\\n",
    "                               .head(topn)\n",
    "\n",
    "        if verbose:\n",
    "            if self.items_df is None:\n",
    "                raise Exception('\"items_df\" is required in verbose mode')\n",
    "\n",
    "            recommendations_df = recommendations_df.merge(self.items_df, how = 'left', \n",
    "                                                          left_on = 'variable', \n",
    "                                                          right_on = 'variable')[['recStrength', 'variable']]\n",
    "\n",
    "\n",
    "        return recommendations_df\n",
    "    \n",
    "cf_recommender_model = CFRecommender(cf_preds_df, None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "custid_input= int(getArgument(\"CustId_input\"))\n",
    "try:\n",
    "  renamer = {'variable': 'Produto', 'recPerc': '% de chance'}\n",
    "  person_recs_df = cf_recommender_model.recommend_items(custid_input, \n",
    "                                                        get_items_interacted(custid_input, interactions_full_indexed_df))\n",
    "  rec_strengths = person_recs_df[person_recs_df['recStrength'] >0 ]['recStrength']\n",
    "  person_recs_df['recPerc'] = ((rec_strengths*100/rec_strengths.sum()))\n",
    "  person_recs_df = person_recs_df.dropna(axis=0)\n",
    "  person_recs_df = person_recs_df.drop('recStrength', axis=1).rename(index=str, columns=renamer)\n",
    "  rec_html = person_recs_df.to_html(index=False)\n",
    "except Exception:\n",
    "  ord_dict = OrderedDict([('variable', ['Poup', 'CDB']), ('recPerc', [50, 50])])\n",
    "  rec_html = pd.DataFrame.from_dict(ord_dict).rename(index=str, columns=renamer).to_html(index=False)\n",
    "\n",
    "try:\n",
    "  products_html = 'possui os produtos {}'.format(', '.join(get_items_interacted(custid_input, interactions_full_indexed_df)))\n",
    "except KeyError:\n",
    "  products_html = 'nao possui produtos'\n",
    "\n",
    "displayHTML(\"\"\"<center style=\"font-family:verdana;\">O cliente {}<br>\n",
    "Ofereça os produtos abaixo:<br> {}\n",
    "</center>\"\"\".format(products_html, rec_html))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cf_recommender_model.recommend_items(custid_input, \n",
    "                                                        get_items_interacted(custid_input, interactions_full_indexed_df))"
   ]
  }
 ],
 "metadata": {
  "name": "bigData_UI",
  "notebookId": 2189168469067636
 },
 "nbformat": 4,
 "nbformat_minor": 0
}