自適應下鑽：從分鐘到原始交易的可變粒度回測

分鐘級K線是回測的標準粒度。但在一根分鐘K線內，價格的波動幅度各不相同：有時僅為0.01%，有時卻達到2%。當止損和止盈同時落在一根分鐘K線的[最低價, 最高價]範圍內時，回測無法知道哪個先被觸發。這就是成交歧義（fill ambiguity）問題。

樸素的解決方案是對整個回測使用秒級資料。但兩年的資料意味著約6300萬根秒級K線，而非約100萬根分鐘K線。儲存空間增加60倍，速度也按比例下降。

自適應下鑽解決了這個問題：僅在真正需要的地方使用更細粒度。

問題：大K線上的成交歧義

考慮一個具體場景。策略以3000 USDT開多。止損：2970（-1%）。止盈：3060（+2%）。

14:37的分鐘K線：

• 開盤價：3010
• 最高價：3065
• 最低價：2965
• 收盤價：3050

止損（2970）和止盈（3060）都落在[2965, 3065]範圍內。哪個先被觸發？

可能的結果：

• 價格先下行 → 觸發止損 → 虧損 -1%
• 價格先上行 → 觸發止盈 → 盈利 +2%

單筆交易的差異：3個百分點。若使用10倍槓桿則為30%。對於包含數百筆交易的回測，錯誤的成交歧義解析會系統性地扭曲結果。

框架的預設處理方式

大多數回測引擎使用以下兩種啟發式方法之一：

1. 樂觀式： 止盈先觸發 → 結果偏高
2. 悲觀式： 止損先觸發 → 結果偏低

兩種方法都是猜測。真實資料在秒級甚至毫秒級是可獲取的，既然可以檢視真實資料，就沒有理由去猜測。

下鑽：四級策略

下鑽的思路：從分鐘級開始，僅在存在歧義時"下鑽"到更低級別——基於價格波動或成交量異常。

第1级：1m（分钟K线）
  → 如果止损或止盈明确在[最低价, 最高价]范围之外——当场解决
  → 如果两者都在范围内——下钻 ↓

第2级：1s（秒级K线）
  → 加载该分钟的60根秒级K线
  → 逐秒遍历：哪个先被触发？
  → 如果秒级K线有歧义，或 price_move >= min_pct，或 volume >= median_1s * vol_mult——下钻 ↓

第3级：100ms（毫秒级K线）
  → 加载该秒最多10根100ms的K线
  → 逐100ms遍历
  → 如果100ms K线有歧义，或 price_move >= min_pct，或 volume >= median_100ms * vol_mult——下钻 ↓

第4级：原始交易（raw trades）
  → 加载该100ms桶中的单笔交易
  → 逐笔解析成交——最大可能的精度

何時不需要下鑽

95%的情況下不需要下鑽。典型場景：

明確的止損： K線最高價未達到止盈，最低價擊穿止損 → 止損觸發，無需下鑽。

明確的止盈： 最低價未達到止損，最高價突破止盈 → 止盈觸發，無需下鑽。

都未觸發： 兩個價位都在範圍之外 → 持倉繼續。

跳空檢測： 下一根K線的開盤價跳過止損或止盈 → 按開盤價成交，無需下鑽。

下鑽僅在約5%的K線中需要——即兩個價位都落在單根K線範圍內時。

