Skip to main content

Python client for Kraken API REST and Kraken Websockets API using httpx and websockets. Supports both sync and async for API REST.

Project description

Total alerts Language grade: Python

Kraky

Python client for Kraken API REST and Kraken Websockets API using httpx and websockets. Supports both sync and async for API REST.

Installation

pip install kraky

Docs

https://kraky.readthedocs.io/en/latest/

Usage

CLI

Kraky provides a CLI that matches the API function names and args.

You can use it like the following:

kraky get_ohlc_data pair=XBTUSD interval=240

You can replace get_ohlc_data by any kraky API function and pair=XBTUSD or interval=240 by any function argument. Please respect the format key=value.

It also supports .env files so you can also create a .env file with your api_key and secret and kraky will pick them up.

Example:

KRAKEN_API_KEY=""
KRAKEN_SECRET=""

Sync REST API

from kraky import KrakyApiClient


def get_web_sockets_token():
    kraken_api_key = ""
    kraken_secret = ""
    kraky_api_client = KrakyApiClient(
        api_key=kraken_api_key, secret=kraken_secret
    )

    ws_token = kraky_api_client.get_web_sockets_token()
    return ws_token

get_web_sockets_token()

Async REST API

from kraky import KrakyApiAsyncClient


async def get_web_sockets_token():
    kraken_api_key = ""
    kraken_secret = ""
    kraky_api_client = KrakyApiAsyncClient(
        api_key=kraken_api_key, secret=kraken_secret
    )

    ws_token = await kraky_api_client.get_web_sockets_token()
    return ws_token

asyncio.run(get_web_sockets_token)

Websocket

import asyncio
from kraky import KrakyApiAsyncClient, KrakyWsClient


async def get_web_sockets_token():
    kraken_api_key = ""
    kraken_secret = ""
    kraky_api_client = KrakyApiAsyncClient(
        api_key=kraken_api_key, secret=kraken_secret
    )

    ws_token = await kraky_api_client.get_web_sockets_token()
    return ws_token


async def public_handler(response):
    print(response)


async def private_handler(response):
    print(response)


async def main():

    interval = 30

    ws_pairs = ["XBT/USD", "ETH/USD"]

    ws_token = await get_ws_token()

    kraky_public_ws_client = KrakyWsClient("production")
    kraky_private_ws_client = KrakyWsClient("production-auth")

    asyncio.create_task(
        kraky_public_ws_client.connect(public_handler, connection_name="public")
    )

    asyncio.create_task(
        kraky_private_ws_client.connect(private_handler, connection_name="private")
    )

    await kraky_public_ws_client.subscribe(
        {"name": "ohlc", "interval": interval},
        ws_pairs,
        connection_name="public",
    )

    await kraky_private_ws_client.subscribe(
        {
            "interval": interval,
            "token": ws_token,
            "name": "openOrders",
        },
        connection_name="private",
    )


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.create_task(main())
    loop.run_forever()

Compatibility

  • Python 3.7 and above

Licence

MIT License

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

kraky-2022.3.26.tar.gz (11.0 kB view hashes)

Uploaded Source

Built Distribution

kraky-2022.3.26-py3-none-any.whl (10.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page