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