Futures REST

The examples presented below serve to demonstrate the usage of the Futures REST clients provided by python-kraken-sdk to access Kraken’s REST API.

For questions, feedback, additions, suggestions for improvement or problems python-kraken-sdk/discussions or python-kraken-sdk/issues may be helpful.

See https://docs.kraken.com/api/docs/guides/global-intro for information about the available endpoints and their usage.

The Futures client provides access to all un-and authenticated endpoints of Kraken’s Futures API.

Example: Spot Client Usage (1)

from kraken.futures import FuturesClient

client = FuturesClient(key="<your-api-key>", secret="<your-secret-key>")
print(client.request("GET", "/derivatives/api/v3/accounts"))

The async Futures client allows for asynchronous access to Kraken’s Futures endpoints. Below are two examples demonstrating its usage.

Using FuturesAsyncClient without a context manager; In this example, the client is manually closed after the request is made.

Example: Spot Client Usage (2)

import asyncio
from kraken.futures import FuturesAsyncClient

async def main():
   client = FuturesAsyncClient(key="<your-api-key>", secret="<your-secret-key>")
   try:
      response = await client.request("GET", "/derivatives/api/v3/accounts")
      print(response)
   finally:
      await client.async_close()

asyncio.run(main())

Using FuturesAsyncClient as context manager; This example demonstrates the use of the context manager, which ensures the client is automatically closed after the request is completed.

Example: Spot Client Usage (3)

import asyncio
from kraken.futures import FuturesAsyncClient

async def main():
   async with FuturesAsyncClient(
      key="<your-api-key>", secret="<your-secret-key>"
   ) as client:
      response = await client.request("GET", "/derivatives/api/v3/accounts")
      print(response)

asyncio.run(main())

The following legacy examples are not maintained on a regular basis. They serve only for demonstration purposes - make sure to checkout the documentation of the individual functions.

Example usage of Futures REST clients

# !/usr/bin/env python3
# -*- mode: python; coding: utf-8 -*-
#
# Copyright (C) 2023 Benjamin Thomas Schwertfeger
# All rights reserved.
# https://github.com/btschwertfeger
#
# SPDX-License-Identifier: Apache-2.0
#

"""
Module that implements *some* examples for the Kraken Futures REST clients
usage.
"""

import logging
import os
import time
from pathlib import Path

from kraken.futures import Funding, Market, Trade, User

logging.basicConfig(
    format="%(asctime)s %(module)s,line: %(lineno)d %(levelname)8s | %(message)s",
    datefmt="%Y/%m/%d %H:%M:%S",
    level=logging.INFO,
)
logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)

key = os.getenv("FUTURES_SANDBOX_KEY")
secret = os.getenv("FUTURES_SANDBOX_SECRET")


def market_examples() -> None:
    """Example Futures Market client usage"""

    # Usage of the Market client to access public endpoints:
    market = Market()
    print(market.get_tick_types())
    print(market.get_tradeable_products(tick_type="trade"))
    print(market.get_resolutions(tick_type="trade", tradeable="PI_XBTUSD"))
    print(
        market.get_ohlc(
            tick_type="trade",
            symbol="PI_XBTUSD",
            resolution="5m",
            from_="1668989233",
        ),
    )
    print(market.get_fee_schedules())
    print(
        market.get_orderbook(symbol="fi_xbtusd_180615"),
    )  # might need adjustment of the symbol
    print(market.get_tickers())
    print(market.get_instruments())
    print(market.get_instruments_status())
    print(market.get_instruments_status(instrument="PI_XBTUSD"))
    print(market.get_trade_history(symbol="PI_XBTUSD"))
    print(market.get_historical_funding_rates(symbol="PI_XBTUSD"))
    time.sleep(2)  # Just to avoid rate limits

    # Usage of the Market client to access private endpoints:
    # (commented out to avoid accidental usage)
    priv_market = Market(key=key, secret=secret, sandbox=True)
    # print(priv_market.get_fee_schedules_vol())
    print(priv_market.get_leverage_preference())
    # print(priv_market.set_leverage_preference(symbol='PF_XBTUSD', maxLeverage=2)) # set max leverage
    # print(priv_market.set_leverage_preference(symbol='PF_XBTUSD')) # reset max leverage
    # print(priv_market.set_pnl_preference(symbol='PF_XBTUSD', pnlPreference='BTC'))

    # time.sleep(2)
    # print(priv_market.get_execution_events())
    # print(market.get_public_execution_events(tradeable='PI_XBTUSD'))
    # print(market.get_public_order_events(tradeable='PI_XBTUSD'))
    # print(market.get_public_mark_price_events(tradeable='PI_XBTUSD'))
    # print(priv_market.get_order_events())
    # print(priv_market.get_trigger_events())


