Spot Trading Bot Example

The template presented below serves as a starting point for the development of a trading algorithm for Spot trading on the cryptocurrency platform Kraken using the python-kraken-sdk.

The ManagedBot class is a helper class that instantiates the trading strategy. The trading strategy can be implemented in the TradingBot class. This class has access to all REST clients and gets gets via the on_message method all messages that are sent via the subscribed websocket feeds.

This is the starting point from which a strategy can be implemented and applied.

  1#!/usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# Copyright (C) 2023 Benjamin Thomas Schwertfeger
  4# GitHub: https://github.com/btschwertfeger
  5
  6
  7"""Module that provides an example Futures trading bot data structure."""
  8
  9from __future__ import annotations
 10
 11import asyncio
 12import logging
 13import logging.config
 14import os
 15import sys
 16import traceback
 17from typing import Optional, Union
 18
 19import requests
 20import urllib3
 21
 22from kraken.exceptions import KrakenException
 23from kraken.spot import Funding, KrakenSpotWSClient, Market, Staking, Trade, User
 24
 25
 26class TradingBot(KrakenSpotWSClient):
 27    """
 28    Class that implements the trading strategy
 29
 30    * The on_message function gets all events from the websocket feed
 31    * Decisions can be made based on these events
 32    * Can place trades using the self.__trade client
 33    * Do everything you want
 34
 35    ====== P A R A M E T E R S ======
 36    config: dict
 37        configuration like: {
 38            'key' 'kraken-futures-key',
 39            'secret': 'kraken-secret-key',
 40            'products': ['PI_XBTUSD']
 41        }
 42    """
 43
 44    def __init__(self: "TradingBot", config: dict) -> None:
 45        super().__init__(
 46            key=config["key"], secret=config["secret"]
 47        )  # initialize the KakenFuturesWSClient
 48        self.__config: dict = config
 49
 50        self.__user: User = User(key=config["key"], secret=config["secret"])
 51        self.__trade: Trade = Trade(key=config["key"], secret=config["secret"])
 52        self.__market: Market = Market(key=config["key"], secret=config["secret"])
 53        self.__funding: Funding = Funding(key=config["key"], secret=config["secret"])
 54        self.__staking: Staking = Staking(key=config["key"], secret=config["secret"])
 55
 56    async def on_message(self: "TradingBot", event: Union[dict, list]) -> None:
 57        """Receives all events that came form the websocket connection"""
 58        if isinstance(event, dict) and "event" in event:
 59            if event["event"] == "heartbeat":
 60                return
 61            if event["event"] == "pong":
 62                return
 63            if "error" in event:
 64                # handle exceptions/errors sent by websocket connection ...
 65                pass
 66
 67        logging.info(event)
 68
 69        # ... apply your trading strategy here
 70
 71        # call functions from self.__trade and other clients if conditions met...
 72
 73        # try:
 74        #     print(self.__trade.create_order(
 75        #         ordertype='limit',
 76        #         side='buy',
 77        #         volume=2,
 78        #         pair='XBTUSD',
 79        #         price=12000
 80        #     ))
 81        # except KrakenException.KrakenPermissionDeniedError:
 82        #    # ... handle exceptions
 83        #    pass
 84
 85        # The spot websocket client also allow sending orders via websockets
 86        # this is way faster than using REST endpoints.
 87
 88        # await self.create_order(
 89        #     ordertype='limit',
 90        #     side='buy',
 91        #     pair='BTC/EUR',
 92        #     price=20000,
 93        #     volume=200
 94        # )
 95
 96        # you can also un-/subscribe here using `self.subscribe(...)` or `self.unsubscribe(...)`
 97
 98        # more can be found in the documentation
 99
