15 September 2024
Photo by Christopher Gower
As a algorithmic trader you may be interested in developing more sophisticated trading bots that incorporate a variety of functions like multiple data analysis, news sentiment analysis, AI, etc. A clean, well-structured software architecture can help to manage the increased complexity of the bot.
This story is solely for general information purposes, and should not be relied upon for trading recommendations or financial advice. Source code and information is provided for educational purposes only, and should not be relied upon to make an investment decision. Please review my full cautionary guidance before continuing.
This post proposes a generic trading bot architecture that can be used for any type of market. The architecture consists of an object-oriented, typed data model implemented in Python.
The model contains the following components/classes:
If you are interesting in building your own trading bot in Python, this post is for you!
The architecture proposed in this post is based on the Python backtesting framework ‘backtesting.py’. I selected it because it had a simple, object-oriented and typed data model that is also suitable for a bot software architecture.
The Python script used in this post is a working example currently in the ‘concept’ state. It has not been tested in production.
You can download the complete script from my blog ‘B/O Trading Blog’.
Here the necessary Python packages I used for this prototype.
from typing import Callable, Dict, List, Optional, Sequence, Tuple, Type, Union
import numpy as np
import datetime
import os
from loguru import logger
import sys
import time
import requests
import pandas as pd
from pandas import DataFrame
import pandas_ta as ta
import uuid
from enum import EnumI defined the following enums for Position, Side and Order Status to be used by the Bot.
class Positions(Enum):
LONG = 'LONG'
SHORT = 'SHORT'
class Side(Enum):
BUY = 'BUY'
SELL = 'SELL'
class OrderStatus(Enum):
PENDING = 'PENDING'
CANCELED = 'CANCELED'
FILLED = 'FILLED'
The Config class is a convenience class to contain all the configuration parameters the Bot needs. Here you may want to specify which symbols to trade, the available cash for trading or the interval in which to fetch new price data.
class Config():
"""
Class for Bot configuration
"""
def __init__(self, symbols, total_cash, cash_per_trade, data_fetch_interval):
self._symbols = symbols
self._cash_per_trade = cash_per_trade
self._total_cash = total_cash
self._data_fetch_interval = data_fetch_interval # Getters
@property
def symbols(self) -> 'Tuple[str, ...]':
return self._symbols @property
def cash_per_trade(self) -> float:
return self._cash_per_trade @property
def total_cash(self) -> float:
return self._total_cash @property
def data_fetch_interval(self) -> int:
return self._data_fetch_interval
The Bot class is the main class which owns all the secondary classes and controls the actions that a bot must perform, like setting up logging, fetching data at regular intervals, or executing a strategy.
class Bot():
""" Main Trading Bot class """
def __init__(self, config: 'Config'):
self._config = config
self._price_loader = PriceDataLoader()
self._broker = Broker(config.total_cash, self._price_loader)
self._data: None
self._is_running = False # Initial configurations
self.setup_logging() def setup_logging(self):
today = datetime.datetime.today()
today_str = today.strftime("%Y-%m-%d")
log_file_name = f"{today_str}_bot_log.txt"
log_full_path = os.path.join(LOG_DIR, log_file_name)
open(log_full_path, 'w').close()
logger.remove()
logger.add(sys.stdout, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> <red>{level}</red> {message}",colorize=True)
logger.add(log_full_path, format="{time:YYYY-MM-DD HH:mm:ss} {level} {message}") def start(self):
logger.info(f"Starting bot")
self._is_running = True while self._is_running:
# For each symbol -> Get OHLC
symbols = self._config.symbols
for symbol in symbols:
data_df = self._price_loader.load(symbol, 'usd')
strategy = Strategy(symbol, self._broker, self._config.cash_per_trade)
strategy.execute(data_df) # Check order status
self._broker.check_order_status() # Log some stats
logger.info(self._broker.describe()) time.sleep(self._config.data_fetch_interval) def stop(self):
logger.info(f"Stopping bot")
self._is_running = True
The logging system is currently using the Python library loguru but can of course swapped out with other logging frameworks.
The logging system in set up by the Bot during initialization and can then be used by other classes.
def setup_logging(self):
today = datetime.datetime.today()
today_str = today.strftime("%Y-%m-%d")
log_file_name = f"{today_str}_bot_log.txt"
log_full_path = os.path.join(LOG_DIR, log_file_name)
open(log_full_path, 'w').close()
logger.remove()
logger.add(sys.stdout, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> <red>{level}</red> {message}",colorize=True)
logger.add(log_full_path, format="{time:YYYY-MM-DD HH:mm:ss} {level} {message}")The Data Loader class can be used to fetch one or multiple types of data either from your exchange or data API.
In this example I’m fetching OHLC crypto data from CoinGecko.com but you probably want to modify this section get other types of data you need.
class PriceDataLoader():
"""
Loader for OLHC
"""
def __init__(self):
self._last_refresh_date = None
self._data_map = {} def fetch_coin_price_data(self, symbol, currency):
try:
fetch_url = f"https://api.coingecko.com/api/v3/coins/{symbol}/ohlc?vs_currency={currency}&days=7"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:101.0) Gecko/20100101 Firefox/101.0',
'Accept': 'application/json',
'Accept-Language': 'en-US,en;q=0.5',
} # Send GET request
response = requests.get(fetch_url, headers=headers)
data = response.json() df = pd.DataFrame({'Unix': [], 'Open': [], 'High': [], 'Low': [], 'Close': []})
for item in data:
row = pd.DataFrame(
{'Unix': [item[0]], 'Open': [item[1]], 'High': [item[2]], 'Low': [item[3]], 'Close': [item[4]]})
row['Date'] = pd.to_datetime(row['Unix'])
df = pd.concat([df, row], axis=0, ignore_index=True) df.set_index(df['Date']) # Store for last price lookup
self._data_map[symbol] = df return df
except Exception as e:
print('Error loading trending coins, ', e)
return None def load(self, symbol, currency):
logger.info(f"DataLoader: Starting data load at {get_now_date_str()} for symbol {symbol}")
self._data = self.fetch_coin_price_data(symbol, currency)
self._last_refresh_date = get_now_date_str()
return self._data # Getters
@property
def data(self) -> DataFrame:
return self._data @property
def last_refresh_date(self) -> str:
return self._last_refresh_date def last_price(self, symbol: str) -> float:
if symbol in self._data_map:
df = self._data_map[symbol]
return df['Close'].iloc[-1]
else:
return 0
The Order class is used to store information needed to place an order with an exchange, for example the symbol, position, size, limit and stop loss.
class Order:
""" Class for managing orders """
def __init__(self, symbol: str,
position: Positions,
side: Side,
size: float,
limit_price: float = None,
stop_price: float = None):
self._id = str(uuid.uuid4())
self._symbol = symbol
self._position = position
self._side = side
self._size = size
self._price = 0
self._limit_price = limit_price
self._stop_price = stop_price
self._status = OrderStatus.PENDING def describe(self):
return f"<Order id: {self._id} symbol: {self._symbol} position: {self._position} limit: {self._limit_price} stop: {self._stop_price}>" # Getters
@property
def id(self) -> str:
return self._id @property
def symbol(self) -> str:
return self._symbol @property
def position(self) -> Positions:
return self._position @property
def side(self) -> Side:
return self._side @property
def size(self) -> float:
return self._size @property
def price(self) -> float:
return self._price @property
def limit_price(self) -> float:
return self._limit_price @property
def stop_price(self) -> float:
return self._stop_price @property
def status(self) -> OrderStatus:
return self._status # Setters
def set_status(self, status):
self._status = status def set_price(self, price):
self._price = price
When an entry order has been executed by the exchange and we confirmed the order status, we create a trade.
When an exit order is placed and the status was confirmed by the exchange, we look up the last matching trade based on symbol, size, and position and close it.
Closed trades are moved to a ‘closed_trades’ list owned by the Broker.
Here I’m also passing the Price Loader into the trade so we can calculate the profit/loss of the open trade at any time.
class Trade:
""" When an `Order` is filled, it results in an active `Trade` """
def __init__(self, symbol: str, position: Positions, side: Side, size: int, entry_price: float, price_loader: 'PriceDataLoader'):
self._id = str(uuid.uuid4())
self._symbol = symbol
self._position = position
self._side = side
self._size = size
self._entry_price = entry_price
self._exit_price: Optional[float] = None
self._entry_date = get_now_date_str()
self._exit_date: str = None
self._price_loader = price_loader def describe(self):
return f'<Trade id: {self._id} symbol: {self._symbol} size: {self._size} entry_date: {self._entry_date} exit date: {self._exit_date or ""}' \
f'entry price: {self._entry_price} exit price: {self._exit_price or ""} pl: {self.profit_loss_pct}>' # Getters
@property
def id(self) -> str:
return self._id @property
def symbol(self) -> str:
return self._symbol @property
def position(self) -> Positions:
return self._position @property
def side(self) -> Side:
return self._side @property
def size(self) -> float:
return self._size @property
def entry_date(self) -> str:
return self._entry_date @property
def exit_date(self) -> Optional[str]:
return self._exit_date @property
def entry_price(self) -> float:
return self._entry_price @property
def exit_price(self) -> Optional[float]:
return self._exit_price @property
def profit_loss(self):
""" Trade profit (positive) or loss (negative) in cash units. """
price = self._exit_price or self._price_loader.last_price(self._symbol)
return self._size * (price - self._entry_price) @property
def profit_loss_pct(self):
""" Trade profit (positive) or loss (negative) in percent."""
price = self._exit_price or self._price_loader.last_price(self._symbol)
return self._size * (price / self._entry_price - 1) # Setters
def set_exit_price(self, exit_price):
self._exit_price = exit_price
The Position class is used to identify all held positions of an asset — so it is basically all open trades grouped by symbol.
This class dynamically calculates the size and P&L of the position with the help of the Broker.
class Position:
""" Currently held asset position. """ def __init__(self, broker: 'Broker', symbol: str, position: Positions):
self._id = str(uuid.uuid4())
self._symbol = symbol
self._broker = broker
self._position = position def describe(self):
return f"<Position id: {self._id} symbol: {self._symbol} position: {self._position} Size: {self.size} stop: {self.profit_loss}>" # Getters
@property
def size(self) -> float:
total_size = 0
for trade in self._broker.trades:
if trade.symbol != self.symbol or trade.position != self._position:
continue
total_size += trade.size return total_size @property
def profit_loss(self) -> float:
""" P/L of current position in cash units. """
return sum(trade.profit_loss for trade in self._broker.trades) @property
def position(self) -> Positions:
return self._position
The Broker maintains the list of orders, trades, positions and closed trades.
Here we could also maintain the list of Positions. However, it may be more accurate to pull that from the exchange.
The Broker has two main functions. In the ‘submit_order()’ function, we submit buy/sell orders for long or short positions to the exchange.
For closing orders, we look up the size of the order in the open trades list.
In the ‘check_order_status()’ function, we check on the order status with the exchange at regular intervals. Once the order is filled, we either open a new trade if it is an entry order or we close the trade if it is an exit order.
You also see here that we increase and decrease the available cash amount so we can keep track the funds we have available for investment. Alternatively, we could pull this amount from the exchange to get an exact number.
Note that the API integration with the exchange you are using will need to be implemented here.
class Broker:
def __init__(self, cash, price_loader):
# Assert inputs
assert 0 < cash, f"Cash should be > 0, is {cash}" self._cash = cash
self._price_loader = price_loader
self._orders: List[Order] = []
self._trades: List[Trade] = []
self._closed_trades: List[Trade] = [] def describe(self):
# Calculate Profit & Loss
open_trades_profit_loss = 0
for trade in self._trades:
open_trades_profit_loss += trade.profit_loss closed_trades_profit_loss = 0
for trade in self._trades:
closed_trades_profit_loss += trade.profit_loss return f'<Broker: Cash: {self._cash:.0f} Orders: {len(self._orders)} Trades: {len(self._trades)} Closed Trades: {len(self._closed_trades)} \
Open Trades P&L: {open_trades_profit_loss:.0f} Closed Trades P&L: {closed_trades_profit_loss:.0f}>' def submit_order(self,
symbol: str,
position: Positions,
side: Side,
size: float = None,
limit: float = None,
stop: float = None):
size = size and float(size)
stop = stop and float(stop)
limit = limit and float(limit) # Look up the trade to get the size
if position == Positions.LONG and side == Side.SELL:
for trade in reversed(self._trades):
if trade.symbol == symbol and trade.position == position:
size = trade.size # If size is not specified, look up the last trade and close it
elif position == Positions.SHORT and side == Side.BUY:
for trade in reversed(self._trades):
if trade.symbol == symbol and trade.position == position:
size = trade.size if size is None:
logger.error(f"Position {position}, side: {side} - Size not specified for sell request, matching trade not found")
return # Create a new order
order = Order(symbol, position, side, size, limit, stop)
logger.info(f"Sending order: {order.describe()}") # todo: Send order to exchange # Put the new order in the order queue,
self._orders.insert(0, order) return order def check_order_status(self):
logger.info('Checking order status')
for order in list(self._orders):
# todo: Check order status with exchange
order.set_status(OrderStatus.FILLED)
# Use the execution price of the transaction instead
current_price = self._price_loader.last_price(order.symbol)
order.set_price(current_price) # Process different order states
if order.status == OrderStatus.FILLED:
# Long buy or short sell
if (order.position == Positions.LONG and order.side == Side.BUY) or \
(order.position == Positions.SHORT and order.side == Side.SELL):
# Open a new trade
trade = Trade(order.symbol, order.position, order.side, order.size, order.price, self._price_loader)
self._trades.insert(0, trade)
logger.info(f"Opened new trade {trade.describe()}") # Decrease cash reserve
if order.position == Positions.LONG:
self._cash -= order.size * order.price
else:
self._cash += order.size * order.price # Remove order
self._orders.remove(order)
logger.info(f"Removed order {order.describe()}")
# Long sell or short sell
elif (order.position == Positions.LONG and order.side == Side.SELL) or \
(order.position == Positions.SHORT and order.side == Side.BUY):
# Close the trade and add it to the 'closed trades' list
for trade in reversed(self._trades):
if trade.position == order.position and trade.side != order.side and trade.size == order.size:
trade.set_exit_price(order.price)
self._trades.remove(trade)
self._closed_trades.append(trade)
logger.info(f"Closed trade {trade.describe()}") if order.position == Positions.LONG:
self._cash += order.size * order.price
else:
self._cash -= order.size * order.price # Remove order
self._orders.remove(order)
logger.info(f"Removed order {order.describe()}")
Finally, the main() function, which instantiates and configures the Bot and then calls the start() function to start the loop to pull prices and execute the strategy.
# Main
if __name__ == '__main__':
# Create new bot
config = Config(symbols=['litecoin', 'ethereum'], cash_per_trade=100, total_cash=10000, data_fetch_interval=60)
bot = Bot(config)
bot.start()Here a sample of the log output the bot generates. As you can see the output is very detailed, which may be just right while you are developing the bot. Eventually you may manage the logging volume by using different log levels.
2022-08-10 04:45:21 INFO Starting bot
2022-08-10 04:45:21 INFO DataLoader: Starting data load at 2022-08-10 04:45:21 for symbol litecoin
2022-08-10 04:45:22 INFO Executing strategy for symbol litecoin
2022-08-10 04:45:22 INFO Sending order: <Order id: c0706e9c-9e78-4f34-b816-dbc4e33da596 symbol: litecoin position: Positions.LONG limit: None stop: None>
2022-08-10 04:45:22 INFO Checking order status
2022-08-10 04:45:22 INFO Opened new trade <Trade id: a3465439-7866-4956-8bbf-a6f501bca2a1 symbol: litecoin size: 1.7056114617090226 entry_date: 2022-08-10 04:45:22 exit date: entry price: 58.63 exit price: pl: 0.0>
2022-08-10 04:45:22 INFO Removed order <Order id: c0706e9c-9e78-4f34-b816-dbc4e33da596 symbol: litecoin position: Positions.LONG limit: None stop: None>
2022-08-10 04:45:22 INFO DataLoader: Starting data load at 2022-08-10 04:45:22 for symbol ethereum
2022-08-10 04:45:22 INFO Executing strategy for symbol ethereum
2022-08-10 04:45:22 INFO Sending order: <Order id: fb119b88-834f-4e5c-9326-3bee3782a1a9 symbol: ethereum position: Positions.LONG limit: None stop: None>
2022-08-10 04:45:22 INFO Checking order status
2022-08-10 04:45:22 INFO Opened new trade <Trade id: 14296d54-7459-44cc-8b6b-8392510a98cd symbol: ethereum size: 0.05915619601997113 entry_date: 2022-08-10 04:45:22 exit date: entry price: 1690.44 exit price: pl: 0.0>
2022-08-10 04:45:22 INFO Removed order <Order id: fb119b88-834f-4e5c-9326-3bee3782a1a9 symbol: ethereum position: Positions.LONG limit: None stop: None>
2022-08-10 04:45:22 INFO <Broker: Cash: 9800 Orders: 0 Trades: 2 Closed Trades: 0 Open Trades P&L: 0 Closed Trades P&L: 0>
BTC
ETH
USDT
USDC
TRX