def user_examples() -> None:
    """Example Futures User client usage"""
    # NOTE: This only works if you have set valid credentials for the the
    #       Futures demo environment. Remove the `sandbox=True` argument to use
    #       the production environment.
    #
    # Usage of the User client to access private endpoints:
    user = User(key=key, secret=secret, sandbox=True)
    print(user.get_wallets())
    print(user.get_subaccounts())
    print(user.get_unwind_queue())
    print(user.get_notifications())
    print(user.get_open_positions())
    print(user.get_open_orders())

    # You can retrieve the account log like so:
    print(user.get_account_log(before="1604937694000"))
    print(user.get_account_log(info="futures liquidation"))
    time.sleep(2)  # Just to avoid rate limits

    response = user.get_account_log_csv()
    assert response.status_code in {200, "200"}
    with Path("account_log.csv").open("wb") as file:
        for chunk in response.iter_content(chunk_size=512):
            if chunk:
                file.write(chunk)


def trade_examples() -> None:
    """Example Futures Trade client usage"""
    print(
        "Attention: Please check if you want to execute the trade endpoints!"
        " Check the script manually before running this example.",
    )
    return
    # return
    # NOTE: This only works if you have set valid credentials for the the
    #       Futures demo environment. Remove the `sandbox=True` argument to use
    #       the production environment.
    trade = Trade(key=key, secret=secret, sandbox=True)
    print(trade.get_fills())
    print(trade.get_fills(lastFillTime="2020-07-21T12:41:52.790Z"))
    print(
        trade.create_batch_order(
            batchorder_list=[
                {
                    "order": "send",
                    "order_tag": "1",
                    "orderType": "lmt",
                    "symbol": "PI_XBTUSD",
                    "side": "buy",
                    "size": 1,
                    "limitPrice": 1.00,
                },
                {
                    "order": "send",
                    "order_tag": "2",
                    "orderType": "stp",
                    "symbol": "PI_XBTUSD",
                    "side": "buy",
                    "size": 1,
                    "limitPrice": 2.00,
                    "stopPrice": 3.00,
                },
                {
                    "order": "cancel",
                    "order_id": "e35d61dd-8a30-4d5f-a574-b5593ef0c050",
                },
                {
                    "order": "cancel",
                    "cliOrdId": 123456789,
                },
            ],
        ),
    )
    print(trade.cancel_all_orders())
    print(trade.cancel_all_orders(symbol="pi_xbtusd"))
    print(trade.dead_mans_switch(timeout=60))
    print(trade.dead_mans_switch(timeout=0))  # to deactivate
    print(trade.cancel_order(order_id="some order id"))
    print(
        trade.edit_order(
            orderId="some order id",
            size=300,
            limitPrice=401,
            stopPrice=350,
        ),
    )
    print(trade.get_orders_status(orderIds=["orderid1", "orderid2"]))
    print(
        trade.create_order(
            orderType="lmt",
            side="buy",
            size=1,
            limitPrice=4,
            symbol="pf_bchusd",
        ),
    )
    print(
        trade.create_order(
            orderType="take_profit",
            side="buy",
            size=1,
            symbol="pf_bchusd",
            stopPrice=100,
            triggerSignal="mark",
        ),
    )


def funding_examples() -> None:
    """Example Funding client usage"""
    funding = Funding(key=key, secret=secret, sandbox=True)
    print(funding.get_historical_funding_rates(symbol="PF_SOLUSD"))


def main() -> None:
    """Uncomment the examples you want to run:"""
    # user_examples()
    # market_examples()
    # trade_examples()
    # funding_examples()


if __name__ == "__main__":
    main()