Spot Trading Bot Templates
The templates presented below serve as starting points for the development of a trading algorithms for Spot trading on the cryptocurrency exchange Kraken using the python-kraken-sdk.
The trading strategy can be implemented using the TradingBot class. This
class has access to all REST clients and receives all messages that are sent by
the subscribed websocket feeds via the on_message function.
Template to build a trading bot using the Kraken Spot Websocket API v2
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# ruff: noqa: RUF027
9
10"""
11Module that provides a template to build a Spot trading algorithm using the
12python-kraken-sdk and Kraken Spot websocket API v2.
13"""
14
15from __future__ import annotations
16
17import asyncio
18import logging
19import logging.config
20import os
21import sys
22import traceback
23
24import requests
25import urllib3
26
27from kraken.exceptions import KrakenAuthenticationError # , KrakenPermissionDeniedError
28from kraken.spot import Funding, Market, SpotWSClient, Trade, User
29
30logging.basicConfig(
31 format="%(asctime)s %(module)s,line: %(lineno)d %(levelname)8s | %(message)s",
32 datefmt="%Y/%m/%d %H:%M:%S",
33 level=logging.INFO,
34)
35logging.getLogger("requests").setLevel(logging.WARNING)
36logging.getLogger("urllib3").setLevel(logging.WARNING)
37
38LOG: logging.Logger = logging.getLogger(__name__)
39
40
41class TradingBot(SpotWSClient):
42 """
43 Class that implements the trading strategy
44
45 * The on_message function gets all messages sent by the websocket feeds.
46 * Decisions can be made based on these messages
47 * Can place trades using the self.__trade client or self.send_message
48 * Do everything you want
49
50 ====== P A R A M E T E R S ======
51 config: dict
52 configuration like: {
53 "key": "kraken-spot-key",
54 "secret": "kraken-spot-secret",
55 "pairs": ["DOT/USD", "BTC/USD"],
56 }
57 """
58
59 def __init__(
60 self: TradingBot,
61 config: dict,
62 **kwargs: object | dict | set | tuple | list | str | float | None,
63 ) -> None:
64 super().__init__(
65 key=config["key"],
66 secret=config["secret"],
67 **kwargs,
68 )
69 self.__config: dict = config
70
71 self.__user: User = User(key=config["key"], secret=config["secret"])
72 self.__trade: Trade = Trade(key=config["key"], secret=config["secret"])
73 self.__market: Market = Market(key=config["key"], secret=config["secret"])
74 self.__funding: Funding = Funding(key=config["key"], secret=config["secret"])
75
76 async def on_message(self: TradingBot, message: dict) -> None:
77 """Receives all messages of the websocket connection(s)"""
78 if message.get("method") == "pong" or message.get("channel") == "heartbeat":
79 return
80 if "error" in message:
81 # handle exceptions/errors sent by websocket connection …
82 pass
83
84 LOG.info(message)
85
86 # == apply your trading strategy here ==
87
88 # Call functions of `self.__trade` and other clients if conditions met …
89 # try:
90 # print(self.__trade.create_order(
91 # ordertype='limit',
92 # side='buy',
93 # volume=2,
94 # pair='XBTUSD',
95 # price=12000
96 # ))
97 # except KrakenPermissionDeniedError:
98 # # … handle exceptions
99 # pass
100
101 # The spot websocket client also allow sending orders via websockets
102 # this is way faster than using REST endpoints.
103 # await self.send_message(
104 # message={
105 # "method": "add_order",
106 # "params": {
107 # "limit_price": 1234.56,
108 # "order_type": "limit",
109 # "order_userref": 123456789,
110 # "order_qty": 1.0,
111 # "side": "buy",
112 # "symbol": "BTC/USD",
113 # "validate": True,
114 # },
115 # }
116 # )
117
118 # You can also un-/subscribe here using `self.subscribe(...)` or
119 # `self.unsubscribe(...)`.
120 #
121 # … more can be found in the documentation
122 # (https://python-kraken-sdk.readthedocs.io/en/stable/).
123
124 # Add more functions to customize the trading strategy …
125
126 def save_exit(self: TradingBot, reason: str | None = "") -> None:
127 """controlled shutdown of the strategy"""
128 LOG.warning("Save exit triggered, reason: %s", reason)
129 # Some ideas:
130 # * Save the current data
131 # * Close trades
132 # * Enable dead man's switch
133 sys.exit(1)
134
135
136class Manager:
137 """
138 Class to manage the trading strategy
139
140 … subscribes to desired feeds, instantiates the strategy and runs as long
141 as there is no error.
142
143 ====== P A R A M E T E R S ======
144 config: dict
145 configuration like: {
146 "key" "kraken-spot-key",
147 "secret": "kraken-secret-key",
148 "pairs": ["DOT/USD", "BTC/USD"],
149 }
150 """
151
152 def __init__(self: Manager, config: dict) -> None:
153 self.__config: dict = config
154 self.__trading_strategy: TradingBot | None = None
155
156 def run(self: Manager) -> None:
157 """Starts the event loop and bot"""
158 if not self.__check_credentials():
159 sys.exit(1)
160
161 try:
162 asyncio.run(self.__main())
163 except KeyboardInterrupt:
164 self.save_exit(reason="KeyboardInterrupt")
165 else:
166 self.save_exit(reason="Asyncio loop left")
167
168 async def __main(self: Manager) -> None:
169 """
170 Instantiates the trading strategy and subscribes to the desired
171 websocket feeds. While no exception within the strategy occur run the
172 loop.
173
174 The variable `exception_occur` which is an attribute of the SpotWSClient
175 can be set individually but is also being set to `True` if the websocket
176 connection has some fatal error. This is used to exit the asyncio loop -
177 but you can also apply your own reconnect rules.
178 """
179 try:
180 self.__trading_strategy = TradingBot(config=self.__config)
181 await self.__trading_strategy.start()
182
183 await self.__trading_strategy.subscribe(
184 params={"channel": "ticker", "symbol": self.__config["pairs"]},
185 )
186 await self.__trading_strategy.subscribe(
187 params={
188 "channel": "ohlc",
189 "interval": 15,
190 "symbol": self.__config["pairs"],
191 },
192 )
193
194 await self.__trading_strategy.subscribe(params={"channel": "executions"})
195
196 while not self.__trading_strategy.exception_occur:
197 # Check if the algorithm feels good
198 # Send a status update every day via Telegram or Mail
199 # …
200 await asyncio.sleep(6)
201
202 except Exception as exc:
203 LOG.error(message := f"Exception in main: {exc} {traceback.format_exc()}")
204 self.__trading_strategy.save_exit(reason=message)
205 finally:
206 # Close the sessions properly.
207 await self.__trading_strategy.close()
208
209 def __check_credentials(self: Manager) -> bool:
210 """Checks the user credentials and the connection to Kraken"""
211 try:
212 User(self.__config["key"], self.__config["secret"]).get_account_balance()
213 LOG.info("Client credentials are valid.")
214 return True
215 except urllib3.exceptions.MaxRetryError:
216 LOG.error("MaxRetryError, can't connect.")
217 return False
218 except requests.exceptions.ConnectionError:
219 LOG.error("ConnectionError, Kraken not available.")
220 return False
221 except KrakenAuthenticationError:
222 LOG.error("Invalid credentials!")
223 return False
224
225 def save_exit(self: Manager, reason: str = "") -> None:
226 """Invoke the save exit function of the trading strategy"""
227 print(f"Save exit triggered - {reason}")
228 if self.__trading_strategy is not None:
229 self.__trading_strategy.save_exit(reason=reason)
230 else:
231 sys.exit(1)
232
233
234def main() -> None:
235 """Example main - load environment variables and run the strategy."""
236 manager: Manager = Manager(
237 config={
238 "key": os.getenv("SPOT_API_KEY"),
239 "secret": os.getenv("SPOT_SECRET_KEY"),
240 "pairs": ["DOT/USD", "BTC/USD"],
241 },
242 )
243
244 try:
245 manager.run()
246 except Exception:
247 manager.save_exit(
248 reason=f"manageBot.run() has ended: {traceback.format_exc()}",
249 )
250
251
252if __name__ == "__main__":
253 main()