Quant Trading Robot #1 - API Calls I'm (re)building a Quantitative Trading Robot in Python.



Description:

This is my module for API Calls. The API documentation can be found here. This is the direct interface layer between my code and the exchange server. Any abstraction on top of the direct API calls is in a separate module.

Classes:

GDAXExchangeAuth - Authentication for restricted API Calls

Functions:

get_server_time - Retrieve the curent server time

get_account - Retrieve Account balances

get_book - Retrieve the order book for a specified currency pair

get_fills_after_order - Retrieve all filled orders that have occurred after some specific order id.

post_limit_order - Post an order to the exchange

get_price_history - Retrieve the price history for a specific currency pair



Areas of interest:

The usual, am I following conventions, how's my naming, how's my documentation, how well thought-out is my structure?



import json
import hmac
import hashlib
import requests
import base64
from requests.auth import AuthBase

API_KEY = 'redacted'
API_SECRET = 'redacted'
API_PASSPHRASE = 'redacted'
API_URL = 'https://api.gdax.com/'


class GDAXExchangeAuth(AuthBase):
    SIGNATURE_HTTP_HEADER = 'CB-ACCESS-SIGN'
    TIMESTAMP_HTTP_HEADER = 'CB-ACCESS-TIMESTAMP'
    KEY_HTTP_HEADER = 'CB-ACCESS-KEY'
    PASSPHRASE_HTTP_HEADER = 'CB-ACCESS-PASSPHRASE'

    def __init__(self, api_key, secret_key, passphrase, server_time):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
        self.server_time = server_time

    def __call__(self, request):
        epoch = self.server_time['epoch']
        timestamp = str(epoch)
        request.headers.update(
            {
                'CB-ACCESS-TIMESTAMP': timestamp,
                'CB-ACCESS-KEY': self.api_key,
                'CB-ACCESS-PASSPHRASE': self.passphrase,
                'Content-Type': 'application/json'
            }
        )

        message = timestamp + request.method + request.path_url + (request.body or '')
        message = message.encode()
        hmac_key = base64.b64decode(self.secret_key)
        signature = hmac.new(hmac_key, message, hashlib.sha256)
        signature_b64 = base64.b64encode(signature.digest())

        request.headers.update(
            {
                'CB-ACCESS-SIGN': signature_b64,
                'Content-Type': 'application/json'
            }
        )

        return request


def get_server_time():
    url = API_URL + 'time'
    r = requests.get(url)
    return r


def get_account(server_time):
    auth = GDAXExchangeAuth(API_KEY, API_SECRET, API_PASSPHRASE, server_time)
    url = API_URL + 'accounts'
    r = requests.get(url, auth=auth)
    return r


def get_book(product, level):
    # level 1 = Bid/Ask
    url = API_URL + '/products/' + product + '/book?level=' + str(level)
    r = requests.get(url)
    return r


def get_fills_after_order(server_time, order_id):
    # get all filled orders after the order id.
    # max 100 results returned
    auth = GDAXExchangeAuth(API_KEY, API_SECRET, API_PASSPHRASE, server_time)
    url = API_URL + 'fills?before=' + str(order_id)
    r = requests.get(url, auth=auth)
    return r


def post_limit_order(server_time, side, price, size, product_id):
    order_type = 'limit'

    order = {}
    order['side'] = side
    order['price'] = price
    order['size'] = size
    order['product_id'] = product_id
    order['type'] = order_type

    auth = GDAXExchangeAuth(API_KEY, API_SECRET, API_PASSPHRASE, server_time)
    url = API_URL + 'orders'
    data = json.dumps(order)
    r = requests.post(url, data=data, auth=auth)
    return r


def get_price_history(product_id, start, end, granularity):
    params = {}
    params['start'] = start
    params['end'] = end
    params['granularity'] = granularity

    url = API_URL + '/products/' + product_id + '/candles'
    r = requests.get(url, params=params)
    return r
To check your code against coding standard and some error detection, I'd suggest you pylint, if you don't know it already.
Running pylint on your original code gives a list of things to check and a score 7.29/10 which is a good start.

What you should fix:


add docstrings at module, class and function level.
remove temporary variables when used just once.

r = requests.get(url, params=params)
return r

return requests.get(url, params=params)

prefer string formatting over concatenation.

url = API_URL + '/products/' + product_id + '/candles'
url = '%s/products/%s/candles' % (API_URL, product_id)

no need to initialize each dictionary's key separately.

order = {}
order['side'] = side
order['price'] = price
order['size'] = size
order['product_id'] = product_id
order['type'] = order_type

order = {
    'side': side,
    'price': price,
    'size': size,
    'product_id': product_id,
    'type': order_type
}

consider parameter injection for the GDAXExchangeAuth instance in all your functions, it will simplify testing or reusing of the code with a different Authentication method.


Pylint complains about GDAXExchangeAuth class having too few public methods, but in this case I like the usage of a function class and there is no real need of other methods, so for me it's fine.

I assume this is probably just an example, but in any case I suggest also getting sensitive information from environment variables, something like

    import os

    API_PASSPHRASE = os.environ['API_PASSPHRASE']