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

**Published by:** [quantbang](https://paragraph.com/@quantbang/)
**Published on:** 2022-03-24
**URL:** https://paragraph.com/@quantbang/python-api-2

## Content

在本系列的第一部分中，我们讨论了文件夹结构，添加了我们的第一个Price业务对象，并为策略抽象层进行编码。 下面将通过编码我们需要的货币和订单的业务对象开始。 此处，货币业务对象存储所有不同的加密货币或法定货币。 ./models/order.pyfrom models.model import AbstractModel class Currency(AbstractModel): name: str = '' symbol: str = '' fiat: bool def __init__(self, **kwargs): super().__init__(**kwargs) 然后是订单业务对象处理我们在交易所创建的订单。 ./models/order.pyfrom 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.pyimport 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.pyfrom 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() 现在，只需调用我们的策略启动方法，我们便拥有了一个极简而强大的交易机器人系统。使用自己的指标和策略，您可以通过传递订单来开始买卖。但是，在开始编写自己的策略代码之前，请确保对其进行测试并安全运行。所以你需要一个系统来做到这一点。 在下一部分中，我们将实现回测模式，以针对交易所的历史数据来测试您的策略，并提供导入这些数据的服务。 另外，我们将所有这些部分与全局配置和一些依赖项一起打包进命令行工具中。 最后我们将通过查看如何对该代码进行容器化和工业化实现来进一步介绍。

## Publication Information

- [quantbang](https://paragraph.com/@quantbang/): Publication homepage
- [All Posts](https://paragraph.com/@quantbang/): More posts from this publication
- [RSS Feed](https://api.paragraph.com/blogs/rss/@quantbang): Subscribe to updates
- [Twitter](https://twitter.com/andywong1024): Follow on Twitter