Sentiment_predictor.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 181,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "import sys\n",
    "assert sys.version_info >= (3, 5) # make sure we have Python 3.5+\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "from textblob import TextBlob\n",
    "\n",
    "from pyspark.sql import SparkSession, functions, types\n",
    "spark = SparkSession.builder.appName('Emotion prediction').getOrCreate()\n",
    "spark.sparkContext.setLogLevel('WARN')\n",
    "assert spark.version >= '2.3' # make sure we have Spark 2.3+\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.ml.feature import StopWordsRemover\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler, IndexToString\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.feature import CountVectorizer\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.ml.feature import HashingTF, IDF, Tokenizer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 155,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!pip install textblob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 156,
   "metadata": {},
   "outputs": [],
   "source": [
    "inputs = '/Users/syedikram/Documents/BigData-733/cryptointel/news_cc/btcnews.csv'\n",
    "data = spark.read.csv(inputs,header=True, mode=\"DROPMALFORMED\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.withColumn(\"title_body\", concat_ws(' ', data.title, data.body))\n",
    "lower_udf = udf(lambda x: \" \".join(x.lower() for x in x.split()))\n",
    "data = data.withColumn(\"title_body\", lower_udf(data['title_body']))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {},
   "outputs": [],
   "source": [
    "def remove_punct(text):\n",
    "    regex = re.compile('[' + re.escape(string.punctuation) + '0-9\\\\r\\\\t\\\\n]')\n",
    "    nopunct = regex.sub(\" \", text)\n",
    "    return nopunct"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "metadata": {},
   "outputs": [],
   "source": [
    "punct_remover = udf(lambda x: remove_punct(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 160,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.withColumn(\"title_body\", punct_remover(data['title_body']))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 161,
   "metadata": {},
   "outputs": [],
   "source": [
    "spaces_udf = udf(lambda x: \" \".join(x.split()))\n",
    "data = data.withColumn(\"title_body\", spaces_udf(data['title_body']))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 162,
   "metadata": {},
   "outputs": [],
   "source": [
    "sentiment_udf = udf(lambda x: 1 if TextBlob(str(x)).sentiment.polarity > 0 else (0 if TextBlob(str(x)).sentiment.polarity == 0 else -1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.withColumn(\"sentiment\", sentiment_udf(data['title_body']))\n",
    "data = data.select('title_body','sentiment')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 164,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Tokenizing and Vectorizing\n",
    "tok = Tokenizer(inputCol=\"title_body\", outputCol=\"words\")\n",
    "tokenized = tok.transform(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 165,
   "metadata": {},
   "outputs": [],
   "source": [
    "stopword_rm = StopWordsRemover(inputCol='words', outputCol='words_rm')\n",
    "rm_tokenized = stopword_rm.transform(tokenized)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 166,
   "metadata": {},
   "outputs": [],
   "source": [
    "cv = CountVectorizer(inputCol='words_rm', outputCol='tf')\n",
    "cvModel = cv.fit(rm_tokenized)\n",
    "count_vectorized = cvModel.transform(rm_tokenized)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 167,
   "metadata": {},
   "outputs": [],
   "source": [
    "idf_ngram = IDF().setInputCol('tf').setOutputCol('tfidf')\n",
    "tfidfModel_ngram = idf_ngram.fit(count_vectorized)\n",
    "tfidf_df = tfidfModel_ngram.transform(count_vectorized)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 168,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+--------------+\n",
      "|          title_body|sentiment|sentimentIndex|\n",
      "+--------------------+---------+--------------+\n",
      "|quantum computing...|        1|           0.0|\n",
      "|bitcoin trading p...|        1|           0.0|\n",
      "|crypto analyst as...|        1|           0.0|\n",
      "|bitcoin btc price...|        1|           0.0|\n",
      "|bitcoin sv bsv pr...|       -1|           1.0|\n",
      "|cboe halts bitcoi...|       -1|           1.0|\n",
      "|ripple xrp price ...|       -1|           1.0|\n",
      "|ripple xrp and ca...|        1|           0.0|\n",
      "|bitcoin price ana...|        1|           0.0|\n",
      "|bitcoin ethereum ...|       -1|           1.0|\n",
      "|craig wright rage...|        1|           0.0|\n",
      "|million in weeks ...|        0|           2.0|\n",
      "|bitcoin btc becom...|        1|           0.0|\n",
      "|stablecoins are n...|        1|           0.0|\n",
      "|winklevoss capita...|        1|           0.0|\n",
      "|bitcoin btc price...|        1|           0.0|\n",
      "|binance announces...|        0|           2.0|\n",
      "|canadian regulato...|       -1|           1.0|\n",
      "|binance unveils a...|        0|           2.0|\n",
      "|vaneck’s bitcoin ...|       -1|           1.0|\n",
      "+--------------------+---------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "indexer = StringIndexer(inputCol=\"sentiment\", outputCol=\"sentimentIndex\")\n",
    "indexed = indexer.fit(data).transform(data)\n",
    "indexed.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 169,
   "metadata": {},
   "outputs": [],
   "source": [
    "splits = tfidf_df.randomSplit([0.8,0.2],seed=100)\n",
    "train = splits[0]\n",
    "val = splits[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 170,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+--------------------+--------------------+--------------------+--------------------+\n",
      "|          title_body|sentiment|               words|            words_rm|                  tf|               tfidf|\n",
      "+--------------------+---------+--------------------+--------------------+--------------------+--------------------+\n",
      "|a bad week for th...|       -1|[a, bad, week, fo...|[bad, week, crypt...|(30169,[0,4,6,7,1...|(30169,[0,4,6,7,1...|\n",
      "|a better bitcoin ...|        1|[a, better, bitco...|[better, bitcoin,...|(30169,[0,1,2,3,1...|(30169,[0,1,2,3,1...|\n",
      "|a bitcoin backed ...|        1|[a, bitcoin, back...|[bitcoin, backed,...|(30169,[0,10,13,2...|(30169,[0,10,13,2...|\n",
      "|a bitcoin etf is ...|        1|[a, bitcoin, etf,...|[bitcoin, etf, ‘v...|(30169,[0,1,2,3,2...|(30169,[0,1,2,3,2...|\n",
      "|a bitcoin etf – o...|        1|[a, bitcoin, etf,...|[bitcoin, etf, –,...|(30169,[0,1,2,3,1...|(30169,[0,1,2,3,1...|\n",
      "|a bitcoin rat is ...|       -1|[a, bitcoin, rat,...|[bitcoin, rat, oc...|(30169,[0,1,2,3,7...|(30169,[0,1,2,3,7...|\n",
      "|a cautionary tale...|        1|[a, cautionary, t...|[cautionary, tale...|(30169,[0,1,2,3,6...|(30169,[0,1,2,3,6...|\n",
      "|a crypto apocalyp...|        1|[a, crypto, apoca...|[crypto, apocalyp...|(30169,[0,1,2,3,6...|(30169,[0,1,2,3,6...|\n",
      "|a cryptocurrency ...|        1|[a, cryptocurrenc...|[cryptocurrency, ...|(30169,[0,1,2,3,4...|(30169,[0,1,2,3,4...|\n",
      "|a decentralized b...|        0|[a, decentralized...|[decentralized, b...|(30169,[0,14,52,1...|(30169,[0,14,52,1...|\n",
      "|a dozen bitcoin a...|        1|[a, dozen, bitcoi...|[dozen, bitcoin, ...|(30169,[0,1,2,3,7...|(30169,[0,1,2,3,7...|\n",
      "|a guide to buildi...|        1|[a, guide, to, bu...|[guide, building,...|(30169,[0,1,2,3,4...|(30169,[0,1,2,3,4...|\n",
      "|a heartbroken mot...|        1|[a, heartbroken, ...|[heartbroken, mot...|(30169,[0,1,2,3,7...|(30169,[0,1,2,3,7...|\n",
      "|a hedge fund mana...|        1|[a, hedge, fund, ...|[hedge, fund, man...|(30169,[0,9,12,39...|(30169,[0,9,12,39...|\n",
      "|a look at openbaz...|        1|[a, look, at, ope...|[look, openbazaar...|(30169,[0,1,2,3,4...|(30169,[0,1,2,3,4...|\n",
      "|a look at some of...|        1|[a, look, at, som...|[look, ’s, popula...|(30169,[0,1,2,3,4...|(30169,[0,1,2,3,4...|\n",
      "|a look at the gat...|        1|[a, look, at, the...|[look, gateway, c...|(30169,[0,1,2,3,7...|(30169,[0,1,2,3,7...|\n",
      "|a look at the mul...|        1|[a, look, at, the...|[look, multi, cur...|(30169,[0,1,2,3,4...|(30169,[0,1,2,3,4...|\n",
      "|a maritime crypto...|        1|[a, maritime, cry...|[maritime, crypto...|(30169,[1,2,3,6,1...|(30169,[1,2,3,6,1...|\n",
      "|a millennial and ...|        1|[a, millennial, a...|[millennial, cryp...|(30169,[0,1,2,3,6...|(30169,[0,1,2,3,6...|\n",
      "+--------------------+---------+--------------------+--------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 174,
   "metadata": {},
   "outputs": [],
   "source": [
    "hm_assembler = VectorAssembler(inputCols=[ \"tfidf\"], outputCol=\"features\")\n",
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0,labelCol=\"sentimentIndex\",featuresCol = \"features\")\n",
    "hm_pipeline = Pipeline(stages=[hm_assembler, indexer, lr])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 175,
   "metadata": {},
   "outputs": [],
   "source": [
    "paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.1, 0.3]).addGrid(lr.elasticNetParam, [0.0, 0.1]).build()\n",
    "crossval = CrossValidator(estimator=hm_pipeline,estimatorParamMaps=paramGrid,\\\n",
    "            evaluator=MulticlassClassificationEvaluator(labelCol=\"sentimentIndex\", predictionCol=\"prediction\",metricName=\"accuracy\"),numFolds=5)\n",
    "\n",
    "model = crossval.fit(train)\n",
    "prediction_train = model.transform(val)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 176,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Model Accuracy = 0.8192949907235622\n"
     ]
    }
   ],
   "source": [
    "evaluator = MulticlassClassificationEvaluator(labelCol=\"sentimentIndex\", predictionCol=\"prediction\",\n",
    "                                              metricName=\"accuracy\")\n",
    "accuracy = evaluator.evaluate(prediction_train)\n",
    "print(\"Model Accuracy = \" + str(accuracy))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 186,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+-----------------+\n",
      "|          title_body|sentiment|originalSentiment|\n",
      "+--------------------+---------+-----------------+\n",
      "|quantum computing...|        1|                1|\n",
      "|bitcoin trading p...|        1|                1|\n",
      "|crypto analyst as...|        1|                1|\n",
      "|bitcoin btc price...|        1|                1|\n",
      "|bitcoin sv bsv pr...|       -1|               -1|\n",
      "|cboe halts bitcoi...|       -1|               -1|\n",
      "|ripple xrp price ...|       -1|               -1|\n",
      "|ripple xrp and ca...|        1|                1|\n",
      "|bitcoin price ana...|        1|                1|\n",
      "|bitcoin ethereum ...|       -1|               -1|\n",
      "+--------------------+---------+-----------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "converter = IndexToString(inputCol=\"sentimentIndex\", outputCol=\"originalSentiment\")\n",
    "converted = converter.transform(indexed)\n",
    "converted.select('title_body','sentiment','originalSentiment').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}