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