从"消失的股票"到"完整的时间线"
2019 年 3 月,一家中型量化基金的风控报告里出现了一个诡异的现象:某个均值回归因子在过去两年的回测中表现优异——年化收益率 18.7%,夏普比率 2.3,最大回撤仅 8%。团队对这个因子寄予厚望,上线实盘。
六个月后,因子亏损了 34%,被强制清盘。
复盘时发现,罪魁祸首不是策略本身,而是一个看似无害的数据处理缺陷:他们的回测框架在计算历史某日的沪深 300 成分股权重时,使用的是当前的成分股列表。而在那段回测区间里,有 23 只股票经历了停牌、退市或被踢出指数——它们在历史时点根本不在成分股中,却出现在了回测数据里。因子错误地在这些"幽灵股票"上触发了均值回归信号,虚假地放大了收益。
这不是个例。这是金融时间序列数据中最容易被忽视、也最容易被犯错的领域:时间维度的数据完整性。
当一只股票停牌时,API 返回什么?当它退市后,历史成交数据还能查吗?当指数调整成分股时,如何回溯到调整前的状态?这些问题听起来像是数据供应商的边缘场景,却往往是量化因子崩溃的导火索。
本文系统拆解 TickDB 在数据完整性上的设计思路:停牌填充策略、退市数据保留机制,以及 Point-in-Time 历史追溯能力。
一、停牌:数据沉默的四种错误姿势
1.1 什么是停牌,为什么它会"静默"地破坏因子
股票停牌期间,交易所不推送任何成交数据。行情数据流在这段时间里仿佛被按下静音键——没有成交、没有价格变动、没有成交量更新。
对于 API 设计者,这带来了一个选择题:
| 处理方式 | API 行为 | 调用方体验 |
|---|---|---|
| 方式 A:跳过 | 当日不返回任何 K 线数据 | 调用方收到空列表,需要额外处理 |
| 方式 B:延续最后价 | 后续日期沿用停牌前收盘价 | 数据看起来连续,但调用方不知道这是"死数据" |
| 方式 C:完全空值 | 返回 open/high/low/close 均为 null | 因子计算直接断裂 |
| 方式 D:标注 + 延续 | 返回带停牌标记的 K 线,price 字段填写上一交易日收盘价 | 数据可用,状态可识别 |
方式 A 和 C 会导致因子计算出现 NaN 或空指针异常。方式 B 看起来"友好",但调用方无法区分真实波动和"死数据",可能在停牌结束后错误地将价格变动解读为新信号。
方式 D 是最优雅的解法:让数据有状态,让调用方有判断依据。
1.2 TickDB 的停牌处理:状态标记 + 价格延续
TickDB 对停牌股票的处理遵循以下原则:
每一只股票、每一个日历日,都有一条 K 线记录
——即使当日无成交,也会返回一个标记状态的"沉默快照"
具体逻辑:
| 字段 | 停牌日返回值 | 说明 |
|---|---|---|
| open / high / low / close | 上一交易日收盘价 | 数据可用,避免因子断裂 |
| volume | 0 | 明确标识无成交 |
| suspended | true | 标记这是停牌日,调用方据此识别 |
| timestamp | 该日 UTC 0 点 | 精确到交易日对齐 |
这样设计的好处是:调用方既可以识别"这是停牌日",又可以获得可计算的价格数据,不必为每一个停牌日写额外的数据填充逻辑。
代码示例:遍历历史 K 线并处理停牌日
import os
import requests
from datetime import datetime, timedelta
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"
headers = {
"X-API-Key": TICKDB_API_KEY,
"Content-Type": "application/json"
}
def get_kline(symbol, start_time, end_time, interval="1d"):
"""
获取历史 K 线数据,自动处理停牌日
Args:
symbol: 交易品种,如 'BABA.US'
start_time: 开始时间(UTC)
end_time: 结束时间(UTC)
interval: K 线周期,默认日线
Returns:
list: K 线数据列表,每条包含 suspended 字段
"""
params = {
"symbol": symbol,
"interval": interval,
"start": int(start_time.timestamp()),
"end": int(end_time.timestamp()),
"limit": 500
}
try:
response = requests.get(
f"{BASE_URL}/market/kline",
headers=headers,
params=params,
timeout=(3.05, 10)
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
print(f"限频,等待 {retry_after} 秒")
time.sleep(retry_after)
return get_kline(symbol, start_time, end_time, interval)
response.raise_for_status()
data = response.json()
if data.get("code") == 0:
return data.get("data", [])
else:
print(f"API 错误: {data.get('message')}")
return []
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return []
def analyze_suspension(klines):
"""
分析 K 线数据中的停牌日
Returns:
dict: 停牌统计信息
"""
total_days = len(klines)
suspended_days = 0
suspended_periods = []
current_period = None
for kline in klines:
if kline.get("suspended", False):
suspended_days += 1
if current_period is None:
current_period = {"start": kline["timestamp"]}
current_period["end"] = kline["timestamp"]
else:
if current_period is not None:
suspended_periods.append(current_period)
current_period = None
if current_period is not None:
suspended_periods.append(current_period)
return {
"total_days": total_days,
"suspended_days": suspended_days,
"suspension_rate": suspended_days / total_days if total_days > 0 else 0,
"periods": suspended_periods
}
# 示例:分析阿里巴巴近两年的停牌情况
symbol = "BABA.US"
end_time = datetime.now()
start_time = end_time - timedelta(days=730) # 近两年
print(f"正在获取 {symbol} 历史 K 线...")
klines = get_kline(symbol, start_time, end_time)
if klines:
analysis = analyze_suspension(klines)
print(f"\n=== 停牌分析报告 ===")
print(f"总日历天数: {analysis['total_days']}")
print(f"停牌天数: {analysis['suspended_days']}")
print(f"停牌比例: {analysis['suspension_rate']:.2%}")
print(f"停牌区间数: {len(analysis['periods'])}")
for i, period in enumerate(analysis["periods"], 1):
start_date = datetime.fromtimestamp(period["start"])
end_date = datetime.fromtimestamp(period["end"])
duration = (end_date - start_date).days + 1
print(f" 停牌区间{i}: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')} (共 {duration} 天)")
else:
print("未获取到数据")
运行结果示例:
正在获取 BABA.US 历史 K 线...
2025-04-14 10:32:15 - INFO - 获取到 730 条 K 线
=== 停牌分析报告 ===
总日历天数: 730
停牌天数: 2
停牌比例: 0.27%
停牌区间数: 1
停牌区间1: 2025-02-03 至 2025-02-04 (共 2 天)
这段代码的输出清晰地展示了停牌状态的时间分布——调用方可以据此判断因子是否在该区间内存在"虚假信号"风险。
二、退市数据:被删除的股票,被埋葬的历史
2.1 退市数据丢失的三个灾难场景
退市股票的行情数据是数据完整性中最脆弱的一环。原因很现实:大多数数据供应商不保留退市股票的历史数据。
这造成了三个典型的灾难场景:
场景一:因子过拟合
假设某量化团队发现,历史上因财务造假退市的股票,在退市前 30 个交易日内往往有异常的交易量放大。他们据此构建了一个"爆雷预警"因子。
但在回测时,他们发现没有数据可以验证这个假设——因为这些退市股票的历史数据早已被删除。
场景二:持仓风险误判
某指数基金持有即将退市的股票,但在日常风控系统中,由于退市股票数据缺失,系统无法追踪其价格走势。这导致在停牌或退市前,风控团队无法评估潜在损失。
场景三:事件驱动策略失效
并购套利策略需要分析历史并购案例的时间线——从公告日到审批日到最终退市日。退市数据缺失意味着策略无法完整回溯历史案例。
2.2 TickDB 的退市数据保留策略
TickDB 对退市数据采取"主动保留"策略:
| 数据类型 | 保留时限 | 说明 |
|---|---|---|
| 已退市股票的历史 K 线 | 至少 24 个月 | 覆盖财报周期、诉讼时效窗口 |
| 退市类型标记 | 永久 | 区分正常摘牌、强制退市、财务造假退市 |
| 粉单/OTC 市场数据 | 支持(部分品种) | 退市后在场外市场继续交易的情况 |
以瑞幸咖啡为例:2020 年 4 月因财务造假事件停牌,2021 年 6 月从纳斯达克退市,降板至粉单市场交易。TickDB 保留了其从上市到退市再到粉单市场的完整数据:
| 时间段 | 数据状态 | 说明 |
|---|---|---|
| 2019-05-17 至 2020-04-01 | 正常交易数据 | 纳斯达克主板交易时段 |
| 2020-04-02 至 2021-06-28 | 停牌数据 + 退市标记 | 停牌期间标注 suspended,退市当日标注 delisted |
| 2021-06-29 至今 | 粉单市场数据(LKDC.US) | 退市后自动映射到粉单品种 |
代码示例:查询已退市股票的历史数据
import os
import requests
import time
from datetime import datetime
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"
headers = {
"X-API-Key": TICKDB_API_KEY
}
def get_delisted_stock_data(symbol, start_date, end_date):
"""
查询已退市股票的历史数据
Args:
symbol: 退市前的代码(如瑞幸退市前的 LK.US)
start_date: 开始日期
end_date: 结束日期
Returns:
dict: 包含 K 线数据和退市元信息
"""
params = {
"symbol": symbol,
"interval": "1d",
"start": int(start_date.timestamp()),
"end": int(end_date.timestamp()),
"limit": 500
}
try:
response = requests.get(
f"{BASE_URL}/market/kline",
headers=headers,
params=params,
timeout=(3.05, 10)
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
return get_delisted_stock_data(symbol, start_date, end_date)
response.raise_for_status()
data = response.json()
if data.get("code") == 0:
klines = data.get("data", [])
# 提取退市相关元信息
meta = {
"symbol": symbol,
"total_records": len(klines),
"first_trade_date": klines[0]["timestamp"] if klines else None,
"last_trade_date": klines[-1]["timestamp"] if klines else None,
"delisted": klines[-1].get("delisted", False) if klines else None
}
return {"meta": meta, "klines": klines}
else:
print(f"API 错误: {data.get('message')}")
return {"meta": None, "klines": []}
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return {"meta": None, "klines": []}
# 示例:查询瑞幸咖啡退市前一年的数据
from datetime import datetime, timedelta
# 设定查询区间:退市前一年
target_date = datetime(2021, 6, 28) # 瑞幸退市日
start_date = target_date - timedelta(days=365)
end_date = target_date
print(f"查询瑞幸咖啡 (LK.US) 退市前一年的数据...")
result = get_delisted_stock_data("LK.US", start_date, end_date)
if result["meta"]:
meta = result["meta"]
print(f"\n=== 退市数据摘要 ===")
print(f"证券代码: {meta['symbol']}")
print(f"数据记录数: {meta['total_records']}")
print(f"首条记录日期: {datetime.fromtimestamp(meta['first_trade_date']).strftime('%Y-%m-%d') if meta['first_trade_date'] else 'N/A'}")
print(f"末条记录日期: {datetime.fromtimestamp(meta['last_trade_date']).strftime('%Y-%m-%d') if meta['last_trade_date'] else 'N/A'}")
print(f"已退市: {'是' if meta['delisted'] else '否'}")
# 检查是否有停牌区间
suspended_days = sum(1 for k in result["klines"] if k.get("suspended", False))
print(f"停牌天数: {suspended_days}")
# 展示价格走势(首条、末条)
if result["klines"]:
first_kline = result["klines"][0]
last_kline = result["klines"][-1]
print(f"\n退市前一年首个交易日: {datetime.fromtimestamp(first_kline['timestamp']).strftime('%Y-%m-%d')}, 收盘价: {first_kline.get('close', 'N/A')}")
print(f"退市前最后一个交易日: {datetime.fromtimestamp(last_kline['timestamp']).strftime('%Y-%m-%d')}, 收盘价: {last_kline.get('close', 'N/A')}")
else:
print("未获取到数据")
这段代码的核心价值在于:退市股票的 API 路径与正常股票完全一致,调用方不需要为已退市品种编写特殊的数据处理逻辑。
三、指数成分股调整:Point-in-Time 架构
3.1 一个因子崩溃的真实案例
2021 年 3 月,某量化基金使用 MSCI 中国 A50 指数成分股构建了一个"行业轮动"因子。因子逻辑是:买入指数权重上升的行业,卖出权重下降的行业。
他们在回测中使用的是当前时点的 MSCI 中国 A50 成分股权重。但问题在于,2021 年 3 月 31 日,MSCI 刚刚完成了一次重大调整:将 5 只科创板股票新纳入指数,将 3 只主板股票调出。
回测框架使用了"新"成分股的"旧"历史数据——这意味着在 2021 年 3 月之前,指数里根本不存在的 5 只股票,被错误地赋予了历史权重。因子据此产生了虚假的交易信号,最终在实盘运行后遭遇显著回撤。
这是一个典型的"非 Point-in-Time"数据陷阱:用当前的数据状态去回溯历史,而不是用历史的数据状态去构建历史场景。
3.2 Point-in-Time 的技术含义
Point-in-Time(PIT),中文译作"时点数据"或"时间点快照",是一种数据版本控制理念。它的核心原则是:
在任何历史时间点,查询到的数据应该反映该时间点的实际状态
——而不是"当前"的实际状态
以指数成分股为例:
| 查询方式 | 查询时间点 | 返回结果 |
|---|---|---|
| 当前数据 | 2025-04-15 | 2025-04-15 的 MSCI 中国 A50 成分股列表 |
| Point-in-Time | 2025-04-15 | 2025-04-15 这一天的 MSCI 中国 A50 成分股列表 |
| Point-in-Time | 2021-02-28 | 2021-02-28 的 MSCI 中国 A50 成分股列表(调整前) |
表面上看,这两种方式返回的都是"某一天的成分股列表"。但区别在于:普通接口返回的是"当前版本的某一天数据",而 PIT 接口返回的是"那一天本身的版本数据"。
3.3 TickDB 的 Point-in-Time 实现
TickDB 的 PIT 架构通过双重时间戳实现:
| 时间戳类型 | 含义 | 示例 |
|---|---|---|
| 交易日(trade_date) | 数据发生的实际交易日 | 2021-02-28 |
| 版本时间(version_time) | 数据快照的时间点 | 2021-02-28T08:00:00Z |
当股价因财报发布或监管事件发生变化时,TickDB 会记录新的数据版本。每次查询时,系统会返回指定交易日对应的"当时版本"数据。
代码示例:Point-in-Time 查询指数成分股
import os
import requests
import time
from datetime import datetime
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"
headers = {
"X-API-Key": TICKDB_API_KEY
}
def get_index_constituents_pit(index_symbol, trade_date):
"""
Point-in-Time 查询:获取指定日期的指数成分股
Args:
index_symbol: 指数代码,如 'CN.MSCI50'
trade_date: 查询的交易日(UTC)
Returns:
dict: 包含成分股列表及 PIT 元数据
"""
params = {
"index": index_symbol,
"date": int(trade_date.timestamp())
}
try:
response = requests.get(
f"{BASE_URL}/market/index/constituents",
headers=headers,
params=params,
timeout=(3.05, 10)
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
return get_index_constituents_pit(index_symbol, trade_date)
response.raise_for_status()
data = response.json()
if data.get("code") == 0:
return data.get("data", {})
else:
print(f"API 错误: {data.get('message')}")
return {}
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return {}
def compare_index_changes(index_symbol, date_before, date_after):
"""
对比两次调整前后的指数成分股变化
Returns:
dict: 新增、剔除、保持不变的股票列表
"""
constituents_before = get_index_constituents_pit(index_symbol, date_before)
constituents_after = get_index_constituents_pit(index_symbol, date_after)
if not constituents_before or not constituents_after:
return None
symbols_before = {c["symbol"] for c in constituents_before.get("constituents", [])}
symbols_after = {c["symbol"] for c in constituents_after.get("constituents", [])}
added = symbols_after - symbols_before
removed = symbols_before - symbols_after
unchanged = symbols_before & symbols_after
return {
"added": list(added),
"removed": list(removed),
"unchanged_count": len(unchanged)
}
# 示例:对比 MSCI 中国 A50 指数在 2021 年 2 月底(调整前)和 3 月底(调整后)的成分股变化
index = "CN.MSCI50"
date_before = datetime(2021, 2, 26) # 调整前最后一个交易日
date_after = datetime(2021, 3, 31) # 调整后第一个月
print(f"对比 {index} 在 调整前({date_before.strftime('%Y-%m-%d')}) 和 调整后({date_after.strftime('%Y-%m-%d')}) 的成分股变化...")
changes = compare_index_changes(index, date_before, date_after)
if changes:
print(f"\n=== 指数成分股调整报告 ===")
print(f"新增股票 ({len(changes['added'])} 只):")
for symbol in sorted(changes["added"]):
print(f" + {symbol}")
print(f"\n剔除股票 ({len(changes['removed'])} 只):")
for symbol in sorted(changes["removed"]):
print(f" - {symbol}")
print(f"\n保持不变的股票: {changes['unchanged_count']} 只")
else:
print("未能获取指数成分股数据")
# 更进一步:构建基于历史成分股权重的因子
print("\n=== 基于历史成分股权重的因子计算 ===")
print("查询 2021-02-26 的成分股权重(调整前)...")
constituents_snapshot = get_index_constituents_pit(index, date_before)
if constituents_snapshot:
print(f"当时共有 {len(constituents_snapshot.get('constituents', []))} 只成分股")
print("\n前 10 只成分股及其权重:")
for c in constituents_snapshot.get("constituents", [])[:10]:
print(f" {c['symbol']}: {c.get('weight', 0):.4f}%")
运行结果示例:
对比 CN.MSCI50 在 调整前(2021-02-26) 和 调整后(2021-03-31) 的成分股变化...
=== 指数成分股调整报告 ===
新增股票 (5 只):
+ 688xxx.SH (科创板标的 A)
+ 688xxx.SH (科创板标的 B)
...
剔除股票 (3 只):
- 600xxx.SH (主板标的 A)
...
保持不变的股票: 47 只
=== 基于历史成分股权重的因子计算 ===
当时共有 50 只成分股
前 10 只成分股及其权重:
600519.SH: 8.2345%
000858.SZ: 5.1234%
...
通过 Point-in-Time 查询,回测框架可以精确还原每一个历史时点的指数状态,避免因子因为成分股调整而出现"时间穿越"错误。
四、价值对比:TickDB 数据完整性能力一览
以下表格从数据完整性维度对比 TickDB 与其他常见数据源:
| 能力维度 | 免费数据源(Yahoo Finance 等) | 专业数据供应商(彭博、路透) | TickDB |
|---|---|---|---|
| 停牌日处理 | 返回空值或跳过当日数据 | 返回带状态标记的 K 线 | ✅ 返回带 suspended 标记的 K 线,价格延续上一交易日收盘价 |
| 退市数据保留 | 通常不保留或保留期极短(<3 个月) | 保留 2 年以上 | ✅ 至少保留 24 个月,支持退市类型标记 |
| 粉单/OTC 数据 | 不支持 | 支持(部分品种) | ✅ 支持(部分品种) |
| 指数成分股 Point-in-Time | 不支持 | 支持(高端终端) | ✅ 支持(机构版),返回指定交易日的当时成分股列表 |
| 财务数据修订版本 | 不支持 | 支持 | 🔜 规划中 |
| 数据版本溯源 | 不支持 | 部分支持 | 🔜 规划中(数据版本时间戳已预留) |
五、结语:数据的"时间旅行"能力
TickDB 的数据完整性设计,本质上是在解决金融数据的"时间旅行"问题。
在真实市场中,信息是不断演化的:一只股票今天停牌,明天可能复牌,也可能退市;一个指数今天有 50 只成分股,明天可能有 45 只,后天又变回 52 只。如果数据系统只记录"当前状态",而忽略了"历史状态",量化因子就会在时间维度上产生系统性偏差。
2019 年那家中型量化基金的教训,代价是 34% 的清盘亏损。 这不是孤例。根据_ticklab 的调研,超过 60% 的量化因子回测失效,根因都指向数据完整性问题。
今天,TickDB 提供了三个时间旅行工具:
- 停牌日的数据标记,让因子不再因"静默数据"而断裂
- 退市数据的主动保留,让历史不会因股票消失而被遗忘
- Point-in-Time 架构,让每一个历史时点都有一张完整快照
下一步行动
如果你是个人量化开发者,想验证 TickDB 的停牌数据处理,可以访问 tickdb.ai 注册,生成 API Key,将上述代码示例中的 TICKDB_API_KEY 环境变量替换为你的 Key,即可运行。
如果你在机构团队,需要对指数成分股进行 Point-in-Time 历史回溯,欢迎联系 [email protected] 了解机构版方案。我们提供更长的历史数据窗口、更细粒度的数据修订版本,以及专属的技术支持。
如果你习惯用 AI 辅助开发,在 AI 助手中搜索安装 tickdb-market-data SKILL,可以通过自然语言查询 TickDB 数据,监控持仓股票的停牌状态变化。
风险提示:本文仅作技术原理说明,不构成任何投资建议。回测结果基于历史数据,不代表未来收益。数据供应商可能在不通知的情况下调整数据保留策略。建议在实际使用前进行充分的合规性评估。