Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Kraken Limit Orders
import time
import sys
import json
import base64
import hashlib
import hmac
import urllib.request
import requests
import datetime
def placeLimitOrder(trade_symbol, trade_size, trade_leverage, limit_price):
# Infinite loop (can be exited via keyboard interrupt)
try:
api_path = '/0/private/AddOrder'
api_nonce = str(int(time.time()*1000))
api_post = 'nonce=%(api_nonce)s&pair=%(symbol)s&type=%(direction)s&ordertype=limit&price=%(price)f&volume=%(volume)f&leverage=%(leverage)s' % {'api_nonce':api_nonce, 'symbol':trade_symbol, 'direction':'buy', 'price':limit_price,'volume':trade_size, 'leverage':str(trade_leverage) if trade_leverage > 0 else 'none'}
api_sha256 = hashlib.sha256(api_nonce.encode('utf8') + api_post.encode('utf8'))
api_hmac = hmac.new(base64.b64decode(api_key_private), api_path.encode('utf8') + api_sha256.digest(), hashlib.sha512)
api_signature = base64.b64encode(api_hmac.digest())
api_request = urllib.request.Request('https://api.kraken.com/0/private/AddOrder', api_post.encode('utf8'))
api_request.add_header('API-Key', api_key_public)
api_request.add_header('API-Sign', api_signature)
api_request.add_header('User-Agent', 'Kraken trading bot example')
api_response = urllib.request.urlopen(api_request).read().decode()
api_data = json.loads(api_response)
except Exception as error:
print('Failed (%s)' % error)
else:
print('Done (%s)' % api_response if len(api_data['error']) == 0 else 'Error (%s)' % api_data['error'])
return True
#Prints list of tradable currency pairs
def getTradableAssetPairs():
resp = requests.get('https://api.kraken.com/0/public/AssetPairs')
return resp.json()['result'].keys()
#takes asset like "XBT" or "ETH" and returns asset details
def getAssetInfo(asset):
resp = requests.get(f'https://api.kraken.com/0/public/Assets?asset={asset}')
return resp.json()
#takes trading pair like XXBTZUSD or ETHUSD and gives details
def getTickerInfo(trading_pair):
resp = requests.get(f'https://api.kraken.com/0/public/Ticker?pair={trading_pair}')
values = list(resp.json()['result'].values())
return {'symbol' : trading_pair,
'ask':values[0]['a'],
'bid':values[0]['b'],
'last_trade_closed':values[0]['c'],
'volume':values[0]['v'],
'volume_weighted_average_price':values[0]['p'],
'num_trades':values[0]['t'],
'low':values[0]['l'],
'high':values[0]['h'],
'open':values[0]['o']}
#Get balances for all accounts
def getAccountBalances():
try:
print("trying")
api_path = '/0/private/Balance'
api_nonce = str(int(time.time()*1000))
api_post = 'nonce=%(api_nonce)s' % {'api_nonce':api_nonce}
api_sha256 = hashlib.sha256(api_nonce.encode('utf8') + api_post.encode('utf8'))
api_hmac = hmac.new(base64.b64decode(api_key_private), api_path.encode('utf8') + api_sha256.digest(), hashlib.sha512)
api_signature = base64.b64encode(api_hmac.digest())
api_request = urllib.request.Request('https://api.kraken.com'+api_path, api_post.encode('utf8'))
api_request.add_header('API-Key', api_key_public)
api_request.add_header('API-Sign', api_signature)
api_response = urllib.request.urlopen(api_request).read().decode()
api_data = json.loads(api_response)
except Exception as error:
print('Failed (%s)' % error)
else:
#print('Done (%s)' % api_response if len(api_data['error']) == 0 else 'Error (%s)' % api_data['error'])
return api_data['result']
return True
#Get account coin balance passing in the coin
def getCoinBalance(coin):
accounts = getAccountBalances()
return accounts[coin]
#Gets bid price of coin pair (ie. XXBTZUSD, ETHUSD, etc)
def getCoinBid(trading_pair):
return getTickerInfo(trading_pair)['bid'][0]
# Configure API key (copy/paste from account management)
api_key_public = ''
api_key_private = ''
def lambda_handler(event, context):
symbol = "XXBTZUSD"
usdAmount = 10
trade_size = round(usdAmount/float(getCoinBid("XXBTZUSD")),6)
print(trade_size)
leverage = 0
limit_price = float(getCoinBid("XXBTZUSD"))
print(limit_price)
placeLimitOrder(symbol, trade_size, leverage, limit_price)
return {
'statusCode': 200,
'body': json.dumps('End of script')
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment