Добро пожаловать в третью и последнюю часть этого поста. Первая часть здесь, а вторая часть здесь.
Набор данных бизнес -объект
Сначала давайте представим новый бизнес -объект «набора данных» для группировки цен.
./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)
Затем нам нужно создать услугу для разбора и загрузки исторических данных из бинанса или любого другого обмена с API и такой исторической конечной точкой тикера.
./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()
Эта ответственность за обслуживание действительно проста и ясна, его имя говорит, что мы импортируем и храним исторические данные о тикерах на биржах.
Здесь вы можете напрямую хранить свои объекты в реляционной базе данных, такой как PostgreSQL, например, вы также можете создавать и использовать внутренний API REST в качестве прокси для вашей базы данных для высокопроизводительных целей.
Западный тестирование является наиболее важным инструментом для написания вашего будущего пуленепробиваемого бота и проверить его на всех рыночных ситуациях на основе исторических данных тикера.
Для этой цели мы создадим сервис бэктеста, его обязанности будут заключаться в загрузке набора данных из ваших текущих локальных данных, и, если не найдено, он загружает его непосредственно с обмена (Binance по умолчанию). Затем запустите данную стратегию против каждой свечи данных о ценах из исторического набора данных.
/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 и соглашения для управления переменными среды. Вот значения проекта по умолчанию:
./.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:password@127.0.0.1:15432/cryptobot"
Затем сложите все эти части на основной потоке, в основном команда CLI, используя ARG, а также переменные среды.
При этом мы можем переопределить любые настройки среды по умолчанию и настраивать все входные параметры непосредственно с помощью клиента на основе командной строки.
Например, действительно полезно при использовании инструмента для контейнеризации, такого как Docker, просто запустите этот основной поток, и он будет работать с конкретными переменными среды контейнера.
Мы динамически загружаем и импортируем все компоненты, которые мы создали в соответствии с настройками.
./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
Вы можете легко переопределить любые настройки при вызове, например,:
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” Используйте с осторожностью в ваших собственных рисках.
TRADING_MODE=real ./main.py BTC_EUR
Контейнерный проект
Мы можем контейнерить эту программу с помощью Docker. Вот мертвый простой самостоятельно объясняющий файл Docker Build.
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" ]
Используя старый четырехъядерный процессор AMD II 955 с 16GO с DDR3 RAM, а другой процесс работает.
импорт
Импорт и упорные цены во внутренний API
1 -дневный билет разделен на 1 минуту свечи:
Execution time: 82.716666 seconds
1 -недельный тикер выплетен на 1 минуту свечи:
Execution time: 9,423079183 minutes
1 месяц тикер на 1 минуту свечи:
Execution time: 27,48139456 minutes
6 -месячный тикер плюнул на 1 минуту свечи:
Execution time: 3.032364739 hours
Бэк -тест
Из импортного набора данных
1 -дневный билет разделен на 1 минуту свечи:
Execution time: 3.746787 seconds
1 -недельный тикер выплетен на 1 минуту свечи:
Execution time: 46.900068 seconds
1 месяц тикер на 1 минуту свечи:
Execution time: 1.8953 seconds
6 -месячный тикер плюнул на 1 минуту свечи:
Execution time: 12,15175435 minutes
Мы построили бот с крипто -торговой торговлей в реальном времени. Он способен поддержать ваши стратегии по сравнению с большим набором данных о рынке, действительно, используя небольшое количество процессора и оперативной памяти. Импортируйте наборы данных с обменов, выполняйте живую торговлю с настраиваемыми размерами свечей или даже в реальном времени, используя WebSocket.
Код A Tests Suite, который охватывает поведение всех программ, чтобы обеспечить отсутствие будущей регрессии.
Создайте и используйте внутренний API REST, чтобы сохранить все данные Crypto Exchange Markets в режиме реального времени.
Создайте клиента конечного пользователя, например, мобильное приложение или веб -приложение. Использование WebSocket или сервер, отправленных событиями, для отображения метрик в реальном времени.
Хотите начать свою собственную стратегию с ваших пользовательских индикаторов или просто внести свой вклад и улучшить этот проект, вы можете найти полный проект исходный код на GitHub Анкет
Использовать с стабильная ветвь и внести свой вклад с использованием Основная ветвь развивается Анкет
Как заканчивая этот последний пост, я выпустил стабильную версию 0,4
Все взносы приветствуются!
Спасибо за то, что прочитали этот пост с тремя частями о том, как построить крипто -бот с Python 3 и Binance API.
Оригинал: “https://dev.to/nicolasbonnici/how-to-build-a-crypto-bot-with-python-3-and-the-binance-api-part-3-1c53”