摘要

很多人第一次做 A 股回测，最关心的是策略逻辑、参数和收益曲线。但真正让回测结果失真的，往往不是模型，而是那份看起来很完整的历史 K 线数据。拿到数据后，先验五件事——symbol、时间范围、K 线周期、OHLCV 字段、异常和缺口——再让它进入回测。本文给出一套可复现的 Python 校验流程、字段检查表和排错表。

1. 为什么要验：10 年数据不等于可信数据

拿到一份“10 年历史 K 线数据”，几千个交易日、几十万根 K 线，看起来很厚，很完整。但只要其中几个关键口径没对齐，问题就会被藏得更深。

• 你以为自己在查 600519.SH，但某个环节实际入库的是另一个 symbol；
• 你以为拿到的是完整日线，但某些停牌、节假日、异常返回没有被解释；
• 字段名都叫 volume，但没有确认它在你的策略里应该如何理解；
• 某次查询失败后，系统没有报错，而是用空值、旧值或缓存值填过去。

回测框架不会替你判断这些。它只会读取数据、计算指标、输出曲线。数据错的时候，回测也会很认真。

2. 环境准备

本文代码为教学示例，非生产级完整实现。所有 API 路径、字段和 Header 以官方文档或实测为准。

pip install requests==2.31.0

TickDB 在此处的工程能力：统一 symbol 格式、结构化字段返回、UTC 毫秒时间戳、REST 查询接口和明确的错误处理——这些是回测前数据校验的基础。TickDB 支持 REST、WebSocket、MCP 等多种接入方式，本文聚焦 REST 历史 K 线接口的校验流程。

3. 请求参数检查

调用 K 线接口前，先确认以下参数。参数写错，后面所有校验都白做。

参数	核对点	失败处理
symbol	格式为 CODE.EXCHANGE（如 600519.SH），逐字符与预期一致	不匹配则阻断，不依赖接口自动修正
interval	接口支持的周期列表（1d/1w/1M 等），策略周期与接口周期一致	不支持的 interval 会返回错误
时间范围	起止日期在接口支持的历史范围内；停牌日、节假日是否有数据返回	超范围请求验证接口是否返回明确错误

Python 示例：

# 查询可用 interval（教学伪代码，实际路径以官方文档为准）
# resp = requests.get(f"{BASE_URL}/market/kline/intervals", headers={"X-API-Key": API_KEY})
# intervals = resp.json().get("data", [])
# print(intervals)  # 示例：['1m','3m','5m','15m','30m','1h','2h','4h','1d','1w','1M']

4. 返回字段检查

K 线返回体中的每个字段，逐项核对类型和取值。以下字段以 TickDB REST K 线接口返回结构为示例。

字段	核对点	失败处理
symbol	与请求逐字符一致	不匹配则阻断
time	整数且非 bool，UTC 毫秒	类型异常则阻断
open	非空字符串，可解析为有限 Decimal	解析失败阻断，不默认成 0
high	同上，且 high >= open, close, low	不满足则标记异常
low	同上，且 low <= open, close, high	不满足则标记异常
close	同上	解析失败阻断
volume	非空字符串，可解析为有限 Decimal	解析失败阻断

Python 示例：

from decimal import Decimal, InvalidOperation

def validate_kline_bar(bar: dict, expected_symbol: str) -> dict:
    """单根 K 线的字段校验。"""
    # symbol
    if bar.get("symbol") != expected_symbol:
        return {"ok": False, "reason": f"symbol mismatch: expected {expected_symbol}, got {bar.get('symbol')}"}

    # time
    ts = bar.get("time")
    if isinstance(ts, bool) or not isinstance(ts, int):
        return {"ok": False, "reason": f"time invalid: {ts}"}

    # OHLCV
    for field in ("open", "high", "low", "close", "volume"):
        raw = bar.get(field)
        if not isinstance(raw, str) or not raw.strip():
            return {"ok": False, "reason": f"{field} missing or empty"}
        try:
            val = Decimal(raw)
            if not val.is_finite():
                return {"ok": False, "reason": f"{field} not finite: {raw}"}
        except (InvalidOperation, ValueError):
            return {"ok": False, "reason": f"{field} unparseable: {raw}"}

    # high >= low 基本逻辑
    if Decimal(bar["high"]) < Decimal(bar["low"]):
        return {"ok": False, "reason": "high < low"}

    return {"ok": True, "data": bar}

5. 异常场景排错表

失败场景	现象	处理方向
symbol 不存在	返回 symbol not found 或类似错误	核对 symbol 格式，确认后缀（.SH/.SZ）
interval 不支持	返回 interval 无效或空数据	先查询可用 interval 列表
时间范围超限	返回最早可用时间限制	确认接口的历史数据覆盖范围
data 为空	正常请求但返回空数组	检查 symbol 状态、权限和当前时段
OHLCV 类型异常	字段为 null、"N/A" 或非数字字符串	阻断，不默认成 0
raw_snapshot 未保存	事后无法复盘	每次请求保存原始 JSON

6. 缺口检查

K 线序列拿到后，逐根检查时间间隔是否稳定。

Python 示例：

def check_kline_gaps(klines: list, expected_interval_ms: int) -> list:
    """检查 K 线序列中的缺口。klines 已按 time 升序排列。"""
    gaps = []
    for i in range(len(klines) - 1):
        curr_ts = klines[i].get("time")
        next_ts = klines[i + 1].get("time")
        if not isinstance(curr_ts, int) or not isinstance(next_ts, int):
            continue
        interval = next_ts - curr_ts
        # 允许一个周期的容差（如日线允许 ±1 小时）
        if interval > expected_interval_ms * 1.5:
            gaps.append({
                "from_index": i,
                "to_index": i + 1,
                "expected_ms": expected_interval_ms,
                "actual_ms": interval,
            })
    return gaps

注意：expected_interval_ms 需根据 interval 换算（如 1d = 86400000ms）。停牌日和节假日会导致正常间隔变大，实际使用时需结合交易日历区分“正常缺口”和“异常缺口”。

7. 最小验收报告

每次取数后，生成一份最小验收报告：

def build_acceptance_report(symbol: str, interval: str, 
                            total_bars: int, gaps: list, 
                            errors: list, snapshot_id: str) -> dict:
    """生成 K 线数据验收报告。"""
    return {
        "symbol": symbol,
        "interval": interval,
        "total_bars": total_bars,
        "gap_count": len(gaps),
        "gaps": gaps,
        "error_count": len(errors),
        "errors": errors,
        "raw_snapshot_id": snapshot_id,
        "checked_at": "ISO8601 时间戳",
        "verdict": "通过" if len(errors) == 0 else "存疑",
    }

8. 回测前 K 线验收清单

序号	检查项	通过标准
1	symbol 格式统一	带市场后缀，未被自动修正
2	interval 明确	策略周期与接口周期一致
3	时间范围清晰	起止日期、停牌处理方式明确
4	OHLCV 字段稳定	抽查不同年份字段含义和类型一致
5	异常有明确返回	无效 symbol、超范围、空数据不静默
6	缺口已记录	区分正常缺口（节假日）和异常缺口
7	raw_snapshot 已保存	每次请求保留原始 JSON
8	验收报告已生成	包含总条数、缺口数、错误数和结论

回测从来不是先问“策略能不能赚钱”。更底层的问题是：你喂进去的数据，配不配得上这个策略。

📡 本文 K 线校验示例以 TickDB.ai 作为候选行情入口，具体字段以官方文档和实测为准
⚠️ 本文为技术教程，不构成任何投资建议