Futures Trading Bot Template

The template presented below serves as a starting point for the development of a trading algorithm for trading futures contracts on the crypto asset exchange Kraken using the python-kraken-sdk.

The trading strategy can be implemented in the TradingBot class. This class has access to all REST clients and receives all messages that are sent via the subscribed websocket feeds via the on_message function.

Template to build a trading bot using the Kraken Futures Websocket API
  1# !/usr/bin/env python3
  2# -*- mode: python; coding: utf-8 -*-
  3#
  4# Copyright (C) 2023 Benjamin Thomas Schwertfeger
  5# All rights reserved.
  6# https://github.com/btschwertfeger
  7#
  8
  9"""
 10Module that provides a template to build a Futures trading algorithm using the
 11python-kraken-sdk.
 12"""
 13
 14from __future__ import annotations
 15
 16import asyncio
 17import logging
 18import os
 19import sys
 20import traceback
 21
 22import requests
 23import urllib3
 24
 25from kraken.exceptions import KrakenAuthenticationError
 26from kraken.futures import FuturesWSClient, User
 27
 28logging.basicConfig(
 29    format="%(asctime)s %(module)s,line: %(lineno)d %(levelname)8s | %(message)s",
 30    datefmt="%Y/%m/%d %H:%M:%S",
 31    level=logging.INFO,
 32)
 33logging.getLogger("requests").setLevel(logging.WARNING)
 34logging.getLogger("urllib3").setLevel(logging.WARNING)
 35LOG: logging.Logger = logging.getLogger(__name__)
 36
 37
 38class TradingBot(FuturesWSClient):
 39    """
 40    Class that implements the trading strategy
 41
 42    * The on_message function gets all messages sent by the websocket feeds.
 43    * Decisions can be made based on these messages
 44    * Can place trades using the self.__trade client
 45    * Do everything you want
 46
 47    ====== P A R A M E T E R S ======
 48    config: dict
 49        configuration like: {
 50            'key' 'kraken-futures-key',
 51            'secret': 'kraken-secret-key',
 52            'products': ['PI_XBTUSD']
 53        }
 54    """
 55
 56    def __init__(self: TradingBot, config: dict) -> None:
 57        super().__init__(  # initialize the KrakenFuturesWSClient
 58            key=config["key"],
 59            secret=config["secret"],
 60        )
 61
 62    async def on_message(self: TradingBot, message: list | dict) -> None:
 63        """Receives all messages that came form the websocket feed(s)"""
 64        LOG.info(message)
 65
 66        # == apply your trading strategy here ==
 67        # Hint: You can execute requests using the `request` function directly:
 68        # print(await self.request(
 69        #     "GET", "/api/charts/v1/spot/PI_XBTUSD/1d",
 70        # ))
 71
 72        # You can also un-/subscribe here using `self.subscribe(...)` or
 73        # `self.unsubscribe(...)`
 74        # … more can be found in the documentation
 75        #        (https://python-kraken-sdk.readthedocs.io/en/stable/).
 76
 77    # Add more functions to customize the trading strategy …
 78
 79    def save_exit(self: TradingBot, reason: str = "") -> None:
 80        """Controlled shutdown of the strategy"""
 81        LOG.warning("Save exit triggered, reason: %s", reason)
 82        # Some ideas:
 83        #   * Save the bots data
 84        #   * Close trades
 85        #   * Enable dead man's switch
 86        sys.exit(1)
 87
 88
 89class ManagedBot:
 90    """
 91    Class to manage the trading strategy
 92
 93    … subscribes to desired feeds, instantiates the strategy and runs as long
 94    as there is no error.
 95
 96    ====== P A R A M E T E R S ======
 97    config: dict
 98        bot configuration like: {
 99            'key' 'kraken-futures-key',
100            'secret': 'kraken-secret-key',
101            'products': ['PI_XBTUSD']
102        }
103    """
104
105    def __init__(self: ManagedBot, config: dict) -> None:
106        self.__config: dict = config
107        self.__trading_strategy: TradingBot | None = None
108
109    def run(self: ManagedBot) -> None:
110        """Runner function"""
111        if not self.__check_credentials():
112            sys.exit(1)
113
114        try:
115            asyncio.run(self.__main())
116        except KeyboardInterrupt:
117            self.save_exit(reason="KeyboardInterrupt")
118        else:
119            self.save_exit(reason="Asyncio loop left")
120
121    async def __main(self: ManagedBot) -> None:
122        """
123        Instantiates the trading strategy/algorithm and subscribes to the
124        desired websocket feeds. Run the loop while no exception occur.
125
126        The variable `exception_occur` which is an attribute of the
127        KrakenFuturesWSClient can be set individually but is also being set to
128        `True` if the websocket connection has some fatal error. This is used to
129        exit the asyncio loop - but you can also apply your own reconnect rules.
130        """
131        try:
132            self.__trading_strategy = TradingBot(config=self.__config)
133            await self.__trading_strategy.start()
134
135            await self.__trading_strategy.subscribe(
136                feed="ticker",
137                products=self.__config["products"],
138            )
139            await self.__trading_strategy.subscribe(
140                feed="book",
141                products=self.__config["products"],
142            )
143
144            await self.__trading_strategy.subscribe(feed="fills")
145            await self.__trading_strategy.subscribe(feed="open_positions")
146            await self.__trading_strategy.subscribe(feed="open_orders")
147            await self.__trading_strategy.subscribe(feed="balances")
148
149            while not self.__trading_strategy.exception_occur:
150                # Check if the algorithm feels good
151                # Send a status update every day via Telegram or Mail
152                # …
153                await asyncio.sleep(6)
154
155        except Exception as exc:
156            LOG.error(message := f"Exception in main: {exc} {traceback.format_exc()}")
157            self.__trading_strategy.save_exit(reason=message)
158        finally:
159            # Close the sessions properly.
160            await self.__trading_strategy.close()
161
162    def __check_credentials(self: ManagedBot) -> bool:
163        """Checks the user credentials and the connection to Kraken"""
164        try:
165            User(self.__config["key"], self.__config["secret"]).get_wallets()
166            LOG.info("Client credentials are valid.")
167            return True
168        except urllib3.exceptions.MaxRetryError:
169            LOG.error("MaxRetryError, cannot connect.")
170            return False
171        except requests.exceptions.ConnectionError:
172            LOG.error("ConnectionError, Kraken not available.")
173            return False
174        except KrakenAuthenticationError:
175            LOG.error("Invalid credentials!")
176            return False
177
178    def save_exit(self: ManagedBot, reason: str = "") -> None:
179        """Calls the save exit function of the trading strategy"""
180        print(f"Save exit triggered - {reason}")
181        if self.__trading_strategy is not None:
182            self.__trading_strategy.save_exit(reason=reason)
183        else:
184            sys.exit(1)
185
186
187def main() -> None:
188    """Example main - load environment variables and run the strategy."""
189
190    managed_bot: ManagedBot = ManagedBot(
191        config={
192            "key": os.getenv("FUTURES_API_KEY"),
193            "secret": os.getenv("FUTURES_SECRET_KEY"),
194            "products": ["PI_XBTUSD", "PF_SOLUSD"],
195        },
196    )
197
198    try:
199        managed_bot.run()
200    except Exception:
201        managed_bot.save_exit(
202            reason=f"manageBot.run() has ended: {traceback.format_exc()}",
203        )
204
205
206if __name__ == "__main__":
207    main()