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