keras_training.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%reload_ext autoreload\n",
    "%matplotlib inline\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Notebook explaining how to train model on market data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As exaplained in the README, first start up Kafka and related contaniers with `docker-compose up` and then run the services for gathering data. Once that is done, you can then download the data from Kafka and perform offline analysis (i.e. model training). As the project evolves, the goal is having the model being trained continuously and being re-deployed automatically. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Download data from Kafka\n",
    "For now, I'll use market data only"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from kryptoflow.models.streamer_base import AvroAsync\n",
    "\n",
    "a = AvroAsync(topic='gdax')\n",
    "a.read_from_start(persist=True, path='/media/carlo/HDD/kafka_local/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transform data into suitable format and train with Keras "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'price': 7890.0, 'volume_24h': 9903.30859375, 'spread': 0.009999999776482582, 'ts': '2018-03-29 02:31:31', 'side': 'sell'}\n",
      "{'price': 7890.0, 'volume_24h': 9903.3671875, 'spread': 0.009999999776482582, 'ts': '2018-03-29 02:31:33', 'side': 'sell'}\n",
      "{'price': 7890.0, 'volume_24h': 9903.3681640625, 'spread': 0.009999999776482582, 'ts': '2018-03-29 02:31:33', 'side': 'sell'}\n",
      "{'price': 7890.0, 'volume_24h': 9903.3681640625, 'spread': 1.5099999904632568, 'ts': '2018-03-29 02:31:34', 'side': 'sell'}\n",
      "{'price': 7888.08984375, 'volume_24h': 9903.369140625, 'spread': 2.009999990463257, 'ts': '2018-03-29 02:31:34', 'side': 'sell'}\n",
      "{'price': 7885.0, 'volume_24h': 9904.306640625, 'spread': 5.010000228881836, 'ts': '2018-03-29 02:31:34', 'side': 'sell'}\n",
      "{'price': 7885.0, 'volume_24h': 9904.3076171875, 'spread': 0.009999999776482582, 'ts': '2018-03-29 02:31:39', 'side': 'sell'}\n",
      "{'price': 7885.0, 'volume_24h': 9904.537109375, 'spread': 1.9900000095367432, 'ts': '2018-03-29 02:31:40', 'side': 'sell'}\n",
      "{'price': 7884.990234375, 'volume_24h': 9904.974609375, 'spread': 1.9900000095367432, 'ts': '2018-03-29 02:31:40', 'side': 'buy'}\n"
     ]
    }
   ],
   "source": [
    "from kryptoflow.analysis.dataset import get_data, ForecastTransformer, TimeEmbedder\n",
    "import pandas\n",
    "import numpy\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.pipeline import Pipeline\n",
    "from sklearn.preprocessing import MinMaxScaler\n",
    "from kryptoflow.analysis.model import KerasModel\n",
    "from kryptoflow.analysis.export import ModelExporter\n",
    "\n",
    "\n",
    "local_df = get_data('gdax', remote=False, keep_keys=['ts', 'price', 'volume_24h', 'spread', 'side'])\n",
    "remote_df = get_data('gdax', remote=True, keep_keys=['ts', 'price', 'volume_24h', 'spread', 'side'])\n",
    "\n",
    "# exp = ModelExporter()\n",
    "\n",
    "pipe = Pipeline([\n",
    "    ('tr', ForecastTransformer(prediction_steps=1)),\n",
    "    ('sc', MinMaxScaler()),\n",
    "    ('time', TimeEmbedder(inital_dims=len(local_df.columns)))\n",
    "])\n",
    "\n",
    "X, y = pipe.fit_transform(local_df)\n",
    "# exp.store(pipe, 'pipeline', model_type='sklearn')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[7.89339990e+03, 4.36000013e+00, 9.94000977e+03, 0.00000000e+00,\n",
       "        0.00000000e+00, 1.00000000e+00, 7.89304004e+03, 6.28999996e+00,\n",
       "        9.94001074e+03, 0.00000000e+00, 0.00000000e+00, 1.00000000e+00,\n",
       "        7.89110986e+03, 6.28999996e+00, 9.94001465e+03, 0.00000000e+00,\n",
       "        0.00000000e+00, 1.00000000e+00, 7.89058008e+03, 9.99999978e-03,\n",
       "        9.94001660e+03, 2.00000000e+00, 0.00000000e+00, 1.00000000e+00,\n",
       "        7.89058008e+03, 9.99999978e-03, 9.94002051e+03, 0.00000000e+00,\n",
       "        0.00000000e+00, 1.00000000e+00, 7.89727002e+03, 6.67000008e+00,\n",
       "        9.94003516e+03, 1.00000000e+00, 1.00000000e+00, 0.00000000e+00,\n",
       "        7.89004004e+03, 9.99999978e-03, 9.94004785e+03, 6.00000000e+00,\n",
       "        0.00000000e+00, 1.00000000e+00, 7.89004980e+03, 9.99999978e-03,\n",
       "        9.94006738e+03, 6.00000000e+00, 1.00000000e+00, 0.00000000e+00,\n",
       "        1.00000000e+00]])"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pipe.named_steps['tr'].transform(remote_df)\n",
    "# remote_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.75 ms, sys: 0 ns, total: 2.75 ms\n",
      "Wall time: 2.2 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Pipeline(memory=None,\n",
       "     steps=[('tr', ForecastTransformer(n_time_steps=None, prediction_steps=None)), ('sc', MinMaxScaler(copy=True, feature_range=(0, 1))), ('time', TimeEmbedder(inital_dims=None))])"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.externals import joblib\n",
    "% time joblib.load('/home/carlo/projects/kryptoflow/stored_models/8/sklearn/pipeline.mdl')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9881294722414049"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "y[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9153.8896484375"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def inv_transf(x):\n",
    "    x -= pipe.named_steps['sc'].min_[-1]\n",
    "    x /= pipe.named_steps['sc'].scale_[-1]\n",
    "    return x\n",
    "inv_transf(y[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Train on 19377 samples, validate on 6459 samples\n",
      "Epoch 1/2\n",
      " - 5s - loss: 0.0118 - val_loss: 3.1284e-05\n",
      "Epoch 2/2\n",
      " - 5s - loss: 2.3012e-05 - val_loss: 2.1331e-05\n",
      "Saved Keras model to disk\n",
      "INFO:tensorflow:No assets to save.\n",
      "INFO:tensorflow:No assets to write.\n",
      "INFO:tensorflow:SavedModel written to: b'/home/carlo/projects/kryptoflow/kryptoflow/../stored_models/8/tf/saved_model.pb'\n",
      "Saved tf model to disk\n"
     ]
    }
   ],
   "source": [
    "X, x_test, y, y_test = train_test_split(X, y)\n",
    "model = KerasModel(dims=X.shape[1:])\n",
    "model.fit(X, y, x_test, y_test, epochs=2)\n",
    "exp.store(model.model, 'lstm', model_type='keras')"
   ]
  }
 ],
 "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}