在股票模拟（Stocksim）管道中使用生成器

简介

在本次实验中，你将学习如何利用 Python 生成器（generator）构建高效的数据处理管道。生成器是 Python 中一项强大的特性，它支持按需生成数据，无需将所有数据同时存储在内存中。你将了解如何连接生成器，以创建类似于 Unix 管道的数据处理工作流。

本次实验的目标是理解基于生成器的处理管道的基本原理，使用 Python 生成器创建数据处理工作流，以及过滤和格式化实时数据流。在本次实验中，你将创建 ticker.py 文件。请注意，在本次练习中，stocksim.py 程序应在后台运行，并且你将使用之前练习中的 follow() 函数。

Skills Graph

使用 CSV 数据的基本生成器管道

在这一步中，你将学习如何使用生成器创建一个基本的处理管道。不过，首先让我们了解一下什么是生成器。生成器是 Python 中一种特殊的迭代器（iterator）。与可能会一次性将所有数据加载到内存中的常规迭代器不同，生成器按需生成值。在处理大数据流时，这一特性非常有用，因为它可以节省内存。生成器无需将整个数据集存储在内存中，而是根据你的需求逐个生成值。

理解生成器

生成器本质上是一个返回迭代器的函数。当你对这个迭代器进行迭代时，它会生成一系列值。编写生成器函数的方式与常规函数类似，但有一个关键区别。生成器函数不使用 return 语句，而是使用 yield 语句。yield 语句具有独特的行为。它会暂停函数并保存其当前状态。当请求下一个值时，函数会从暂停的位置继续执行。这使得生成器能够逐步生成值，而无需每次都从头开始。

使用 follow() 函数

你之前创建的 follow() 函数的工作方式类似于 Unix 的 tail -f 命令。tail -f 命令会持续监控文件的新内容，follow() 函数也是如此。现在，让我们使用它来创建一个简单的处理管道。

步骤 1：打开一个新的终端窗口

首先，在 WebIDE 中打开一个新的终端窗口。你可以通过选择 Terminal → New Terminal 来实现。这个新终端将用于运行 Python 命令。

步骤 2：启动 Python 交互式 shell

新终端打开后，启动 Python 交互式 shell。你可以在终端中输入以下命令来启动：

python3

Python 交互式 shell 允许你逐行运行 Python 代码并立即查看结果。

步骤 3：导入 follow 函数并设置管道

现在，我们将导入 follow 函数并设置一个基本的管道来读取股票数据。在 Python 交互式 shell 中，输入以下代码：

>>> from follow import follow
>>> import csv
>>> lines = follow('stocklog.csv')
>>> rows = csv.reader(lines)
>>> for row in rows:
...     print(row)
...

以下是每行代码的作用：

• from follow import follow：从 follow 模块中导入 follow 函数。
• import csv：导入 csv 模块，该模块用于在 Python 中读取和写入 CSV 文件。
• lines = follow('stocklog.csv')：调用 follow 函数并传入文件名 stocklog.csv。follow 函数返回一个生成器，该生成器会在文件有新行添加时生成这些新行。
• rows = csv.reader(lines)：csv.reader() 函数接收 follow 函数生成的行，并将它们解析为 CSV 数据行。
• for 循环遍历这些行并打印每一行。

步骤 4：检查输出

运行代码后，你应该会看到类似以下的输出（你的数据会有所不同）：

['BA', '98.35', '6/11/2007', '09:41.07', '0.16', '98.25', '98.35', '98.31', '158148']
['AA', '39.63', '6/11/2007', '09:41.07', '-0.03', '39.67', '39.63', '39.31', '270224']
['XOM', '82.45', '6/11/2007', '09:41.07', '-0.23', '82.68', '82.64', '82.41', '748062']
['PG', '62.95', '6/11/2007', '09:41.08', '-0.12', '62.80', '62.97', '62.61', '454327']
...

这个输出表明你已经成功创建了一个数据管道。follow() 函数从文件中生成行，然后这些行被传递给 csv.reader() 函数，该函数将它们解析为数据行。

如果你已经看到足够的输出，可以通过按 Ctrl+C 停止执行。

原理剖析

让我们详细分析一下这个管道中发生了什么：

1. follow('stocklog.csv') 创建了一个生成器。这个生成器会跟踪 stocklog.csv 文件，并在文件有新行添加时生成这些新行。
2. csv.reader(lines) 接收 follow 函数生成的行，并将它们解析为 CSV 行数据。它了解 CSV 文件的结构，并将行拆分为单个值。
3. for 循环遍历这些行并打印每一行。这使你能够以可读的格式查看数据。

这是一个使用生成器进行数据处理管道的简单示例。在接下来的步骤中，我们将构建更复杂、更实用的管道。

创建 Ticker 类

在数据处理中，处理原始数据可能颇具挑战。为了让我们对股票数据的处理更加有序和高效，我们将定义一个合适的类来表示股票报价。这个类将作为我们股票数据的蓝图，使我们的数据处理管道更加健壮且易于管理。

创建 ticker.py 文件

