project_get_pred.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "balanced-eagle",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Looking in indexes: http://pypi-service.pypi:8080/\n",
      "Collecting sklearn\n",
      "  Downloading http://pypi-service.pypi:8080/packages/sklearn-0.0.tar.gz (1.1 kB)\n",
      "Requirement already satisfied: scikit-learn in /opt/conda/lib/python3.8/site-packages (from sklearn) (0.24.1)\n",
      "Requirement already satisfied: numpy>=1.13.3 in /opt/conda/lib/python3.8/site-packages (from scikit-learn->sklearn) (1.20.1)\n",
      "Requirement already satisfied: scipy>=0.19.1 in /opt/conda/lib/python3.8/site-packages (from scikit-learn->sklearn) (1.6.1)\n",
      "Requirement already satisfied: joblib>=0.11 in /opt/conda/lib/python3.8/site-packages (from scikit-learn->sklearn) (1.0.1)\n",
      "Requirement already satisfied: threadpoolctl>=2.0.0 in /opt/conda/lib/python3.8/site-packages (from scikit-learn->sklearn) (2.1.0)\n",
      "Building wheels for collected packages: sklearn\n",
      "  Building wheel for sklearn (setup.py) ... \u001b[?25ldone\n",
      "\u001b[?25h  Created wheel for sklearn: filename=sklearn-0.0-py2.py3-none-any.whl size=1316 sha256=0032c65054c4e01ada60f1421a435db9510828b5c93c3b409c786f67b92f9d60\n",
      "  Stored in directory: /home/jovyan/.cache/pip/wheels/5a/66/e7/f2bbd87b7ae75f6159d4fd57ec3cc1c530c1b19b304a1716f6\n",
      "Successfully built sklearn\n",
      "Installing collected packages: sklearn\n",
      "Successfully installed sklearn-0.0\n"
     ]
    }
   ],
   "source": [
    "!pip install --user --trusted-host pypi-service.pypi --index-url http://pypi-service.pypi:8080/ sklearn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 188,
   "id": "affiliated-disney",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, Row\n",
    "from pyspark.sql.functions import pandas_udf, udf, size, length, when, col, array_contains, row_number, lit\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import FloatType, IntegerType, ArrayType, StringType, StructType, StructField, DateType\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import pandas_udf\n",
    "from textblob import TextBlob\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "from statistics import mean, stdev\n",
    "\n",
    "import datetime, time \n",
    "import emoji\n",
    "import re\n",
    "import socket\n",
    "import pandas as pd\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "current-provision",
   "metadata": {},
   "outputs": [],
   "source": [
    "LOCAL_IP = socket.gethostbyname(socket.gethostname())\n",
    "spark = (\n",
    "    SparkSession\n",
    "    .builder\n",
    "    .master('k8s://https://10.32.7.103:6443')\n",
    "    .config(\"spark.driver.host\", LOCAL_IP)\n",
    "    .config(\"spark.driver.bindAddress\", \"0.0.0.0\")\n",
    "    .config('spark.executor.instences', '2')\n",
    "    .config('spark.executor.cores', '2')\n",
    "    .config('spark.cores.max', '4')\n",
    "    .config('spark.executor.memory', '4g')\n",
    "    .config('spark.sql.execution.arrow.enabled', 'true')\n",
    "    .config('spark.kubernetes.namespace', 'zpinaev-244202')\n",
    "    .config('spark.kubernetes.container.image', 'node03.st:5000/spark-executor:zpinaev-244202')\n",
    "    .config(\"spark.kubernetes.container.image.pullPolicy\", \"Always\")\n",
    "    .config('spark.kubernetes.executor.deleteOnTermination', 'false')\n",
    "    .getOrCreate()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "norman-intelligence",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+\n",
      "|             target|          prediction|\n",
      "+-------------------+--------------------+\n",
      "| 1.5434267216982664|  2.0427243539198217|\n",
      "| 0.4209528633973711|  0.4725503325269045|\n",
      "|0.48446295930137595|-0.20330700243302816|\n",
      "|0.33664444540659016|  0.9999394547350783|\n",
      "|0.12772982981743183|  0.3417761784670134|\n",
      "|-0.6164723371777932| -0.6536933077937467|\n",
      "|-1.2582528850942512|  -1.266468295761551|\n",
      "| 0.6766494524474193|  0.6827303031464702|\n",
      "| 1.1071552154750104|  1.2259051877517029|\n",
      "|  0.996971267045149|  2.0472631660977854|\n",
      "+-------------------+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "preds_df = spark.read\\\n",
    "                .parquet(\"hdfs:///tmp/vshishkov-243998/project_data/preds/LinearRegression_aave-usd.parquet\")\n",
    "                \n",
    "preds_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "located-behavior",
   "metadata": {},
   "outputs": [],
   "source": [
    "coins_list = ['lrc-usd', 'axs-usd', 'xrp-usd', 'uni1-usd', 'ada-usd', 'fil-usd', 'dot-usd', 'rep-usd', 'icp-usd', 'bnb-usd', 'xmr-usd', 'doge-usd', 'aave-usd', 'ltc-usd', 'zil-usd', 'link-usd', 'vet-usd']\n",
    "models_list = ['LinearRegression', 'RandomForestRegressor']\n",
    "\n",
    "coins_list_Bcs = spark.sparkContext.broadcast(coins_list)\n",
    "models_list_Bcs = spark.sparkContext.broadcast(models_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "id": "orange-consultancy",
   "metadata": {},
   "outputs": [],
   "source": [
    "def std_cor(data):\n",
    "    data_mean = mean(data)\n",
    "    res = 0\n",
    "    for x in data:\n",
    "        res += pow((x - data_mean),2)\n",
    "    res = pow((res / len(data)),(1/2))\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 225,
   "id": "heavy-catch",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_pred(coin_name, model_type, date):\n",
    "    \n",
    "    models_list = models_list_Bcs.value\n",
    "    coins_list = coins_list_Bcs.value\n",
    "    \n",
    "    if not model_type in models_list:\n",
    "        print(\"Error: wrong model type\")\n",
    "        return\n",
    "    if not coin_name in coins_list:\n",
    "        print(\"Error: wrong coin name\")\n",
    "        return\n",
    "    \n",
    "    coin_data = spark.read\\\n",
    "                   .csv(f\"hdfs:///tmp/vshishkov-243998/project_data/currencies/{coin_name.upper()}.csv\", header=True)\\\n",
    "                   .withColumn(\"id_n\", F.monotonically_increasing_id())\n",
    "    \n",
    "    windowSpec = Window.orderBy(col(\"id_n\"))\n",
    "    coin_data = coin_data.withColumn(\"id\",row_number().over(windowSpec)-1).drop(\"id_n\")\n",
    "    \n",
    "    \n",
    "    if coin_data.where(col(\"Date\") == date).count() == 0:\n",
    "        print(\"Error: wrong date or no pred for passed date\")\n",
    "        return\n",
    "    \n",
    "    pred_df = spark.read\\\n",
    "                   .parquet(f\"hdfs:///tmp/vshishkov-243998/project_data/preds/{model_type}_{coin_name}.parquet\") \\\n",
    "                   .withColumn(\"id_n\", F.monotonically_increasing_id())\n",
    "    \n",
    "    pred_df = pred_df.withColumn(\"id\",row_number().over(windowSpec)-1).drop(\"id_n\")\n",
    "    \n",
    "    id_1 = coin_data.where(col(\"Date\") == date).select(\"id\").head()[0]\n",
    "    \n",
    "    id_2 = pred_df.count() - ( coin_data.count() - id_1 )\n",
    "    \n",
    "    target_sc = pred_df.where(col(\"id\") == id_2).select(\"target\").head()[0]\n",
    "    pred_sc = pred_df.where(col(\"id\") == id_2).select(\"prediction\").head()[0]\n",
    "    \n",
    "    coin_close_data = [float(row[\"Close\"]) for row in coin_data.select(\"Close\").collect()]\n",
    "    N_elem = coin_data.count() - pred_df.count() - 1\n",
    "    coin_close_data = coin_close_data[:N_elem]\n",
    "    \n",
    "    data_mean = mean(coin_close_data)\n",
    "    data_std = std_cor(coin_close_data)\n",
    "    \n",
    "\n",
    "    target = target_sc * data_std + data_mean\n",
    "    pred = pred_sc * data_std + data_mean\n",
    "    \n",
    "    return target, pred\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 228,
   "id": "molecular-madonna",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Target:  108.20206448522916\n",
      "Prediction:  108.98950356890762\n"
     ]
    }
   ],
   "source": [
    "model_type = \"LinearRegression\"\n",
    "coin = \"aave-usd\"\n",
    "date = \"2022-06-06\"\n",
    "\n",
    "target, pred = get_pred(coin, model_type, date)\n",
    "print(\"Target: \", target)\n",
    "print(\"Prediction: \", pred)"
   ]
  }
 ],
 "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}