第4章数据基础设施（二）

第4章：数据基础设施（二）：多资产数据管理、时间序列数据库选型与接入

上一章我们把单资产的数据管道搭起来了。但做宏观量化，你手里不可能只有一支股票或一个期货品种。我见过不少新手，一开始只跑一两个品种，觉得数据管理很简单。等扩展到几十个资产时，代码就乱成一锅粥了。

这一章，我们就来解决这个问题。

4.1 多资产数据管理的核心挑战

先说说多资产数据管理到底难在哪。说白了，就三个字：杂、大、乱。

杂：股票、期货、期权、外汇、债券，每个品种的数据结构都不一样。股票有除权除息，期货有换月，期权有行权价和到期日。
大：分钟级数据，几十个品种跑一年，数据量轻松上亿条。普通的关系型数据库扛不住。
乱：不同数据源的时间戳格式不同，有的用UTC，有的用本地时间。合并时经常对不齐。

我在项目中遇到过最头疼的事：两个数据源对同一只股票的同一条tick数据，时间戳差了1毫秒。就这1毫秒，导致回测结果完全不一样。嗯，这里要注意，时间精度问题在量化里是致命的。

4.2 多资产数据模型设计

怎么设计一个能支撑多资产的数据模型？我个人习惯用标签+字段的方式。

举个例子，一条数据记录可以这样设计：

{
  "time": "2024-01-15T09:30:00Z",
  "tags": {
    "symbol": "AAPL",
    "exchange": "NASDAQ",
    "asset_type": "stock"
  },
  "fields": {
    "open": 185.20,
    "high": 186.50,
    "low": 184.80,
    "close": 186.20,
    "volume": 1250000
  }
}

你想想看，这种设计的好处是什么？

标签（tags）：用来做过滤和分组。比如你想查所有NASDAQ的股票，或者只查期货数据，直接按标签过滤就行。
字段（fields）：存储实际的数值。不同资产类型的字段可以不一样，股票有volume，期权有implied_volatility，互不干扰。

这种设计思路，其实就是时间序列数据库的核心理念。我们接下来要选的数据库，天生就支持这种结构。

4.3 时间序列数据库选型：InfluxDB vs ClickHouse

市面上时间序列数据库不少，但真正适合量化回测的，我个人觉得就两个：InfluxDB 和 ClickHouse。

先看一张对比表：

特性	InfluxDB	ClickHouse
写入性能	极高，单机每秒百万级	极高，列式存储批量写入
查询性能	按时间范围查询极快	复杂聚合查询极快
SQL支持	类SQL（Flux/InfluxQL）	标准SQL
存储压缩比	中等	高（列式压缩）
部署复杂度	简单，单机即可	中等，适合集群
适用场景	实时监控、高频数据	大规模分析、历史回测

怎么选？我建议这样：

如果你的回测以分钟级或更高频的数据为主，数据量在几十亿条以内，InfluxDB 就够了。部署简单，查询快。
如果你要做全市场、多年份的大规模回测，数据量上百亿，需要复杂的聚合分析，ClickHouse 更合适。

我曾经在一个项目里用InfluxDB存了3年的全市场tick数据，结果查询一次要等十几秒。后来换成ClickHouse，同样的查询，1秒内出结果。这就是选型不对的代价。

4.4 InfluxDB接入实战

我们直接上代码。假设你已经装好了InfluxDB 2.x版本。

第一步，安装客户端库：

pip install influxdb-client

第二步，写一个数据写入器：

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import pandas as pd

class InfluxWriter:
    def __init__(self, url="http://localhost:8086", token="my-token", org="my-org", bucket="market_data"):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(write_type=SYNCHRONOUS)
        self.bucket = bucket

    def write_ohlcv(self, symbol, exchange, asset_type, df):
        """
        写入OHLCV数据
        df必须包含: time, open, high, low, close, volume
        """
        points = []
        for _, row in df.iterrows():
            point = Point("ohlcv") \
                .tag("symbol", symbol) \
                .tag("exchange", exchange) \
                .tag("asset_type", asset_type) \
                .field("open", float(row["open"])) \
                .field("high", float(row["high"])) \
                .field("low", float(row["low"])) \
                .field("close", float(row["close"])) \
                .field("volume", float(row["volume"])) \
                .time(row["time"])
            points.append(point)

        self.write_api.write(bucket=self.bucket, record=points)
        print(f"写入 {len(points)} 条数据到 {symbol}")

