alpacamarkets/marketstore

MarketStore

概述

MarketStore是一个针对***时间序列数据优化的数据库服务器。您可以将其视为一个可扩展的DataFrame服务，可从系统中的任何位置访问，并且具有更高的可扩展性。

它从头开始设计，旨在解决处理大量市场数据时的可扩展性问题，这些数据用于算法交易回测、图表绘制和价格历史分析，数据跨度可达多年，粒度可降至所有美国或加密货币领域的tick级别。如果您正在为管理大量HDF5文件而烦恼，这是解决您问题的理想方案。

基本安装包含了必要的组件 - 您可以通过简单的插件配置开始从GDAX拉取加密货币价格数据并将其写入数据库。

MarketStore使您能够通过网络查询DataFrame内容，延迟与从磁盘读取本地HDF5文件相当，而向末尾追加新数据的速度比DataFrame快两个数量级。这是因为存储格式针对数据类型和使用场景以及现代文件系统/硬件特性进行了优化。

MarketStore已准备好投入生产！在Alpaca，它已在生产环境中使用多年，用于重要业务。

核心功能和特性

• 专为***时间序列数据优化的存储格式
• 高性能的数据读写操作，特别是追加新数据
• 支持从1秒到1天的多种时间框架
• 支持变长记录，适用于非固定间隔的数据
• 插件架构，支持扩展功能
• 支持WebSocket实时数据推送
• 数据复制功能，实现主从架构
• 支持多种客户端，包括Python和命令行工具

安装方法

Docker

如果您想立即开始使用，可以使用我们最新的Docker镜像引导一个marketstore数据库实例。该镜像预装了默认的mkts.yml文件，并将/data声明为其根目录的卷。要使用默认设置运行容器：

sh
docker run -i -p 5993:5993 alpacamarkets/marketstore:latest

如果您想运行自定义的mkts.yml，可以创建一个新容器并将您的mkts.yml文件加载到其中：

sh
docker create --name mktsdb -p 5993:5993 alpacamarkets/marketstore:latest
docker cp mkts.yml mktsdb:/etc/mkts.yml
docker start -i mktsdb

您还可以将容器绑定挂载到本地主机配置文件：一个自定义的mkts.yml：

sh
docker run -v /full/path/to/mkts.yml:/etc/mkts.yml -i -p 5993:5993 alpacamarkets/marketstore:latest

默认情况下，容器不会将任何写入的数据持久化到容器的主机存储中。要实现这一点，将data目录绑定到本地位置：

sh
docker run -v "/path/to/store/data:/data" -i -p 5993:5993 alpacamarkets/marketstore:latest

一旦数据写入服务器，您应该看到如下文件树布局，它将在容器运行之间保持：

sh
>>> tree /<path_to_data>/marketstore
/<path_to_data>/marketstore
├── category_name
├── WALFile.1590868038674814776.walfile
├── SYMBOL_1
├── SYMBOL_2
├── SYMBOL_3

如果您已在本地构建了cmd包，可以使用以下命令打开与正在运行的docker实例的会话：

sh
marketstore connect --url localhost:5993

配置说明

要运行MarketStore，需要一个YAML配置文件。可以使用marketstore init创建一个默认文件(mkts.yml)。此文件的路径通过--config标志传递给start命令，或者默认情况下，它会在运行目录中查找名为mkts.yml的文件。

配置选项

默认mkts.yml

yml
root_directory: data
listen_port: 5993
grpc_listen_port: 5995
log_level: info
queryable: true
stop_grace_period: 0
wal_rotate_interval: 5
stale_threshold: 5

使用方法

您可以通过运行以下命令列出可用命令：

marketstore

或

$GOPATH/bin/marketstore

取决于您的GOPATH设置。

您可以通过运行以下命令创建一个名为mkts.yml的新配置文件，其中填充了默认值：

$GOPATH/bin/marketstore init

然后使用以下命令启动marketstore服务器：

$GOPATH/bin/marketstore start

输出将类似于：

example@alpaca:~/go/bin/src/github.com/alpacahq/marketstore$ marketstore
I0619 16:29:30.102101    7835 log.go:14] Disabling "enable_last_known" feature until it is fixed...
I0619 16:29:30.102980    7835 log.go:14] Initializing MarketStore...
I0619 16:29:30.103092    7835 log.go:14] WAL Setup: initCatalog true, initWALCache true, backgroundSync true, WALBypass false:
I0619 16:29:30.103179    7835 log.go:14] Root Directory: /example/go/bin/src/github.com/alpacahq/marketstore/project/data/mktsdb
I0619 16:29:30.144461    7835 log.go:14] My WALFILE: WALFile.1529450970104303654.walfile
I0619 16:29:30.144486    7835 log.go:14] Found a WALFILE: WALFile.1529450306968096708.walfile, entering replay...
I0619 16:29:30.244778    7835 log.go:14] Beginning WAL Replay
I0619 16:29:30.244861    7835 log.go:14] Partial Read
I0619 16:29:30.244882    7835 log.go:14] Entering replay of TGData
I0619 16:29:30.244903    7835 log.go:14] Replay of WAL file /example/go/bin/src/github.com/alpacahq/marketstore/project/data/mktsdb/WALFile.1529450306968096708.walfile finished
I0619 16:29:30.289401    7835 log.go:14] Finished replay of TGData
I0619 16:29:30.340760    7835 log.go:14] Launching rpc data server...
I0619 16:29:30.340792    7835 log.go:14] Initializing websocket...
I0619 16:29:30.340814    7835 plugins.go:14] InitializeTriggers
I0619 16:29:30.340824    7835 plugins.go:42] InitializeBgWorkers

