用 Python 和幣安 API 構(gòu)建數(shù)字貨幣交易機器人（三）
數(shù)據(jù)集創(chuàng)建
首先，讓我們介紹一個新的“數(shù)據(jù)集”業(yè)務(wù)對象來對價格進行分組。
./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)
引入服務(wù)對象
我們需要構(gòu)建一個服務(wù)來解析和加載Binance交易所或任何其他具有API和此類歷史行情自動收錄器端點的歷史數(shù)據(jù)。
./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()
這項服務(wù)職責非常簡單明了，他的名字說明了一切，我們從交易所導(dǎo)入和存儲歷史行情自動收錄器數(shù)據(jù)。
在這里，您可以將對象直接存儲在諸如PostgreSQL之類的關(guān)系數(shù)據(jù)庫中，也可以構(gòu)建內(nèi)部REST API并將其用作數(shù)據(jù)庫的代理，以實現(xiàn)高性能。
回測
回測是編寫您未來的機器人并根據(jù)歷史行情自動收錄器數(shù)據(jù)針對所有市場情況進行測試的最重要工具。
為此，我們將創(chuàng)建一個回測服務(wù)，他的職責是從您當前的本地數(shù)據(jù)中加載數(shù)據(jù)集，如果找不到，則直接從交易所加載（默認情況下為Binance）。然后針對歷史數(shù)據(jù)集中的每個價格數(shù)據(jù)運行給定策略。
/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()
項目配置
我們將使用dotenv庫來管理環(huán)境變量。這是項目的默認值：
./.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:[email protected]:15432/cryptobot"
主線程
然后將所有這些部分放到一個主線程上，主要是使用args以及環(huán)境變量的CLI命令。
這樣，我們可以覆蓋任何默認環(huán)境設(shè)置，并直接使用基于命令行的客戶端直接調(diào)整所有輸入?yún)?shù)。
例如，在使用諸如Docker之類的容器化工具時，它也確實非常有用，只需啟動此主線程，它將與特定容器的環(huán)境變量一起運行。
我們將根據(jù)設(shè)置動態(tài)加載和導(dǎo)入我們創(chuàng)建的每個組件。
./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)
用法
# 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
您可以輕松地覆蓋任何設(shè)置，如下所示：
PERIOD_START="2021-04-16 00:00" PERIOD_END="2021-04-16 00:00" STRATEGY=myCustomStrategy MODE=backtest ./main.py BTC_EUR
要退出測試模式并進行真實交易，只需將“ trading_mode”從“ test”切換為“ real”。謹慎使用，后果自負。
TRADING_MODE=real ./main.py BTC_EUR
集成化項目
我們可以使用Docker對該程序進行容器化。這是一個簡單的自我解釋Docker構(gòu)建文件的例子。
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" ]
基準
使用帶有16go DDR3 ram的舊AMD Phenom II 955四核CPU，并運行其他進程。
導(dǎo)入
導(dǎo)入價格并將其部署到內(nèi)部API
1天的行情
Execution time: 82.716666 seconds
一周的行情
Execution time: 9,423079183 minutes
一個月的行情
Execution time: 27,48139456 minutes
六個月的行情
Execution time: 3.032364739 hours
回測
從導(dǎo)入的數(shù)據(jù)集
1天的行情
Execution time: 3.746787 seconds
一周的行情
Execution time: 46.900068 seconds
一個月的行情
Execution time: 1.8953 seconds
六個月的行情
Execution time: 12,15175435 minutes
結(jié)論
我們構(gòu)建了一個kickass性能的實時加密交易機器人。他能夠使用少量的CPU和RAM在大型市場數(shù)據(jù)集上對您的策略進行回測。從交易所導(dǎo)入數(shù)據(jù)集，甚至使用WebSocket實時執(zhí)行實時交易。
更進一步
編寫一個涵蓋所有程序行為的測試套件，以確保將來不會退化。
構(gòu)建并使用內(nèi)部Rest API實時持久保存所有加密貨幣交易所市場數(shù)據(jù)。
建立最終用戶客戶端，例如移動應(yīng)用或網(wǎng)絡(luò)應(yīng)用。使用WebSocket或服務(wù)器發(fā)送事件，以顯示實時指標。
感謝您閱讀這三部分的文章，內(nèi)容涉及如何使用python 3和Binance API構(gòu)建加密機器人。
評論
圖片
表情