Getting Kraken tickers

API/Krakenfutures_api.py

@@ -0,0 +1,292 @@
+import time
+import base64
+import hashlib
+import hmac
+import json
+import urllib.request as urllib2
+import urllib.parse as urllib
+import ssl
+
+
+class KrakenFutureClient(object):
+    def __init__(self, api_key="", api_secret="", timeout=10, checkCertificate=True, useNonce=False):
+        self.apiPath = 'https://futures.kraken.com'
+        self._api_key = api_key
+        self._api_secret = api_secret
+        self.timeout = timeout
+        self.nonce = 0
+        self.checkCertificate = checkCertificate
+        self.useNonce = useNonce
+
+    ''' public endpoints '''
+
+    # returns all instruments with specifications
+    def get_instruments(self):
+        endpoint = "/derivatives/api/v3/instruments"
+        return self.make_request("GET", endpoint)['instruments']
+
+    # returns market data for all instruments
+    def get_tickers(self):
+        endpoint = "/derivatives/api/v3/tickers"
+        return self.make_request("GET", endpoint)
+
+    # returns the entire order book of a futures
+    def get_orderbook(self, symbol):
+        endpoint = "/derivatives/api/v3/orderbook"
+        postUrl = "symbol=%s" % symbol
+        return self.make_request("GET", endpoint, postUrl=postUrl)
+
+    # returns historical data for futures and indices
+    def get_history(self, symbol, lastTime=""):
+        endpoint = "/derivatives/api/v3/history"
+        if lastTime != "":
+            postUrl = "symbol=%s&lastTime=%s" % (symbol, lastTime)
+        else:
+            postUrl = "symbol=%s" % symbol
+        return self.make_request("GET", endpoint, postUrl=postUrl)
+
+    ''' private endpoints '''
+
+    # returns key account information
+    # Deprecated because it returns info about the Futures margin account
+    # Use get_accounts instead
+    def get_account(self):
+        endpoint = "/derivatives/api/v3/account"
+        return self.make_request("GET", endpoint)
+
+    # returns key account information
+    def get_accounts(self):
+        endpoint = "/derivatives/api/v3/accounts"
+        return self.make_request("GET", endpoint)['accounts']
+
+    # places an order
+    def send_order(self, orderType, symbol, side, size, limitPrice, stopPrice=None, clientOrderId=None):
+        endpoint = "/derivatives/api/v3/sendorder"
+        postBody = "orderType=%s&symbol=%s&side=%s&size=%s&limitPrice=%s" % (
+            orderType, symbol, side, size, limitPrice)
+
+        if orderType == "stp" and stopPrice is not None:
+            postBody += "&stopPrice=%s" % stopPrice
+
+        if clientOrderId is not None:
+            postBody += "&cliOrdId=%s" % clientOrderId
+
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # places an order
+    def send_order_1(self, order):
+        endpoint = "/derivatives/api/v3/sendorder"
+        postBody = urllib.urlencode(order)
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # edit an order
+    def edit_order(self, edit):
+        endpoint = "/derivatives/api/v3/editorder"
+        postBody = urllib.urlencode(edit)
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # cancels an order
+    def cancel_order(self, order_id=None, cli_ord_id=None):
+        endpoint = "/derivatives/api/v3/cancelorder"
+
+        if order_id is None:
+            postBody = "cliOrdId=%s" % cli_ord_id
+        else:
+            postBody = "order_id=%s" % order_id
+
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # cancel all orders
+    def cancel_all_orders(self, symbol=None):
+        endpoint = "/derivatives/api/v3/cancelallorders"
+        if symbol is not None:
+            postbody = "symbol=%s" % symbol
+        else:
+            postbody = ""
+
+        return self.make_request("POST", endpoint, postBody=postbody)
+
+    # cancel all orders after
+    def cancel_all_orders_after(self, timeoutInSeconds=60):
+        endpoint = "/derivatives/api/v3/cancelallordersafter"
+        postbody = "timeout=%s" % timeoutInSeconds
+
+        return self.make_request("POST", endpoint, postBody=postbody)
+
+    # places or cancels orders in batch
+    def send_batchorder(self, jsonElement):
+        endpoint = "/derivatives/api/v3/batchorder"
+        postBody = "json=%s" % jsonElement
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # returns all open orders
+    def get_openorders(self):
+        endpoint = "/derivatives/api/v3/openorders"
+        return self.make_request("GET", endpoint)
+
+    # returns filled orders
+    def get_fills(self, lastFillTime=""):
+        endpoint = "/derivatives/api/v3/fills"
+        if lastFillTime != "":
+            postUrl = "lastFillTime=%s" % lastFillTime
+        else:
+            postUrl = ""
+        return self.make_request("GET", endpoint, postUrl=postUrl)
+
+    # returns all open positions
+    def get_openpositions(self):
+        endpoint = "/derivatives/api/v3/openpositions"
+        return self.make_request("GET", endpoint)
+
+    # sends an xbt withdrawal request
+    def send_withdrawal(self, targetAddress, currency, amount):
+        endpoint = "/derivatives/api/v3/withdrawal"
+        postBody = "targetAddress=%s&currency=%s&amount=%s" % (
+            targetAddress, currency, amount)
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # returns xbt transfers
+    def get_transfers(self, lastTransferTime=""):
+        endpoint = "/derivatives/api/v3/transfers"
+        if lastTransferTime != "":
+            postUrl = "lastTransferTime=%s" % lastTransferTime
+        else:
+            postUrl = ""
+        return self.make_request("GET", endpoint, postUrl=postUrl)
+
+    # returns all notifications
+    def get_notifications(self):
+        endpoint = "/derivatives/api/v3/notifications"
+        return self.make_request("GET", endpoint)
+
+    # makes an internal transfer
+    def transfer(self, fromAccount, toAccount, unit, amount):
+        endpoint = "/derivatives/api/v3/transfer"
+        postBody = "fromAccount=%s&toAccount=%s&unit=%s&amount=%s" % (
+            fromAccount, toAccount, unit, amount)
+        return self.make_request("POST", endpoint, postBody=postBody)
+
+    # accountlog csv
+    def get_accountlog(self):
+        endpoint = "/api/history/v2/accountlogcsv"
+        return self.make_request("GET", endpoint)
+
+    def _get_partial_historical_elements(self, elementType, since=None, contToken=None):
+        endpoint = "/api/history/v2/%s" % elementType
+
+        if contToken is not None:
+            return self.make_request_raw("GET", endpoint, postUrl="continuationToken=%s" % contToken)
+        else:
+            if since is not None:
+                return self.make_request_raw("GET", endpoint, postUrl="since=%s" % since)
+
+        return self.make_request_raw("GET", endpoint)
+
+    def _get_historical_elements(self, elementType, since=None):
+        elements = []
+
+        more = True
+        contToken = None
+
+        while more:
+            res = self._get_partial_historical_elements(elementType, since, contToken)
+            body = json.loads(res.read().decode('utf-8'))
+            elements = elements + body['elements']
+
+            if res.headers['is-truncated'] is None or res.headers['is-truncated'] == "false":
+                more = False
+                contToken = None
+            else:
+                contToken = res.headers['next-continuation-token']
+
+        elements.sort(key=lambda el: el['timestamp'], reverse=True)
+        return elements
+
+    # historical orders after a certain point in reverse chronological order
+    def get_historical_orders(self, since=None):
+        return self._get_historical_elements('orders', since)
+
+    # recent orders in reverse chronological order
+    def get_recent_orders(self):
+        return self.get_historical_orders(None)
+
+    # historical executions after a certain point in reverse chronological order
+    def get_historical_executions(self, since=None):
+        return self._get_historical_elements('executions', since)
+
+    # recent executions in reverse chronological order
+    def get_recent_executions(self):
+        return self.get_historical_executions(None)
+
+    # signs a message
+    def sign_message(self, endpoint, postData, nonce=""):
+        if endpoint.startswith('/derivatives'):
+            endpoint = endpoint[len('/derivatives'):]
+
+        # step 1: concatenate postData, nonce + endpoint
+        message = postData + nonce + endpoint
+
+        # step 2: hash the result of step 1 with SHA256
+        sha256_hash = hashlib.sha256()
+        sha256_hash.update(message.encode('utf8'))
+        hash_digest = sha256_hash.digest()
+
+        # step 3: base64 decode apiPrivateKey
+        secretDecoded = base64.b64decode(self._api_secret)
+
+        # step 4: use result of step 3 to has the result of step 2 with HMAC-SHA512
+        hmac_digest = hmac.new(secretDecoded, hash_digest,
+                               hashlib.sha512).digest()
+
+        # step 5: base64 encode the result of step 4 and return
+        return base64.b64encode(hmac_digest)
+
+    # creates a unique nonce
+    def get_nonce(self):
+        # https://en.wikipedia.org/wiki/Modulo_operation
+        self.nonce = (self.nonce + 1) & 8191
+        return str(int(time.time() * 1000)) + str(self.nonce).zfill(4)
+
+    # sends an HTTP request
+    def make_request_raw(self, requestType, endpoint, postUrl="", postBody=""):
+        # create authentication headers
+        postData = postUrl + postBody
+
+        if self.useNonce:
+            nonce = self.get_nonce()
+            signature = self.sign_message(endpoint, postData, nonce=nonce)
+            authentHeaders = {"APIKey": self._api_key,
+                              "Nonce": nonce, "Authent": signature}
+        else:
+            signature = self.sign_message(endpoint, postData)
+            authentHeaders = {
+                "APIKey": self._api_key, "Authent": signature}
+
+        authentHeaders["User-Agent"] = "cf-api-python/1.0"
+
+        # create request
+        if postUrl != "":
+            url = self.apiPath + endpoint + "?" + postUrl
+        else:
+            url = self.apiPath + endpoint
+
+        request = urllib2.Request(url, str.encode(postBody), authentHeaders)
+        request.get_method = lambda: requestType
+
+        # read response
+        if self.checkCertificate:
+            response = urllib2.urlopen(request, timeout=self.timeout)
+        else:
+            ctx = ssl.create_default_context()
+            ctx.check_hostname = False
+            ctx.verify_mode = ssl.CERT_NONE
+            response = urllib2.urlopen(
+                request, context=ctx, timeout=self.timeout)
+
+        # return
+        return response
+
+    # sends an HTTP request and read response body
+    def make_request(self, requestType, endpoint, postUrl="", postBody=""):
+        output = self.make_request_raw(requestType, endpoint, postUrl, postBody)
+        return json.loads(output.read().decode('utf-8'))

