DEV Community

Cover image for How to build a crypto bot with Python 3 and the Binance API (part 3)
Nicolas Bonnici
Nicolas Bonnici

Posted on • Edited on

How to build a crypto bot with Python 3 and the Binance API (part 3)

#crypto #bot #trading #python

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)
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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" ]
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

1 week ticker spitted onto 1 minutes candles: 

Execution time: 9,423079183 minutes
Enter fullscreen mode Exit fullscreen mode

1 month ticker spitted onto 1 minutes candles: 

Execution time: 27,48139456 minutes
Enter fullscreen mode Exit fullscreen mode

6 months ticker spitted onto 1 minutes candles: 

Execution time: 3.032364739 hours
Enter fullscreen mode Exit fullscreen mode

Backtest

From imported dataset

1 day ticker spitted onto 1 minutes candles:

Execution time: 3.746787 seconds
Enter fullscreen mode Exit fullscreen mode

1 week ticker spitted onto 1 minutes candles:

Execution time: 46.900068 seconds
Enter fullscreen mode Exit fullscreen mode

1 month ticker spitted onto 1 minutes candles:

Execution time: 1.8953 seconds
Enter fullscreen mode Exit fullscreen mode

6 months ticker spitted onto 1 minutes candles:

Execution time: 12,15175435 minutes
Enter fullscreen mode Exit fullscreen mode

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)

Collapse
 
blahdeblah profile image
blahdeBlah •

OK...excellent.

Having a little trouble tracking this down, as it throws an error when I try to explore the Arbitrage strategy:

response = self.exchange.get_client().symbol_ticker()
Enter fullscreen mode Exit fullscreen mode

I don't see exchange.get_client() so was this a "ToDo"? If so, do you know what was intended, roughly?

Collapse
 
blahdeblah profile image
blahdeBlah • • Edited

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.

Collapse
 
adcueto profile image
Adrian Cueto •

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'

Collapse
 
donjonson profile image
donjonson • • Edited

How would you implement machine learning or perhaps just optimization using backrest?

Collapse
 
nicolasbonnici profile image
Nicolas Bonnici •

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.

Collapse
 
tontonconvolution2000 profile image
Tontonconvolution2000 •

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.

Collapse
 
buscon profile image
buscon •

thanks for your tutorials!
one question: do you need a REST API to log the data to the database?

Collapse
 
nicolasbonnici profile image
Nicolas Bonnici •

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.

Read next

nicobistolfi profile image

9 Coding Principles to Write Great Python Code 🐍🚀

Nico Bistolfi -

yuheshpandian profile image

Wonders with Click: Streamline your CLI Experience

Yuhesh Pandian P J -

siddharthbhalsod profile image

Case Study on Python for AI: Unlocking Business Success through Advanced Analytics

Siddharth Bhalsod -

chiefremote profile image

Part 4 - Using Cursor and Claude to Create Automated Tests with Playwright

Joe Giglio, Chief Remote Officer -