使用 Python 构建高效的加密交易软件

您好，欢迎阅读本文，我将讨论使用 Python 编程语言构建生产级交易软件的现代方法。

Python因其简单、干净、可读的代码而被广泛使用。超过一百万的软件开发人员可以理解和支持用 python 编写的代码。

但是python以一些缺点而闻名，例如速度。大多数时候你可以忽略这个缺点，因为大多数用于交易目的的库都使用用 C 编写的代码片段，并且对于大多数常见任务来说足够快。

我们将讨论三个主要概念：异步编程、事件驱动架构和微服务架构。

至于我们的软件将使用不同的数据源、API 服务等，我们将不得不使其异步，因此我们不会等待来自任何远程服务的数据延迟。异步执行是必须的。

事件驱动的方法可以帮助我们更好地理解和控制系统中正在发生的事情，并且运行速度很快。

微服务架构让我们有机会专注于单个问题，而使用单体架构可能会更难，从而使开发和支持具有数千行代码的项目变得更加容易。

异步编程

如果您至少对异步编程有基本的了解来构建快速反应的交易软件，那将是最好的。

大多数时候使用的同步方法的问题是我们的程序必须等待来自专用源的数据输入，然后才能继续执行进一步的操作。使用同步软件连接到 WebSockets 并从交易所接收分时数据更新或开放订单更新几乎是不可能的。

异步编程允许我们跳过等待数据输入并继续进行控制流或数据处理流。这种方法从根本上提高了我们的软件速度和性能。

我们将使用 asyncio python 库来构建异步应用程序。我们需要学习一些概念来开始构建异步软件。

首先，我们的函数用 async word 标记为异步，就像在 JavaScript 中一样。例子：

async def my_function()

其次，我们使用 await 字来告诉程序它可以将控制从一个流切换到另一个流。例子：

await asyncio.sleep(0)

这行代码只是告诉我们的程序不要等待数据输入或一些冗长的计算，因此我们的程序可以切换到另一个任务。

第三，在构建异步软件时，我们不仅操作函数，还操作任务。一个任务内部可以包含一个或多个函数。

例子：

async def input_func():
    pass


async def input_task():
    await asyncio.sleep(0)
    await self.input_func()

最后，循环。我们的任务在循环中运行，可以称为“控制流”。

例子：

ioloop = asyncio.get_event_loop()
tasks = [ioloop.create_task(input_task()), ioloop.create_task(another_task())]
wait_tasks = asyncio.wait(tasks)
ioloop.run_until_complete(wait_tasks)

这是一个简单的异步应用程序示例。你可以在这个 GitHub 存储库中找到它。

事件驱动的方法

事件驱动的方法是一种思考软件架构的方法。在考虑程序和功能范式的同时开发和维护一个相当大的交易应用程序可能效率低下，甚至是不可能的。

我们可以将交易软件视为对现实世界中的一些随机事件做出反应的系统。市场上发生了一些事情；价格变化，这是一个事件。我们可以在我们的系统内部对其进行处理并产生另一个事件，订单，例如，它的条件得到满足。我们还可以根据需要创建日志记录或通知事件。

事件驱动的方法可以帮助我们更好地了解系统内部正在发生的事情，保持控制，并支持更清晰、更高效的软件架构。

事件是我们架构的基本元素。您可以将事件视为数据类或数据结构。它们可以包含很多信息，例如事件类型、时间戳、特定数据等。

例子：

class Event(object):
    #base class
    pass

class AccountUpdateEvent(Event):
    def __init__(self, timestamp, username, total):
        self.type = 'ACCOUNT_UPDATE'
        self.timestamp = timestamp
        self.username = username
        self.total = float(total)

现在我们将构建一个简单的应用程序，它可以在不到 100 行 Python 代码中生成和处理事件！

我们的第一个模块将进行 API 调用并产生事件。

示例代码（整个类在我的 GitHub 存储库中可用）：

class GetAccountData:
    def create_event(self, timestamp, username, total):
        self.events.put(AccountUpdateEvent(timestamp=timestamp, username=username, total=total))

第二个模块处理事件。

示例代码（整个类在我的 GitHub 存储库中可用）：

class ProcessAccountData:
    def update_account_data(self, event):
        print('Account update received at: ', event.timestamp, 'Username: ', event.username, 'Total balance: ', event.total)

最后，我们的应用程序。它将包含 GetAccountData 和 ProcessAccountData 以及一个事件队列。

示例代码（整个类在我的 GitHub 存储库中可用）：