100    # add more functions to customize the trading strategy
101    # ...
102
103    def save_exit(self: "TradingBot", reason: Optional[str] = "") -> None:
104        """controlled shutdown of the strategy"""
105        logging.warning(f"Save exit triggered, reason: {reason}")
106        # ideas:
107        #   * save the bots data
108        #   * maybe close trades
109        #   * enable dead man's switch
110        sys.exit(1)
111
112
113class ManagedBot:
114    """
115    Class to manage the trading strategy
116
117    subscribes to desired feeds, instantiates the strategy and runs until condition met
118
119    ====== P A R A M E T E R S ======
120    config: dict
121        configuration like: {
122            'key' 'kraken-futures-key',
123            'secret': 'kraken-secret-key',
124            'products': ['PI_XBTUSD']
125        }
126    """
127
128    def __init__(self: "ManagedBot", config: dict):
129        self.__config: dict = config
130        self.__trading_strategy: Optional[TradingBot] = None
131
132    def run(self: "ManagedBot") -> None:
133        """Starts the event loop and bot"""
134        if not self.__check_credentials():
135            sys.exit(1)
136
137        try:
138            asyncio.run(self.__main())
139        except KeyboardInterrupt:
140            pass
141        finally:
142            if self.__trading_strategy is not None:
143                self.__trading_strategy.save_exit(reason="Asyncio loop left")
144
145    async def __main(self: "ManagedBot") -> None:
146        """
147        Instantiates the trading strategy (bot) and subscribes to the
148        desired websocket feeds. While no exception within the strategy occur
149        run the loop.
150
151        This variable `exception_occur` which is an attribute of the KrakenSpotWSClient
152        can be set individually but is also being set to True if the websocket connection
153        has some fatal error. This is used to exit the asyncio loop.
154        """
155        self.__trading_strategy = TradingBot(config=self.__config)
156
157        await self.__trading_strategy.subscribe(
158            subscription={"name": "ticker"}, pair=self.__config["pairs"]
159        )
160        await self.__trading_strategy.subscribe(
161            subscription={"name": "ohlc", "interval": 15}, pair=self.__config["pairs"]
162        )
163
164        await self.__trading_strategy.subscribe(subscription={"name": "ownTrades"})
165        await self.__trading_strategy.subscribe(subscription={"name": "openOrders"})
166
167        while not self.__trading_strategy.exception_occur:
168            try:
169                # check if bot feels good
170                # maybe send a status update every day
171                # ...
172                pass
173
174            except Exception as exc:
175                message: str = f"Exception in main: {exc} {traceback.format_exc()}"
176                logging.error(message)
177                self.__trading_strategy.save_exit(reason=message)
178
179            await asyncio.sleep(6)
180        self.__trading_strategy.save_exit(
181            reason="Left main loop because of exception in strategy."
182        )
183        return
184
185    def __check_credentials(self: "ManagedBot") -> bool:
186        """Checks the user credentials and the connection to Kraken"""
187        try:
188            User(self.__config["key"], self.__config["secret"]).get_account_balance()
189            logging.info("Client credentials are valid")
190            return True
191        except urllib3.exceptions.MaxRetryError:
192            logging.error("MaxRetryError, cannot connect.")
193            return False
194        except requests.exceptions.ConnectionError:
195            logging.error("ConnectionError, Kraken not available.")
196            return False
197        except KrakenException.KrakenAuthenticationError:
198            logging.error("Invalid credentials!")
199            return False
200
201    def save_exit(self: "ManagedBot", reason: str = "") -> None:
202        """Invoke the save exit function of the trading strategy"""
203        print(f"Save exit triggered - {reason}")
204        if self.__trading_strategy is not None:
205            self.__trading_strategy.save_exit(reason=reason)
206
207
208def main() -> None:
209    """Main"""
210    bot_config: dict = {
211        "key": os.getenv("API_KEY"),
212        "secret": os.getenv("SECRET_KEY"),
213        "pairs": ["DOT/EUR", "XBT/USD"],
214    }
215    managed_bot: ManagedBot = ManagedBot(config=bot_config)
216    try:
217        managed_bot.run()
218    except Exception:
219        managed_bot.save_exit(
220            reason=f"manageBot.run() has ended: {traceback.format_exc()}"
221        )
222
223
224if __name__ == "__main__":
225    main()