第二章：数据采集与存储——API接口对接与数据落地实战

做期权做市，数据就是你的弹药。没有干净、稳定的数据流，再好的策略也是空中楼阁。这一章，我跟你聊聊数据怎么来、怎么存、怎么管。说白了，就是解决「数据从哪里来，来了放哪里，放了怎么找回来」这三个核心问题。

核心要点：数据采集与存储的三大支柱——协议对接、存储选型、压缩归档。这三件事做扎实了，后面的分析工作才能站得住脚。

2.1 API接口对接：FIX协议 vs REST API

对接交易所或数据商，你逃不开两个东西：FIX协议和REST API。我个人的习惯是，高频场景用FIX，中低频用REST。为什么？往下看。

2.1.1 FIX协议——专业玩家的选择

FIX协议是金融领域的「老炮儿」。它诞生于1992年，专门为证券交易设计。做市商用它，因为延迟低、消息结构稳定。

我在项目中遇到过一个问题：某次对接一家交易所的FIX网关，对方文档里写的是「支持FIX 4.4」，结果实际跑起来，某些字段的标签号跟标准不一样。嗯，这里要注意——FIX协议虽然标准化，但各家交易所都会做「方言化」扩展。你对接前，一定要拿到对方的FIX字典文件，仔细核对。

一个典型的FIX消息长这样：

8=FIX.4.4|9=78|35=D|49=CLIENT1|56=EXCHANGE|34=2|52=20240101-12:00:00.000|
11=ORD001|21=1|55=600519.SH|54=1|38=100|40=2|44=150.00|59=0|10=128|

你看，每个字段用竖线分隔，标签号+等号+值。第35字段是消息类型，D代表新订单。第55字段是证券代码。第44字段是价格。

我的小技巧：调试FIX连接时，先用QuickFIX/J或stunnel做日志分析。我曾经花了一整天排查一个连接超时问题，最后发现是对方防火墙只开放了特定IP段。所以，先做连通性测试，再看协议细节。

2.1.2 REST API——灵活但慢一点

REST API更适合获取历史数据、查询账户信息、执行非高频交易。它的优势是简单——用HTTP请求就能搞定，不需要维护长连接。

举个例子，获取某期权合约的实时行情：

import requests
import json

url = "https://api.exchange.com/v1/market/option/quotes"
params = {
    "symbol": "AAPL230616C00150000",
    "type": "snapshot"
}
headers = {
    "X-API-Key": "your_api_key_here"
}

response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
    data = response.json()
    print(f"最新买价: {data['bid']}, 最新卖价: {data['ask']}")
else:
    print(f"请求失败: {response.status_code}")

你想想看，REST API的瓶颈在哪？每次请求都要建立TCP连接、发送HTTP头、等待响应。对于毫秒级的做市场景，这太慢了。但如果你只是每天收盘后拉一次历史数据，REST完全够用。

避坑指南：我曾经遇到过某数据商的REST API在盘中时段频繁返回429（Too Many Requests）。后来发现，他们的限流策略是按IP+API Key双重计算的。解决方案是：使用多个API Key轮询，或者降低请求频率。做市系统里，我建议对REST请求做本地缓存，避免重复请求。

2.2 数据落地存储方案

数据拿到手了，放哪？我见过有人把所有数据塞进一个CSV文件，也见过有人用Oracle存行情快照。选存储方案，要看你的数据量、查询频率、分析需求。

下面这张图，是我对存储选型的理解：

数据存储选型决策树 原始行情数据 数据用途：实时分析 vs 历史回测 实时分析场景 推荐：内存数据库（Redis）+ 时序数据库（InfluxDB） 历史回测场景 推荐：Parquet + 列式存储（ClickHouse） 小数据量（<100GB）：CSV + 压缩 大数据量（>100GB）：Parquet + 分区 最终建议：实时用Redis+InfluxDB，历史用Parquet+ClickHouse

2.2.1 CSV——简单但别滥用

CSV是最简单的存储格式。我刚开始做量化时，所有数据都存CSV。但后来发现，当数据量超过10GB时，CSV的读写效率直线下降。

CSV适合什么场景？

  • 每日收盘后的行情快照（几十MB级别）
  • 策略参数的配置文件
  • 临时数据交换

但要注意：CSV没有schema约束，字段类型全靠猜。我曾经因为一个日期字段被读成字符串，导致回测结果全错。所以，用CSV时一定要显式指定数据类型。

import pandas as pd

# 正确做法：指定数据类型
df = pd.read_csv('option_quotes.csv',
                 dtype={
                     'symbol': 'string',
                     'bid_price': 'float64',
                     'ask_price': 'float64',
                     'bid_size': 'int32',
                     'ask_size': 'int32'
                 },
                 parse_dates=['timestamp'])

