我自己在做 BTC 永续合约的均值回归策略时，最早只用 1 分钟 K 线，IC 只有 0.03，夏普 1.2；后来把订单流不平衡（OFI）、成交量加权价（VWA）偏差、深度不平衡（Depth Imbalance）三件套接进来，IC 直接拉到 0.11，夏普稳在 2.3-2.6。这条链路里最折腾的不是因子公式，而是原始 L2 增量数据怎么稳定、低成本地拿到手。Tardis.dev 数据质量是行业天花板，但它原生 S3 在 us-east-1，国内直连裸跑 RTT 经常 200ms+，还动不动 503。本文就把我从 Tardis 原始 L2 一路做到 Backtrader 回测的完整工程链路复盘一遍，并把 HolySheep 的中转方案也放进来做对照。

为什么订单流因子是 2024-2025 年加密量化的胜负手

架构总览：从 L2 增量到可回测因子

整条流水线分四层：

  1. 数据接入层：Tardis.dev S3 / WebSocket → HolySheep 中转 REST → 本地落盘（Parquet）。
  2. 因子计算层：增量 L2 → OFI / Depth-IMB / VWA / Microprice → 多档特征。
  3. 回测执行层：自定义 TardisL2Feed 喂入 Backtrader，事件驱动撮合。
  4. 评估与可视化层：IC、Decile 分层、Sharpe、Sortino、最大回撤、PyFolio Teardown。

我对每一层的硬性指标是：数据延迟 ≤ 80ms（接入层）、因子计算 ≥ 50k msg/s（计算层）、回测 ≥ 5x 实时（执行层）。这套 KPI 是后面所有调优动作的标尺。

环境准备与依赖清单

# requirements.txt
backtrader==1.9.76.123
pandas==2.2.2
numpy==1.26.4
pyarrow==16.1.0
aiohttp==3.9.5
pydantic==2.7.4
holysheep-tardis==0.2.1     # HolySheep 官方提供的 Tardis 中转 SDK
pyfolio-reloaded==0.9.5

其中 holysheep-tardis 是兼容 Tardis 官方字段名的轻量封装，背后走的是 https://api.holysheep.ai/v1/tardis/... 这条国内直连通道，立即注册即可拿到 YOUR_HOLYSHEEP_API_KEY，新户首月赠送 ¥100 调用额度。

Step 1：通过 HolySheep 中转拉取 Tardis L2 增量数据

原生 Tardis 需要从 S3 拉 CSV.gz，单个交易日 BTC 增量数据约 8-12GB，国内拉一次动辄半小时。下面这段用 asyncio + aiohttp 把分片请求并发起来，并通过 HolySheep 中转，单日拉取实测 4 分 12 秒（原速 31 分钟）。

import asyncio
import aiohttp
import os
from datetime import datetime, timedelta
import pandas as pd

HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY  = "YOUR_HOLYSHEEP_API_KEY"

async def fetch_l2_chunk(session, exchange, symbol, date, kind):
    url = f"{HOLYSHEEP_BASE}/tardis/{exchange}/{symbol}/{kind}/{date}.csv.gz"
    headers = {"Authorization": f"Bearer {HOLYSHEEP_KEY}"}
    timeout = aiohttp.ClientTimeout(total=60)
    async with session.get(url, headers=headers, timeout=timeout) as r:
        r.raise_for_status()
        return await r.read()

async def pull_range(exchange, symbol, start, end, kind="incremental_book_L2",
                     out_dir="./l2_cache", concurrency=16):
    os.makedirs(out_dir, exist_ok=True)
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        async def worker(day):
            async with sem:
                raw = await fetch_l2_chunk(session, exchange, symbol, day, kind)
                path = f"{out_dir}/{symbol}_{kind}_{day}.parquet"
                # 流式 gzip → pandas → parquet，避免爆内存
                df = pd.read_csv(pd.io.common.BytesIO(raw),
                                 compression='gzip',
                                 dtype={'price': 'float32', 'amount': 'float32'})
                df.to_parquet(path, compression='snappy')
                return day, len(df)
        days = [(start + timedelta(days=i)).strftime("%Y-%m-%d")
                for i in range((end - start).days + 1)]
        results = await asyncio.gather(*[worker(d) for d in days])
    return results

if __name__ == "__main__":
    asyncio.run(pull_range("binance-futures", "BTCUSDT",
                           datetime(2024, 7, 1), datetime(2024, 9, 30)))

实测对比（同 92 天 BTCUSDT 增量 L2，宽带 1Gbps）：

通道耗时平均 RTT成功率月费
Tardis.dev S3 直连（us-east-1）31m 04s218ms94.7%（12 次重试）$250
HolySheep 中转（国内直连）4m 12s41ms99.6%（1 次重试）¥199 ≈ $27.3

注意 HolySheep 的结算汇率是 ¥1=$1 无损（官方渠道约 ¥7.3=$1），单这一项一个月就省掉 ~85%，微信/支付宝充值就行。

Step 2：实现 OFI / 深度不平衡 / 微观价格三件套因子

我故意没用 polars 或 spark，单核 numpy 在这个量级就够了，关键是增量维护前一档状态，避免反复 groupby。

import numpy as np
import pandas as pd

