回測與實盤一致性：為什麼你的機器人交易表現與回測不同

你用回測跑了一個策略。夏普比率 2.1，最大回撤 -8%，PnL +67%。你啟動了機器人。一個月後對比：相同的訊號，相同的時間段 — 但實盤 PnL 低了40%。回撤深了一倍半。十筆交易中有兩筆根本沒有執行。

這不是 bug。這是 回測-實盤差異（backtest-live divergence） — 回測結果與真實交易之間的系統性偏差。每個人都有這個問題。唯一的問題是你是否知道它的存在，以及你能否控制它。

本文提供差異的完整分類、最小化差異的架構模式，以及在生產環境中監控一致性的實用清單。

"回測中好用"綜合徵

每個演算法交易者都會經歷這個迴圈：

1. 在 Jupyter notebook 中編寫了策略
2. 在歷史 CSV 上運行了回測 — 結果很好
3. 將邏輯重寫為機器人（通常使用不同的語言或框架）
4. 啟動 — 結果不匹配
5. 開始找 bug，沒找到 — "市場變了"

問題不在於市場。問題在於回測和機器人是兩個不同的軟體產品，它們以不同的方式建模同一個現實。差異是不可避免的，但可以被系統化和最小化。

差異分類

所有差異來源分為四類。每一類都有嚴重程度評級（1到5）和對 PnL 差異的典型貢獻。

1. 資料差異（嚴重程度：3/5）

回測看到的資料和機器人即時看到的資料不是同一回事。

時間戳。 交易所以不同的規則為K線分配時間戳。一個交易所用週期開始時間標記K線，另一個用結束時間。REST API 可能在實際收盤後延遲1-3秒才返回K線。回測使用歷史檔案中的"理想"時間戳。

OHLCV 聚合。 歷史資料通常由資料提供商以不同於交易所即時聚合的方式進行聚合。差異在最後一位 — 但對於閾值訊號（MA 交叉、突破水平），這決定了策略是否建倉。

缺口和缺失資料。 歷史資料通常是乾淨的 — 缺失的K線通過插值填充。在即時環境中，WebSocket 可能斷開，機器人錯過30秒的資料。

對 PnL 差異的典型貢獻：年度 PnL 的 2-5%。

2. 執行差異（嚴重程度：5/5）

最危險的差異類別。回測完美模擬執行 — 現實遠非理想。

滑點。 回測以收盤價（或訊號價）成交訂單。在現實中，市價單以最佳買賣價加上滑點執行，滑點取決於交易量和流動性。對於中等流動性山寨幣上的$10K倉位，滑點可能為0.05-0.3%。

$N$ 筆交易的累積滑點公式：

$\text{Slippage}_{total} = \sum_{i=1}^{N} \text{size}_i \times s_i$

其中 $s_i$ 是第 $i$ 筆交易的滑點，取決於訂單簿深度：

$s_i \approx \frac{\text{size}_i}{\text{Liquidity}(t_i)} \times k$

延遲。 從訊號生成到訂單執行之間有時間：訊號計算（1-50 ms）、請求傳送（10-200 ms）、交易所撮合（1-10 ms）。在回測中，延遲 = 0。在實盤中 — 價格可能已經變動。

部分成交。 回測假設100%的訂單立即成交。在現實中，限價單可能部分成交 — 或者如果價格反轉則完全未成交。對於流動性差的市場上的市價單，訂單"滑過"訂單簿的多個價格層級。

佇列優先順序。 以最佳買價掛出的限價單不會立即成交 — 它排在該價位所有先前掛出的訂單之後。認為"價格觸及 = 訂單成交"的回測系統性地高估了成交率。

對 PnL 差異的典型貢獻：年度 PnL 的 10-30%。

3. 邏輯差異（嚴重程度：4/5）

這是回測和機器人之間策略程式碼本身的差異。

獨立的程式碼庫。 經典的反模式：backtests/strategy_a.py 和 bot/strategy_a.py — 兩個"做同樣事情"的獨立檔案。經過三個月的修改，它們不可避免地出現分歧。有人在回測中添加了過濾器，忘了在機器人中複製。或者相反 — 在機器人中修復了一個 bug，但回測中仍然存在。

不同的框架。 回測使用 pandas 的向量化操作，機器人使用 asyncio 的事件驅動邏輯。即使策略相同，邊界情況的處理也不同：舍入、條件檢查順序、NaN 處理。

狀態管理。 回測通常是無狀態的 — 遍歷資料陣列。機器人是有狀態的 — 儲存倉位、餘額、訂單歷史。機器人重啟、狀態丟失、與交易所不同步 — 所有這些都是差異的來源。