2.2.2 Parquet——大数据量的首选

Parquet是列式存储格式，压缩率高、查询快。我个人的习惯是，所有超过1GB的历史数据都用Parquet存。

为什么选Parquet？

  • 压缩比高：同样的数据，Parquet比CSV小5-10倍
  • 列式读取：如果你只查某几列，不需要读整行数据
  • 内置schema：数据类型、字段名都写在文件里

举个例子，存储一天的期权行情数据：

import pandas as pd
import pyarrow.parquet as pq

# 写入Parquet
df.to_parquet('option_data_20240101.parquet',
              compression='snappy',
              row_group_size=10000)

# 读取时只选需要的列
df = pd.read_parquet('option_data_20240101.parquet',
                     columns=['timestamp', 'symbol', 'bid_price', 'ask_price'])

我的经验：Parquet的row_group_size设置很关键。太小了（比如1000行），元数据开销大；太大了（比如100万行），随机查询慢。我一般设成10000-50000行，具体要看你的数据宽度。

2.2.3 数据库——适合需要复杂查询的场景

如果你需要做SQL查询、关联分析、权限管理，那就得上数据库。我推荐时序数据库（如InfluxDB、TimescaleDB）来处理行情数据，关系型数据库（如PostgreSQL）来处理交易记录和账户信息。

存储方案 适用场景 优点 缺点
CSV 小数据量、临时数据 简单、通用 无schema、查询慢
Parquet 大数据量、分析场景 压缩高、列式读取 不支持随机写入
InfluxDB 实时行情、时序数据 写入快、查询灵活 存储成本高
PostgreSQL 交易记录、账户信息 ACID、支持复杂查询 写入吞吐有限

2.3 数据压缩与归档策略

数据存下来了，但磁盘空间不是无限的。尤其是期权做市，每天产生的行情数据可能几十GB。怎么压缩？怎么归档？

2.3.1 压缩算法选型

我常用的压缩算法有几种：

  • Snappy：速度快，压缩比一般（2-3倍）。适合实时写入场景。
  • Gzip：速度慢，压缩比高（5-10倍）。适合归档场景。
  • Zstd：平衡型，速度和压缩比都不错。我最近越来越喜欢用它。

你想想看，如果每天的数据量是50GB，用Snappy压缩后大概20GB，用Gzip压缩后可能只有8GB。但Gzip压缩要花3倍的时间。怎么选？看你的业务需求。

我的建议：热数据（最近7天）用Snappy，保证读写速度；冷数据（超过30天）用Gzip或Zstd，最大化节省空间。

2.3.2 归档策略——按时间分区

归档的核心思想是：把数据按时间切分，定期迁移到冷存储。我一般这样做：

  1. 日分区：每天一个文件或表分区。比如option_data_20240101.parquet
  2. 周归档：每周把7天的数据打包成一个压缩文件。比如week_01_2024.tar.gz
  3. 月清理：超过3个月的数据，迁移到对象存储（如S3、MinIO）或磁带库

我曾经犯过一个错误：把所有数据都放在一个目录下，结果一年后目录里有3000多个文件，ls命令都要卡几秒。后来我改成按年/月/日的目录结构，问题就解决了。

# 归档脚本示例
import os
import shutil
import gzip
from datetime import datetime, timedelta

def archive_old_data(base_path, days_old=30):
    """归档30天前的数据"""
    cutoff = datetime.now() - timedelta(days=days_old)
    
    for root, dirs, files in os.walk(base_path):
        for file in files:
            file_path = os.path.join(root, file)
            file_time = datetime.fromtimestamp(os.path.getmtime(file_path))
            
            if file_time < cutoff:
                # 压缩归档
                archive_name = f"{file}.gz"
                with open(file_path, 'rb') as f_in:
                    with gzip.open(archive_name, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                
                # 删除原文件
                os.remove(file_path)
                print(f"已归档: {file_path} -> {archive_name}")

# 每周执行一次
archive_old_data('/data/option_quotes', days_old=30)

避坑指南：我曾经在归档时忘记检查文件是否被其他进程占用，结果压缩出来的文件是损坏的。所以，归档前一定要先检查文件锁。另外，归档后的数据要定期做完整性校验——我一般用MD5或CRC32。

2.4 本章小结

数据采集与存储，说白了就是三个字：接、存、管。

  • 接：FIX协议适合高频，REST API适合中低频。对接前一定要做连通性测试和协议校验。
  • 存：小数据用CSV，大数据用Parquet，实时场景用时序数据库。别把鸡蛋放在一个篮子里。
  • 管：热数据用Snappy，冷数据用Gzip。按时间分区归档，定期做完整性校验。

这些经验，是我在实盘做市系统里一点点踩坑踩出来的。你照着做，至少能少走一半弯路。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →