This repository has been archived by the owner on Oct 1, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 37
/
api.py
122 lines (90 loc) · 3.62 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# This file is part of krakenex.
#
# krakenex is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# krakenex is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser
# General Public LICENSE along with krakenex. If not, see
# <http://www.gnu.org/licenses/gpl-3.0.txt>.
import json
import urllib
# private query nonce
import time
# private query signing
import hashlib
import hmac
import base64
from krakenex import connection
class API(object):
"""Kraken.com cryptocurrency Exchange API.
Public methods:
load_key
query_public
query_private
"""
def __init__(self, key = '', secret = ''):
"""Create an object with authentication information.
Arguments:
key -- key required to make queries to the API (default: '')
secret -- private key used to sign API messages (default: '')
"""
self.key = key
self.secret = secret
self.uri = 'https://api.kraken.com'
self.apiversion = '0'
def load_key(self, path):
"""Load key and secret from file.
Argument:
path -- path to file (string, no default)
"""
with open(path, "r") as f:
self.key = f.readline().strip()
self.secret = f.readline().strip()
def _query(self, urlpath, req = {}, conn = None, headers = {}):
"""Low-level query handling.
Arguments:
urlpath -- API URL path sans host (string, no default)
req -- additional API request parameters (default: {})
conn -- kraken.Connection object (default: None)
headers -- HTTPS headers (default: {})
"""
url = self.uri + urlpath
if conn is None:
conn = connection.Connection()
ret = conn._request(url, req, headers)
return json.loads(ret)
def query_public(self, method, req = {}, conn = None):
"""API queries that do not require a valid key/secret pair.
Arguments:
method -- API method name (string, no default)
req -- additional API request parameters (default: {})
conn -- connection object to reuse (default: None)
"""
urlpath = '/' + self.apiversion + '/public/' + method
return self._query(urlpath, req, conn)
def query_private(self, method, req={}, conn = None):
"""API queries that require a valid key/secret pair.
Arguments:
method -- API method name (string, no default)
req -- additional API request parameters (default: {})
conn -- connection object to reuse (default: None)
"""
urlpath = '/' + self.apiversion + '/private/' + method
req['nonce'] = int(1000*time.time())
postdata = urllib.urlencode(req)
message = urlpath + hashlib.sha256(str(req['nonce']) +
postdata).digest()
signature = hmac.new(base64.b64decode(self.secret),
message, hashlib.sha512)
headers = {
'API-Key': self.key,
'API-Sign': base64.b64encode(signature.digest())
}
return self._query(urlpath, req, conn, headers)