存了50TB!巨能“装”的量化数据存储方案

发布时间:2024年01月03日

R50
在上一篇笔记中,我们指出,如果我们只在日线级别上存储行情数据和因子,HDF5 无论如何都是够用了。即使是在存储了 40 年分钟线的单个股数据集上,查询时间也只花了 0.2 秒 – 这个速度已经足够快了,如果我们不需要在分钟级别上进行横截面数据查询的话。

但是,如果个人交易者确实有条件(网速和硬件)做高频交易,处理 tick 级和 level 2 的数据将是必要的。如此一来,我们处理数据的规模就达到了TB级别,但我们还想保持查询时间在毫秒级。或者,您确实需要在分钟级别上进行横截面数据查询,这些场景下,HDF5 就难堪大任了。


现在我们得拜托 pyarrow + parquet 了。它们本来就是大数据解决方案之一。本文的标题是存了 50TB。博主并没有这个存储设施进行实验。但这一数字并非杜撰:

!!! tip level 2 数据有多大?
据 dolphin(一种高效的时序数据库)的报告,A 股的 level 2 数据达到了一天 10GB 的数据量。

pyarrow

要介绍 pyarrow,我们先得从 Apache Arrow 说起。Apache Arrow 定义了一种与开发语言无关的列存储(内存中)格式,以便在 CPU/GPU 上高效地执行数据读取和分析任务。

pyarrow 是 Apache Arrow 库的一个 python 实现。Apache Arrow 还有 R, Java 等多种其它实现。但是, Pyarrow 是 Apache Arrow 实现中的头等公民,许多先进的功能都率先在这里实现了,然后才是 R 等其它语言。


Pyarrow 支持从 3.8 到 3.11 的 Python 版本。我们使用下面的命令来安装它:

pip install pyarrow

Parquet

parquet 是一种数据存储格式。在大数据存储语境下,我们还常常看到 feather 这种格式。两者的区别是,parquet 提供了 RLE 压缩,字典编码(与 pandas 中的 category 替换类似)和数据页压缩。因此,在读写速度上要慢于 feather,但更省磁盘空间。

基本概念

pyarrow 中最基本的概念主要有:

  1. array – 它是一列同构的数据,但通常允许出现 None。
  2. 一系列等长的 arry 实例构成 Record Batch。batch 可以像 array 一样进行切片
  3. 在 batch 之上,还有 Table 的概念。table 由一些列构成,每一列都是一个 ChunkedArray。

接下来我们还要接触到 schema 的概念,这将在后面结合示例进行说明。


pyarrow 的主要功能:

  1. 提供各种 I/O 接口 (memory and IO interfaces),比如与常见的其它格式,比如 CSV, dataframe, S3, minio,本地文件等之间的读写转换。
  2. 为数据提供表格(tabluar datasets)化的视图和相关操作。
  3. 提供基础函数 (compute Functions),如 group, 聚合查询,join 操作,查询操作(表达式,参照 pandas)。

上述功能分类中,括号中的英文也正对应着 Arrow 的文档,以便大家在需要时查询。

行情数据存储方案

尽管我们介绍 pyarrow + parquet 是为了存储 level 2 的数据,但我手头并没有相应的 level 2 数据源,所以,也无法做出可实际运行的示例。因此,我们将通过构造 1 分钟行情的数据集来演示如何通过 pyarrow 来写入、增加和查询数据。

我们把每一天所有证券品种的 1 分钟行情放在同一个文件中,所有的分钟线都放在 1m 目录下。这里我们没有使用 partition,但在实际应用中,我们要考虑按年进行分钟线的 patition。对 level 2 的数据,可能要按月或者周进行 partition。

另一方面,每个 parquet 文件的大小最好在 20M 和 2GB 之间,所以,对于分钟线以上级别的数据,可以考虑按周或者更大的尺度写入一个文件。


这是磁盘文件结构示例:

