用 Python 和币安 API 构建数字货币交易机器人（二）-技术圈

在本系列的第一部分中：用 Python 和币安 API 构建数字货币交易机器人（一），我们讨论了文件夹结构，添加了我们的第一个Price业务对象，并为策略抽象层进行编码。

下面将通过编码我们需要的货币和订单的业务对象开始。

此处，货币业务对象存储所有不同的加密货币或法定货币。

./models/order.py

from models.model import AbstractModel


class Currency(AbstractModel):
    name: str = ''
    symbol: str = ''
    fiat: bool

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

然后是订单业务对象处理我们在交易所创建的订单。

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

然后构建交换抽象层，并开发我们的第一个连接器。由于这里不是编写API包装器的重点，因此为了方便起见，我们将使用非官方的Binance API包装器库python-binance（https://python-binance.readthedocs.io/en/latest/binance.html）。

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

我们的第一个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()

现在，只需调用我们的策略启动方法，我们便拥有了一个极简而强大的交易机器人系统。使用自己的指标和策略，您可以通过传递订单来开始买卖。但是，在开始编写自己的策略代码之前，请确保对其进行测试并安全运行。所以你需要一个系统来做到这一点。

在下一部分中，我们将实现回测模式，以针对交易所的历史数据来测试您的策略，并提供导入这些数据的服务。

另外，我们将所有这些部分与全局配置和一些依赖项一起打包进命令行工具中。

最后我们将通过查看如何对该代码进行容器化和工业化实现来进一步介绍。

欢迎添加下方二维码进群交流

Python数字货币量化交易技术