如何使用 Python 3 和 Binance API 构建加密机器人（第 1 部分）

模型

策略部分

交易加密货币或任何资产的首要原则是设定目标并制定策略。这里我并非在写交易策略，而是想构建一个简单实用的加密货币交易机器人来应用你的策略。“谨慎交易”系列文章更像是一个自动化加密货币交易机器人框架。

我们将使用 python 3.9（3.9.2）首先创建项目文件结构。

/交易所
/策略
/模型

这里的“exchanges”文件夹存储了交易所 API 包装器、策略、您的策略以及我们将要使用的业务对象的模型。

模型

我们将为这个项目定义几个业务对象，例如价格、货币或订单。首先，让我们从价格对象开始，并将其用于我们即将创建的策略中。

在我们为我们的业务对象编写一个公共抽象层之前。

./模型/模型.py

from datetime import datetime


class AbstractModel:
    created: datetime

    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)

然后是我们的第一堂 Price 课。

./型号/价格

from models.model import AbstractModel


class Price(AbstractModel):
    pair: str = ''
    exchange: str = ''
    current: float = 0
    lowest: float = 0
    highest: float = 0
    currency: str = ''
    asset: str = ''

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.pair = self.get_pair()

    def get_pair(self):
        return self.currency + '_' + self.asset

策略部分

如果这个系统的最大部分是策略，它将负责运行您自己的交易逻辑，并且要做到这一点，有两种方式，一种是经典的使用间隔，然后运行 API 调用到外部交易所 API 或甚至内部 webhook 和路由，以及使用 WebSocket 的基于实时事件的调用。

为此，我们首先创建一个抽象策略类，我们的策略将对其进行扩展。

./策略/strategy.py

import json
import threading
import time
from datetime import datetime
from decouple import config
from models.price import Price


class Strategy(object):
    price: Price

    def __init__(self, exchange, interval=60, *args, **kwargs):
        self._timer = None
        self.interval = interval
        self.args = args
        self.kwargs = kwargs
        self.is_running = False
        self.next_call = time.time()
        self.portfolio = {}
        self.exchange = exchange
        # Load account portfolio for pair at load
        self.get_portfolio()

    def _run(self):
        self.is_running = False
        self.start()
        self.run(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            print(datetime.now())
            if self._timer is None:
                self.next_call = time.time()
            else:
                self.next_call += self.interval

            self._timer = threading.Timer(self.next_call - time.time(), self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

    def get_portfolio(self):
        self.portfolio = {'currency': self.exchange.get_asset_balance(self.exchange.currency),
                          'asset': self.exchange.get_asset_balance(self.exchange.asset)}

    def get_price(self):
        try:
            self.price = self.exchange.symbol_ticker()
        except Exception as e:
            pass

...

这里我们的策略抽象层构造函数签名需要一个 Exchange 实例，我们稍后将使用 Binance API 编写第一个包装器来完成这部分。

我们定义了一个简单但功能齐全的无限间隔运行器，无需任何额外的库。需要注意的是，每次运行都会在另一个线程上启动下一个调用，但实际上，如果您的策略需要大量外部调用或繁重的计算，则您的策略将使用最多两个线程：一个主线程和一个当前运行迭代线程。每个线程消耗 0.3% 的 RAM 和 0 或 0.1% 的 CPU 使用率，这还涉及策略获取股票代码和订单，然后将价格和订单相关数据存储在另一个内部 API 上。

然而间隔运行精度可能会在微秒级上稍微漂移，但会稳定在秒级。

这里是该层的简单用法，该策略基本上会打印您的交易所账户投资组合和交易所价格。我们还提供了从稍后要连接的外部交易所检索代码的方法，以及检索您当前在连接交易所的投资组合的方法。

./strategies/watcher.py

from exchanges.exchange import Exchange
from strategies.strategy import Strategy


class Watcher(Strategy):
    def __init__(self, exchange: Exchange, timeout=60, *args, **kwargs):
        super().__init__(exchange, timeout, *args, **kwargs)

    def run(self):
        self.get_price()
        print('*******************************')
        print('Exchange: ', self.exchange.name)
        print('Pair: ', self.exchange.get_symbol())
        print('Available: ', self.portfolio['currency'] + ' ' + self.exchange.currency)
        print('Available: ', self.portfolio['asset'] + ' ' + self.exchange.asset)
        print('Price: ', self.price.current)

在接下来的部分中，我们将使用官方币安 API https://github.com/binance/binance-spot-api-docs和 Python 请求库https://pypi.org/project/requests/编写一个简单的包装器来连接我们的第一个交易所币安。然后编写我们缺少的业务对象 order 和 currency ，分别用于存储您发送给交易所的订单和我们将要操纵的法定货币或加密货币，然后将所有这些不同的部分组合在一起。

感谢阅读，敬请期待下一部分。

鏂囩珷鏉ユ簮锛�https://dev.to/nicolasbonnici/how-to-build-a-crypto-bot-with-python-3-and-the-binance-api-part-1-1864