對 PnL 差異的典型貢獻：年度 PnL 的 5-20%。

4. 成本差異（嚴重程度：3/5）

交易成本建模中的差異。

資金費率。 大多數永續合約回測根本不考慮資金費率。在10倍槓桿和平均每8小時0.01%的費率下，這是 $0.01\% \times 3 \times 365 \times 10 = 109.5\%$ 每年 — 超過大多數策略的 PnL。詳細分析見文章資金費率扼殺你的槓桿。

佣金。 Maker/taker 佣金通常會被建模，但往往使用錯誤的費率。VIP 等級、BNB 折扣、返佣 — 所有這些都影響最終結果。

價差。 基於K線的回測看不到買賣價差。在分鐘K線上，收盤價 = 3000，但實際上買價 = 2999.5，賣價 = 3000.5。每筆交易"花費"半個價差。

對 PnL 差異的典型貢獻：年度 PnL 的 5-15%。

累積效應

所有四個類別同時作用，並且通常朝一個方向 — 對交易者不利：

$\text{PnL}_{live} \approx \text{PnL}_{backtest} - \Delta_{data} - \Delta_{execution} - \Delta_{logic} - \Delta_{costs}$

對於設計不完善的系統，回測 PnL 的20-50%總差異是正常的。使用槓桿時，效應會被放大。

實現一致性的架構模式

模式1：共享核心（提取公共核心）

思路：將策略核心 — 訊號生成和執行邏輯 — 提取到一個獨立模組中，供回測和機器人共同使用。只有周圍的基礎設施不同：資料來源和訂單提交機制。

┌─────────────────────────────────────┐
│         strategy_core.py            │
│  ┌─────────────┐ ┌───────────────┐  │
│  │ SignalEngine │ │ OrderManager  │  │
│  └──────┬──────┘ └──────┬────────┘  │
│         │               │           │
│    generate_signal()  create_order()│
└─────────┬───────────────┬───────────┘
          │               │
    ┌─────┴─────┐   ┌─────┴──────┐
    │  回测      │   │  实盘      │
    │ DataFeed   │   │ DataFeed   │
    │ FillModel  │   │ Exchange   │
    └────────────┘   └────────────┘

from dataclasses import dataclass
from typing import Optional
import numpy as np

@dataclass
class Signal:
    side: str          # 'long' | 'short'
    entry_price: float
    sl_price: float
    tp_price: float
    size: float
    timestamp: int

@dataclass
class OrderRequest:
    side: str
    order_type: str    # 'market' | 'limit'
    price: float
    size: float

class StrategyCore:
    """
    策略核心。回测和实盘使用相同的代码。
    只依赖数据，不依赖基础设施。
    """
    def __init__(self, params: dict):
        self.fast_period = params.get('fast_ma', 20)
        self.slow_period = params.get('slow_ma', 50)
        self.sl_pct = params.get('sl_pct', 0.02)
        self.tp_pct = params.get('tp_pct', 0.04)
        self.position: Optional[Signal] = None
        self._closes: list[float] = []

    def on_candle(self, timestamp: int, o: float, h: float,
                  l: float, c: float, v: float) -> Optional[OrderRequest]:
        """
        处理新K线。返回 OrderRequest 或 None。
        此方法在回测和机器人中以相同方式调用。
        """
        self._closes.append(c)

        if len(self._closes) < self.slow_period:
            return None

        fast_ma = np.mean(self._closes[-self.fast_period:])
        slow_ma = np.mean(self._closes[-self.slow_period:])

        if self.position is not None:
            exit_order = self._check_exit(h, l, c)
            if exit_order:
                self.position = None
                return exit_order

        if self.position is None:
            if fast_ma > slow_ma and self._prev_fast_ma <= self._prev_slow_ma:
                self.position = Signal(
                    side='long', entry_price=c,
                    sl_price=c * (1 - self.sl_pct),
                    tp_price=c * (1 + self.tp_pct),
                    size=1.0, timestamp=timestamp,
                )
                return OrderRequest('buy', 'market', c, 1.0)

        self._prev_fast_ma = fast_ma
        self._prev_slow_ma = slow_ma
        return None

    def _check_exit(self, high: float, low: float,
                    close: float) -> Optional[OrderRequest]:
        pos = self.position
        if pos.side == 'long':
            if low <= pos.sl_price:
                return OrderRequest('sell', 'market', pos.sl_price, pos.size)
            if high >= pos.tp_price:
                return OrderRequest('sell', 'market', pos.tp_price, pos.size)
        return None

現在回測和機器人使用同一個 StrategyCore：

from strategy_core import StrategyCore

