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)
Great initiative. You can also look at dataclasses to simplify the code even more.
from dataclasses import dataclass
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?
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.
Hi I'm really interested in this project, when it will be released the part 3 ? I would like to use it.
Hey there thank you the third and last part will come soon this week or the next.
Amazing article, waiting for part 3
Some comments may only be visible to logged-in visitors. Sign in to view all comments.