CoinList Pro Python3 API

coinlist.py

'''
Copyright (c) 2020 Lucas Ryan

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
'''

# Based on https://trade-docs.coinlist.co/ (v1.3.0)

import requests # https://pypi.org/project/requests/
import time
import hmac
import hashlib
import base64
import json

def sha265hmac(data, key):
    h = hmac.new(key, data.encode('utf-8'), digestmod=hashlib.sha256)
    return base64.b64encode(h.digest()).decode('utf-8')


class CoinList(object):
    def __init__(self, access_key, access_secret, endpoint_url='https://trade-api.coinlist.co'):
        self.access_key = access_key
        self.access_secret = access_secret
        self.endpoint_url = endpoint_url

    def request(self, method, path, params={}, body={}):
        timestamp = str(int(time.time()))
        # build the request path with any GET params already included
        path_with_params = requests.Request(method, self.endpoint_url + path, params=params).prepare().path_url
        json_body = json.dumps(body, separators=(',', ':')).strip()
        message = timestamp + method + path_with_params + ('' if not body else json_body)
        secret = base64.b64decode(self.access_secret).strip()
        signature = sha265hmac(message, secret)
        headers = {
            'Content-Type': 'application/json',
            'CL-ACCESS-KEY': self.access_key,
            'CL-ACCESS-SIG': signature,
            'CL-ACCESS-TIMESTAMP': timestamp
        }
        url = self.endpoint_url + path_with_params
        r = requests.request(method, url, headers=headers, data=json_body)
        return r.json()


'''
# Uncomment to enable wire-level debugging
import logging
logging.basicConfig(level=logging.DEBUG)
from http.client import HTTPConnection
HTTPConnection.debuglevel = 1
'''

## Example usage
if __name__ == "__main__":

    import csv    
    def objects_to_csv(objects, fields, filename):
        with open(filename, 'w') as f:
            c = csv.writer(f)
            c.writerow(fields)
            for o in objects:
                c.writerow([o[f] for f in fields])
    
    # provide access_key and access_secret on command line
    import sys
    access_key = sys.argv[1]
    access_secret = sys.argv[2]

    coinlist = CoinList(access_key, access_secret)

    order_fields = ['epoch_timestamp', 'order_id', 'trader_id', 'symbol', 'side', 'size_filled', 'cost', 'price', 'fill_fees']
    # note that the orders endpoint has a bug - it only returns orders after 2020-11-01
    r = coinlist.request('GET', '/v1/orders', {'count': '400', 'status': 'done', 'descending': 'true'})
    objects_to_csv(r['orders'], order_fields, 'orders.csv')

    fill_fields = ['symbol', 'auction_code', 'order_id', 'quantity', 'fee', 'fee_type', 'price', 'logical_time']
    r = coinlist.request('GET', '/v1/fills', {'count': '400', 'descending': 'true'})
    objects_to_csv(r['fills'], fill_fields, 'fills.csv')
    
    transfer_fields = ['transfer_id', 'created_at', 'confirmed_at', 'asset', 'amount', 'status']
    r = coinlist.request('GET', '/v1/transfers')
    objects_to_csv(r['transfers'], transfer_fields, 'transfers.csv')
    
    account_fields = ['trader_id', 'asset', 'balance']
    r = coinlist.request('GET', '/v1/accounts')
    trader_ids = [account['trader_id'] for account in r['accounts']]
    accounts = []
    for trader_id in trader_ids:
        r = coinlist.request('GET', f'/v1/accounts/{trader_id}')
        for asset, balance in r['asset_balances'].items():
            accounts.append({
                'trader_id': trader_id,
                'asset': asset, 
                'balance': balance
            })
    objects_to_csv(accounts, account_fields, 'accounts.csv')