El siguiente código Python 2 implementa un cliente API de línea de comandos para nuestro API de Exchange WebSocket (https://www.kraken.com/en-gb/features/websocket-api).

El código se puede utilizar como un cliente API independiente, o se puede llamar desde otros lenguajes de programación interpretados o compilados (Bash, C++, etc.).

La librería estándar de WebSocket para Python (websocket-client (https://pypi.org/project/websocket_client/)) es necesaria para el manejo de redes a través de WebSocket, pero no se requieren otros módulos/librerías API.

El cliente API WebSocket para línea de comandos también puede ser descargado como krakenws.py (https://support.kraken.com/hc/article_attachments/360032452691/krakenws.py).

#!/usr/bin/env python

# Kraken WebSocket API
#
# Usage: ./krakenws.py feed symbols
# Example: ./krakenws.py trade xbt/usd
# Example: ./krakenws.py ticker xlm/usd
# Example: ./krakenws.py book eth/usd eth/eur
# Example: ./krakenws.py ohlc xbt/eur
# Example: ./krakenws.py spread ltc/usd ltc/eur

import sys
from websocket import create_connection

api_status = {"ping"}
api_public = {"trade", "book", "ticker", "spread", "ohlc"}
api_domain = "wss://ws.kraken.com/"

try:
ws = create_connection(api_domain)
print(ws.recv())
except Exception as error:
print("WebSocket connection failed (%s)" % error)
sys.exit(1)

if len(sys.argv) < 2:
api_feed = "ping"
else:
api_feed = sys.argv[1]

if api_feed in api_status:
api_data = '{"event":"%(feed)s"}' % {"feed":api_feed}
elif api_feed in api_public:
if len(sys.argv) < 3:
print("Usage: %s feed symbols" % sys.argv[0])
print("Example: %s ticker xbt/usd ltc/eur" % sys.argv[0])
ws.close()
sys.exit(1)
api_symbols = sys.argv[2].upper()
for count in range(3, len(sys.argv)):
api_symbols = api_symbols + '","' + sys.argv[count].upper()
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s"}, "pair":["%(symbols)s"]}' % {"feed":api_feed, "symbols":api_symbols}
else:
print("Usage: %s feed symbols" % sys.argv[0])
print("Example: %s ticker xbt/usd ltc/eur" % sys.argv[0])
ws.close()
sys.exit(1)

try:
ws.send(api_data)
except Exception as error:
print("Feed subscription failed (%s)" % error)
ws.close()
sys.exit(1)

for count in range(300):
try:
api_data = ws.recv()
print(api_data)
except KeyboardInterrupt:
ws.close()
sys.exit(0)
except Exception as error:
print("WebSocket message failed (%s)" % error)
ws.close()
sys.exit(1)

ws.close()
sys.exit(0)