客户端使用

启动MarketStore实例后，您就可以读取和写入tick数据了。

Python客户端

https://github.com/alpacahq/pymarketstore%E6%98%AF%E6%A0%87%E5%87%86%E7%9A%84Python%E5%AE%A2%E6%88%B7%E7%AB%AF%E3%80%82%E7%A1%AE%E4%BF%9D%E5%9C%A8%E5%8F%A6%E4%B8%80%E4%B8%AA%E7%BB%88%E7%AB%AF%E4%B8%AD%E8%BF%90%E8%A1%8Cmarketstore%EF%BC%9A

• 查询数据

python
import pymarketstore as pymkts
param = pymkts.Params('BTC', '1Min', 'OHLCV', limit=10)
cli = pymkts.Client()
reply = cli.query(param)
reply.first().df()

显示：

python
Out[5]:
                               Open      High       Low     Close     Volume
Epoch
2018-01-17 17:19:00+00:00  10400.00  10400.25  10315.00  10337.25   7.772154
2018-01-17 17:20:00+00:00  10328.22  10359.00  10328.22  10337.00  14.206040
2018-01-17 17:21:00+00:00  10337.01  10337.01  10180.01  10192.15   7.906481
2018-01-17 17:22:00+00:00  10199.99  10200.00  10129.88  10160.08  28.119562
2018-01-17 17:23:00+00:00  10140.01  10161.00  10115.00  10115.01  11.283704
2018-01-17 17:24:00+00:00  10115.00  10194.99  10102.35  10194.99  10.617131
2018-01-17 17:25:00+00:00  10194.99  10240.00  10194.98  10220.00   8.586766
2018-01-17 17:26:00+00:00  10210.02  10210.02  10101.00  10138.00   6.616969
2018-01-17 17:27:00+00:00  10137.99  10138.00  10108.76  10124.94   9.962978
2018-01-17 17:28:00+00:00  10124.95  10142.39  10124.94  10142.39   2.262249

• 写入数据

python
import numpy as np
import pandas as pd
data = np.array([(pd.Timestamp('2017-01-01 00:00').value / 10**9, 10.0)], dtype=[('Epoch', 'i8'), ('Ask', 'f4')])
cli.write(data, 'TEST/1Min/Tick')
# Out[10]: {'responses': None}

cli.query(pymkts.Params('TEST', '1Min', 'Tick')).first().df()

显示：

python
                            Ask
Epoch
2017-01-01 00:00:00+00:00  10.0

• 变长记录

Marketstore是一个通过将时间范围内的记录数限制为一个来实现高性能的数据库。支持的时间范围从1D（1天）到1Sec（1秒），基本上，时间范围越长，读写数据的速度就越快。

但是，它也支持不按特定间隔到达或比每秒更频繁的数据，例如盘口数据和TICK数据。这种数据在marketstore中称为变长记录。

您可以通过在使用pymarketstore写入时指定isvariablelength=True来使用变长记录功能。

python
import numpy as np, pandas as pd, pymarketstore as pymkts

symbol, timeframe, attribute_group = "TEST", "1Sec", "Tick"
data_type = [('Epoch', 'i8'), ('Bid', 'f4'), ('Ask', 'f4'), ('Nanoseconds', 'i4')]
tbk = "{}/{}/{}".format(symbol, timeframe, attribute_group)
client = pymkts.Client()

# --- 写入变长记录（=单个时间范围（1秒）内的多个记录）
data = np.array([
    (pd.Timestamp('2021-01-01 00:00:00').value / 10 ** 9, 10.0, 20.0, 1000000),
    (pd.Timestamp('2021-01-01 00:00:00').value / 10 ** 9, 30.0, 40.0, 2000000),
    (pd.Timestamp('2021-01-01 00:00:00').value / 10 ** 9, 50.0, 60.0, 3000000),
], dtype=data_type)
client.write(data, tbk, isvariablelength=True)

# --- 查询变长记录
params = pymkts.Params(symbol, timeframe, attribute_group)
print(client.query(params=params).first().df())

# --- 清理
client.destroy(tbk)

显示：

                            Bid  Ask  Nanoseconds
Epoch                                             
2021-01-01 00:00:00+00:00  10.0  20.0      1000000
2021-01-01 00:00:00+00:00  30.0  40.0      2000000
2021-01-01 00:00:00+00:00  50.0  60.0      3000000

因为Epoch列的数据类型始终设置为'i8'（=int64），不足以描述亚秒级精度的日期，所以亚秒级信息存储在marketstore中的另一个列（= Nanoseconds列，数据类型为'i4'）中。

命令行客户端

使用以下命令连接到marketstore实例：

// 对于本地数据库
marketstore connect --dir <path>
// 对于服务器
marketstore connect --url <address>

并通过sql会话运行命令。

插件

Go插件架构在Linux上使用Go1.10+效果最佳。有关插件的更多信息，请参见plugins包。这里介绍一些特色插件：

流处理

您可以通过WebSocket流功能接收实时K线更新。数据库服务器在/ws上接受WebSocket连接，我们构建了一个推送数据的插件。有关更多详细信息，请查看contrib/