def run_backtest(candles, params, fill_model):
    core = StrategyCore(params)
    trades = []

    for candle in candles:
        order = core.on_candle(
            candle['timestamp'], candle['open'], candle['high'],
            candle['low'], candle['close'], candle['volume'],
        )
        if order:
            fill_price = fill_model.simulate_fill(order, candle)
            trades.append({'price': fill_price, 'side': order.side})

    return trades

from strategy_core import StrategyCore

async def run_live(exchange, symbol, params):
    core = StrategyCore(params)

    async for candle in exchange.stream_candles(symbol, '1m'):
        order = core.on_candle(
            candle['timestamp'], candle['open'], candle['high'],
            candle['low'], candle['close'], candle['volume'],
        )
        if order:
            await exchange.place_order(symbol, order.side,
                                       order.order_type, order.size)

關鍵規則：StrategyCore 不知道資料來自哪裡，也不知道訂單傳送到哪裡。它接收 OHLCV 並返回 OrderRequest。其他一切都是基礎設施層的職責。

模式2：事件驅動統一（NautilusTrader 方法）

NautilusTrader 通過統一的 NautilusKernel 實現一致性 — 一個 Rust 原生引擎，具有確定性的事件驅動核心和納秒級解析度。同一策略實現在回測和實盤交易中都能工作。

架構基於埠和介面卡模式（六角架構）：

┌──────────────────────────────────┐
│        NautilusKernel            │
│  ┌───────────┐  ┌─────────────┐  │
│  │ Strategy   │  │ RiskEngine  │  │
│  │ (Python)   │  │ (Rust)      │  │
│  └─────┬─────┘  └──────┬──────┘  │
│        │               │         │
│  ┌─────┴───────────────┴──────┐  │
│  │      Message Bus (Rust)    │  │
│  └─────┬───────────────┬──────┘  │
└────────┼───────────────┼─────────┘
         │               │
   ┌─────┴─────┐   ┌─────┴──────┐
   │  回测      │   │  实盘      │
   │ Adapter    │   │ Adapter    │
   │ FillModel  │   │ Exchange   │
   │ (L2 book)  │   │ Gateway    │
   └────────────┘   └────────────┘

優勢：

• 確定性回放。 事件按嚴格定義的順序處理 — 回測結果可按位重現。
• 自定義 FillModel。 每次執行都使用 L2 訂單簿模擬 — 滑點基於真實訂單簿深度模擬。
• 效能。 每秒高達500萬行，可處理不適合 RAM 的資料。
• Redis + PostgreSQL。 通過 Redis 實現快取和訊息匯流排，通過 PostgreSQL 實現持久化 — 回測和實盤使用相同的基礎設施。

模式3：策略介面（Freqtrade 方法）

Freqtrade 使用統一的 IStrategy 介面：同一個策略類在回測和實盤中都能工作。唯一的區別是持久化層。

class IStrategy:
    """统一接口 — 实现不知道这是回测还是实盘。"""

    def populate_indicators(self, dataframe, metadata):
        """计算指标。"""
        dataframe['fast_ma'] = dataframe['close'].rolling(20).mean()
        dataframe['slow_ma'] = dataframe['close'].rolling(50).mean()
        return dataframe

    def populate_entry_trend(self, dataframe, metadata):
        """确定入场信号。"""
        dataframe.loc[
            (dataframe['fast_ma'] > dataframe['slow_ma']) &
            (dataframe['fast_ma'].shift(1) <= dataframe['slow_ma'].shift(1)),
            'enter_long'
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe, metadata):
        """确定出场信号。"""
        dataframe.loc[
            (dataframe['fast_ma'] < dataframe['slow_ma']),
            'exit_long'
        ] = 1
        return dataframe

Freqtrade 還提供：

• 通過 Optuna 進行超參數最佳化 — 策略參數最佳化
• --timeframe-detail — 深入更細的時間框架以最佳化成交（類似於自適應逐層細化）

模式對比

成交模擬精度

成交模擬是執行差異的主要來源。三個精度級別：

級別1：簡單模式（以收盤價成交）

fill_price = candle['close']

誤差： 不考慮滑點、價差、部分成交。系統性地高估 PnL。

級別2：滑點模型

def simulate_fill(order, candle, slippage_bps=5):
    """带滑点的成交。"""
    base_price = candle['close']
    slip = base_price * slippage_bps / 10000

    if order.side == 'buy':
        return base_price + slip  # 以更高价格买入
    else:
        return base_price - slip  # 以更低价格卖出

誤差： 固定滑點不考慮流動性和訂單大小。比簡單模式好，但仍然是粗糙的模型。

