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