/tmp/pyarrow
├── 1d
├── 1m
│   ├── 2023-12-27.parquet
│   ├── 2023-12-28.parquet
│   └── 2023-12-29.parquet
└── factors

首先,我们定义存储字段:

import pyarrow as pa

schema = pa.schema([
        ("symbol", pa.string()),
        ("frame", pa.date64()),
        ("open", pa.float32()),
        ("high", pa.float32()),
        ("low", pa.float32()),
        ("close", pa.float32()),
        ("volume", pa.float64()),
        ("money", pa.float64()),
        ("factor", pa.float64())
])

pyarrow 支持字符串类型和日期类型!

接下来,我们就实现数据追加和读写部分。由于我们是将每天的分钟线存为一个 parquet 文件,所以代码非常简单:


import arrow
import pyarrow.parquet as pq

async def save_1m_bars(codes, dt: datetime.datetime):
    tables = None

    for code in codes:
        bars = await Stock.get_bars(code, 240, FrameType.MIN1, end=dt)
        data = [[code] * len(bars)]

        data.extend([
                    bars[key] for key in bars.dtype.names
                ])
        table = pa.Table.from_arrays(data, schema=schema)
        if tables is None:
            tables = table
        else: # 拼接表
            tables = pa.concat_tables([tables, table])

    # 写入磁盘
    name = arrow.get(dt).format("YYYY-MM-DD")
    pq.write_table(tables, f"/tmp/pyarrow/1m/{name}.parquet")

在这里插入图片描述

omicron 返回的 bars 是一个 numpy structured array,在转换成 pyarrow Table 时,我们需要先将它拆成 List[array] 的格式,然后通过 from_arrays 来生成一个子 table,它包含了某支个股的 1 分钟行情数据。


由于我们要将所有的个股、指数数据存放在同一张大表里,所以,我们还调用了 concat_tables 来实现拼接。在其它常见的类似数据结构中,无论是 numpy, 还是 pandas 或者 hdf5,这种拼接都有比较昂贵的代价,但在 pyarrow 中,这种拼接几乎是零成本的!没有数据被拷贝。

接下来,我们就演示一下如何调用 save_1m_bars 来保存数据:

codes = ["000001.XSHE", "600000.XSHG"]
for i in (25, 26, 27, 28, 29):
    dt = datetime.datetime(2023, 12, i, 15)
    await save_1m_bars(codes, dt)

查询数据

现在,我们把刚刚写进去的数据,再读回来:

import pyarrow.dataset as ds

dataset = ds.dataset("/tmp/pyarrow/1m")
dataset.files

dataset 在这里,只是一种元数据。到此为止,它并不会真的从磁盘上加载任何数据。加载数据是从 to_table, 或者某个查询开始的。


上述代码将显示该数据集所包含的磁盘文件。

我们可以将 dataset 转换成一张大表:

table = dataset.to_table()
table

这将输出 table 的数据字段定义,以及部分数据。

to_table 将会把所有的数据都加载进内存。如果数据量很大,这会导致内存不够用。因此,很多情况下,我们可能使用 RecordBatch:

for rb in dataset.to_batches():
    print(rb.to_pandas())

每一个 RecordBatch, 都有一个 to_pandas 方法,这样就进入了你所熟悉的领域。

但更多的时候,我们会通过查询,只加载我们需要的那部分数据。此时,我们需要使用 pyarrow.compute 中的表达式语法,主要是field函数,逻辑运算符和bitwise运算符(此处未演示,即 & | 等):


import pyarrow.compute as pc

filter = (pc.field("frame") > pc.scalar(datetime.datetime(2023, 12, 28, 15)))
dataset.filter(filter).to_table().to_pandas()

溺水三千,只取一瓢。这次我们只把其中的 480 条记录加载的了内存中。


在这里插入图片描述

文章来源:https://blog.csdn.net/hbaaron/article/details/135352101
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。