級別3：使用1秒/100毫秒資料的自適應逐層細化

最佳方案：使用真實的細粒度資料精確確定止損/止盈的成交順序。詳細描述見文章自適應逐層細化：可變粒度回測。

class RealisticFillModel:
    """
    组合成交模型：滑点 + 价差 + 交易量冲击。
    """
    def __init__(self, avg_spread_bps=3, impact_coeff=0.1):
        self.avg_spread_bps = avg_spread_bps
        self.impact_coeff = impact_coeff

    def simulate_fill(self, order, candle, order_size_usd):
        base_price = candle['close']

        spread_cost = base_price * self.avg_spread_bps / 20000

        candle_volume_usd = candle['volume'] * candle['close']
        participation_rate = order_size_usd / max(candle_volume_usd, 1)
        impact = base_price * self.impact_coeff * np.sqrt(participation_rate)

        if order.side == 'buy':
            return base_price + spread_cost + impact
        else:
            return base_price - spread_cost - impact

市場衝擊公式（簡化的 Almgren-Chriss 模型）：

$\Delta p = \sigma \cdot k \cdot \sqrt{\frac{V_{order}}{V_{market}}}$

其中 $\sigma$ 是波動率，$k$ 是衝擊係數，$V_{order}$ 是訂單量，$V_{market}$ 是該時間段的市場成交量。

實用一致性清單

在實盤啟動機器人之前，檢查每一項：

程式碼：

• 策略使用共享核心（回測和實盤使用同一個模組）
• 沒有在兩個地方重複訊號邏輯
• 單元測試驗證相同輸入時核心輸出相同
• 條件檢查順序相同（先止損還是先止盈？）

資料：

• 時間戳格式相同（UTC，相同的資料提供商）
• OHLCV 聚合使用相同的規則
• 缺失K線的處理方式相同
• 沒有前視偏差 — 回測不會窺視未來

執行：

• 滑點模型已根據真實資料校準
• 部分成交已建模（或至少進行了悲觀估計）
• 限價單有佇列優先順序模型
• 延遲已考慮（訊號到成交100-500 ms延遲）

成本：

• Maker/taker 佣金已按當前費率包含
• 永續合約的資金費率已考慮
• 價差已建模（至少是平均值）

基礎設施：

• 狀態持久化：機器人重啟後可恢復倉位
• 重連邏輯：WebSocket 斷線重連不丟失資料
• 日誌：所有訂單和成交都有記錄，用於事後分析

在生產環境中監控差異

一致性不是一次性檢查，而是一個持續的過程。啟動機器人後，必須即時追蹤差異。

影子模式（模擬交易）

在相同資料上並行執行機器人和回測。機器人生成訊號但不傳送訂單 — 只記錄日誌。同時回測處理相同的資料。比較：

class DivergenceMonitor:
    """
    实时比较回测和实盘机器人的信号。
    """
    def __init__(self, tolerance_pct=0.5):
        self.tolerance = tolerance_pct / 100
        self.divergences = []

    def compare_signal(self, backtest_signal, live_signal, timestamp):
        """比较回测和实盘信号。"""
        if backtest_signal is None and live_signal is None:
            return  # 两者都沉默 — OK

        if (backtest_signal is None) != (live_signal is None):
            self.divergences.append({
                'timestamp': timestamp,
                'type': 'signal_mismatch',
                'backtest': backtest_signal,
                'live': live_signal,
                'severity': 'HIGH',
            })
            return

        price_diff = abs(
            backtest_signal.entry_price - live_signal.entry_price
        ) / backtest_signal.entry_price

        if price_diff > self.tolerance:
            self.divergences.append({
                'timestamp': timestamp,
                'type': 'price_divergence',
                'diff_pct': price_diff * 100,
                'severity': 'MEDIUM',
            })

    def compare_fill(self, backtest_fill, live_fill, timestamp):
        """比较执行情况。"""
        if backtest_fill and live_fill:
            slippage = (live_fill['price'] - backtest_fill['price']
                        ) / backtest_fill['price']
            self.divergences.append({
                'timestamp': timestamp,
                'type': 'fill_divergence',
                'slippage_bps': slippage * 10000,
                'severity': 'LOW' if abs(slippage) < 0.001 else 'MEDIUM',
            })

    def report(self):
        """每周差异报告。"""
        from collections import Counter
        severity_counts = Counter(d['severity'] for d in self.divergences)
        return {
            'total_divergences': len(self.divergences),
            'by_severity': dict(severity_counts),
            'avg_slippage_bps': np.mean([
                d['slippage_bps'] for d in self.divergences
                if d['type'] == 'fill_divergence'
            ]) if any(d['type'] == 'fill_divergence'
                      for d in self.divergences) else 0,
        }

