tforce_colab_demo.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "tforce-colab_demo.ipynb",
      "version": "0.3.2",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "metadata": {
        "id": "r4RkH_ITL0nP",
        "colab_type": "text"
      },
      "cell_type": "markdown",
      "source": [
        "Rough 'demo' of the v0.2 hypersearch funcationality (as of 02/12/19). Imports modified to work off a single file. Open this in colab.research.google.com or modify for a local run. For colab, go to edit-> notebook settings and set GPU for backend to test."
      ]
    },
    {
      "metadata": {
        "id": "8GOYyJrt-1jX",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title installs\n",
        "import os\n",
        "os.chdir('/content/')\n",
        "!wget http://prdownloads.sourceforge.net/ta-lib/ta-lib-0.4.0-src.tar.gz\n",
        "!tar -xzf ta-lib-0.4.0-src.tar.gz\n",
        "os.chdir('/content/ta-lib')\n",
        "! ./configure --prefix=/usr\n",
        "! make\n",
        "! make install\n",
        "os.chdir('/content/')\n",
        "!pip install python-box\n",
        "!pip install ta-lib\n",
        "!pip install gdax"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "id": "y8ZEQ6aAFSjf",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title Unzip data\n",
        "#@markdown Upload bitcoin-historical-data.zip from \n",
        "#@markdown https://www.kaggle.com/mczielinski/bitcoin-historical-data\n",
        "#@markdown to the colab 'Files' directory\n",
        "!unzip /content/bitcoin-historical-data.zip"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "id": "FwbQ9UM0_AZb",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title utils\n",
        "\n",
        "import numpy as np\n",
        "from enum import Enum\n",
        "\n",
        "\n",
        "class ScoreMode(Enum):\n",
        "    \"\"\"Different ways we might consider scoring our runs. This is for BO's sake, not for our RL agent -\n",
        "    ie helps us decide which hyper combos to pursue.\"\"\"\n",
        "    MEAN = 1  # mean of all episodes\n",
        "    LAST = 2  # final episode (the one w/o killing)\n",
        "    POS = 3  # max # positive tests\n",
        "    CONSECUTIVE_POS = 4  # max # *consecutive* positives\n",
        "    TOTAL = 5\n",
        "    MIX = 6\n",
        "\n",
        "\n",
        "MODE = ScoreMode.MEAN\n",
        "\n",
        "\n",
        "def calculate_score(scores):\n",
        "    for i, a in enumerate(scores):\n",
        "        if a == 0.: scores[i] = -1.\n",
        "    if MODE == ScoreMode.MEAN:\n",
        "        return np.mean(scores)\n",
        "    elif MODE == ScoreMode.LAST:\n",
        "        return scores[-1]\n",
        "    elif MODE == ScoreMode.MIX:\n",
        "        return np.mean(scores[:-1]) + scores[-1]\n",
        "    elif MODE == ScoreMode.POS:\n",
        "        return sum(1 for x in scores if x > 0)\n",
        "    elif MODE == ScoreMode.TOTAL:\n",
        "        return sum(x for x in scores)\n",
        "    elif MODE == ScoreMode.CONSECUTIVE_POS:\n",
        "        score, curr_consec = 0, 0\n",
        "        for i, adv in enumerate(scores):\n",
        "            if adv > 0:\n",
        "                curr_consec += 1\n",
        "                continue\n",
        "            if curr_consec > score:\n",
        "                score = curr_consec\n",
        "            curr_consec = 0\n",
        "        return score\n",
        "\n",
        "\n",
        "def add_common_args(parser):\n",
        "    parser.add_argument('-g', '--gpu-split', type=float, default=1, help=\"Num ways we'll split the GPU (how many tabs you running?)\")\n",
        "    parser.add_argument('--autoencode', action=\"store_true\", help=\"If you're running out of GPU memory, try --autoencode which scales things down\")\n",
        "\n",
        "\n",
        "last_good_commit = '6a6e49c'\n",
        "\n",
        "\n",
        "def raise_refactor():\n",
        "    raise NotImplemented(f'Restore from {last_good_commit}')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "id": "ZNtwsOum-tgy",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title data\n",
        "data_file = \"coinbaseUSD_1-min_data_2014-12-01_to_2018-11-11.csv\" #@param {type:\"string\"}\n",
        "\n",
        "import time, json, re, pdb\n",
        "from os import path\n",
        "from enum import Enum\n",
        "import pandas as pd\n",
        "import numpy as np\n",
        "from sqlalchemy import create_engine\n",
        "from sqlalchemy import text\n",
        "# from utils import raise_refactor, last_good_commit\n",
        "from sklearn.preprocessing import robust_scale\n",
        "import os\n",
        "\n",
        "# From connecting source file, `import engine` and run `engine.connect()`. Need each connection to be separate\n",
        "# (see https://stackoverflow.com/questions/3724900/python-ssl-problem-with-multiprocessing)\n",
        "# config_json = json.load(open(os.path.dirname(__file__) + '/../config.json'))\n",
        "# DB = config_json['DB_HISTORY'].split('/')[-1]\n",
        "# engine_runs = create_engine(config_json['DB_RUNS'])\n",
        "\n",
        "# Decide which exchange you want to trade on (significant even in training). Pros & cons; Kraken's API provides more\n",
        "# details than GDAX (bid/ask spread, VWAP, etc) which means predicting its next price-action is easier for RL. It\n",
        "# also has a lower minimum trade (.002 BTC vs GDAX's .01 BTC), which gives it more wiggle room. However, its API is\n",
        "# very unstable and slow, so when you actually go live you'r bot will be suffering. GDAX's API is rock-solid. Look\n",
        "# into the API stability, it may change by the time you're using this. If Kraken is solid, use it instead.\n",
        "class Exchange(Enum):\n",
        "    GDAX = 'gdax'\n",
        "    KRAKEN = 'kraken'\n",
        "EXCHANGE = Exchange.KRAKEN\n",
        "\n",
        "# see {last_good_commit} for imputes (ffill, bfill, zero),\n",
        "# alex database\n",
        "\n",
        "def setup_runs_table():\n",
        "    \"\"\"Run this function once during project setup (see README). Or just copy/paste the SQL into your runs database\n",
        "    \"\"\"\n",
        "    conn_runs = engine_runs.connect()\n",
        "    conn_runs.execute(\"\"\"\n",
        "        create table if not exists runs\n",
        "        (\n",
        "            id uuid not null,\n",
        "            hypers jsonb not null,\n",
        "            returns double precision[],\n",
        "            signals double precision[],\n",
        "            prices double precision[],\n",
        "            uniques double precision[]\n",
        "        );\n",
        "    \"\"\")\n",
        "\n",
        "class Data(object):\n",
        "    def __init__(self, ep_len=5000, window=300, arbitrage=False, indicators={}):\n",
        "        self.ep_len = ep_len\n",
        "        self.window = window\n",
        "        self.arbitrage = arbitrage\n",
        "        self.indicators = indicators\n",
        "\n",
        "        self.ep_stride = ep_len  # disjoint\n",
        "        # self.ep_stride = 100  # overlap; shift each episode by x seconds.\n",
        "        # TODO overlapping stride would cause test/train overlap. Tweak it so train can overlap data, but test gets silo'd\n",
        "\n",
        "        col_renames = {\n",
        "            'Timestamp': 'timestamp',\n",
        "            'Open': 'open',\n",
        "            'High': 'high',\n",
        "            'Low': 'low',\n",
        "            'Close': 'close',\n",
        "            'Volume_(BTC)': 'volume_btc',\n",
        "            'Volume_(Currency)': 'volume',\n",
        "            'Weighted_Price': 'vwap'\n",
        "        }\n",
        "\n",
        "        filenames = {\n",
        "            # 'bitstamp': 'bitstampUSD_1-min_data_2012-01-01_to_2018-06-27.csv',\n",
        "            'coinbase': data_file,\n",
        "            # 'coincheck': 'coincheckJPY_1-min_data_2014-10-31_to_2018-06-27.csv'\n",
        "        }\n",
        "        primary_table = 'coinbase'\n",
        "        self.target = f\"{primary_table}_close\"\n",
        "\n",
        "        df = None\n",
        "        for table, filename in filenames.items():\n",
        "#             df_ = pd.read_csv(path.join(path.dirname('/content/'), 'bitcoin-historical-data', filename))\n",
        "            df_ = pd.read_csv('/content/'+filename)\n",
        "            col_renames_ = {k: f\"{table}_{v}\" for k, v in col_renames.items()}\n",
        "            df_ = df_.rename(columns=col_renames_)\n",
        "            ts = f\"{table}_timestamp\"\n",
        "            df_[ts] = pd.to_datetime(df_[ts], unit='s')\n",
        "            df_ = df_.set_index(ts)\n",
        "            df = df_ if df is None else df.join(df_)\n",
        "\n",
        "        # too quiet before 2015, time waste. copy() to avoid pandas errors\n",
        "        df = df.loc['2015':].copy()\n",
        "\n",
        "        df['month'] = df.index.month\n",
        "        df['day'] = df.index.day\n",
        "        df['hour'] = df.index.hour\n",
        "\n",
        "        # TODO drop null rows? (inner join?)\n",
        "        # TODO arbitrage\n",
        "        # TODO indicators\n",
        "\n",
        "        diff_cols = [\n",
        "            f\"{table}_{k}\" for k in\n",
        "            'open high low close volume_btc volume vwap'.split(' ')\n",
        "            for table in filenames.keys()\n",
        "        ]\n",
        "        df[diff_cols] = df[diff_cols].pct_change()\\\n",
        "            .replace([np.inf, -np.inf], np.nan)\\\n",
        "            .ffill()  # .bfill()?\n",
        "        df = df.iloc[1:]\n",
        "        target = df[self.target]  # don't scale price changes; we use that in raw form later\n",
        "        df = pd.DataFrame(\n",
        "            robust_scale(df.values, quantile_range=(.1, 100-.1)),\n",
        "            columns=df.columns, index=df.index\n",
        "        )\n",
        "        df[self.target] = target\n",
        "\n",
        "        df['cash'], df['value'] = 0., 0.\n",
        "\n",
        "        self.df = df\n",
        "\n",
        "    def offset(self, ep, step):\n",
        "        return ep * self.ep_stride + step\n",
        "\n",
        "    def has_more(self, ep):\n",
        "        return self.offset(ep + 2, 0) + self.window < self.df.shape[0]\n",
        "        # return (ep + 1) * self.ep_stride + self.window < self.df.shape[0]\n",
        "\n",
        "    def get_data(self, ep, step):\n",
        "        offset = self.offset(ep, step)\n",
        "        X = self.df.iloc[offset:offset+self.window]\n",
        "        y = self.df.iloc[offset+self.window]\n",
        "        return X, y\n",
        "\n",
        "    def get_prices(self, ep, step):\n",
        "        offset = self.offset(ep, step)\n",
        "        return self.df.iloc[offset + self.window:self.ep_len][self.target]\n",
        "\n",
        "    def reset_cash_val(self):\n",
        "        self.df['cash'] = 0.\n",
        "        self.df['value'] = 0.\n",
        "\n",
        "    def set_cash_val(self, ep, step, cash, value):\n",
        "        offset = self.offset(ep, step)\n",
        "        self.df.cash.iloc[offset] = cash\n",
        "        self.df.value.iloc[offset] = value\n",
        "\n",
        "    def fetch_more(self):\n",
        "        raise_refactor()\n"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "id": "rg86Yx13-iEZ",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title btc_env\n",
        "\n",
        "\"\"\"BTC trading environment. Trains on BTC price history to learn to buy/sell/hold.\n",
        "\n",
        "This is an environment tailored towards TensorForce, not OpenAI Gym. Gym environments are\n",
        "a standard used by many projects (Baselines, Coach, etc) and so would make sense to use; and TForce is compatible with\n",
        "Gym envs. It's just that there's hoops to go through converting a Gym env to TForce, and it was ugly code. I actually\n",
        "had it that way, you can search through Git if you want the Gym env; but one day I decided \"I'm not having success with\n",
        "any of these other projects, TForce is the best - I'm just gonna stick to that\" and this approach was cleaner.\n",
        "\n",
        "I actually do want to try NervanaSystems/Coach, that one's new since I started developing. Will require converting this\n",
        "env back to Gym format. Anyone wanna give it a go?\n",
        "\"\"\"\n",
        "\n",
        "import random, time, requests, pdb, gdax, math, pickle, os, shutil, copy\n",
        "from sklearn.model_selection import TimeSeriesSplit\n",
        "from enum import Enum\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import talib.abstract as tlib\n",
        "from box import Box\n",
        "from tensorforce.environments import Environment\n",
        "from tensorforce.execution import Runner\n",
        "from sklearn import preprocessing\n",
        "# from data.data import Data, Exchange, EXCHANGE\n",
        "# from utils import raise_refactor\n",
        "\n",
        "\n",
        "class Mode(Enum):\n",
        "    TRAIN = 'train'\n",
        "    TEST = 'test'\n",
        "    LIVE = 'live'\n",
        "    TEST_LIVE = 'test_live'\n",
        "\n",
        "# See 6fc4ed2 for Scaling states/rewards\n",
        "\n",
        "\n",
        "class BitcoinEnv(Environment):\n",
        "    EPISODE_LEN = 1000\n",
        "\n",
        "    def __init__(self, hypers, cli_args={}):\n",
        "        \"\"\"Initialize hyperparameters (done here instead of __init__ since OpenAI-Gym controls instantiation)\"\"\"\n",
        "        self.hypers = h = Box(hypers)\n",
        "        self.cli_args = cli_args\n",
        "\n",
        "        # cash/val start @ about $3.5k each. You should increase/decrease depending on how much you'll put into your\n",
        "        # exchange accounts to trade with. Presumably the agent will learn to work with what you've got (cash/value\n",
        "        # are state inputs); but starting capital does effect the learning process.\n",
        "        self.start_cash, self.start_value = 5., 5.  # .4, .4\n",
        "\n",
        "        # We have these \"accumulator\" objects, which collect values over steps, over episodes, etc. Easier to keep\n",
        "        # same-named variables separate this way.\n",
        "        acc = dict(\n",
        "            ep=dict(\n",
        "                i=-1,  # +1 in reset, makes 0\n",
        "                returns=[],\n",
        "                uniques=[],\n",
        "            ),\n",
        "            step=dict(),  # setup in reset()\n",
        "        )\n",
        "        self.acc = Box(train=copy.deepcopy(acc), test=copy.deepcopy(acc))\n",
        "        self.data = Data(ep_len=self.EPISODE_LEN, arbitrage=h.custom.arbitrage, indicators={})\n",
        "\n",
        "        # gdax min order size = .01btc; kraken = .002btc\n",
        "        self.min_trade = {Exchange.GDAX: .01, Exchange.KRAKEN: .002}[EXCHANGE]\n",
        "        self.update_btc_price()\n",
        "\n",
        "        # Action space\n",
        "        # see {last_good_commit_ for action_types other than 'single_discrete'\n",
        "        # In single_discrete, we allow buy2%, sell2%, hold (and nothing else)\n",
        "        self.actions_ = dict(type='int', shape=(), num_actions=3)\n",
        "\n",
        "        # Observation space\n",
        "        # width = step-window (150 time-steps)\n",
        "        # height = nothing (1)\n",
        "        # channels = features/inputs (price actions, OHCLV, etc).\n",
        "        self.cols_ = self.data.df.shape[1]\n",
        "        shape = (h.custom.net.step_window, 1, self.cols_)\n",
        "        self.states_ = dict(type='float', shape=shape)\n",
        "\n",
        "    def __str__(self): return 'BitcoinEnv'\n",
        "\n",
        "    def close(self): pass\n",
        "\n",
        "    @property\n",
        "    def states(self): return self.states_\n",
        "\n",
        "    @property\n",
        "    def actions(self): return self.actions_\n",
        "\n",
        "    # We don't want random-seeding for reproducibilityy! We _want_ two runs to give different results, because we only\n",
        "    # trust the hyper combo which consistently gives positive results.\n",
        "    def seed(self, seed=None): return\n",
        "\n",
        "    def update_btc_price(self):\n",
        "        self.btc_price = 8000\n",
        "        # try:\n",
        "        #     self.btc_price = int(requests.get(f\"https://api.cryptowat.ch/markets/{EXCHANGE.value}/btcusd/price\").json()['result']['price'])\n",
        "        # except:\n",
        "        #     self.btc_price = self.btc_price or 8000\n",
        "\n",
        "    def xform_data(self, df):\n",
        "        # TODO here was autoencoder, talib indicators, price-anchoring\n",
        "        raise_refactor()\n",
        "\n",
        "    def get_next_state(self):\n",
        "        acc = self.acc[self.mode.value]\n",
        "        X, _ = self.data.get_data(acc.ep.i, acc.step.i)\n",
        "        return X.values[:, np.newaxis, :]  # height, width(nothing), depth\n",
        "\n",
        "    def reset(self):\n",
        "        acc = self.acc[self.mode.value]\n",
        "        acc.step.i = 0\n",
        "        acc.step.cash, acc.step.value = self.start_cash, self.start_value\n",
        "        acc.step.hold_value = self.start_value\n",
        "        acc.step.totals = Box(\n",
        "            trade=[self.start_cash + self.start_value],\n",
        "            hold=[self.start_cash + self.start_value]\n",
        "        )\n",
        "        acc.step.signals = []\n",
        "        if self.mode == Mode.TEST:\n",
        "            acc.ep.i = self.acc.train.ep.i + 1\n",
        "        elif self.mode == Mode.TRAIN:\n",
        "            acc.ep.i += 1\n",
        "\n",
        "        self.data.reset_cash_val()\n",
        "        self.data.set_cash_val(acc.ep.i, acc.step.i, 0., 0.)\n",
        "        return self.get_next_state()\n",
        "\n",
        "    def execute(self, action):\n",
        "        acc = self.acc[self.mode.value]\n",
        "        totals = acc.step.totals\n",
        "        h = self.hypers\n",
        "\n",
        "        act_pct = {\n",
        "            0: -.02,\n",
        "            1: 0,\n",
        "            2: .02\n",
        "        }[action]\n",
        "        act_btc = act_pct * (acc.step.cash if act_pct > 0 else acc.step.value)\n",
        "\n",
        "        fee = {\n",
        "            Exchange.GDAX: 0.0025,  # https://support.gdax.com/customer/en/portal/articles/2425097-what-are-the-fees-on-gdax-\n",
        "            Exchange.KRAKEN: 0.0026  # https://www.kraken.com/en-us/help/fees\n",
        "        }[EXCHANGE]\n",
        "\n",
        "        # Perform the trade. In training mode, we'll let it dip into negative here, but then kill and punish below.\n",
        "        # In testing/live, we'll just block the trade if they can't afford it\n",
        "        if act_pct > 0:\n",
        "            if acc.step.cash < self.min_trade:\n",
        "                act_btc = -(self.start_cash + self.start_value)\n",
        "            elif act_btc < self.min_trade:\n",
        "                act_btc = 0\n",
        "            else:\n",
        "                acc.step.value += act_btc - act_btc*fee\n",
        "            acc.step.cash -= act_btc\n",
        "\n",
        "        elif act_pct < 0:\n",
        "            if acc.step.value < self.min_trade:\n",
        "                act_btc = -(self.start_cash + self.start_value)\n",
        "            elif abs(act_btc) < self.min_trade:\n",
        "                act_btc = 0\n",
        "            else:\n",
        "                acc.step.cash += abs(act_btc) - abs(act_btc)*fee\n",
        "            acc.step.value -= abs(act_btc)\n",
        "\n",
        "        acc.step.signals.append(float(act_btc))  # clipped signal\n",
        "        # acc.step.signals.append(np.sign(act_pct))  # indicates an attempted trade\n",
        "\n",
        "        # next delta. [1,2,2].pct_change() == [NaN, 1, 0]\n",
        "        # pct_change = self.prices_diff[acc.step.i + 1]\n",
        "        _, y = self.data.get_data(acc.ep.i, acc.step.i)  # TODO verify\n",
        "        pct_change = y[self.data.target]\n",
        "\n",
        "        acc.step.value += pct_change * acc.step.value\n",
        "        total_now = acc.step.value + acc.step.cash\n",
        "        totals.trade.append(total_now)\n",
        "\n",
        "        # calculate what the reward would be \"if I held\", to calculate the actual reward's _advantage_ over holding\n",
        "        hold_before = acc.step.hold_value\n",
        "        acc.step.hold_value += pct_change * hold_before\n",
        "        totals.hold.append(acc.step.hold_value + self.start_cash)\n",
        "\n",
        "        reward = 0\n",
        "\n",
        "        acc.step.i += 1\n",
        "\n",
        "        self.data.set_cash_val(\n",
        "            acc.ep.i, acc.step.i,\n",
        "            acc.step.cash/self.start_cash,\n",
        "            acc.step.value/self.start_value\n",
        "        )\n",
        "        next_state = self.get_next_state()\n",
        "\n",
        "        terminal = int(acc.step.i + 1 >= self.EPISODE_LEN)\n",
        "        if acc.step.value < 0 or acc.step.cash < 0:\n",
        "            terminal = True\n",
        "        if terminal and self.mode in (Mode.TRAIN, Mode.TEST):\n",
        "            # We're done.\n",
        "            acc.step.signals.append(0)  # Add one last signal (to match length)\n",
        "            reward = self.get_return()\n",
        "            if np.unique(acc.step.signals).shape[0] == 1:\n",
        "                reward = -(self.start_cash + self.start_value)  # slam if you don't do anything\n",
        "\n",
        "        if terminal and self.mode in (Mode.LIVE, Mode.TEST_LIVE):\n",
        "            raise_refactor()\n",
        "\n",
        "        # if acc.step.value <= 0 or acc.step.cash <= 0: terminal = 1\n",
        "        return next_state, terminal, reward\n",
        "\n",
        "    def get_return(self, adv=True):\n",
        "        acc = self.acc[self.mode.value]\n",
        "        totals = acc.step.totals\n",
        "        trade = (totals.trade[-1] / totals.trade[0] - 1)\n",
        "        hold = (totals.hold[-1] / totals.hold[0] - 1)\n",
        "        return trade - hold if adv else trade\n",
        "\n",
        "    def episode_finished(self, runner):\n",
        "        if self.mode == Mode.TRAIN: return True\n",
        "\n",
        "        acc = self.acc.test\n",
        "        totals = acc.step.totals\n",
        "        signals = np.array(acc.step.signals)\n",
        "        n_uniques = np.unique(signals).shape[0]\n",
        "        ret = self.get_return()\n",
        "        hold_ret = totals.hold[-1] / totals.hold[0] - 1\n",
        "\n",
        "        acc.ep.returns.append(float(ret))\n",
        "        acc.ep.uniques.append(n_uniques)\n",
        "\n",
        "        # Print (limit to note-worthy)\n",
        "        lt_0 = (signals < 0).sum()\n",
        "        eq_0 = (signals == 0).sum()\n",
        "        gt_0 = (signals > 0).sum()\n",
        "        completion = int(acc.ep.i * self.data.ep_stride / self.data.df.shape[0] * 100)\n",
        "        steps = f\"\\tSteps: {acc.step.i}\"\n",
        "\n",
        "        fm = '%.3f'\n",
        "        print(f\"{completion}%{steps}\\tTrade: {fm%ret}\\tHold: {fm%hold_ret}\\tTrades:\\t{lt_0}[<0]\\t{eq_0}[=0]\\t{gt_0}[>0]\")\n",
        "        return True\n",
        "\n",
        "    def run_deterministic(self, runner, print_results=True):\n",
        "        next_state, terminal = self.reset(), False\n",
        "        while not terminal:\n",
        "            next_state, terminal, reward = self.execute(runner.agent.act(next_state, deterministic=True, independent=True))\n",
        "        if print_results: self.episode_finished(None)\n",
        "\n",
        "    def train_and_test(self, agent):\n",
        "        runner = Runner(agent=agent, environment=self)\n",
        "        train_steps = 20000  # TODO something self.data.df.shape[0]... self.EPISODE_LEN...\n",
        "\n",
        "        try:\n",
        "            while self.data.has_more(self.acc.train.ep.i):\n",
        "                self.mode = Mode.TRAIN\n",
        "                # max_episode_timesteps not required, since we kill on (cash|value)<0 or max_repeats\n",
        "                runner.run(timesteps=train_steps)\n",
        "                self.mode = Mode.TEST\n",
        "                self.run_deterministic(runner, print_results=True)\n",
        "        except IndexError:\n",
        "            # FIXME data.has_more() issues\n",
        "            pass\n",
        "        except KeyboardInterrupt:\n",
        "            # Lets us kill training with Ctrl-C and skip straight to the final test. This is useful in case you're\n",
        "            # keeping an eye on terminal and see \"there! right there, stop you found it!\" (where early_stop & n_steps\n",
        "            # are the more methodical approaches)\n",
        "            print('Keyboard interupt, killing training')\n",
        "            pass\n",
        "\n",
        "    def run_live(self, agent, test=True):\n",
        "        raise_refactor()\n"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "id": "Wd6DKBLECSBM",
        "colab_type": "code",
        "colab": {},
        "cellView": "form"
      },
      "cell_type": "code",
      "source": [
        "#@title hypersearch\n",
        "\n",
        "import argparse, json, math, time, pdb, os, copy, uuid\n",
        "from pprint import pprint\n",
        "from box import Box\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import pickle\n",
        "import tensorflow as tf\n",
        "from sqlalchemy.sql import text\n",
        "from tensorforce import TensorForceError\n",
        "from tensorforce.agents import agents as agents_dict\n",
        "from tensorforce.core.networks import layer as TForceLayers\n",
        "from tensorforce.core.networks.network import LayeredNetwork\n",
        "\n",
        "from sqlalchemy.dialects import postgresql as psql\n",
        "from hyperopt import fmin, tpe, hp, Trials\n",
        "from hyperopt.pyll.base import scope\n",
        "\n",
        "# from btc_env import BitcoinEnv\n",
        "# import utils\n",
        "# from data import data\n",
        "\n",
        "\n",
        "def network_spec(hypers):\n",
        "    \"\"\"Builds an array of dicts that conform to TForce's network specification (see their docs) by mix-and-matching\n",
        "    different network hypers\n",
        "    \"\"\"\n",
        "    net = Box(hypers['net'])\n",
        "    batch_norm = {\"type\": \"tf_layer\", \"layer\": \"batch_normalization\"}\n",
        "    arr = []\n",
        "\n",
        "    def add_dense(s):\n",
        "        dense = {\n",
        "            'size': s,\n",
        "            'l2_regularization': net.l2,\n",
        "            'l1_regularization': net.l1\n",
        "        }\n",
        "        if not net.batch_norm:\n",
        "            arr.append({'type': 'dense', 'activation': net.activation, **dense})\n",
        "            return\n",
        "        arr.append({'type': 'linear', **dense})\n",
        "        arr.append(batch_norm)\n",
        "        arr.append({'type': 'nonlinearity','name': net.activation})\n",
        "        # FIXME dense dropout bug https://github.com/reinforceio/tensorforce/issues/317\n",
        "        if net.dropout: arr.append({'type': 'dropout', 'rate': net.dropout})\n",
        "\n",
        "    # Mid-layer\n",
        "    for i in range(net.depth_mid):\n",
        "        arr.append({\n",
        "            'size': net.width,\n",
        "            'window': (net.kernel_size, 1),\n",
        "            'stride': (net.stride, 1),\n",
        "            'type': 'conv2d',\n",
        "            # 'bias': net.bias,\n",
        "            'l2_regularization': net.l2,\n",
        "            'l1_regularization': net.l1\n",
        "        })\n",
        "    arr.append({'type': 'flatten'})\n",
        "\n",
        "    # Post Dense layers\n",
        "    if net.flat_dim:\n",
        "        fc_dim = net.width * (net.step_window / (net.depth_mid * net.stride))\n",
        "    else:\n",
        "        fc_dim = net.width * 4\n",
        "    for i in range(net.depth_post):\n",
        "        size = fc_dim / (i + 1) if net.funnel else fc_dim\n",
        "        add_dense(int(size))\n",
        "\n",
        "    return arr\n",
        "\n",
        "\n",
        "# @scope.define\n",
        "# def two_to_the(x):\n",
        "#     return 2**int(x)\n",
        "\n",
        "# @scope.define\n",
        "# def ten_to_the_neg(x):\n",
        "#     return 10**-int(x)\n",
        "\n",
        "# @scope.define\n",
        "# def min_threshold(x, thresh, fallback):\n",
        "#     \"\"\"Returns x or `fallback` if it doesn't meet the threshold. Note, if you want to turn a hyper \"off\" below,\n",
        "#     set it to \"outside the threshold\", rather than 0.\n",
        "#     \"\"\"\n",
        "#     return x if (x and x > thresh) else fallback\n",
        "\n",
        "# @scope.define\n",
        "# def min_ten_neg(x, thresh, fallback):\n",
        "#     \"\"\"Returns 10**-x, or `fallback` if it doesn't meet the threshold. Note, if you want to turn a hyper \"off\" below,\n",
        "#     set it to \"outside the threshold\", rather than 0.\n",
        "#     \"\"\"\n",
        "#     x = 10**-x\n",
        "#     return x if (x and x > thresh) else fallback\n",
        "\n",
        "def post_process(hypers):\n",
        "    hypers = copy.deepcopy(hypers)  # don't modify original\n",
        "    agent, custom = hypers['ppo_agent'], hypers['custom']\n",
        "\n",
        "    o = agent['update_mode']\n",
        "    o['frequency'] = math.ceil(o['batch_size'] / o['frequency'])\n",
        "    # agent['memory']['capacity'] = BitcoinEnv.EPISODE_LEN * o['batch_size']\n",
        "    agent['memory']['capacity'] = BitcoinEnv.EPISODE_LEN * MAX_BATCH_SIZE + 1\n",
        "\n",
        "    agent.update(agent['baseline_stuff'])\n",
        "    del agent['baseline_stuff']\n",
        "    if agent['baseline_mode']:\n",
        "        o = agent['baseline_optimizer']\n",
        "        # o['num_steps'] = agent['optimization_steps']\n",
        "        o['optimizer']['learning_rate'] = agent['step_optimizer']['learning_rate']\n",
        "        o['optimizer']['type'] = agent['step_optimizer']['type']\n",
        "\n",
        "        agent['baseline']['network'] = network_spec(custom)\n",
        "        # if main['gae_lambda']: main['gae_lambda'] = main['discount']\n",
        "    return hypers\n",
        "\n",
        "\n",
        "# Most hypers come directly from tensorforce/tensorforce/agents/ppo_agent.py, see that for documentation\n",
        "# Note: Name this something other than \"hypers\" (eg \"space\"), easy conflicts with other methods\n",
        "space = {}\n",
        "space['agent'] = {\n",
        "    # 'states_preprocessing': None,\n",
        "    # 'actions_exploration': None,\n",
        "    # 'reward_preprocessing': None,\n",
        "\n",
        "    # I'm pretty sure we don't want to experiment any less than .99 for non-terminal reward-types (which are 1.0).\n",
        "    # .99^500 ~= .6%, so looses value sooner than makes sense for our trading horizon. A trade now could effect\n",
        "    # something 2-5k steps later. So .999 is more like it (5k steps ~= .6%)\n",
        "    'discount': 1.,  # hp.uniform('discount', .9, .99),\n",
        "}\n",
        "\n",
        "MAX_BATCH_SIZE = 15\n",
        "space['memory_model'] = {\n",
        "    'update_mode': {\n",
        "        'unit': 'episodes',\n",
        "        'batch_size': scope.int(hp.quniform('batch_size', 1, MAX_BATCH_SIZE, 1)),  # 5 FIXME\n",
        "        'frequency': scope.int(hp.quniform('frequency', 1, 3, 1)),  # t-shirt sizes, reverse order\n",
        "    },\n",
        "\n",
        "    'memory': {\n",
        "        'type': 'latest',\n",
        "        'include_next_states': False,\n",
        "        'capacity': None,  # 5000  # BitcoinEnv.EPISODE_LEN * MAX_BATCH_SIZE,  # hp.uniform('capacity', 2000, 20000, 500)\n",
        "    }\n",
        "}\n",
        "\n",
        "space['distribution_model'] = {\n",
        "    # 'distributions': None,\n",
        "    'entropy_regularization': hp.choice('entropy_regularization', [None, .01]), # scope.min_ten_neg(hp.uniform('entropy_regularization', 0., 5.), 1e-4, .01),\n",
        "    # 'variable_noise': TODO\n",
        "}\n",
        "\n",
        "space['pg_model'] = {\n",
        "    'baseline_stuff': hp.choice('baseline_stuff', [\n",
        "        {'baseline_mode': None},\n",
        "        {\n",
        "            'baseline': {'type': 'custom'},\n",
        "            'baseline_mode': 'states',\n",
        "            'baseline_optimizer': {\n",
        "                'type': 'multi_step',\n",
        "                # Consider having baseline_optimizer learning hypers independent of the main learning hypers.\n",
        "                # At least with PPO, it seems the step_optimizer learning hypers function quite diff0erently than\n",
        "                # expected; where baseline_optimizer's function more as-expected. TODO Investigate.\n",
        "                'num_steps': scope.int(hp.quniform('num_steps', 1, 20, 1)),  # 5 FIXME\n",
        "                'optimizer': {}  # see post_process()\n",
        "            },\n",
        "            'gae_lambda': hp.choice('gae_lambda', [1., None]),\n",
        "            # scope.min_threshold(hp.uniform('gae_lambda', .8, 1.), .9, None)\n",
        "        }\n",
        "    ])\n",
        "}\n",
        "space['pg_prob_ration_model'] = {\n",
        "    'likelihood_ratio_clipping': .2,  # scope.min_threshold(hp.uniform('likelihood_ratio_clipping', 0., 1.), .05, None),\n",
        "}\n",
        "\n",
        "space['ppo_model'] = {\n",
        "    # Doesn't seem to matter; consider removing\n",
        "    'step_optimizer': {\n",
        "        'type': 'adam',  # hp.choice('type', ['nadam', 'adam']),\n",
        "        'learning_rate': scope.ten_to_the_neg(hp.uniform('learning_rate', 2., 5.)),\n",
        "    },\n",
        "\n",
        "    'optimization_steps': scope.int(hp.quniform('optimization_steps', 1, 50, 1)),  # 5 FIXME\n",
        "\n",
        "    'subsampling_fraction': .1,  # hp.uniform('subsampling_fraction', 0.,  1.),\n",
        "}\n",
        "\n",
        "ppo_agent = {\n",
        "    **space['agent'],\n",
        "    **space['memory_model'],\n",
        "    **space['distribution_model'],\n",
        "    **space['pg_model'],\n",
        "    **space['pg_prob_ration_model'],\n",
        "    **space['ppo_model']\n",
        "}\n",
        "\n",
        "space = {\n",
        "    'ppo_agent': ppo_agent, # 'vpg_agent': ppo_agent, 'trpo_agent': ppo_agent,\n",
        "    # TODO dqn, ddpg (hierarchical hyperopt)\n",
        "}\n",
        "\n",
        "space['custom'] = {\n",
        "    'agent': 'ppo_agent',\n",
        "\n",
        "    # Use a handful of TA-Lib technical indicators (SMA, EMA, RSI, etc). Which indicators used and for what time-frame\n",
        "    # not optimally chosen at all; just figured \"if some randos are better than nothing, there's something there and\n",
        "    # I'll revisit\". Help wanted.\n",
        "\n",
        "    # Currently disabling indicators in general. A good CNN should \"see\" those automatically in the window, right?\n",
        "    # If I'm wrong, experiment with these (see commit 6fc4ed2)\n",
        "    # TODO indicators overhaul\n",
        "    'indicators_count': 0,\n",
        "    'indicators_window': 0,\n",
        "\n",
        "    # This is special. \"Risk arbitrage\" is the idea of watching two exchanges for the same\n",
        "    # instrument's price. Let's say BTC is $10k in GDAX and $9k in Kraken. Well, Kraken is a smaller / less popular\n",
        "    # exchange, so it tends to play \"follow the leader\". Ie, Kraken will likely try to get to $10k\n",
        "    # to match GDAX (oversimplifying obviously). This is called \"risk arbitrage\" (\"arbitrage\"\n",
        "    # by itself is slightly different, not useful for us). Presumably that's golden info for the neural net:\n",
        "    # \"Kraken < GDAX? Buy in Kraken!\". It's not a gaurantee, so this is a hyper in hypersearch.py.\n",
        "    # Incidentally I have found it detrimental, I think due to imperfect time-phase alignment (arbitrage code in\n",
        "    # data.py) which makes it hard for the net to follow.\n",
        "    # Note: not valuable if GDAX is main (ie, not valuable if the bigger exchange is the main, only\n",
        "    # if the smaller exchange (eg Kraken) is main)\n",
        "    'arbitrage': False,  # see 6fc4ed2\n",
        "\n",
        "    # single = one action (-$x to +$x). multi = two actions: (buy|sell|hold) and (how much?). all_or_none = buy/sell\n",
        "    # w/ all the cash or value owned\n",
        "    'action_type': 'single_discrete',  # hp.choice('action_type', ['single_discrete', 'single_continuous', 'multi']),\n",
        "\n",
        "    # Should rewards be as-is (PNL), or \"how much better than holding\" (advantage)? if `sharpe` then we discount 1.0\n",
        "    # and calculate sharpe score at episode-terminal.\n",
        "    # See 6fc4ed2 for handling Sharpe rewards\n",
        "    'reward_type': 'sharpe',  # hp.choice('reward_type', ['raw', 'advantage', 'sharpe']),\n",
        "\n",
        "}\n",
        "space['custom']['net'] = {\n",
        "    # Conv / LSTM layers\n",
        "    'depth_mid': scope.int(hp.quniform('depth_mid', 1, 4, 1)),\n",
        "\n",
        "    # Dense layers\n",
        "    'depth_post': scope.int(hp.quniform('depth_post', 1, 3, 1)),\n",
        "\n",
        "    # Network depth, in broad-strokes of 2**x (2, 4, 8, 16, 32, 64, 128, 256, 512, ..) just so you get a feel for\n",
        "    # small-vs-large. Later you'll want to fine-tune.\n",
        "    'width': scope.two_to_the(hp.quniform('width', 4, 6, 1)),\n",
        "\n",
        "    'batch_norm': hp.choice('batch_norm', [True, False]),\n",
        "\n",
        "    # Whether to expand-in and shrink-out the nueral network. You know the look, narrower near the inputs, gets wider\n",
        "    # in the hidden layers, narrower again on hte outputs.\n",
        "    'funnel': True,  # hp.choice('funnel', [True, False]),\n",
        "\n",
        "    # Is the first FC layer the same size as the last flattened-conv? Or is it something much smaller,\n",
        "    # like depth_mid*4?\n",
        "    'flat_dim': hp.choice('funnel', [True, False]),\n",
        "\n",
        "    # tanh vs \"the relu family\" (relu, selu, crelu, elu, *lu). Broad-strokes here by just pitting tanh v relu; then,\n",
        "    # if relu wins you can fine-tune \"which type of relu\" later.\n",
        "    'activation': hp.choice('activation', ['tanh', 'relu']),\n",
        "\n",
        "    # Regularization: Dropout, L1, L2. You'd be surprised (or not) how important is the proper combo of these. The RL\n",
        "    # papers just role L2 (.001) and ignore the other two; but that hasn't jived for me. Below is the best combo I've\n",
        "    # gotten so far, and I'll update as I go.\n",
        "    # 'dropout': scope.min_threshold(hp.uniform('dropout', 0., .5), .1, None),\n",
        "    # 'l2': scope.min_ten_neg(hp.uniform('l2', 0., 7.), 1e-6, 0.),\n",
        "    # 'l1': scope.min_ten_neg(hp.uniform('l1', 0., 7.), 1e-6, 0.),\n",
        "    'dropout': None,\n",
        "    'l2': 0.,\n",
        "    'l1': 0.,\n",
        "\n",
        "    # LSTM at {last_good_commit}\n",
        "\n",
        "    # T-shirt size window-sizes, smaller # = more destructive. See comments in build_net_spec()\n",
        "    'kernel_size': hp.choice('window', [3, 5]),\n",
        "\n",
        "    # How many ways to divide a window? 1=no-overlap, 2=half-overlap (smaller # = more destructive). See comments\n",
        "    # in build_net_spec()\n",
        "    'stride': 2,\n",
        "\n",
        "    # Size of the window to look at w/ the CNN (ie, width of the image). Would like to have more than 400 \"pixels\" here,\n",
        "    # but it causes memory issues the way PPO's MemoryModel batches things. This is made up for via indicators\n",
        "    'step_window': 300,  # scope.int(hp.quniform('step_window', 200, 500, 50)),\n",
        "}\n",
        "\n",
        "# TODO restore get_winner() from git & fix-up\n",
        "\n",
        "def main():\n",
        "#     parser = argparse.ArgumentParser()\n",
        "#     add_common_args(parser)\n",
        "#     args = parser.parse_args()\n",
        "    args = []\n",
        "\n",
        "    # Specify the \"loss\" function (which we'll maximize) as a single rl_hsearch instantiate-and-run\n",
        "    def loss_fn(hypers):\n",
        "        processed = post_process(hypers)\n",
        "        network = network_spec(processed['custom'])\n",
        "\n",
        "        agent = processed['ppo_agent']\n",
        "        ## GPU split\n",
        "        gpu_split = 1\n",
        "        if gpu_split != 1:\n",
        "            fraction = .9 / gpu_split if gpu_split > 1 else gpu_split\n",
        "            session_config = tf.ConfigProto(gpu_options=tf.GPUOptions(per_process_gpu_memory_fraction=fraction))\n",
        "            agent['execution'] = {'type': 'single', 'session_config': session_config, 'distributed_spec': None}\n",
        "\n",
        "        pprint(processed)\n",
        "        pprint(network)\n",
        "  \n",
        "        env = BitcoinEnv(processed, args)\n",
        "        agent = agents_dict['ppo_agent'](\n",
        "            states=env.states,\n",
        "            actions=env.actions,\n",
        "            network=network,\n",
        "            **agent\n",
        "        )\n",
        "\n",
        "        env.train_and_test(agent)\n",
        "\n",
        "        acc = env.acc.test\n",
        "        adv_avg = calculate_score(acc.ep.returns)\n",
        "        print(hypers, f\"\\nScore={adv_avg}\\n\\n\")\n",
        "\n",
        "        df = pd.DataFrame([dict(\n",
        "            id=uuid.uuid4(),\n",
        "            hypers=json.dumps(hypers),\n",
        "            returns=list(acc.ep.returns),\n",
        "            uniques=list(acc.ep.uniques),\n",
        "            prices=list(env.data.get_prices(acc.ep.i, 0)),\n",
        "            signals=list(acc.step.signals),\n",
        "        )]).set_index('id')\n",
        "        dtype = {\n",
        "            'hypers': psql.JSONB,\n",
        "            **{k: psql.ARRAY(psql.DOUBLE_PRECISION) for k in ['returns', 'signals', 'prices', 'uniques']},\n",
        "        }\n",
        "        with data.engine_runs.connect() as conn:\n",
        "            df.to_sql('runs', conn, if_exists='append', index_label='id', dtype=dtype)\n",
        "\n",
        "        # TODO restore save_model() from git\n",
        "\n",
        "        agent.close()\n",
        "        env.close()\n",
        "        return -adv_avg  # maximize\n",
        "\n",
        "        # TODO restore fetching between runs so can pick up where left off, or\n",
        "        # get updates from other servers\n",
        "\n",
        "    # set initial max_eval, attempt to load a saved trials object from pickle, if that fails start fresh.\n",
        "    # grab how many trials were previously run and add max_evals to it for the next run.\n",
        "    # this allows the hyper parameter search to resume where it left off last.\n",
        "    # TODO save trials to SQL table and restore from there instead of local pickle. \n",
        "    max_evals = 20\n",
        "    try:\n",
        "        trialPickle = open('./trial.pickle','rb')\n",
        "        trials = pickle.load(trialPickle)\n",
        "        max_evals = len(trials.trials) + max_evals\n",
        "    except:\n",
        "        trials = Trials()\n",
        "\n",
        "    best = fmin(loss_fn, space=space, algo=tpe.suggest, max_evals=max_evals, trials=trials)\n",
        "\n",
        "    with open('./trial.pickle', 'wb') as f:\n",
        "            pickle.dump(trials, f)\n",
        "\n",
        "\n",
        "if __name__ == '__main__':\n",
        "    main()\n"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}