class AdaptiveFillSimulator:
    """
    四级下钻，用于确定成交顺序。
    """
    def __init__(self, data_loader):
        self.loader = data_loader
        self.cache_1s = {}  # 按月缓存的秒级数据

    def check_fill(self, timestamp, candle_1m, sl_price, tp_price, side):
        """
        检查给定分钟K线上是否触发了止损或止盈。

        Returns: ('sl', fill_price) | ('tp', fill_price) | None
        """
        low, high = candle_1m['low'], candle_1m['high']

        open_price = candle_1m['open']
        if side == 'long':
            if open_price <= sl_price:
                return ('sl', open_price)
            if open_price >= tp_price:
                return ('tp', open_price)
        else:
            if open_price >= sl_price:
                return ('sl', open_price)
            if open_price <= tp_price:
                return ('tp', open_price)

        sl_hit = self._level_hit(sl_price, low, high, side, 'sl')
        tp_hit = self._level_hit(tp_price, low, high, side, 'tp')

        if sl_hit and not tp_hit:
            return ('sl', sl_price)
        if tp_hit and not sl_hit:
            return ('tp', tp_price)
        if not sl_hit and not tp_hit:
            return None

        return self._drill_down_1s(timestamp, sl_price, tp_price, side)

    def _drill_down_1s(self, minute_ts, sl_price, tp_price, side):
        """第2级：逐秒遍历。"""
        bars_1s = self.loader.load_1s_for_minute(minute_ts)

        if bars_1s is None or len(bars_1s) == 0:
            return self._pessimistic_fill(side, sl_price, tp_price)

        for bar in bars_1s:
            sl_hit = self._level_hit(sl_price, bar['low'], bar['high'], side, 'sl')
            tp_hit = self._level_hit(tp_price, bar['low'], bar['high'], side, 'tp')

            if sl_hit and not tp_hit:
                return ('sl', sl_price)
            if tp_hit and not sl_hit:
                return ('tp', tp_price)
            if sl_hit and tp_hit:
                result = self._drill_down_100ms(bar['timestamp'], sl_price, tp_price, side)
                if result:
                    return result

        return self._pessimistic_fill(side, sl_price, tp_price)

    def _pessimistic_fill(self, side, sl_price, tp_price):
        """悲观假设：多头触发止损，空头触发止损。"""
        if side == 'long':
            return ('sl', sl_price)
        else:
            return ('sl', sl_price)

效能

在兩年約400筆交易的回測中，下鑽大約被呼叫20次。總開銷——整個回測不到1秒。

自適應資料儲存

下鑽需要秒級和毫秒級資料。但以最大粒度儲存所有資料是不切實際的：

兩年的完整1秒存檔約13 GB。100毫秒超過100 GB。全部儲存是可以的，但考慮到下鑽使用的資料不到1%，這是浪費的。

熱點秒檢測

關鍵觀察：價格顯著波動的秒數只佔很小的比例。如果某秒內價格變化不到0.1%——就沒有必要儲存該秒的100ms細分資料。

熱點秒檢測：在下載和處理資料時，分析每一秒並僅為"熱點秒"生成100ms K線——即價格波動超過閾值的秒。

