3、数据基础设施搭建：数据源选择、实时行情API接入、Tick级与分钟级数据存储
做可转债做市策略，说白了就是跟数据打交道。我见过不少团队，策略逻辑写得天花乱坠，结果一上实盘就崩——问题往往出在数据上。数据不干净、延迟高、存储不合理，再好的策略也是白搭。今天咱们就把这块硬骨头啃下来。
核心观点：数据基础设施决定了策略的上限。你想想看，如果行情数据晚了一秒，或者Tick数据缺了几笔，做市策略的报价可能就全错了。这不是危言耸听，我在实盘中就吃过这个亏。
3.1 数据源选择：Wind、聚宽、Tushare
选数据源，得看你的场景。我个人习惯把数据源分成三类：机构级、量化平台级、开源级。咱们一个一个说。
| 数据源 | 类型 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|---|
| Wind | 机构级 | 数据最全、质量最高、客服响应快 | 贵（年费几万起）、API较老 | 机构实盘、回测验证 |
| 聚宽（JoinQuant） | 量化平台级 | 本地化好、文档清晰、支持分钟级 | Tick数据需额外付费 | 策略开发、回测 |
| Tushare | 开源级 | 免费、社区活跃、接口灵活 | 数据延迟、稳定性一般 | 个人研究、原型验证 |
Wind 是我在机构里用得最多的。它的可转债数据非常全，从转股价、下修条款到回售触发价，应有尽有。但有个坑——Wind的实时行情接口是DLL形式的，Python调用起来有点麻烦。我曾经花了一整天，就为了把Wind的Tick数据流接到自己的系统里。
聚宽 更适合做策略开发。它的本地数据包可以直接下载，分钟级数据基本够用。但如果你要做高频做市，Tick数据就得单独买。嗯，这里要注意：聚宽的Tick数据只保留最近几个交易日，历史Tick需要自己攒。
Tushare 是我个人研究时用的。免费、接口简单，拿来做原型验证非常方便。但别指望用它做实盘——数据延迟有时候能到几分钟，而且偶尔会断流。我记得有一次跑回测，发现某天的数据全是空的，排查了半天才发现是Tushare那边的问题。
我的建议：如果你刚开始做可转债做市，先用Tushare搭原型，验证策略逻辑。等要上实盘了，再切换到Wind或聚宽。别一上来就买几万块的Wind，万一策略不行，钱就白花了。
3.2 实时行情API接入
实时行情是做市策略的命脉。接入方式主要有两种：WebSocket 和 轮询。做市策略必须用WebSocket，轮询的延迟你扛不住。
以聚宽的实时行情为例，接入代码大概长这样：
from jqdatasdk import *
import websocket
import json
# 初始化聚宽
auth('你的账号', '你的密码')
# WebSocket回调函数
def on_message(ws, message):
data = json.loads(message)
# 处理Tick数据
process_tick(data)
def on_error(ws, error):
print(f"连接异常: {error}")
# 自动重连逻辑
reconnect()
# 订阅可转债行情
ws = websocket.WebSocketApp(
"wss://data.jqdata.cn/tick",
on_message=on_message,
on_error=on_error
)
ws.run_forever()这段代码看着简单，但实际坑不少。我遇到过最典型的问题：断线重连。WebSocket连接会因为网络波动、服务器维护等原因断开。如果不做重连逻辑，策略就会一直用旧数据报价，那画面太美我不敢看。
避坑指南：我曾经因为没做心跳检测，导致WebSocket断线了整整10分钟都没发现。那10分钟里，策略一直在报错价，亏了大概5万块。后来我加了个心跳机制——每5秒检查一次连接状态，断了就立刻重连，同时清空缓存数据，防止用旧数据报价。
另外，数据对齐也是个头疼的问题。不同数据源的时间戳格式不一样，有的用Unix时间戳，有的用字符串。我习惯统一转成纳秒级时间戳，这样在后续处理中不会出现时间错位。
3.3 Tick级与分钟级数据存储
数据存哪里？怎么存？这取决于你的数据量和查询频率。我一般分两层：热数据 和 冷数据。
3.3.1 Tick级数据存储（InfluxDB）
Tick数据的特点是：量大、写入频繁、查询最近数据多。InfluxDB是时序数据库，天生适合干这个。
我的存储结构是这样的：
# InfluxDB 数据模型
measurement: convertible_bond_tick
tags:
- bond_code: "123456" # 可转债代码
- exchange: "SZ" # 交易所
fields:
- price: 128.50 # 最新成交价
- volume: 1000 # 成交量
- bid_price: 128.48 # 买一价
- ask_price: 128.52 # 卖一价
- bid_volume: 500 # 买一量
- ask_volume: 300 # 卖一量
timestamp: 2024-01-15T09:30:00.123456789Z写入性能是关键。我实测过，单机InfluxDB每秒能处理10万条以上的写入，完全够用。但要注意数据保留策略——Tick数据占空间很大，一天可能就几个GB。我一般只保留最近7天的Tick数据，超过的自动删除或归档。
小技巧：InfluxDB的连续查询（Continuous Query）可以自动降采样。比如把Tick数据聚合成1秒K线，存到另一个measurement里。这样查询历史数据时，就不用扫全量Tick了，速度能快几十倍。
3.3.2 分钟级数据存储（Parquet）
分钟级数据用于回测和策略分析。Parquet格式是我最推荐的——列式存储、压缩率高、支持复杂嵌套结构。
存储结构示例：
# Parquet 文件结构
data/
convertible_bond/
year=2024/
month=01/
day=15/
bond_123456.parquet
bond_123457.parquet
# 每行数据包含
- bond_code: str
- timestamp: datetime
- open: float
- high: float
- low: float
- close: float
- volume: int
- amount: float
- bid_ask_spread: float # 买卖价差，做市策略的关键指标用Parquet存分钟级数据，压缩率能达到5:1以上。我存了一年的可转债分钟数据，原始CSV大概50GB，Parquet只有不到10GB。而且查询速度极快——用Pandas读取一天的分钟数据，基本在毫秒级。
3.4 数据清洗与对齐
数据拿到手，别急着用。清洗这一步，我吃过太多亏了。
常见的问题有：
- 缺失值：某分钟没有成交，数据是空的
- 异常值：价格突然跳变，比如从100跳到1000
- 时间戳错位：不同数据源的时间基准不一样
- 复权问题：可转债有转股、回售等事件，价格需要复权
我的清洗流程是这样的：
def clean_tick_data(df):
# 1. 删除重复数据
df = df.drop_duplicates(subset=['bond_code', 'timestamp'])
# 2. 处理缺失值：用前一个Tick的价格填充
df['price'] = df['price'].ffill()
# 3. 异常值检测：价格变动超过5%的，标记为异常
df['price_change'] = df['price'].pct_change()
df.loc[df['price_change'].abs() > 0.05, 'price'] = None
df['price'] = df['price'].ffill()
# 4. 时间对齐：统一到纳秒级时间戳
df['timestamp'] = pd.to_datetime(df['timestamp']).astype('int64')
return df我曾经踩过的坑：有一次回测结果特别好，年化收益30%+。我高兴坏了，结果仔细一查，发现是数据清洗时把异常值直接删掉了，导致回测里少了很多亏损交易。后来我改成用前值填充，而不是直接删除，回测结果才变得真实。记住：删除数据要谨慎，填充比删除更安全。
数据对齐这块，我建议用时间戳对齐法。把所有数据的时间戳统一到同一个基准（比如交易所的撮合时间），然后按时间排序。不同数据源的时间差超过1毫秒的，就标记为不同步，需要重新获取。
3.5 整体架构图
说了这么多，咱们用一张图把整个数据基础设施串起来：
这张图把整个流程串起来了。从数据源到接入层，再到存储层，最后到清洗层，每一层都有明确的职责。我个人习惯在清洗层之后再加一个数据质量监控模块——每天检查数据完整性、延迟情况、异常比例。如果某个指标超标，就自动报警。
好了，数据基础设施这块就聊到这儿。记住一句话：数据不牢，地动山摇。把底层打扎实了，策略才能在上面跑得稳。