1. 首先，你需要在 WebIDE 中创建一个新文件。你可以点击“New File”图标，或者在文件资源管理器中右键单击并选择“New File”。将这个文件命名为 ticker.py。这个文件将包含我们 Ticker 类的代码。

2. 现在，将以下代码添加到你新创建的 ticker.py 文件中。这段代码将定义我们的 Ticker 类，并设置一个简单的处理管道来测试它。

## ticker.py

from structure import Structure, String, Float, Integer

class Ticker(Structure):
    name = String()
    price = Float()
    date = String()
    time = String()
    change = Float()
    open = Float()
    high = Float()
    low = Float()
    volume = Integer()

if __name__ == '__main__':
    from follow import follow
    import csv
    lines = follow('stocklog.csv')
    rows = csv.reader(lines)
    records = (Ticker.from_row(row) for row in rows)
    for record in records:
        print(record)

3. 添加代码后，保存文件。你可以按 Ctrl+S，或者从菜单中选择“File” → “Save”。保存文件可确保你的更改得以保留，并且之后可以运行。

代码解析

让我们逐步详细了解这段代码的功能：

1. 在代码开头，我们从 structure.py 模块导入 Structure 和字段类型。这个模块已经为你设置好了。这些导入非常重要，因为它们为我们的 Ticker 类提供了构建块。Structure 类将作为我们 Ticker 类的基类，而 String、Float 和 Integer 等字段类型将定义我们股票数据字段的数据类型。

2. 接下来，我们定义一个继承自 Structure 的 Ticker 类。这个类有几个字段，代表了股票数据的不同方面：

   • name：这个字段存储股票代码，例如 “IBM” 或 “AAPL”。它帮助我们识别所处理的是哪家公司的股票。
   • price：它保存股票的当前价格。这对投资者来说是至关重要的信息。
   • date 和 time：这些字段告诉我们股票报价的生成时间。了解时间和日期对于分析股票价格随时间的走势很重要。
   • change：这代表股票的价格变化。它显示了与之前某个时间点相比，股票价格是上涨还是下跌。
   • open、high、low：这些字段分别代表股票在某一时期的开盘价、最高价和最低价。它们让我们了解股票的价格范围。
   • volume：这个字段存储交易的股票数量。高交易量可能表明市场对某只特定股票有强烈的兴趣。

3. 在 if __name__ == '__main__': 代码块中，我们设置了一个处理管道。当我们直接运行 ticker.py 文件时，这段代码将被执行。

   • follow('stocklog.csv') 是一个函数，它从 stocklog.csv 文件中生成行。它允许我们逐行读取文件。
   • csv.reader(lines) 获取这些行并将它们解析为行数据。CSV（逗号分隔值）是一种常见的用于存储表格数据的文件格式，这个函数帮助我们从每行中提取数据。
   • (Ticker.from_row(row) for row in rows) 是一个生成器表达式。它将每行数据转换为一个 Ticker 对象。这样，我们就将原始的 CSV 数据转换为更易于处理的结构化对象。
   • for 循环遍历这些 Ticker 对象并打印每个对象。这让我们可以看到结构化数据的实际效果。

运行代码

让我们运行代码，看看它是如何工作的：

1. 首先，你需要确保在终端中位于项目目录。如果你还不在该目录，可以使用以下命令导航到那里：

   cd /home/labex/project

2. 一旦你位于正确的目录中，使用以下命令运行 ticker.py 脚本：

   python3 ticker.py

3. 运行脚本后，你应该会看到类似以下的输出（你的数据会有所不同）：

   Ticker(IBM, 103.53, 6/11/2007, 09:53.59, 0.46, 102.87, 103.53, 102.77, 541633)
   Ticker(MSFT, 30.21, 6/11/2007, 09:54.01, 0.16, 30.05, 30.21, 29.95, 7562516)
   Ticker(AA, 40.01, 6/11/2007, 09:54.01, 0.35, 39.67, 40.15, 39.31, 576619)
   Ticker(T, 40.1, 6/11/2007, 09:54.08, -0.16, 40.2, 40.19, 39.87, 1312959)

当你看到足够的输出后，可以按 Ctrl+C 停止脚本的执行。

注意原始的 CSV 数据是如何被转换为结构化的 Ticker 对象的。这种转换使得数据在我们的处理管道中更易于处理，因为我们现在可以使用 Ticker 类中定义的字段来访问和操作股票数据。

构建更复杂的数据管道

现在，我们将通过添加过滤功能和改进数据展示方式，将数据管道提升到一个新的水平。这将使你更易于分析和理解所处理的信息。我们将对 ticker.py 脚本进行修改。过滤数据有助于你专注于感兴趣的特定信息，而以格式良好的表格形式展示数据则会使其更具可读性。

更新 ticker.py 文件

1. 首先，在 WebIDE 中打开你的 ticker.py 文件。WebIDE 是一个允许你直接在浏览器中编写和编辑代码的工具。它为修改 Python 脚本提供了一个便捷的环境。

2. 接下来，你需要将 ticker.py 文件中的 if __name__ == '__main__': 代码块替换为以下代码。这个代码块是脚本的入口点，替换它将改变脚本处理和显示数据的方式。

