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