The first point about trading crypto currencies or any asset is to have a goal and a strategy to achieve. Here i’am not writing about trading strategy but just build a simple yet functional crypto trader bot to apply your strategy. Trade with caution this serie of post is just more like an automated crypto trading bot framework.
We’ll use python 3.9 (3.9.2) to first create the project file structure.
/exchanges
/strategies
/models
Here “exchanges” folder store the exchanges API wrappers, strategies your strategies and models the business object we gonna use.
The models
We’ll defined several business objects for this projects, such like price, currency or order for instance. First let’s begin with the Price one to use it on the strategy we’ll create just after.
Before we write a common abstract layer forour business object.
./models/model.py
from datetime import datetime
class AbstractModel:
created: datetime
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
Then our first Price class.
./models/price
from models.model import AbstractModel
class Price(AbstractModel):
pair: str = ''
exchange: str = ''
current: float = 0
lowest: float = 0
highest: float = 0
currency: str = ''
asset: str = ''
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.pair = self.get_pair()
def get_pair(self):
return self.currency + '_' + self.asset
The strategy part
Here if the biggest part of this system, the strategy will be responsible to run your own trading logic and to do so, there’s two way, a classic using an interval then run API calls to external exchanges API or even internal webhooks and route and a real time event based one using WebSocket.
To do so we’ll first create an abstract strategy class that our strategies will extend.
./strategies/strategy.py
import json
import threading
import time
from datetime import datetime
from decouple import config
from models.price import Price
class Strategy(object):
price: Price
def __init__(self, exchange, interval=60, *args, **kwargs):
self._timer = None
self.interval = interval
self.args = args
self.kwargs = kwargs
self.is_running = False
self.next_call = time.time()
self.portfolio = {}
self.exchange = exchange
# Load account portfolio for pair at load
self.get_portfolio()
def _run(self):
self.is_running = False
self.start()
self.run(*self.args, **self.kwargs)
def start(self):
if not self.is_running:
print(datetime.now())
if self._timer is None:
self.next_call = time.time()
else:
self.next_call += self.interval
self._timer = threading.Timer(self.next_call - time.time(), self._run)
self._timer.start()
self.is_running = True
def stop(self):
self._timer.cancel()
self.is_running = False
def get_portfolio(self):
self.portfolio = {'currency': self.exchange.get_asset_balance(self.exchange.currency),
'asset': self.exchange.get_asset_balance(self.exchange.asset)}
def get_price(self):
try:
self.price = self.exchange.symbol_ticker()
except Exception as e:
pass
...
Here our strategy abstract layer constructor signature need an Exchange instance, we’ll do this part later writing of first wrapper using the Binance API.
We define without any extra library a simple yet functional infinite interval runner, note that this every run will launch the next call on another thread but in fact your strategy will use never no more than two threads, one main and the current run iteration if you strategy take a lot of external call or heavy computation. Each thread consume 0.3% of RAM and 0 or 0.1 CPU usage, this also involve the Strategy to fetch ticker and orders, then store price and order related data on an another internal API.
Yet the interval run precision can drift a little bit on microseconds, but will be stable at seconds level.
Here a simple usage of that layer, a strategy that basically print your exchange account portfolio and the exchange price. We got method to also retrieve a symbol ticker from an external exchange we gonna connect later, and also a method to retrieve your current portfolio available on connected exchange.
./strategies/watcher.py
from exchanges.exchange import Exchange
from strategies.strategy import Strategy
class Watcher(Strategy):
def __init__(self, exchange: Exchange, timeout=60, *args, **kwargs):
super().__init__(exchange, timeout, *args, **kwargs)
def run(self):
self.get_price()
print('*******************************')
print('Exchange: ', self.exchange.name)
print('Pair: ', self.exchange.get_symbol())
print('Available: ', self.portfolio['currency'] + ' ' + self.exchange.currency)
print('Available: ', self.portfolio['asset'] + ' ' + self.exchange.asset)
print('Price: ', self.price.current)
In the next parts we gonna connect our first Exchange Binance by coding a simple wrapper using the official Binance API https://github.com/binance/binance-spot-api-docs and Python requests library https://pypi.org/project/requests/. Then code our missing business objects order and currency to store respectively the order you send to exchanges and the currencies we’ll manipulate fiat or cryptocurrencies then put all those different pieces together.
This article has been published from the source link without modifications to the text. Only the headline has been changed.