儀表板指標

滑點模型校準

積累2-4周的資料後，可以根據真實資料校準回測滑點模型：

def calibrate_slippage(live_fills: list[dict]) -> dict:
    """
    使用真实成交数据校准滑点模型。

    live_fills: [{'expected_price': ..., 'actual_price': ..., 'size_usd': ..., 'volume_usd': ...}]
    """
    slippages = []
    participation_rates = []

    for fill in live_fills:
        slip = abs(fill['actual_price'] - fill['expected_price']
                   ) / fill['expected_price']
        part = fill['size_usd'] / max(fill['volume_usd'], 1)
        slippages.append(slip)
        participation_rates.append(part)

    slippages = np.array(slippages)
    participation_rates = np.array(participation_rates)

    from scipy.optimize import curve_fit

    def model(x, k, base):
        return k * np.sqrt(x) + base

    popt, _ = curve_fit(model, participation_rates, slippages,
                        p0=[0.1, 0.0001])

    return {
        'impact_coeff': popt[0],
        'base_slippage': popt[1],
        'mean_slippage_bps': np.mean(slippages) * 10000,
        'p95_slippage_bps': np.percentile(slippages, 95) * 10000,
    }

與其他工具的關聯

回測-實盤一致性不是一個孤立的任務。它與"回測無幻覺"系列中的其他工具交叉：

• 自適應逐層細化 — 提高成交模擬精度，是執行一致性的關鍵元件。
• 資金費率 — 如果回測不模擬資金費率，在槓桿 > 3倍時一致性不可能實現。
• Parquet 快取 — 預計算的時間框架和指標確保回測看到與機器人相同的資料。RunningCandleBuffer 模擬 = 即時更新。
• Polars vs Pandas — 從 pandas（回測）切換到 Polars（實盤）時，需要確保數值結果一致。
• Walk-Forward — 樣本外資料的前進式最佳化展示策略如何退化 — 這比樣本內回測更接近實盤。

建議

1. 共享核心是必須的。 訊號生成使用單一程式碼庫是一致性的最低要求。兩個包含相同邏輯的檔案保證在一個月內出現差異。

2. 校準成交模型。 固定5 bps的滑點比沒有好。根據真實資料校準的滑點模型則好得多。

3. 前2-4周使用影子模式。 在訊號匹配率達到95%+之前不要用真錢交易。

4. 建模資金費率。 對於永續合約，這不是可選的 — 是必須的。資金費率在槓桿 > 5倍時可以吃掉所有 PnL。

5. 記錄一切。 每個訊號、每個訂單、每筆成交 — 帶時間戳。沒有日誌，事後分析就不可能。

6. 自動化比較。 DivergenceMonitor 的每週報告應自動到達。不要等到 PnL 變為負數。

7. 預設悲觀回測。 在回測中低估預期並在實盤中得到驚喜，比反過來好。滑點模型應該是保守的。

結論

回測-實盤一致性不是系統的屬性，而是一個過程。完美的一致性不存在：回測從定義上就是現實的模型，而模型總是簡化的。但"模型差異5%"和"模型差異50%"之間的區別取決於架構。

三個成熟度級別：

1. 基礎級。 共享核心、固定滑點、佣金。差異：10-20%。
2. 進階級。 事件驅動架構、自適應逐層細化、資金費率模型、影子模式。差異：5-10%。
3. 機構級。 L2 訂單簿模擬、校準的衝擊模型、即時差異監控。差異：2-5%。

你的任務是確定你處於哪個級別，並理解對於你的倉位規模和槓桿，什麼程度的差異是可接受的。

有用的連結

1. NautilusTrader — High-Performance Algorithmic Trading Platform
2. Freqtrade — Free, open source crypto trading bot
3. Almgren, R., Chriss, N. — Optimal Execution of Portfolio Transactions (2001)
4. Lopez de Prado — Advances in Financial Machine Learning, Chapter 12: Backtesting
5. Ernest Chan — Quantitative Trading: How to Build Your Own Algorithmic Trading Business
6. Hexagonal Architecture (Ports and Adapters) — Alistair Cockburn
7. Optuna — Hyperparameter Optimization Framework

引用

@article{soloviov2026backtestliveparity,
  author = {Soloviov, Eugen},
  title = {Backtest-live parity: why your bot trades differently from the backtest},
  year = {2026},
  url = {https://marketmaker.cc/ru/blog/post/backtest-live-parity},
  description = {回测与实盘交易之间差异的完整分类：从滑点和部分成交到代码库不同步。实现一致性的架构模式和生产环境监控清单。}
}