<jupyter_start><jupyter_text>Gemini Trading Bot<jupyter_code>!pip install geminipy websocket-client elasticsearch kafka
import json
import pandas as pd
from datetime import datetime, timedelta
import time
from geminipy import Geminipy
from elasticsearch import Elasticsearch
from kafka import KafkaProducer, KafkaConsumer<jupyter_output><empty_output><jupyter_text>Load Forecasts<jupyter_code># load in csv of forecasts
fcst = pd.read_csv('2018_full_trade_forecast.csv')

# generate a datetime field from the string timestamp `ds`
fcst['timestamp'] = fcst.ds.apply(lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'))

# set the index as the timestamp so we can easily make lookups based on time
fcst = fcst.set_index('timestamp')[['side']]
import random
sides = ['buy','sell']
now = datetime.now()
ts_prior = datetime(year=now.year, month=now.month, day=now.day, hour=now.hour, minute=now.minute)
ts = ts_prior + timedelta(minutes=1)

for i in range(100):
    fcst.ix[ts + timedelta(minutes=i)].side = random.choice(sides)
    print(fcst.ix[ts + timedelta(minutes=i)].side)<jupyter_output><empty_output><jupyter_text>Set up Elasticsearch<jupyter_code># first set up Elasticsearch connection
# by default we connect to elasticsearch:9200 
# since we are running this notebook from the Spark-Node we need to use `elasticsearch` instead of `localhost`
# as this is the name of the docker container running Elasticsearch
es = Elasticsearch('elasticsearch:9200')

# if the `gemini` index does not exist, create it
if not es.indices.exists('gemini'):
    es.indices.create(index='gemini')<jupyter_output><empty_output><jupyter_text>Set up Kafka<jupyter_code>consumer = KafkaConsumer('spark.out',
                         bootstrap_servers=['kafka-node:9092'],
                        )<jupyter_output><empty_output><jupyter_text>Set up Gemini<jupyter_code>gemini_api_key = 'some_key'
gemini_api_secret = 'some_secret'<jupyter_output><empty_output><jupyter_text>Set the amount to trade[Gemini's minimum order size](https://docs.gemini.com/rest-api/symbols-and-minimums) for BTC is currently 0.00001. At the time of writing that is about `$0.15`. Let's try trading at 0.0001 which is about `$1.53`.<jupyter_code># trade 50% of our account balance
trade_pct = 0.5

# require that price_volume threshold be between 0.1 and 2
# this indicates that sell side volume is 1/10th and double that of buy side
# suggesting (relatively) normal liquidity thresholds.
# if sell side liquidity is above 2x that of the buy side we might think that 
# price may not move in the next minute.
threshold = (0.1, 2)
con = Geminipy(api_key=gemini_api_key, secret_key=gemini_api_secret, live=False)<jupyter_output><empty_output><jupyter_text>Define functions<jupyter_code>def get_btc_balance(con):
    for i in con.balances().json():
        if i['currency'] == 'BTC':
            i['amount'] = float(i['amount'])
            i['available'] = float(i['available'])
            i['availableForWithdrawal'] = float(i['availableForWithdrawal'])
            i['doc_type'] = 'balance'
            return i
def get_usd_balance(con):
    for i in con.balances().json():
        if i['currency'] == 'USD':
            i['amount'] = float(i['amount'])
            i['available'] = float(i['available'])
            i['availableForWithdrawal'] = float(i['availableForWithdrawal'])
            i['doc_type'] = 'balance'
            return i
def get_ticker(con):
    '''
    NOTE: We need to ensure that numbers are numbers, not strings, for ES.
    Otherwise we would need to specify a mapping.
    '''
    ticker = con.pubticker().json()
    ticker['ask'] = float(ticker['ask'])
    ticker['bid'] = float(ticker['bid'])
    ticker['last'] = float(ticker['last'])
    ticker['volume_BTC'] = float(ticker['volume'].pop('BTC'))
    ticker['volume_USD'] = float(ticker['volume'].pop('USD'))
    ticker['timestamp'] = datetime.fromtimestamp(ticker['volume'].pop('timestamp')/1000)
    ticker.pop('volume')
    ticker['doc_type'] = 'ticker'
    return ticker
def format_order(order):
    '''
    NOTE: We need to ensure that numbers are numbers, not strings, for ES.
    Otherwise we would need to specify a mapping.
    
    Also, we convert epoch time to Python datetime which is natively recognized as a time field by ES.
    Epoch time, without a custom mapping, would appear as a number.
    '''
    order_dict = order.json()
    try:
        order_dict['timestamp'] = datetime.fromtimestamp(int(order_dict['timestamp']))
    except:
        # no timestamp field, try timestampms
        try:
            order_dict['timestamp'] = datetime.fromtimestamp(int(order_dict['timestampms']))
        except:
            # no timestampms, set to now
            order_dict['timestamp'] = datetime.now()
    order_dict['price'] = float(order_dict['price'])
    order_dict['original_amount'] = float(order_dict['original_amount'])
    order_dict['remaining_amount'] = float(order_dict['remaining_amount'])
    order_dict['avg_execution_price'] = float(order_dict['avg_execution_price'])
    order_dict['executed_amount'] = float(order_dict['executed_amount'])
    order_dict['doc_type'] = 'order'
    return order_dict
def make_order(con, amount, side, ticker):
    # if we are buying we should take the last ask price
    if side == 'buy':
        bid_ask = 'ask'

    # if we are selling we should take the last bid price        
    elif side == 'sell':
        bid_ask = 'bid'
        
    order = con.new_order(amount = amount, # set order amount
                      price = ticker[bid_ask], # grab latest bid or ask price
                      side = side, # set side (either buy/sell)
                      options = ['immediate-or-cancel'] # take liquidity with an immediate trade
                     )
    # format order for Elasticsearch
    order_dict = format_order(order)
    
    return order_dict
def lookup_side(fcst):
    # get current timestamp
    now = datetime.now()
    
    # we must add 1 minute to the time we lookup
    # this is because our forecasts are for whether we should have bought/sold in a given minute
    # so, we want trade with our prediction in mind (hence, add 1 minute)
    ts_prior = datetime(year=now.year, month=now.month, day=now.day, hour=now.hour, minute=now.minute)
    ts = ts_prior + timedelta(minutes=1)
    
    last_fcst = fcst.ix[ts_prior].side
    curr_fcst = fcst.ix[ts].side
    return {'last_fcst': last_fcst, 'side': curr_fcst} # return looked up side (either buy or sell)
def get_last_buy(es):
    query = {
      "sort" : [
            { "timestamp" : {"order" : "desc"}}
        ],
      "query": {
        "bool": {
          "must": [
            {
              "match_phrase": {
                "side": {
                  "query": "buy"
                }
              }
            }
          ]
        }
      }
    }

    results = es.search(index='gemini', doc_type='gem', body=query)
    last_buy = results['hits']['hits'][0]
    last_price = last_buy['_source']['price']
    return last_price
def score_trade(es, order, balance):
    last_price = get_last_buy(es)
    
    # get timestamp 
    now = datetime.now()
    timestamp = datetime(year=now.year, month=now.month, day=now.day, hour=now.hour, minute=now.minute)
    
    # grab sell order price
    curr_price = order['price']
    profit = (curr_price - last_price) * balance
    
    return {'profit': profit, 'timestamp': timestamp, 'doc_type': 'score'}
def process_msg(message, fcst, con, es, threshold, trade_pct, last_price, traded):
    # load message as json
    msg = json.loads(message.value)
    
    # convert msg epoch time to datetime
    msg['timestamp'] = datetime.fromtimestamp(msg['timestamp'])
    
    # get ticker
    ticker = get_ticker(con)

    # check side from model
    lookup = lookup_side(fcst)
    side = lookup['side']
    last_fcst = lookup['last_fcst']

    msg['doc_type'] = msg.pop('type')

    # index msg to ES
    es.index(index='gemini', doc_type='gem', body=msg)
    
    # index account balance to ES
    es.index(index='gemini', doc_type='gem',body=get_btc_balance(con))
    
    # index ticker data to ES
    es.index(index="gemini", doc_type='gem', body=ticker)    
    
    if side == 'sell':
        balance = get_btc_balance(con)['amount']
        if balance > 0:
            order = make_order(con, balance, 'sell', ticker)
            score = score_trade(es, order, balance)
            print('Last trade yieled ${} in profit.'.format(score['profit']))

            # index sell order data to ES
            es.index(index="gemini", doc_type='gem', body=order)
            
            # index score to ES
            es.index(index="gemini", doc_type='gem', body=score)
        else:
            print('No balance to sell.')
        return ticker['last'], False

    else: # buy side

        # only trade if we didn't just trade
        if not traded:

            # Execute trade
            if msg['doc_type'] == 'price_volume':

                # only execute trade if price_volume within threshold
                if msg['value'] > threshold[0] and msg['value'] < threshold[1]:

                    # trade with 90% of our balance
                    trade_amount = round((get_usd_balance(con)['available'] * trade_pct) / get_ticker(con)['last'],4)
                    print(trade_amount)
                    
                    # execute order
                    order = make_order(con, trade_amount, side, ticker)

                    # index order data to ES
                    es.index(index="gemini", doc_type='gem', body=order)

                    # print order
                    order.pop('timestamp')
                    print(json.dumps(order, sort_keys=True,
                        indent=4, separators=(',', ': ')))
                    print('\n')

                    return ticker['last'], True
                else:
                    return last_price, False
            else:
                return last_price, False
        else:
            return last_price, True
# get last price for keeping track of performance
last_price = get_ticker(con)['last']
traded = False

for message in consumer:
    try:a
        last_price, traded = process_msg(message, fcst, con, es, threshold, trade_pct, last_price, traded)
    except Exception as e:
        print("Error occurred: {}".format(e))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11870122362462665, 'fraction_numerical': 0.006358994122747856, 'mean_word_length': 2.8189845474613686, 'pattern_counts': {'":': 6, '<': 30, '<?xml version=': 0, '>': 31, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '660165', 'license': 'MIT', 'n_tokens_mistral': 3247, 'n_tokens_neox': 3107, 'n_words': 852, 'path': 'chapter-7/Gemini-Trading.ipynb', 'repo_name': 'brandomr/realtimecrypto', 'size': '10379'}