用录制回放（VCR 思路）给行情接口写稳定的单测

给量化系统写单测有个绕不开的难题：测试要依赖行情接口，而真实接口不稳定——限流、断线、数据格式随版本漂移。每次跑 CI 都去打真实接口，不仅慢，还会因为限流把测试拖红。借鉴 Ruby 圈 VCR 库的思路，我用 Python 实现了一套「录制一次、回放千次」的方案，单元测试再也不碰网络。

一、核心思路：在适配器层做录制开关

关键原则是录制点放在自己的行情适配器（adapter）里，而不是在每个测试里 mock。适配器是全项目唯一接触外部 API 的地方，在这里加一个「先查本地缓存、缓存没有才请求」的开关：

import json, hashlib, pathlib
from pathlib import Path

CASSETTE_DIR = Path("tests/cassettes")

class MarketDataSource:
    def __init__(self, record: bool = False):
        self.record = record

    def _cache_path(self, func: str, **params) -> Path:
        digest = hashlib.md5(
            json.dumps([func, params], sort_keys=True).encode()
        ).hexdigest()[:16]
        return CASSETTE_DIR / f"{func}.{digest}.json"

    def get_kline(self, symbol: str, freq: str, start: str, end: str) -> dict:
        path = self._cache_path("kline", symbol=symbol, freq=freq,
                                start=start, end=end)
        if not self.record and path.exists():
            return json.loads(path.read_text())      # 回放
        data = self._real_request_kline(symbol, freq, start, end)  # 真实请求
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(data, ensure_ascii=False))      # 录制
        return data

测试里 MarketDataSource(record=False) 就是纯离线回放；想刷新样本时切 record=True 跑一遍，新 cassette 自动落盘。

二、参数要全部进 hash

第一个版本我只把 symbol 和日期拼进了缓存 key，结果 freq 参数改动后回放了错误周期的数据，测试全绿但逻辑是错的——这是假阳性最危险的形式。教训：所有影响返回结果的参数必须进 hash（上面 json.dumps([func, params], sort_keys=True)），sort_keys 保证参数顺序不影响 key。

三、 cassette 也要做体检

录制文件本身可能坏：接口升级后返回结构变了、录制当天恰好停牌导致数据稀疏。我给 cassette 加了极简的 schema 校验，加载时顺带检查：

def load_cassette(path: Path) -> dict:
    data = json.loads(path.read_text())
    assert "bars" in data and len(data["bars"]) > 0, f"broken cassette: {path}"
    assert all({"dt", "o", "h", "l", "c"} <= set(b) for b in data["bars"])
    return data

校验失败的 cassette 直接报错，逼你重新录制，而不是让上层拿到畸形数据继续算出错误结论。

四、回放样本的「保鲜期」管理

行情结构可能随接口版本变化，cassette 不能永不过期。我的做法是录制文件里带元信息，测试 setup 时检查年龄：

{"_meta": {"recorded_at": "2026-11-06", "source": "futu", "api_ver": "9.x"},
 "bars": [...]}

超过 90 天的 cassette 打 warning（不硬失败），CI 里黄色提示，提醒下次手动刷新。同时把 cassette 目录纳入 git，团队成员 clone 下来就能跑全量测试，不需要任何账号和密钥——这一条对协作项目的价值比去重本身还大。

五、什么时候该走真实接口

录制回放不是终点。两类测试必须保留真实链路：一是冒烟测试（每天跑一次，验证接口活着、字段没变，发现漂移就触发重新录制）；二是限流与重试逻辑本身——你不可能用回放来测断线重连，这类要用本地起一个假 HTTP 服务模拟 429/超时。分级之后，几百个单测秒级跑完零网络依赖，只有冒烟测试出门打真实接口，稳定性和真实性都保住了。

这套东西一天就能搭完，换来的是此后每一次重构都有真数据兜底，值得。