class App:
    def __init__(self):
        self.events = Queue()
        self.get_data_account = GetAccountData(self.events)
        self.process_data_account = ProcessAccountData()

这个应用程序包含一个处理事件的循环。一旦有新事件进入我们的队列，我们就会将其取出并进行处理。

示例代码（整个类在我的 GitHub 存储库中可用）：

def event_loop(self):
       while True:
            try:
                event = self.events.get(False)
            except Exception as e:
                time.sleep(1)
                break

            else:
                try:
                    if event is not None:
                        if event.type == 'ACCOUNT_UPDATE':
                            self.process_data_account.update_account_data(event)

                except Exception as e:
                    print('event_loop error:', e)

这是一个简单的事件驱动应用程序的示例。您可以在我的 GitHub 存储库中找到源代码。

微服务

微服务软件架构与单体架构相反。对于小型项目，整体架构是考虑其设计的最直接方式。但是任何成熟的项目如果是单体，都无法轻易开发和支持。

微服务模式迫使开发人员不仅将项目划分为单独的模块和类，而且划分为称为微服务的单个微型应用程序。它们可以部署在不同的数据中心，得到其他团队的支持等。

现在我们将为我们的交易软件构建一个简单的微服务。它将通过 API 接收我们的订单数据并将其保存到本地数据库。当然，可以将该数据保存到本地磁盘。尽管如此，假设我们想要构建高频交易软件。在这种情况下，我们不想在日志记录或统计数据上浪费任何本地资源，或者我们希望将功能的开发外包给不同的开发团队。

首先，我们将构建一个基本的数据库连接器。我们将 SQLite 数据库用于教育目的。我们将创建一个包含六个字段的订单表：时间戳、用户名、市场、边、大小和价格。

示例代码：

class Database:
    #this is a DB connector
    #we will use SQLite in this example for simplicity
    def init(self):
        #filename and path to the database are hardcoded for simplicity
        self.connect_to = 'test.db'
    def create_table_orders(self):
        #a func to create our database
        conn = sqlite3.connect(self.connect_to)
        conn.execute('''CREATE TABLE if not exists Orders
        (timestamp TEXT NOT NULL,
        username TEXT NOT NULL,
        market TEXT NOT NULL,
        side TEXT NOT NULL,
        size FLOAT NOT NULL,
        price FLOAT NOT NULL
        );''')
        conn.close()

数据库连接器将仅实现一种方法来将订单数据保存到数据库。

示例代码：

def add_data_orders(self, timestamp, username, market, side, size, price):
    #a func to save orders data
    conn = sqlite3.connect(self.connect_to)
    conn.execute("INSERT INTO Orders (timestamp, username, market, side, size, price) VALUES (?, ?, ?, ?, ?, ?)", (timestamp,
    username, market, side, size, price));
    conn.commit()
    conn.close()

其次，我们需要一个 API 服务器。只需不到 30 行代码就可以使用 Flask 模块创建一个简单的 API 服务器。它将能够接收带有订单数据的 HTTP POST 请求并将其保存到数据库中。

示例代码：

@app.post("/API/orders")
def save_orders():
    if request.is_json:
        response = request.get_json()
        DB.add_data_orders(response['timestamp'], response['username'], response['market'], response['side'],  
        response['size'], response['price'])
        return response, 201
        return {"error": "Request must be JSON"}, 415

您可以在 GitHub 上找到完整的数据库连接器代码和 API SERVER 代码。

最后，我们的服务需要一个 API 连接器。我们的 API 连接器将使用 requests 库向我们的 API 服务器发出 POST HTTP 请求。

示例代码：

def generate_request(self, order):
    try:
        response = requests.post(self.api_url, json=order)
        print(response)
    except Exception as e:
        print('generate_request - Exception', e)

您可以在 GitHub 上找到完整的 API CLIENT 代码。

在大约 100 行代码中，我们创建了一个数据库连接器、API 服务器和 API 客户端，以将订单数据保存到远程服务器上的数据库中。

在本文中，我们讨论了异步编程、事件驱动架构和微服务的基础知识——作为软件开发人员需要了解的三个概念。我希望你能在你的下一个项目中使用这些知识！

-END-

链上数据：https://paragraph.xyz/@0x817F440789a7D15c8F44A29dc069a06F16F0200A/python-crypro-soft

文章哈希：R6ryg2q_dr5alpQzPHyaqp6mCXynP8aOUZSHRRFNoTw

异步编程

事件驱动的方法

微服务

留下评论取消回复

相关文章

Python 是 21 世纪儿童必备的技能