if __name__ == '__main__':
    from follow import follow
    import csv
    from tableformat import create_formatter, print_table

    formatter = create_formatter('text')

    lines = follow('stocklog.csv')
    rows = csv.reader(lines)
    records = (Ticker.from_row(row) for row in rows)
    negative = (rec for rec in records if rec.change < 0)
    print_table(negative, ['name', 'price', 'change'], formatter)

3. 完成这些更改后，保存文件。你可以按键盘上的 Ctrl+S，或者从菜单中选择 “File” → “Save”。保存文件可确保你的更改得以保留，并且之后可以运行。

理解增强型管道

让我们仔细看看这个增强型管道的工作原理。理解每个步骤将帮助你了解代码的不同部分如何协同工作来处理和显示数据。

1. 首先，我们从 tableformat 模块导入 create_formatter 和 print_table。这个模块已经为你设置好了，它提供了一些函数，可帮助你将数据格式化为美观的表格并进行打印。

2. 然后，我们使用 create_formatter('text') 创建一个文本格式化器。这个格式化器将以易于阅读的方式格式化数据。

3. 现在，让我们逐步分解这个管道：

   • follow('stocklog.csv') 是一个函数，它从 stocklog.csv 文件中生成行。它会持续监控文件中的新数据，并逐行提供这些数据。
   • csv.reader(lines) 获取 follow 函数生成的行，并将它们解析为行数据。这是必要的，因为 CSV 文件中的数据是文本格式，我们需要将其转换为可处理的结构化格式。
   • (Ticker.from_row(row) for row in rows) 是一个生成器表达式，它将每行数据转换为一个 Ticker 对象。Ticker 对象代表一只股票，包含股票的名称、价格和价格变化等信息。
   • (rec for rec in records if rec.change < 0) 是另一个生成器表达式，用于过滤 Ticker 对象。它只保留股票价格变化为负的对象。这使你能够专注于价格下跌的股票。
   • print_table(negative, ['name', 'price', 'change'], formatter) 获取过滤后的 Ticker 对象，并使用我们之前创建的格式化器将它们格式化为表格。然后将表格打印到控制台。

这个管道展示了生成器的强大之处。我们不是一次性将文件中的所有数据加载到内存中，而是将多个操作（读取、解析、转换、过滤）链接在一起，逐个处理数据项。这节省了内存，使代码更高效。

运行增强型管道

让我们运行更新后的代码，看看结果如何。

1. 首先，确保你在终端中位于项目目录。如果你还不在该目录，可以使用以下命令导航到那里：

   cd /home/labex/project

2. 一旦你位于项目目录中，使用以下命令运行 ticker.py 脚本：

   python3 ticker.py

3. 运行脚本后，你应该会在终端中看到一个格式良好的表格。这个表格只显示价格变化为负的股票。

          name      price     change
    ---------- ---------- ----------
             C      53.12      -0.21
           UTX      70.04      -0.19
           AXP      62.86      -0.18
           MMM      85.72      -0.22
           MCD      51.38      -0.03
           WMT      49.85      -0.23
            KO       51.6      -0.07
           AIG      71.39      -0.14
            PG      63.05      -0.02
            HD      37.76      -0.19

如果你已经看到足够的输出，并且想停止脚本的执行，可以按键盘上的 Ctrl+C。

生成器管道的强大之处

我们在这里创建的是一个强大的数据处理管道。下面总结一下它的功能：

1. 它持续监控 stocklog.csv 文件中的新数据。这意味着当新数据添加到文件中时，管道会自动处理它。
2. 它将文件中的 CSV 数据解析为结构化的 Ticker 对象。这使得处理数据和对其执行操作更加容易。
3. 它根据特定条件过滤数据，在这种情况下，是价格变化为负的股票。这使你能够专注于正在贬值的股票。
4. 它将过滤后的数据格式化为可读的表格并进行展示。这使得分析数据和得出结论变得容易。

在这个管道中使用生成器的一个关键优势是它使用的内存极少。生成器按需生成值，这意味着它们不会一次性将所有数据存储在内存中。这类似于 Unix 管道，每个组件处理数据并将其传递给下一个组件。

你可以将生成器看作是乐高积木。就像你可以将乐高积木堆叠在一起创建不同的结构一样，你可以组合生成器来创建强大的数据处理工作流。这种模块化方法允许你从简单、可重用的组件构建复杂的系统。

总结

在这个实验中，你学习了如何使用 Python 生成器来构建高效的数据处理管道。你完成了几个重要的任务，例如使用 follow() 函数监控文件中的新数据、创建 Ticker 类来表示股票报价，以及构建一个多阶段的处理管道，该管道可以读取、解析和过滤 CSV 数据，然后对结果进行格式化和展示。

基于生成器的方法具有多个优点，包括内存效率高（因为数据是按需处理的）、模块化（允许轻松组合和重用管道组件），以及能够简单地表达复杂的数据流。这些概念在现实世界的数据处理中经常被应用，特别是在处理大型数据集或流式数据时。