You can find my code at its GitHub repository.
#Code#
##bitex.api##
>>bitex.api.api
# Import Built-Ins
import logging
import requests
import time
# Import Third-Party
# Import Homebrew
log = logging.getLogger(__name__)
class RESTAPI:
def __init__(self, uri, api_version='', key='', secret=''):
"""
Base Class for REST API connections.
"""
self.key = key
self.secret = secret
self.uri = uri
self.apiversion = api_version
self.req_methods = {'POST': requests.post, 'PUT': requests.put,
'GET': requests.get, 'DELETE': requests.delete,
'PATCH': requests.patch}
log.debug("Initialized RESTAPI for URI: %s; "
"Will request on API version: %s" %
(self.uri, self.apiversion))
def load_key(self, path):
"""
Load key and secret from file.
"""
with open(path, 'r') as f:
self.key = f.readline().strip()
self.secret = f.readline().strip()
def nonce(self):
return str(int(1000 * time.time()))
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
"""
Dummy Signature creation method. Override this in child.
URL is required to be returned, as some Signatures use the url for
sig generation, and api calls made must match the address exactly.
"""
url = self.uri
return url, {'params': {'test_param': "authenticated_chimichanga"}}
def query(self, method_verb, endpoint, authenticate=False,
*args, **kwargs):
"""
Queries exchange using given data. Defaults to unauthenticated query.
"""
request_method = self.req_methods[method_verb]
if self.apiversion:
endpoint_path = '/' + self.apiversion + '/' + endpoint
else:
endpoint_path = '/' + endpoint
url = self.uri + endpoint_path
if authenticate: # sign off kwargs and url before sending request
url, request_kwargs = self.sign(url, endpoint, endpoint_path,
method_verb, *args, **kwargs)
else:
request_kwargs = kwargs
log.debug("Making request to: %s, kwargs: %s" % (url, request_kwargs))
r = request_method(url, timeout=5, **request_kwargs)
log.debug("Made %s request made to %s, with headers %s and body %s. "
"Status code %s" %
(r.request.method, r.request.url, r.request.headers,
r.request.body, r.status_code))
return r
>>bitex.api.rest
# Import Built-ins
import logging
import json
import hashlib
import hmac
import base64
import time
import urllib
import urllib.parse
from requests.auth import AuthBase
# Import Third-Party
# Import Homebrew
from bitex.api.api import RESTAPI
log = logging.getLogger(__name__)
class BitfinexREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v1',
url='https://api.bitfinex.com'):
super(BitfinexREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
try:
req = kwargs['params']
except KeyError:
req = {}
req['request'] = endpoint_path
req['nonce'] = self.nonce()
js = json.dumps(req)
data = base64.standard_b64encode(js.encode('utf8'))
h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha384)
signature = h.hexdigest()
headers = {"X-BFX-APIKEY": self.key,
"X-BFX-SIGNATURE": signature,
"X-BFX-PAYLOAD": data}
return url, {'headers': headers}
class BitstampREST(RESTAPI):
def __init__(self, user_id='', key='', secret='', api_version='',
url='https://www.bitstamp.net/api'):
self.id = user_id
super(BitstampREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def load_key(self, path):
"""
Load key and secret from file.
"""
with open(path, 'r') as f:
self.id = f.readline().strip()
self.key = f.readline().strip()
self.secret = f.readline().strip()
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
message = nonce + self.id + self.key
signature = hmac.new(self.secret.encode(), message.encode(),
hashlib.sha256)
signature = signature.hexdigest().upper()
try:
req = kwargs['params']
except KeyError:
req = {}
req['key'] = self.key
req['nonce'] = nonce
req['signature'] = signature
return url, {'data': req}
class BittrexREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v1.1',
url='https://bittrex.com/api'):
super(BittrexREST, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
try:
params = kwargs['params']
except KeyError:
params = {}
nonce = self.nonce()
req_string = endpoint_path + '?apikey=' + self.key + "&nonce=" + nonce + '&'
req_string += urllib.parse.urlencode(params)
headers = {"apisign": hmac.new(self.secret.encode('utf-8'),
(self.uri + req_string).encode('utf-8'),
hashlib.sha512).hexdigest()}
return self.uri + req_string, {'headers': headers, 'params': {}}
class CoincheckREST(RESTAPI):
def __init__(self, key='', secret='', api_version='api',
url='https://coincheck.com'):
super(CoincheckREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
params = json.dumps(params)
# sig = nonce + url + req
data = (nonce + endpoint_path + params).encode('utf-8')
h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha256)
signature = h.hexdigest()
headers = {"ACCESS-KEY": self.key,
"ACCESS-NONCE": nonce,
"ACCESS-SIGNATURE": signature}
return url, {'headers': headers}
class GdaxAuth(AuthBase):
def __init__(self, api_key, secret_key, passphrase):
self.api_key = api_key.encode('utf-8')
self.secret_key = secret_key.encode('utf-8')
self.passphrase = passphrase.encode('utf-8')
def __call__(self, request):
timestamp = str(time.time())
message = (timestamp + request.method + request.path_url +
(request.body or ''))
hmac_key = base64.b64decode(self.secret_key)
signature = hmac.new(hmac_key, message.encode('utf-8'), hashlib.sha256)
signature_b64 = base64.b64encode(signature.digest())
request.headers.update({
'CB-ACCESS-SIGN': signature_b64,
'CB-ACCESS-TIMESTAMP': timestamp,
'CB-ACCESS-KEY': self.api_key,
'CB-ACCESS-PASSPHRASE': self.passphrase,
'Content-Type': 'application/json'
})
return request
class GDAXRest(RESTAPI):
def __init__(self, passphrase='', key='', secret='', api_version='',
url='https://api.gdax.com'):
self.passphrase = passphrase
super(GDAXRest, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def load_key(self, path):
"""
Load key and secret from file.
"""
with open(path, 'r') as f:
self.passphrase = f.readline().strip()
self.key = f.readline().strip()
self.secret = f.readline().strip()
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
auth = GdaxAuth(self.key, self.secret, self.passphrase)
try:
js = kwargs['params']
except KeyError:
js = {}
return url, {'json': js, 'auth': auth}
class KrakenREST(RESTAPI):
def __init__(self, key='', secret='', api_version='0',
url='https://api.kraken.com'):
super(KrakenREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
try:
req = kwargs['params']
except KeyError:
req = {}
req['nonce'] = self.nonce()
postdata = urllib.parse.urlencode(req)
# Unicode-objects must be encoded before hashing
encoded = (str(req['nonce']) + postdata).encode('utf-8')
message = (endpoint_path.encode('utf-8') +
hashlib.sha256(encoded).digest())
signature = hmac.new(base64.b64decode(self.secret),
message, hashlib.sha512)
sigdigest = base64.b64encode(signature.digest())
headers = {
'API-Key': self.key,
'API-Sign': sigdigest.decode('utf-8')
}
return url, {'data': req, 'headers': headers}
class ItbitREST(RESTAPI):
def __init__(self, user_id = '', key='', secret='', api_version='v1',
url='https://api.itbit.com'):
self.userId = user_id
super(ItbitREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def load_key(self, path):
"""
Load user id, key and secret from file.
"""
with open(path, 'r') as f:
self.userId = f.readline().strip()
self.clientKey = f.readline().strip()
self.secret = f.readline().strip()
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
try:
params = kwargs['params']
except KeyError:
params = {}
verb = method_verb
if verb in ('PUT', 'POST'):
body = params
else:
body = {}
timestamp = self.nonce()
nonce = self.nonce()
message = json.dumps([verb, url, body, nonce, timestamp],
separators=(',', ':'))
sha256_hash = hashlib.sha256()
nonced_message = nonce + message
sha256_hash.update(nonced_message.encode('utf8'))
hash_digest = sha256_hash.digest()
hmac_digest = hmac.new(self.secret.encode('utf-8'),
url.encode('utf-8') + hash_digest,
hashlib.sha512).digest()
signature = base64.b64encode(hmac_digest)
auth_headers = {
'Authorization': self.key + ':' + signature.decode('utf8'),
'X-Auth-Timestamp': timestamp,
'X-Auth-Nonce': nonce,
'Content-Type': 'application/json'
}
return url, {'headers': auth_headers}
class OKCoinREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v1',
url='https://www.okcoin.com/api'):
super(OKCoinREST, self).__init__(url, api_version=api_version,
key=key,
secret=secret)
def sign(self,url, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
# sig = nonce + url + req
data = (nonce + url).encode()
h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha256)
signature = h.hexdigest()
headers = {"ACCESS-KEY": self.key,
"ACCESS-NONCE": nonce,
"ACCESS-SIGNATURE": signature}
return url, {'headers': headers}
class BTCERest(RESTAPI):
def __init__(self, key='', secret='', api_version='3',
url='https://btc-e.com/api'):
super(BTCERest, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
post_params = params
post_params.update({'nonce': nonce, 'method': endpoint.split('/', 1)[1]})
post_params = urllib.parse.urlencode(post_params)
signature = hmac.new(self.secret.encode('utf-8'),
post_params.encode('utf-8'), hashlib.sha512)
headers = {'Key': self.key, 'Sign': signature.hexdigest(),
"Content-type": "application/x-www-form-urlencoded"}
# split by tapi str to gain clean url;
url = url.split('/tapi', 1)[0] + '/tapi'
return url, {'headers': headers, 'params': params}
class CCEXRest(RESTAPI):
def __init__(self, key='', secret='', api_version='',
url='https://c-cex.com/t'):
super(CCEXRest, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
params['apikey'] = self.key
params['nonce'] = nonce
post_params = params
post_params.update({'nonce': nonce, 'method': endpoint})
post_params = urllib.parse.urlencode(post_params)
url = uri + post_params
sig = hmac.new(url, self.secret, hashlib.sha512)
headers = {'apisign': sig}
return url, {'headers': headers}
class CryptopiaREST(RESTAPI):
def __init__(self, key='', secret='', api_version='',
url='https://www.cryptopia.co.nz/api'):
super(CryptopiaREST, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
post_data = json.dumps(params)
md5 = base64.b64encode(hashlib.md5().updated(post_data).digest())
sig = self.key + 'POST' + urllib.parse.quote_plus(uri).lower() + nonce + md5
hmac_sig = base64.b64encode(hmac.new(base64.b64decode(self.secret),
sig, hashlib.sha256).digest())
header_data = 'amx' + self.key + ':' + hmac_sig + ':' + nonce
headers = {'Authorization': header_data,
'Content-Type': 'application/json; charset=utf-8'}
return uri, {'headers': headers, 'data': post_data}
class GeminiREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v1',
url='https://api.gemini.com'):
super(GeminiREST, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
payload = params
payload['nonce'] = nonce
payload['request'] = endpoint_path
payload = base64.b64encode(json.dumps(payload))
sig = hmac.new(self.secret, payload, hashlib.sha384).hexdigest()
headers = {'X-GEMINI-APIKEY': self.key,
'X-GEMINI-PAYLOAD': payload,
'X-GEMINI-SIGNATURE': sig}
return uri, {'headers': headers}
class YunbiREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v2',
url='https://yunbi.com/api'):
super(YunbiREST, self).__init__(url, api_version=api_version, key=key,
secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
params['tonce'] = nonce
params['access_key'] = self.key
post_params = urllib.parse.urlencode(params)
msg = '%s|%s|%s' % (method_verb, endpoint_path, post_params)
sig = hmac.new(self.secret, msg, hashlib.sha256).hexdigest()
uri += post_params + '&signature=' + sig
return uri, {}
class RockTradingREST(RESTAPI):
def __init__(self, key='', secret='', api_version='v1',
url='https://api.therocktrading.com'):
super(RockTradingREST, self).__init__(url, api_version=api_version,
key=key,
secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
nonce = self.nonce()
try:
params = kwargs['params']
except KeyError:
params = {}
payload = params
payload['nonce'] = int(nonce)
payload['request'] = endpoint_path
msg = nonce + uri
sig = hmac.new(self.secret.encode(), msg.encode(), hashlib.sha384).hexdigest()
headers = {'X-TRT-APIKEY': self.key,
'X-TRT-Nonce': nonce,
'X-TRT-SIGNATURE': sig, 'Content-Type': 'application/json'}
return uri, {'headers': headers}
class PoloniexREST(RESTAPI):
def __init__(self, key='', secret='', api_version='',
url='https://poloniex.com'):
super(PoloniexREST, self).__init__(url, api_version=api_version,
key=key, secret=secret)
def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
try:
params = kwargs['params']
except KeyError:
params = {}
params['nonce'] = self.nonce()
payload = params
msg = urllib.parse.urlencode(payload).encode('utf-8')
sig = hmac.new(self.secret.encode('utf-8'), msg, hashlib.sha512).hexdigest()
headers = {'Key': self.key, 'Sign': sig}
return uri, {'headers': headers, 'data': params}
##bitex.interfaces##
>>bitex.interfaces.kraken
"""
https:/kraken.com/help/api
"""
# Import Built-Ins
import logging
# Import Third-Party
# Import Homebrew
from bitex.api.rest import KrakenREST
from bitex.utils import return_json
from bitex.formatters.kraken import cancel, trade, order_book
# Init Logging Facilities
log = logging.getLogger(__name__)
class Kraken(KrakenREST):
def __init__(self, key='', secret='', key_file=''):
super(Kraken, self).__init__(key, secret)
if key_file:
self.load_key(key_file)
def make_params(self, *pairs, **kwargs):
q = {'pair': ','.join(pairs)}
q.update(kwargs)
return q
def public_query(self, endpoint, **kwargs):
path = 'public/' + endpoint
return self.query('GET', path, **kwargs)
def private_query(self, endpoint, **kwargs):
path = 'private/' + endpoint
return self.query('POST', path, authenticate=True, **kwargs)
"""
BitEx Standardized Methods
"""
@return_json(None)
def ticker(self, *pairs):
q = self.make_params(*pairs)
return self.public_query('Ticker', params=q)
@return_json(order_book)
def order_book(self, pair, **kwargs):
q = self.make_params(pair, **kwargs)
return self.public_query('Depth', params=q)
@return_json(None)
def trades(self, pair, **kwargs):
q = self.make_params(pair, **kwargs)
return self.public_query('Trades', params=q)
def _add_order(self, pair, side, price, amount, **kwargs):
q = {'pair': pair, 'type': side, 'price': price,
'ordertype': 'limit', 'volume': amount,
'trading_agreement': 'agree'}
q.update(kwargs)
return self.private_query('AddOrder', params=q)
@return_json(trade)
def bid(self, pair, price, amount, **kwargs):
return self._add_order(pair, 'buy', price, amount, **kwargs)
@return_json(trade)
def ask(self, pair, price, amount, **kwargs):
return self._add_order(pair, 'sell', price, amount, **kwargs)
@return_json(cancel)
def cancel_order(self, order_id, **kwargs):
q = {'txid': order_id}
q.update(kwargs)
return self.private_query('CancelOrder', params=q)
@return_json(None)
def order_info(self, *txids, **kwargs):
if len(txids) > 1:
q = {'txid': txids}
elif txids:
txid, *_ = txids
q = {'txid': txid}
else:
q = {}
q.update(kwargs)
return self.private_query('QueryOrders', params=q)
@return_json(None)
def balance(self, **kwargs):
return self.private_query('Balance')
@return_json(None)
def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
raise NotImplementedError()
@return_json(None)
def deposit_address(self, **kwargs):
raise NotImplementedError()
"""
Exchange Specific Methods
"""
@return_json(None)
def time(self):
return self.public_query('Time')
@return_json(None)
def assets(self, **kwargs):
return self.public_query('Assets', params=kwargs)
@return_json(None)
def pairs(self, **kwargs):
return self.public_query('AssetPairs', params=kwargs)
@return_json(None)
def ohlc(self, pair, **kwargs):
q = self.make_params(pair, **kwargs)
return self.public_query('OHLC', params=q)
@return_json(None)
def spread(self, pair, **kwargs):
q = self.make_params(pair, **kwargs)
return self.public_query('Spread', params=q)
@return_json(None)
def orders(self, **kwargs):
q = kwargs
return self.private_query('OpenOrders', params=q)
@return_json(None)
def closed_orders(self, **kwargs):
q = kwargs
return self.private_query('ClosedOrders', params=q)
@return_json(None)
def trade_history(self, **kwargs):
q = kwargs
return self.private_query('TradesHistory', params=q)
@return_json(None)
def fees(self, pair=None):
q = {'fee-info': True}
if pair:
q['pair'] = pair
return self.private_query('TradeVolume', params=q)
>>bitex.interfaces.bitfinex
"""
http://docs.bitfinex.com/
"""
# Import Built-Ins
import logging
# Import Third-Party
# Import Homebrew
from bitex.api.rest import BitfinexREST
from bitex.utils import return_json
from bitex.formatters.bitfinex import trade, cancel, order_status
# Init Logging Facilities
log = logging.getLogger(__name__)
class Bitfinex(BitfinexREST):
def __init__(self, key='', secret='', key_file=''):
super(Bitfinex, self).__init__(key, secret)
if key_file:
self.load_key(key_file)
def public_query(self, endpoint, **kwargs):
return self.query('GET', endpoint, **kwargs)
def private_query(self, endpoint, **kwargs):
return self.query('POST', endpoint, authenticate=True, **kwargs)
"""
BitEx Standardized Methods
"""
@return_json(None)
def order_book(self, pair, **kwargs):
return self.public_query('book/%s' % pair, params=kwargs)
@return_json(None)
def ticker(self, pair, **kwargs):
return self.public_query('pubticker/%s' % pair, params=kwargs)
@return_json(None)
def trades(self, pair, **kwargs):
return self.public_query('trades/%s' % pair, params=kwargs)
def _place_order(self, pair, amount, price, side, replace, **kwargs):
q = {'symbol': pair, 'amount': amount, 'price': price, 'side': side,
'type': 'exchange limit'}
q.update(kwargs)
if replace:
return self.private_query('order/cancel/replace', params=q)
else:
return self.private_query('order/new', params=q)
@return_json(trade)
def bid(self, pair, price, amount, replace=False, **kwargs):
return self._place_order(pair, amount, price, 'buy', replace=replace,
**kwargs)
@return_json(trade)
def ask(self, pair, price, amount, replace=False, **kwargs):
return self._place_order(pair, str(amount), str(price), 'sell',
replace=replace, **kwargs)
@return_json(cancel)
def cancel_order(self, order_id, all=False, **kwargs):
q = {'order_id': int(order_id)}
q.update(kwargs)
if not all:
return self.private_query('order/cancel', params=q)
else:
endpoint = 'order/cancel/all'
return self.private_query(endpoint)
@return_json(order_status)
def order(self, order_id, **kwargs):
q = {'order_id': order_id}
q.update(kwargs)
return self.private_query('order/status', params=q)
@return_json(None)
def balance(self, **kwargs):
return self.private_query('balances', params=kwargs)
@return_json(None)
def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
q = {'withdraw_type': _type, 'walletselected': source_wallet,
'amount': amount, 'address': tar_addr}
q.update(kwargs)
return self.private_query('withdraw', params=q)
@return_json(None)
def deposit_address(self, **kwargs):
q = {'method': currency, 'wallet_name': target_wallet}
q.update(kwargs)
return self.private_query('deposit/new', params=kwargs)
"""
Exchange Specific Methods
"""
@return_json(None)
def statistics(self, pair):
return self.public_query('stats/%s' % pair)
@return_json(None)
def funding_book(self, currency, **kwargs):
return self.public_query('lendbook/%s' % currency, params=kwargs)
@return_json(None)
def lends(self, currency, **kwargs):
return self.public_query('lends/%s' % currency, params=kwargs)
@return_json(None)
def pairs(self, details=False):
if details:
return self.public_query('symbols_details')
else:
return self.public_query('symbols')
@return_json(None)
def fees(self):
return self.private_query('account_infos')
@return_json(None)
def orders(self):
return self.private_query('orders')
@return_json(None)
def balance_history(self, currency, **kwargs):
q = {'currency': currency}
q.update(kwargs)
return self.private_query('history/movements', params=q)
@return_json(None)
def trade_history(self, pair, since, **kwargs):
q = {'symbol': pair, 'timestamp': since}
q.update(kwargs)
return self.private_query('mytrades', params=q)
>>bitex.interfaces.gdax
"""
https://docs.gdax.com/
"""
# Import Built-Ins
import logging
# Import Third-Party
# Import Homebrew
from bitex.api.rest import GDAXRest
from bitex.utils import return_json
# Init Logging Facilities
log = logging.getLogger(__name__)
class GDAX(GDAXRest):
def __init__(self, key='', secret='', key_file=''):
super(GDAX, self).__init__(key, secret)
if key_file:
self.load_key(key_file)
def public_query(self, endpoint, **kwargs):
return self.query('GET', endpoint, **kwargs)
def private_query(self, endpoint, method_verb='POST', **kwargs):
return self.query(method_verb, endpoint, authenticate=True, **kwargs)
"""
BitEx Standardized Methods
"""
@return_json(None)
def ticker(self, pair, **kwargs):
return self.public_query('products/%s/ticker' % pair, params=kwargs)
@return_json(None)
def order_book(self, pair, **kwargs):
return self.public_query('products/%s/book' % pair, params=kwargs)
@return_json(None)
def trades(self, pair, **kwargs):
return self.public_query('products/%s/trades' % pair, params=kwargs)
@return_json(None)
def bid(self, pair, price, size, **kwargs):
q = {'side': 'buy', 'type': 'market', 'product_id': pair,
'price': price, 'size': size}
q.update(kwargs)
return self.private_query('orders', params=q)
@return_json(None)
def ask(self, pair, price, amount, **kwargs):
q = {'side': 'sell', 'type': 'market', 'product_id': pair,
'price': price, 'size': size}
q.update(kwargs)
return self.private_query('orders', params=q)
@return_json(None)
def cancel_order(self, order_id, all=False, **kwargs):
if not all:
return self.private_query('orders/%s' % order_id,
method_verb='DELETE', params=kwargs)
else:
return self.private_query('orders', method_verb='DELETE',
params=kwargs)
@return_json(None)
def order(self, order_id, **kwargs):
return self.private_query('orders/%s' % order_id, method_verb='GET',
params=kwargs)
@return_json(None)
def balance(self, **kwargs):
return self.private_query('accounts', method_verb='GET', params=kwargs)
@return_json(None)
def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
raise NotImplementedError()
@return_json(None)
def deposit_address(self, **kwargs):
raise NotImplementedError()
"""
Exchange Specific Methods
"""
@return_json
def time(self):
return self.public_query('time')
@return_json(None)
def currencies(self):
return self.public_query('currencies')
@return_json(None)
def pairs(self):
return self.public_query('products')
@return_json(None)
def ohlc(self, pair, **kwargs):
return self.public_query('products/%s/candles' % pair, params=kwargs)
@return_json(None)
def stats(self, pair, **kwargs):
return self.public_query('products/%s/stats' % pair, params=kwargs)
##bitex.utils##
# Import Built-Ins
import logging
import json
import requests
# Import Third-Party
# Import Homebrew
# Init Logging Facilities
log = logging.getLogger(__name__)
def return_json(formatter=None):
def decorator(func):
def wrapper(*args, **kwargs):
try:
r = func(*args, **kwargs)
except Exception as e:
log.error("return_json(): Error during call to "
"%s(%s, %s) %s" % (func.__name__, args, kwargs, e))
raise
try:
r.raise_for_status()
except requests.HTTPError as e:
log.error("return_json: HTTPError for url %s: "
"%s" % (r.request.url, e))
return None, r
try:
data = r.json()
except json.JSONDecodeError:
log.error('return_json: Error while parsing json. '
'Request url was: %s, result is: '
'%s' % (r.request.url, r.text))
return None, r
except Exception as e:
log.error("return_json(): Unexpected error while parsing json "
"from %s: %s" % (r.request.url, e))
raise
# Apply formatter and return
if formatter is not None:
return formatter(data, *args, **kwargs), r
else:
return data, r
return wrapper
return decorator
You can also find the code at its GitHub repository. I've omitted some of the interface classes - the three I've provided are about as diverse as they come anyway.
