Sources Contact Advanced Search Tutorials

An Interest In:

Web News this Week

Search Archive

Some of Our Sources

View All Sources

Help Webnuz

Referal links:

April 19, 2021 07:43 pm GMT

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

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 datetimefrom api import utilsfrom models.model import AbstractModelfrom models.exchange import Exchangefrom models.currency import Currencyclass 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 sysfrom datetime import datetimefrom models.dataset import Datasetclass 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 sysfrom datetime import datetimefrom exchanges.exchange import Exchangefrom models.dataset import Datasetfrom models.price import Priceclass 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 periodMODE="trade"STRATEGY="logger"# Allow trading "test" mode or "real" tradingTRADING_MODE="test"# Default candle size in secondsCANDLE_INTERVAL=60CURRENCY="BTC"ASSET="EUR"# Default period for backtesting: string in UTC formatPERIOD_START="2021-02-28T08:49"PERIOD_END="2021-03-09T08:49"DATABASE_URL="postgresql://postgres:[email protected]: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/python3import importlibimport signalimport sysimport threadingfrom decouple import configfrom services.backtest import Backtestfrom services.importer import Importerexchange_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 argumentif len(sys.argv) > 1:    currencies = sys.argv[1].split('_')    if len(currencies) > 1:        currency = currencies[0]        asset = currencies[1]# Load exchangeprint("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 currenciesexchange.set_currency(currency)exchange.set_asset(asset)# Load strategystrategyModule = importlib.import_module('strategies.' + strategy, package=None)strategyClass = getattr(strategyModule, strategy[0].upper() + strategy[1:])exchange.set_strategy(strategyClass(exchange, interval))# modeprint("{} 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 eventsignal.signal(signal.SIGINT, signal_handler)forever = threading.Event()forever.wait()exchange.strategy.stop()sys.exit(0)

Usage

# Real time trading mode via WebSocketMODE=live ./main.py BTC_EUR# Trading mode with default 1 minute candleMODE=trade ./main.py BTC_EUR# Import data from ExchangeMODE=import ./main.py BTC_EUR# Backtest with an imported dataset or Binance Exchange APIMODE=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.9WORKDIR /usr/src/appCOPY requirements.txt ./RUN pip install --no-cache-dir -r requirements.txtCOPY . .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.

Original Link: https://dev.to/nicolasbonnici/how-to-build-a-crypto-bot-with-python-3-and-the-binance-api-part-3-1c53

Share this article:

View Full Article

Dev To

An online community for sharing and discovering great ideas, having debates, and making friends

More About this Source Visit Dev To