Welcome to the third and last part of this post. The first part is here and the second part is here.
Dataset creation
Dataset business object
First let's introduce a new "dataset" business object to group prices.
./models/dataset.py
from datetime import datetime
from api import utils
from models.model import AbstractModel
from models.exchange import Exchange
from models.currency import Currency
class Dataset(AbstractModel):
resource_name = 'datasets'
pair: str = ''
exchange: str = ''
period_start: str = ''
period_end: str = ''
currency: str = ''
asset: str = ''
relations = {'exchange': Exchange, 'currency': Currency, 'asset': Currency}
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.pair = self.get_pair()
def get_pair(self):
return utils.format_pair(self.currency, self.asset)
Import service
then we need to build a service to parse and load historical data from the Binance exchange or any other exchange with an API and such historical ticker endpoint.
./services/importer.py
import sys
from datetime import datetime
from models.dataset import Dataset
class Importer:
def __init__(self, exchange, period_start: datetime, period_end=None, interval=60, *args, **kwargs):
self.exchange = exchange
self.interval = interval
self.period_start = period_start
self.period_end = period_end
self.start = datetime.now()
self.dataset = Dataset().create(
data={'exchange': '/api/exchanges/'+self.exchange.name.lower(), 'periodStart': self.period_start, 'periodEnd': self.period_end,
'candleSize': 60,
'currency': '/api/currencies/'+self.exchange.currency.lower(), 'asset': '/api/currencies/'+self.exchange.asset.lower()})
def process(self):
for price in self.exchange.historical_symbol_ticker_candle(self.period_start, self.period_end, self.interval):
print(price.create({'dataset': '/api/datasets/'+self.dataset.uuid}))
execution_time = datetime.now() - self.start
print('Execution time: ' + str(execution_time.total_seconds()) + ' seconds')
sys.exit()
This service responsibility is really simple and clear, his name say it all, we import and store historical ticker data from exchanges.
Here you can directly store your objects on a relational database like PostgreSQL for instance, you can also build and use an internal REST API as proxy to your database for high performance purposes.
Backtesting
Backtesting is the most important tool to write your future bulletproof bot and test it against all market situations from historical ticker data.
For that purpose we'll create a backtest service, his responsibilities will be to load a dataset from your current local data, and if not found then it load it directly from an exchange (Binance by default). Then run a given strategy against each price data candle from the historical dataset.
/services/backtest.py
import sys
from datetime import datetime
from exchanges.exchange import Exchange
from models.dataset import Dataset
from models.price import Price
class Backtest:
def __init__(self, exchange: Exchange, period_start: datetime, period_end=None, interval=60):
self.launchedAt = datetime.now()
# Try to find dataset
dataset = Dataset().query('get', {"exchange": '/api/exchanges/' + exchange.name.lower(),
"currency": '/api/currencies/' + exchange.currency.lower(),
"asset": '/api/currencies/' + exchange.asset.lower(),
"period_start": period_start, "period_end": period_end, "candleSize": interval})
if dataset and len(dataset) > 0:
print(dataset[0])
price = Price()
for price in price.query('get', {"dataset": dataset[0]['uuid']}):
newPrice = Price()
newPrice.populate(price)
exchange.strategy.set_price(newPrice)
exchange.strategy.run()
else:
print("Dataset not found, external API call to " + exchange.name)
for price in exchange.historical_symbol_ticker_candle(period_start, period_end, interval):
exchange.strategy.set_price(price)
exchange.strategy.run()
execution_time = datetime.now() - self.launchedAt
print('Execution time: ' + str(execution_time.total_seconds()) + ' seconds')
sys.exit()
Project's configuration
We'll using dotenv library and conventions to manage environment variables. Here's the project's default values:
./.env.local
AVAILABLE_EXCHANGES="coinbase,binance"
EXCHANGE="binance"
BINANCE_API_KEY="Your Binance API KEY"
BINANCE_API_SECRET="Your Binance API SECRET"
COINBASE_API_KEY="Your Coinbase API KEY""
COINBASE_API_SECRET="Your Coinbase API SECRET""
# Available modes
# "trade" to trade on candlesticks
# "live" to live trade throught WebSocket
# "backtest" to test a strategy for a given symbol pair and a period
# "import" to import dataset from exchanges for a given symbol pair and a period
MODE="trade"
STRATEGY="logger"
# Allow trading "test" mode or "real" trading
TRADING_MODE="test"
# Default candle size in seconds
CANDLE_INTERVAL=60
CURRENCY="BTC"
ASSET="EUR"
# Default period for backtesting: string in UTC format
PERIOD_START="2021-02-28T08:49"
PERIOD_END="2021-03-09T08:49"
DATABASE_URL="postgresql://postgres:password@127.0.0.1:15432/cryptobot"
Main thread
Then put all those parts together on a main thread, mostly a CLI command using args and also environment variables.
By doing so we can override any default environment settings and tweak all input parameters directly with the command line based client.
Really useful too when using containerization tool like Docker for instance, just launch this main thread and it will run with the specific container's environment variables.
We'll dynamically load and import each components we created according to the settings.
./main.py
#!/usr/bin/python3
import importlib
import signal
import sys
import threading
from decouple import config
from services.backtest import Backtest
from services.importer import Importer
exchange_name = config('EXCHANGE')
available_exchanges = config('AVAILABLE_EXCHANGES').split(',')
mode: str = config('MODE')
strategy: str = config('STRATEGY')
trading_mode: str = config('TRADING_MODE')
interval: int = int(config('CANDLE_INTERVAL'))
currency: str = config('CURRENCY')
asset: str = config('ASSET')
if trading_mode == 'real':
print("*** Caution: Real trading mode activated ***")
else:
print("Test mode")
# Parse symbol pair from first command argument
if len(sys.argv) > 1:
currencies = sys.argv[1].split('_')
if len(currencies) > 1:
currency = currencies[0]
asset = currencies[1]
# Load exchange
print("Connecting to {} exchange...".format(exchange_name[0].upper() + exchange_name[1:]))
exchangeModule = importlib.import_module('exchanges.' + exchange_name, package=None)
exchangeClass = getattr(exchangeModule, exchange_name[0].upper() + exchange_name[1:])
exchange = exchangeClass(config(exchange_name.upper() + '_API_KEY'), config(exchange_name.upper() + '_API_SECRET'))
# Load currencies
exchange.set_currency(currency)
exchange.set_asset(asset)
# Load strategy
strategyModule = importlib.import_module('strategies.' + strategy, package=None)
strategyClass = getattr(strategyModule, strategy[0].upper() + strategy[1:])
exchange.set_strategy(strategyClass(exchange, interval))
# mode
print("{} mode on {} symbol".format(mode, exchange.get_symbol()))
if mode == 'trade':
exchange.strategy.start()
elif mode == 'live':
exchange.start_symbol_ticker_socket(exchange.get_symbol())
elif mode == 'backtest':
period_start = config('PERIOD_START')
period_end = config('PERIOD_END')
print(
"Backtest period from {} to {} with {} seconds candlesticks.".format(
period_start,
period_end,
interval
)
)
Backtest(exchange, period_start, period_end, interval)
elif mode == 'import':
period_start = config('PERIOD_START')
period_end = config('PERIOD_END')
print(
"Import mode on {} symbol for period from {} to {} with {} seconds candlesticks.".format(
exchange.get_symbol(),
period_start,
period_end,
interval
)
)
importer = Importer(exchange, period_start, period_end, interval)
importer.process()
else:
print('Not supported mode.')
def signal_handler(signal, frame):
if (exchange.socket):
print('Closing WebSocket connection...')
exchange.close_socket()
sys.exit(0)
else:
print('stopping strategy...')
exchange.strategy.stop()
sys.exit(0)
# Listen for keyboard interrupt event
signal.signal(signal.SIGINT, signal_handler)
forever = threading.Event()
forever.wait()
exchange.strategy.stop()
sys.exit(0)
Usage
# Real time trading mode via WebSocket
MODE=live ./main.py BTC_EUR
# Trading mode with default 1 minute candle
MODE=trade ./main.py BTC_EUR
# Import data from Exchange
MODE=import ./main.py BTC_EUR
# Backtest with an imported dataset or Binance Exchange API
MODE=backtest ./main.py BTC_EUR
You can easily override any settings at call like so:
PERIOD_START="2021-04-16 00:00" PERIOD_END="2021-04-16 00:00" STRATEGY=myCustomStrategy MODE=backtest ./main.py BTC_EUR
To exit test mode and trade for real just switch "trading_mode" from "test" to "real". Use with caution at your own risks.
TRADING_MODE=real ./main.py BTC_EUR
Containerize project
We can containerize this program using Docker. Here's a dead simple self explaining Docker build file.
FROM python:3.9
WORKDIR /usr/src/app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD [ "python", "./main.py" ]
Benchmark
Using an old AMD Phenom II 955 quad core CPU with 16go of DDR3 ram, with other process running.
Import
Import and persist prices to an internal API
1 day ticker spitted onto 1 minutes candles:
Execution time: 82.716666 seconds
1 week ticker spitted onto 1 minutes candles:
Execution time: 9,423079183 minutes
1 month ticker spitted onto 1 minutes candles:
Execution time: 27,48139456 minutes
6 months ticker spitted onto 1 minutes candles:
Execution time: 3.032364739 hours
Backtest
From imported dataset
1 day ticker spitted onto 1 minutes candles:
Execution time: 3.746787 seconds
1 week ticker spitted onto 1 minutes candles:
Execution time: 46.900068 seconds
1 month ticker spitted onto 1 minutes candles:
Execution time: 1.8953 seconds
6 months ticker spitted onto 1 minutes candles:
Execution time: 12,15175435 minutes
Conclusions
We built a kickass performances real time crypto trading bot. He is able to backtest your strategies over big market dataset REALLY QUICLY using a small amount of CPU and RAM. Import datasets from exchanges, perform live trading with customizable candle sizes or even real time using WebSocket.
To go further
Code a tests suite that cover all program's behaviors to ensure no future regression.
Build and use an internal Rest API to persist all crypto exchange markets data in real time.
Build a end user client such like mobile app or web app. Using WebSocket or Server Sent Events, to display real time metrics.
Source code
Want to start your own strategy with your custom indicators, or just contribute and improve this project, you can find the full project source code on github.
Use with the stable branch and contribute using the main branch develop.
As finishing this last post, I released the 0.4 stable version
All contributions are welcome!
Thank's for reading this three parts post on how to build a crypto bot with python 3 and the Binance API.
Top comments (8)
OK...excellent.
Having a little trouble tracking this down, as it throws an error when I try to explore the Arbitrage strategy:
I don't see exchange.get_client() so was this a "ToDo"? If so, do you know what was intended, roughly?
I admire this code very much, but while the level of abstraction is admirable, it's blowing my mind so I'm going to dumb it down for my own beta. I hope to get a better grasp of this coding approach for future revisions.
Would be grateful if you could point to a resource where I may learn about it.
Thank you.
Hello thanks for your contribution, I liked your code. only that when executing the = import mode it gives me the following error .. Could you help me?
Thanks for everything.
self.dataset = Dataset().create(
AttributeError: 'Dataset' object has no attribute 'create'
How would you implement machine learning or perhaps just optimization using backrest?
You can connect an AI prediction system for instance, just create a custom then you can send requests to another API for instance.
To get better perf with backtest, you first need to store them on a database server for instance, then directly query your datasets from. I personally use it with one another instance with an API and an another for database. On my benchmark all was containerized on the same host FYI.
Thanks a lot for your tutorials !
I'm struggling to run the project, I'm wondering what is missing.
I'm a newbie in dev. This is the error is generated when I try to run main.py: "ImportError: cannot import name 'utils' from 'api' (C:\Users\ASUS\anaconda3\envs\Kraken_trader\lib\s
ite-packages\api_init_.py)"
I'm testing your code on jupyterlab is that good ?
Not understood how to use dotenv and docker yet.
Do you have any suggestions to help me go forward on this tutoriel ?
Thanks a lot for your help.
thanks for your tutorials!
one question: do you need a REST API to log the data to the database?
You can directly store on a database, using queries or an ORM layer like SQLAlchemy for instance.
Sending a simple POST request to a REST API is a much faster solution to persist entities.