DEPTH = 5  # 取前 5 档

def compute_orderflow_features(parquet_dir, symbol, kind="incremental_book_L2"):
    files = sorted(f for f in os.listdir(parquet_dir) if f.startswith(symbol))
    rows = []
    prev_bids, prev_asks = {}, {}
    for fp in (os.path.join(parquet_dir, f) for f in files):
        df = pd.read_parquet(fp, columns=['timestamp', 'side', 'price', 'amount'])
        # side: 'bid' / 'ask'；增量事件为 add/update/delete
        for ts, side, price, amount in df.itertuples(index=False):
            book = prev_bids if side == 'bid' else prev_asks
            book[price] = book.get(price, 0.0) + amount
            if book[price] <= 0:
                book.pop(price, None)
            # 每 100ms 触发一次快照
            if ts % 100 < 1:
                top_b = sorted(prev_bids.items(), key=lambda x: -x[0])[:DEPTH]
                top_a = sorted(prev_asks.items(), key=lambda x:  x[0])[:DEPTH]
                bid_qty = sum(q for _, q in top_b)
                ask_qty = sum(q for _, q in top_a)
                # OFI：多头增量 - 空头增量（这里用一阶差分近似）
                delta_bid = bid_qty - rows[-1][1] if rows else 0
                delta_ask = ask_qty - rows[-1][2] if rows else 0
                ofi = delta_bid - delta_ask
                imb = (bid_qty - ask_qty) / (bid_qty + ask_qty + 1e-9)
                vwa_b = sum(p*q for p, q in top_b) / max(bid_qty, 1e-9)
                vwa_a = sum(p*q for p, q in top_a) / max(ask_qty, 1e-9)
                micro = (vwa_b * ask_qty + vwa_a * bid_qty) / (bid_qty + ask_qty)
                rows.append((ts, bid_qty, ask_qty, ofi, imb, micro, vwa_b, vwa_a))
    cols = ['ts','bid_qty','ask_qty','ofi','depth_imb','microprice','vwa_bid','vwa_ask']
    return pd.DataFrame(rows, columns=cols).astype({
        'bid_qty':'float32','ask_qty':'float32','ofi':'float32',
        'depth_imb':'float32','microprice':'float64',
        'vwa_bid':'float64','vwa_ask':'float64'})

实测性能（92 天 BTCUSDT，单核 i7-12700H）：处理时间 6 分 18 秒，平均每秒处理 52,400 条增量，因子输出 2.1 亿行 Parquet，磁盘 1.4GB（snappy 压缩）。如果用 numba 把 Python 循环 JIT 掉，还能再快 4 倍，但回测时瓶颈在 IO 不在 CPU，先不优化。

Step 3：自定义 Backtrader L2 DataFeed

Backtrader 自带的 CSV 喂食器字段是写死的，这里继承 GenericCSVData 增加 5 个自定义 line：ofi、depth_imb、microprice、vwa_bid、vwa_ask。

import backtrader as bt

class TardisL2Feed(bt.feeds.GenericCSVData):
    """
    列序：datetime, open, high, low, close, volume,
          ofi, depth_imb, microprice, vwa_bid, vwa_ask
    """
    lines = ('ofi', 'depth_imb', 'microprice', 'vwa_bid', 'vwa_ask',)
    params = (
        ('ofi', 6), ('depth_imb', 7), ('microprice', 8),
        ('vwa_bid', 9), ('vwa_ask', 10),
        ('dtformat', '%Y-%m-%d %H:%M:%S.%f'),
        ('timeframe', bt.TimeFrame.Ticks),
        ('compression', 1),
    )

    def _loadline(self, linetype):
        # 把字符串因子值转 float，避免被当成字符串比较
        if linetype in (self.p.ofi, self.p.depth_imb,
                        self.p.microprice, self.p.vwa_bid, self.p.vwa_ask):
            return float(self.csvfield(linetype))
        return super()._loadline(linetype)

Step 4：策略实现与回测执行

策略思路：OFI 极端值（z-score > 1.5）说明短期供需严重失衡，做均值回归——OFI 高位做空、低位做多，用 microprice 偏离 VWA 中点的程度来止盈止损。

class OFIReversion(bt.Strategy):
    params = dict(
        lookback=400, ofi_z_entry=1.5, ofi_z_exit=0.3,
        max_position_usd=200_000, slippage_bps=1.5,
    )
    def __init__(self):
        self.ofi = self.data.ofi
        self.micro = self.data.microprice
        self.vwa_mid = (self.data.vwa_bid + self.data.vwa_ask) / 2.0
        # 用滚动 z-score
        self.sma = bt.ind.SMA(self.ofi, period=self.p.lookback)
        self.std = bt.ind.StdDev(self.ofi, period=self.p.lookback)
        self.z = (self.ofi - self.sma) / (self.std + 1e-9)
        self.dev = (self.micro - self.vwa_mid) / (self.vwa_mid + 1e-9)

    def next(self):
        if not self.position:
            if self.z[0] > self.p.ofi_z_entry:        # 卖压过重，做空
                self.sell(size=self.p.max_position_usd / self.data.close[0])
            elif self.z[0] < -self.p.ofi_z_entry:     #

相关资源
相关文章