第三步，查询数据：

class InfluxReader:
    def __init__(self, url="http://localhost:8086", token="my-token", org="my-org"):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.query_api = self.client.query_api()

    def query_ohlcv(self, symbol, start, stop, asset_type=None):
        flux = f'''
        from(bucket: "market_data")
          |> range(start: {start}, stop: {stop})
          |> filter(fn: (r) => r["_measurement"] == "ohlcv")
          |> filter(fn: (r) => r["symbol"] == "{symbol}")
        '''
        if asset_type:
            flux += f'|> filter(fn: (r) => r["asset_type"] == "{asset_type}")'

        flux += '|> pivot(rowKey:["_time"], columnKey:["_field"], valueColumn:"_value")'

        tables = self.query_api.query(flux)
        records = []
        for table in tables:
            for record in table.records:
                records.append(record.values)

        return pd.DataFrame(records)

小提示：写入时尽量用批量写入，别一条一条insert。我见过有人用for循环逐条写入，10万条数据写了半小时。改成批量后，10秒搞定。

4.5 ClickHouse接入实战

ClickHouse的接入稍微复杂一点，但性能确实强。

安装驱动：

pip install clickhouse-driver

建表语句：

CREATE TABLE market_data.ohlcv (
    time DateTime,
    symbol String,
    exchange String,
    asset_type String,
    open Float64,
    high Float64,
    low Float64,
    close Float64,
    volume UInt64
) ENGINE = MergeTree()
ORDER BY (symbol, time)
PARTITION BY toYYYYMM(time);

这里要注意几个设计点：

ORDER BY (symbol, time)：这是ClickHouse的排序键，决定了数据物理存储的顺序。按symbol+time排序，查询单只股票的时间序列时，性能极佳。
PARTITION BY toYYYYMM(time)：按月分区。查询某个月的数据时，只扫描对应分区，速度飞快。

写入代码：

from clickhouse_driver import Client

class ClickHouseWriter:
    def __init__(self, host="localhost", port=9000, database="market_data"):
        self.client = Client(host=host, port=port)
        self.client.execute(f"CREATE DATABASE IF NOT EXISTS {database}")
        self.database = database

    def write_ohlcv(self, symbol, exchange, asset_type, df):
        # 准备数据
        data = []
        for _, row in df.iterrows():
            data.append((
                row["time"].to_pydatetime(),
                symbol,
                exchange,
                asset_type,
                float(row["open"]),
                float(row["high"]),
                float(row["low"]),
                float(row["close"]),
                int(row["volume"])
            ))

        # 批量写入
        self.client.execute(
            f"INSERT INTO {self.database}.ohlcv VALUES",
            data,
            types_check=True
        )
        print(f"写入 {len(data)} 条数据到 {symbol}")

警告：ClickHouse的写入是异步的，默认每2-3秒刷一次盘。如果你写入后立刻查询，可能查不到最新数据。可以手动执行 OPTIMIZE TABLE 强制合并，但生产环境不建议频繁操作。

4.6 多资产数据管理的架构设计

最后，我们来看一张整体架构图。这是我个人比较推荐的多资产数据管理方案。

这张图的核心思路是分层解耦。数据源只管采集，清洗层负责标准化，存储层负责持久化，访问层提供统一接口。每一层都可以独立替换，互不影响。

举个例子，你今天用Yahoo Finance做数据源，明天想换成Wind，只需要改数据源层，上层的回测代码一行都不用动。这就是架构设计的力量。

  核心要点：
  多资产数据管理的关键是「标签+字段」模型
InfluxDB适合高频实时，ClickHouse适合大规模分析
写入一定要批量，查询一定要按标签过滤
架构设计要分层，每层职责单一

好了，这一章的内容就到这里。数据基础设施搭好了，下一章我们就可以开始构建真正的回测引擎了。

公众号:蓝海数据掘金营，微信deep3321

📚 课程目录