用 Python 和币安 API 构建数字货币交易机器人

Posted 2021-06-02 Python中文社区

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了用 Python 和币安 API 构建数字货币交易机器人相关的知识，希望对你有一定的参考价值。

欢迎来到本系列的第三部分。第一部分在这里，第二部分在这里。

数据集创建

首先，让我们介绍一个新的“数据集”业务对象来对价格进行分组。

./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)

引入服务对象

我们需要构建一个服务来解析和加载Binance交易所或任何其他具有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之类的关系数据库中，也可以构建内部REST API并将其用作数据库的代理，以实现高性能。

回测

回测是编写您未来的机器人并根据历史行情自动收录器数据针对所有市场情况进行测试的最重要工具。

为此，我们将创建一个回测服务，他的职责是从您当前的本地数据中加载数据集，如果找不到，则直接从交易所加载（默认情况下为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"

主线程

然后将所有这些部分放到一个主线程上，主要是使用args以及环境变量的CLI命令。

这样，我们可以覆盖任何默认环境设置，并直接使用基于命令行的客户端直接调整所有输入参数。

例如，在使用诸如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”切换为“ real”。谨慎使用，后果自负。

TRADING_MODE=real ./main.py BTC_EUR

集成化项目

我们可以使用Docker对该程序进行容器化。这是一个简单的自我解释Docker构建文件的例子。

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，并运行其他进程。

导入

导入价格并将其部署到内部API

1天的行情