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

#crypto #bot #trading #python

In this post first part, that you can find here: (https://dev.to/nicolasbonnici/how-to-build-a-crypto-bot-with-python-3-and-the-binance-api-part-1-1864), we talked about the global folder structure, add our first Price business object and code our strategies abstract layer.

We'll begin by code the next business objects we'll need currency and order.

Here the currency business object store all the different crypto or fiat currencies.

./models/order.py

from models.model import AbstractModel


class Currency(AbstractModel):
    name: str = ''
    symbol: str = ''
    fiat: bool

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

Then our order business object to handle order created on exchanges.

./models/order.py

from models.model import AbstractModel


class Order(AbstractModel):
    BUY = 'BUY'
    SELL = 'SELL'

    TYPE_LIMIT = 'LIMIT'
    TYPE_MARKET = 'MARKET'
    TYPE_STOP_LOSS = 'STOP_LOSS'
    TYPE_STOP_LOSS_LIMIT = 'STOP_LOSS_LIMIT'
    TYPE_TAKE_PROFIT = 'TAKE_PROFIT'
    TYPE_TAKE_PROFIT_LIMIT = 'TAKE_PROFIT_LIMIT'
    TYPE_LIMIT_MAKER = 'LIMIT_MAKER'

    uuid = ''
    side: str = ''
    type: str = TYPE_LIMIT
    symbol: str = ''
    currency: str = ''
    asset: str = ''
    price: float = 0
    quantity: int = 0
    test: bool = False

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

Then build the exchange abstract layer and develop our first connector for Binance. Since coding an API wrapper is not the point here we gonna use the unofficial Binance API wrapper library python-binance (https://python-binance.readthedocs.io/en/latest/binance.html) for convenience.

./exchanges/exchange.py

import datetime
from api import utils
from abc import ABC, abstractmethod
from twisted.internet import reactor
from strategies.strategy import Strategy
from models.order import Order


class Exchange(ABC):
    currency: str
    asset: str
    strategy: Strategy

    def __init__(self, key: str, secret: str):
        self.apiKey = key
        self.apiSecret = secret
        self.name = None
        self.client = None
        self.socketManager = None
        self.socket = None
        self.currency = ''
        self.asset = ''
        self.strategy = None

    def set_currency(self, symbol: str):
        self.currency = symbol

    def set_asset(self, symbol: str):
        self.asset = symbol

    def set_strategy(self, strategy: Strategy):
        self.strategy = strategy

    def compute_symbol_pair(self):
        return utils.format_pair(self.currency, self.asset)

    # abstract methods

    # Override to set current exchange symbol pair notation (default with _ separator currency_asset ex: eur_btc)
    @abstractmethod
    def get_symbol(self):
        return self.compute_symbol_pair(self)

    # Get current symbol ticker
    @abstractmethod
    def symbol_ticker(self):
        pass

    # Get current symbol ticker candle for given interval
    @abstractmethod
    def symbol_ticker_candle(self, interval):
        pass

    # Get current symbol historic value
    @abstractmethod
    def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=60):
        pass

    # Get balance for a given currency
    @abstractmethod
    def get_asset_balance(self, currency):
        pass

    # Create an exchange order
    @abstractmethod
    def order(self, order: Order):
        pass

    # Create an exchange test order
    @abstractmethod
    def test_order(self, order: Order):
        pass

    # Check an exchange order status
    @abstractmethod
    def check_order(self, orderId):
        pass

    # Cancel an exchange order
    @abstractmethod
    def cancel_order(self, orderId):
        pass

    # WebSocket related methods

    @abstractmethod
    def get_socket_manager(self, purchase):
        pass

    @abstractmethod
    def websocket_event_handler(self, msg):
        pass

    def start_socket(self):
        print('Starting WebSocket connection...')
        self.socketManager.start()

    def close_socket(self):
        self.socketManager.stop_socket(self.socket)
        self.socketManager.close()
        # properly terminate WebSocket
        reactor.stop()

    @abstractmethod
    def start_symbol_ticker_socket(self, symbol: str):
        pass

Our first connector for the Binance API.

./exchanges/binance.py

from datetime import datetime
from math import floor

from binance.client import Client
from binance.enums import *
from binance.websockets import BinanceSocketManager

from api import utils
from exchanges import exchange
from models.order import Order
from models.price import Price


class Binance(exchange.Exchange):
    def __init__(self, key: str, secret: str):
        super().__init__(key, secret)

        self.client = Client(self.apiKey, self.apiSecret)
        self.name = self.__class__.__name__

    def get_client(self):
        return self.client

    def get_symbol(self):
        return self.currency + self.asset

    def symbol_ticker(self):
        response = self.client.get_symbol_ticker(symbol=self.get_symbol())
        return Price(pair=self.get_symbol(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
                     current=response['price'])

    def symbol_ticker_candle(self, interval=Client.KLINE_INTERVAL_1MINUTE):
        return self.client.get_klines(symbol=self.get_symbol(), interval=interval)

    def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=Client.KLINE_INTERVAL_1MINUTE):
        # Convert default seconds interval to string like "1m"
        if isinstance(interval, int):
            interval = str(floor(interval/60)) + 'm'

        output = []
        for candle in self.client.get_historical_klines_generator(self.get_symbol(), interval, start, end):
            output.append(
                Price(pair=self.compute_symbol_pair(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
                      current=candle[1], lowest=candle[3], highest=candle[2], volume=candle[5], openAt=utils.format_date(datetime.fromtimestamp(int(candle[0])/1000)))
            )

        return output

    def get_asset_balance(self, currency):
        response = self.client.get_asset_balance(currency)
        return response['free']

    def order(self, order: Order):
        return self.client.create_order(
            symbol=order.symbol,
            side=order.side,
            type=order.type,
            timeInForce=TIME_IN_FORCE_GTC,
            quantity=order.quantity,
            price=order.price
        )

    def test_order(self, order: Order):
        return self.client.create_test_order(
            symbol=order.symbol,
            side=order.side,
            type=order.type,
            timeInForce=TIME_IN_FORCE_GTC,
            quantity=order.quantity,
            price=order.price
        )

    def check_order(self, orderId):
        return self.client.get_order(
            symbol=self.get_symbol(),
            orderId=orderId
        )

    def cancel_order(self, orderId):
        return self.client.cancel_order(
            symbol=self.get_symbol(),
            orderId=orderId
        )

    def get_socket_manager(self):
        return BinanceSocketManager(self.client)

    def start_symbol_ticker_socket(self, symbol: str):
        self.socketManager = self.get_socket_manager()
        self.socket = self.socketManager.start_symbol_ticker_socket(
            symbol=self.get_symbol(),
            callback=self.websocket_event_handler
        )

        self.start_socket()

    def websocket_event_handler(self, msg):
        if msg['e'] == 'error':
            print(msg)
            self.close_socket()
        else:
            self.strategy.set_price(
                Price(pair=self.compute_symbol_pair(), currency=self.currency, asset=self.asset, exchange=self.name,
                      current=msg['b'], lowest=msg['l'], highest=msg['h'])
            )
            self.strategy.run()

We now have a minimalist but powerful trading bot system by just calling our strategy start method. Using your own indicators and stratgy you can start buying and selling by passing orders. But before launching you to code your own kick ass genius strategy, be sure to test it again and again. So you'll need a system to do that

In the next part we gonna implement a backtest mode, to test your strategies against historical exchange's data and also a service to import them.

Plus we gonna put all those pieces together with a global configuration and some dependencies injection in a command line tool.

Then later we'll how to go further by seeing how to containerize and industrialize this code.

Feel free to comment, if you have questions.

Thank you.

Top comments (6)

Rohan Singh • Apr 20 '21

Great initiative. You can also look at dataclasses to simplify the code even more.
from dataclasses import dataclass

svenkerstjens • Apr 7 '21

Hi, very interesting post! I just keep getting the error that the reference for utils in the api module cannot be found. Do you have a suggestion to resolve this?

Nicolas Bonnici • Apr 9 '21

Hi thank for your feedback, there's a repository you can find here on Github github.com/nicolasbonnici/cryptobot, you have two branch develop the default one and a stable one. I'll take the time this weekend to write the third and last part of this post, feel free to use and contribute on the repository.