conf/config.ini

@@ -0,0 +1,3 @@
+[KRAKEN]
+api_key = i2fnn1LqzTqfoChwdoak/JbrpNytaCb6OAAvkgvdMbB+f9yR+mY7b49H
+api_secret = 8UgJ4fYIs2BIBAHuy0f2kRBWF+4qcrxcMtF9bAQAA/7u+dZzqbZskNxsM9pWDQGELx2wq67Bqx+QdphHvCYitg==

main.py

@@ -0,0 +1,77 @@
+import configparser
+import os
+import sys
+
+# ------------------------------------------------------------------
+# 1. Gestion des imports depuis le dossier ./API
+# ------------------------------------------------------------------
+sys.path.append(os.path.join(os.path.dirname(__file__), 'API'))
+
+try:
+    from API.Krakenfutures_api import KrakenFutureClient
+except ImportError:
+    # Alternative si le dossier API n'est pas vu comme un module
+    from Krakenfutures_api import KrakenFutureClient
+
+# ------------------------------------------------------------------
+# 2. Lecture de la configuration
+# ------------------------------------------------------------------
+config = configparser.ConfigParser()
+config_path = os.path.join(os.path.dirname(__file__), 'conf', 'config.ini')
+
+# Vérification que le fichier existe
+if not os.path.exists(config_path):
+    raise FileNotFoundError(f"Le fichier de configuration est introuvable ici : {config_path}")
+
+config.read(config_path)
+
+# Récupération des clés
+try:
+    api_key = config['KRAKEN']['api_key']
+    api_secret = config['KRAKEN']['api_secret']
+except KeyError as e:
+    raise ValueError(f"Erreur de lecture du config.ini. La clé ou la section est manquante : {e}")
+
+# ------------------------------------------------------------------
+# 3. Initialisation du Client et Test
+# ------------------------------------------------------------------
+if __name__ == "__main__":
+    print("Initialisation du client Kraken Futures...")
+    
+    # Création de l'instance du client
+    kraken_client = KrakenFutureClient(
+        api_key=api_key, 
+        api_secret=api_secret
+    )
+
+    try:
+        print("Tentative de connexion à l'API...")
+        response = kraken_client.get_tickers()
+        
+        # Vérification si la clé 'tickers' existe
+        if 'tickers' in response:
+            tickers_list = response['tickers']
+            print("Connexion réussie !")
+            
+            if len(tickers_list) > 0:
+                # On prend le premier élément de la liste
+                first_ticker = tickers_list[0] 
+                
+                # On affiche le contenu de ce premier ticker
+                print("\n--- Exemple de premier ticker reçu ---")
+                print(first_ticker)
+                
+                # Si vous voulez afficher spécifiquement le nom du symbole (souvent 'symbol' ou 'pair')
+                # Vous pouvez faire : print(f"Symbole : {first_ticker.get('symbol', 'Inconnu')}")
+            else:
+                print("La liste des tickers est vide.")
+
+        else:
+            print("Réponse reçue, mais format inattendu (pas de clé 'tickers') :", response)
+            
+    except Exception as e:
+        print(f"Une erreur est survenue lors de l'appel API : {e}")
+
+    # Pour tester un appel PRIVÉ (nécessite que les clés soient valides)
+    # accounts = kraken_client.get_accounts()
+    # print(accounts)