def process_trades_adaptive(
    trades: pd.DataFrame,
    min_price_change_pct: float = 1.0,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    将原始交易数据处理为自适应结构：
    - 所有秒的1秒K线
    - 仅"热点秒"的100ms K线

    Args:
        trades: 包含 [timestamp, price, quantity] 列的 DataFrame
        min_price_change_pct: 下钻到100ms的阈值

    Returns:
        (df_1s, df_100ms_hot) — 秒级K线和热点秒的100ms K线
    """
    trades['second'] = trades['timestamp'].dt.floor('1s')
    df_1s = trades.groupby('second').agg(
        open=('price', 'first'),
        high=('price', 'max'),
        low=('price', 'min'),
        close=('price', 'last'),
        volume=('quantity', 'sum'),
    )

    df_1s['price_change_pct'] = (df_1s['high'] - df_1s['low']) / df_1s['open'] * 100
    hot_seconds = df_1s[df_1s['price_change_pct'] >= min_price_change_pct].index

    hot_trades = trades[trades['second'].isin(hot_seconds)]
    hot_trades['bucket_100ms'] = hot_trades['timestamp'].dt.floor('100ms')

    df_100ms = hot_trades.groupby('bucket_100ms').agg(
        open=('price', 'first'),
        high=('price', 'max'),
        low=('price', 'min'),
        close=('price', 'last'),
        volume=('quantity', 'sum'),
    )

    return df_1s, df_100ms

儲存節省

以 ETHUSDT 典型月份為例：

當閾值 min_price_change_pct = 1.0% 時，熱點秒佔所有秒的不到1%。它們的100ms資料僅在550 MB秒級資料基礎上增加約50 MB——幾乎可以忽略不計。

如果秒級資料也採用自適應儲存（僅當分鐘內波動超過0.1%時），儲存量還可再減少3-5倍。

Parquet 儲存結構

data/{SYMBOL}/
├── source.json                # 数据来源：{"exchange": "binance"} 或 {"exchange": "bybit"}
├── stats.json                 # 预计算的成交量中位数：{"median_volume_1s": ..., "median_volume_100ms": ...}
├── klines_1m/
│   ├── 2024-01.parquet       # ~1 MB
│   ├── 2024-02.parquet
│   └── ...
├── klines_1s/
│   ├── 2024-01.parquet       # ~550 MB
│   └── ...
├── klines_100ms_hot/
│   ├── 2024-01.parquet       # ~50 MB（仅热点秒）
│   └── ...
├── trades_hot/
│   ├── 2024-01.parquet       # 热点100ms桶的原始交易
│   └── ...
└── states_1m.parquet          # 预计算的滚动状态缓存（~112 MB）

每個檔案包含一個月的資料。秒級、毫秒級資料和原始交易採用延遲載入——僅在下鑽請求時才載入。stats.json 檔案包含預計算的成交量中位數，用於基於成交量的下鑽觸發。

針對金融資料的 Parquet 最佳化

金融資料有其特殊性：時間戳單調遞增，價格變化平滑，成交量波動較大。最優配置：

import pyarrow as pa
import pyarrow.parquet as pq

schema = pa.schema([
    pa.field("timestamp", pa.int32()),    # 从 epoch 起的秒数——int32 足够
    pa.field("open",      pa.float32()),
    pa.field("high",      pa.float32()),
    pa.field("low",       pa.float32()),
    pa.field("close",     pa.float32()),
    pa.field("volume",    pa.float32()),
])

column_encodings = {
    "timestamp": "DELTA_BINARY_PACKED",   # 单调整数 → 差分压缩
    "open":      "BYTE_STREAM_SPLIT",     # 浮点数 → 字节流分割
    "high":      "BYTE_STREAM_SPLIT",
    "low":       "BYTE_STREAM_SPLIT",
    "close":     "BYTE_STREAM_SPLIT",
    "volume":    "BYTE_STREAM_SPLIT",
}

def save_optimized_parquet(df, path):
    table = pa.Table.from_pandas(df, schema=schema)
    pq.write_table(
        table, path,
        compression="zstd",
        compression_level=9,
        use_dictionary=False,
        write_statistics=False,
        column_encoding=column_encodings,
    )

為什麼使用這些配置：

• DELTA_BINARY_PACKED 用於時間戳：連續的時間戳之間差值固定（1m為60，1s為1）。差分編碼將其壓縮到接近零。
• BYTE_STREAM_SPLIT 用於浮點數：將 float32 的位元組分流（所有第一位元組在一起，所有第二位元組在一起，依此類推）。對於平滑變化的價格，比標準編碼壓縮率高2-3倍。
• ZSTD level 9：在可接受的解壓速度下實現良好的壓縮。
• float32 代替 float64：對於價格和成交量足夠，節省50%記憶體。

帶快取的延遲載入

下鑽請求特定分鐘的秒級資料。每次請求都載入一個 parquet 檔案太慢。解決方案——按月進行 LRU 快取的延遲載入。

from functools import lru_cache
import pyarrow.parquet as pq
import pandas as pd

class AdaptiveDataLoader:
    """
    带缓存的延迟加载器：按月加载秒级数据，
    在内存中保留最近 N 个月。
    """
    def __init__(self, symbol: str, data_dir: str = "data", cache_months: int = 2):
        self.symbol = symbol
        self.data_dir = data_dir
        self.cache_months = cache_months
        self._cache_1s: dict[str, pd.DataFrame] = {}

    def load_1s_for_minute(self, minute_ts: pd.Timestamp) -> pd.DataFrame | None:
        """加载特定分钟的1秒数据。"""
        month_key = minute_ts.strftime("%Y-%m")

        if month_key not in self._cache_1s:
            self._load_month_1s(month_key)

        if month_key not in self._cache_1s:
            return None

        df = self._cache_1s[month_key]
        minute_start = minute_ts.floor('1min')
        minute_end = minute_start + pd.Timedelta(minutes=1)

        return df[(df.index >= minute_start) & (df.index < minute_end)]

    def load_100ms_for_second(self, second_ts: pd.Timestamp) -> pd.DataFrame | None:
        """加载热点秒的100ms数据。"""
        month_key = second_ts.strftime("%Y-%m")
        path = f"{self.data_dir}/{self.symbol}/klines_100ms_hot/{month_key}.parquet"

        try:
            df = pd.read_parquet(path)
            second_start = second_ts.floor('1s')
            second_end = second_start + pd.Timedelta(seconds=1)
            return df[(df.index >= second_start) & (df.index < second_end)]
        except FileNotFoundError:
            return None

    def _load_month_1s(self, month_key: str):
        """加载一个月的1秒数据，从缓存中淘汰旧数据。"""
        path = f"{self.data_dir}/{self.symbol}/klines_1s/{month_key}.parquet"
        try:
            df = pd.read_parquet(path)
            df.index = pd.to_datetime(df['timestamp'], unit='s')

            if len(self._cache_1s) >= self.cache_months:
                oldest = min(self._cache_1s.keys())
                del self._cache_1s[oldest]

            self._cache_1s[month_key] = df
        except FileNotFoundError:
            pass

將下鑽應用於回測

整合到回測迴圈中：

def backtest_with_adaptive_fill(
    states: pd.DataFrame,
    strategy_params: dict,
    data_loader: AdaptiveDataLoader,
) -> list:
    """
    使用自适应下钻进行成交模拟的回测。
    """
    fill_sim = AdaptiveFillSimulator(data_loader)
    trades = []
    position = None

    for i in range(len(states)):
        row = states.iloc[i]
        ts = states.index[i]

        candle_1m = {
            'open': row['open'], 'high': row['high'],
            'low': row['low'], 'close': row['close'],
            'timestamp': ts,
        }

        if position is not None:
            fill = fill_sim.check_fill(
                ts, candle_1m,
                position['sl'], position['tp'],
                position['side'],
            )

            if fill is not None:
                fill_type, fill_price = fill
                trades.append({
                    'entry_time': position['entry_time'],
                    'exit_time': ts,
                    'side': position['side'],
                    'entry_price': position['entry_price'],
                    'exit_price': fill_price,
                    'exit_type': fill_type,
                    'drill_down': fill_sim.last_drill_depth,  # 0、1 或 2
                })
                position = None
                continue

        signal = check_entry_signal(row, strategy_params)
        if signal and position is None:
            position = {
                'side': signal['side'],
                'entry_price': row['close'],
                'entry_time': ts,
                'sl': signal['sl'],
                'tp': signal['tp'],
            }

    return trades

與滾動狀態快取的關係

下鑽與聚合 parquet 快取互為補充——它們解決不同的問題：

兩種方法都消除了在日線級別不可見但對真實回測至關重要的錯誤。

總結：成交模擬方法對比

下鑽以1分鐘回測的速度提供了完整1秒回測的精度。關鍵觀察：高粒度並非處處需要——只在決策點需要。

基於成交量的下鑽

原始的下鑽僅基於價格波動觸發——當K線的[最低價, 最高價]範圍足夠寬以產生成交歧義時。但價格並不是某個K線內發生重要事件的唯一訊號。

成交量飆升是同樣重要的觸發條件。當某秒的成交量達到中位數的500倍時，通常對應著大額市價單、連環爆倉或閃崩。即使K線實體看起來很小，該秒內的實際價格軌跡可能非常劇烈——觸及OHLC表示法所隱藏的極值。

下鑽條件現在基於或邏輯：顯著的價格波動或異常的成交量飆升都會觸發向更細粒度的下鑽。

def is_hot(bar, median_volume, min_pct=0.1, vol_mult=500):
    """
    判断一根K线是否需要下钻到下一级。
    两个独立触发条件（或逻辑）：
      - K线内价格波动 >= min_pct
      - 成交量超过 median * vol_mult
    """
    price_move = (bar['high'] - bar['low']) / bar['open'] * 100
    return price_move >= min_pct or bar['volume'] >= median_volume * vol_mult

這捕獲了僅靠價格檢測無法發現的場景：一根open=3000、close=3001但成交量是正常值50000倍的K線，可能在毫秒內曾觸及2950和3050。沒有基於成交量的下鑽，回測永遠不會更仔細地檢查這一秒。

原始交易：第四級

原始的三級層次（1m → 1s → 100ms）仍然留有空白：在單個100ms桶內，多筆交易可能以不同價格成交。對於high=3060和low=2965的桶，我們仍然不知道確切的順序。

解決方案：下鑽到原始交易作為第四級也是最終級別。

1m K线（基础）
  └─> 1s K线      （当 price_move >= min_pct 或 volume >= median_1s * vol_mult）
      └─> 100ms K线    （当检测到热点秒）
          └─> 原始交易     （当100ms显示 price_move >= min_pct 或 volume >= median_100ms * vol_mult）

在原始交易級別，沒有歧義——每筆交易都有精確的價格和時間戳。成交被最終確定：

def resolve_from_trades(trades, sl_price, tp_price, side):
    """
    按时间顺序遍历单笔交易。
    第一笔穿越止损或止盈的交易决定成交。
    """
    for trade in trades:
        price = trade['price']
        if side == 'long':
            if price <= sl_price:
                return ('sl', price)
            if price >= tp_price:
                return ('tp', price)
        else:  # short
            if price >= sl_price:
                return ('sl', price)
            if price <= tp_price:
                return ('tp', price)
    return None

原始交易級別極少被呼叫——不到所有K線的0.1%——但當被呼叫時，它提供了任何K線近似都無法匹配的真實資料。

每個轉換的獨立閾值

不同解析度之間的轉換具有不同的特徵。1秒內0.1%的價格波動是顯著的；100ms桶內同樣的0.1%則是極端的。同樣，成交量分佈在每個時間尺度上也不同。

每個級別轉換現在有自己的 min_pct 和 vol_mult 參數：

1s → 100ms:     --min-pct-1s 0.1     --vol-mult-1s 500
100ms → trades:  --min-pct-100ms 0.1  --vol-mult-100ms 500

這允許獨立地微調每個轉換的靈敏度。實際上，100ms到交易的轉換可以使用更嚴格的閾值，因為載入單個100ms桶的原始交易的成本很小。

@dataclass
class DrillDownConfig:
    min_pct_1s: float = 0.1
    vol_mult_1s: float = 500
    min_pct_100ms: float = 0.1
    vol_mult_100ms: float = 500

持久化中位數統計

基於成交量的下鑽需要知道每個時間尺度的成交量中位數。為每次回測即時計算中位數會抵消效能優勢。解決方案：預計算一次中位數並快取。

對於每個交易對，1秒和100ms粒度的成交量中位數從歷史資料中計算並存儲在 stats.json 檔案中：

{
  "ETHUSDT": {
    "median_volume_1s": 12.5,
    "median_volume_100ms": 1.8
  },
  "BTCUSDT": {
    "median_volume_1s": 0.45,
    "median_volume_100ms": 0.06
  }
}

統計資料在首次下載資料時為每個交易對計算一次，並在所有後續回測中複用。當資料更新（下載新月份）時，統計資料增量重新計算。

def compute_median_stats(symbol, data_dir):
    """为交易对计算并缓存成交量中位数统计。"""
    stats_path = f"{data_dir}/{symbol}/stats.json"

    all_1s = load_all_months(f"{data_dir}/{symbol}/klines_1s/")
    median_1s = all_1s['volume'].median()

    all_100ms = load_all_months(f"{data_dir}/{symbol}/klines_100ms_hot/")
    median_100ms = all_100ms['volume'].median()

    stats = {
        "median_volume_1s": float(median_1s),
        "median_volume_100ms": float(median_100ms),
    }

    with open(stats_path, 'w') as f:
        json.dump(stats, f, indent=2)

    return stats

多交易所支援：Bybit

並非所有交易對都在Binance上可用。對於XAUTUSDT（黃金）等資產，資料必須來自其他交易所。下鑽系統現在支援 Bybit 作為替代資料來源。

對於Bybit交易對，所有K線級別（1m、1s、100ms）和原始交易都從Bybit的原始交易流構建。過程相同——原始交易在每個時間尺度上聚合為K線——但資料來源不同。

data/{SYMBOL}/
├── source.json              # {"exchange": "bybit"} 或 {"exchange": "binance"}
├── klines_1m/
│   └── ...
├── klines_1s/
│   └── ...
├── klines_100ms_hot/
│   └── ...
└── trades_hot/              # 热点100ms桶的原始交易
    └── ...

資料載入器檢查 source.json 並使用相應的下載管道。從回測引擎的角度來看，無論資料來源是哪個交易所，資料格式都是相同的——下鑽邏輯與交易所無關。

這對於跨交易所策略或僅在特定平臺上交易的交易對尤為重要。

結論

自適應下鑽是一個簡單原則的應用：按資料重要性的比例投入計算資源和儲存空間。

四個粒度級別：

1. 1m — 95%K線的基礎遍歷
2. 1s — 成交歧義或成交量飆升時的下鑽
3. 100ms — 極端波動或異常成交量熱點秒的下鑽
4. 原始交易 — 熱點100ms桶的下鑽，在單筆交易級別解析成交

四個儲存級別：

1. 全部1m — 完整存檔，兩年約15 MB
2. 全部1s — 完整或自適應存檔，~550 MB/月
3. 僅熱點100ms — 不到1%的秒，~50 MB/月
4. 僅熱點交易 — 最極端100ms桶的原始交易

兩個下鑽觸發條件（或邏輯）：

• 基於價格：K線價格範圍超過 min_pct
• 基於成交量：K線成交量超過 median * vol_mult

結果：以分鐘級速度獲得逐筆模擬器的精度。儲存空間線性增長而非指數增長。並且支援多個交易所——Binance和Bybit——下鑽邏輯與交易所無關。

關於多時間框架策略的預計算快取，請參閱文章 聚合 Parquet 快取。關於資金費率在高槓杆下對結果的影響 — 資金費率正在摧毀你的槓桿。

參考連結

1. Apache Parquet — 資料儲存格式
2. Apache Arrow — BYTE_STREAM_SPLIT 編碼
3. Zstandard — 壓縮演算法
4. Lopez de Prado — Advances in Financial Machine Learning
5. Binance — 歷史市場資料

引用

@article{soloviov2026adaptivedrilldown,
  author = {Soloviov, Eugen},
  title = {Adaptive Drill-Down: Backtest with Variable Granularity from Minutes to Milliseconds},
  year = {2026},
  url = {https://marketmaker.cc/ru/blog/post/adaptive-resolution-drill-down-backtest},
  description = {自适应数据粒度如何加速回测并节省存储空间：仅在价格显著波动或成交量异常的位置从1分钟下钻到1秒、100毫秒和原始交易。}
}