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