从"消失的股票"到"完整的时间线"

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 数据,监控持仓股票的停牌状态变化。


风险提示:本文仅作技术原理说明,不构成任何投资建议。回测结果基于历史数据,不代表未来收益。数据供应商可能在不通知的情况下调整数据保留策略。建议在实际使用前进行充分的合规性评估。