在量化系統的開發過程中,行情數據通常被視為一項已經解決的基礎能力。接口穩定、字段齊全,策略就可以開始驗證。
但在一些實際項目中,經常會出現這樣的情況:
策略在回測或測試階段表現正常,上線運行後卻逐漸與預期產生偏差。排查邏輯、參數、執行模塊都沒有明顯問題,最後才發現,影響結果的並不是策略本身,而是策略所依賴的行情數據運行在不同的時間假設之上。
這個差異往往發生在系統架構層面,而不是策略代碼裏。
行情數據在系統中的兩種角色
從工程實現上看,行情數據大致可以分為兩類使用方式,但它們在系統中的職責並不相同。
一種是以推送流形式進入系統的數據。這類數據通常通過長連接持續到達,時間戳與真實市場狀態非常接近,更多地參與在線計算和狀態更新。
另一種是以輪詢或歷史快照形式獲取的數據。它們在獲取時已經與真實市場存在時間間隔,常被用於分析、展示或離線計算。
兩者在字段結構上可能高度相似,但在系統中承擔的角色完全不同。
決策時序帶來的結構性差異
將策略流程簡化,可以得到一個常見的結構:
行情數據 → 狀態計算 → 行為判斷
當行情數據與市場狀態基本同步時,系統中的狀態計算反映的是“當前環境”。
而當行情本身存在延遲時,系統實際上是在基於一個已經過去的狀態做判斷。
這並不是邏輯錯誤,而是系統所處的時間參考系發生了變化。
在回測中,這種差異往往不明顯,因為歷史數據天然是完整的,信號和結果在時間軸上總是能對齊。但在在線系統中,這種時間錯位會逐漸累積,最終表現為執行位置偏移、狀態判斷滯後等現象。
一個簡化的工程示例
以下代碼並不是一個完整策略,只用於説明數據到達方式的差異。
# 推送式行情示意(WebSocket)
async def on_market_data(message):
# 數據到達時,系統狀態會被立即更新
process_state(message)
# 輪詢式行情示意(HTTP)
def poll_market_data():
# 獲取的是某個時間點之前的快照
data = fetch_snapshot()
process_state(data)
從代碼結構上看,兩者差異不大,但它們對系統狀態的影響路徑並不一致。
前者更多參與實時狀態演化,後者更像是對歷史狀態的補充。
指標計算在延時條件下的變化
以常見的移動平均為例,指標本身並不複雜,但它依賴的價格序列是否包含“當前信息”,會直接影響信號出現的時間位置。
def moving_average(prices, window):
return sum(prices[-window:]) / window
如果輸入序列本身已經滯後,那麼指標計算得到的結果也只是對過去狀態的再次加工。這類偏差在短週期系統中更容易放大,在長週期中則相對不明顯。
問題並不在於指標是否正確,而在於指標運行的時間前提是否成立。
架構層面的一個實踐建議
在系統設計階段,將行情數據源抽象為接口,而不是直接綁定到某一種實現,通常是一個比較穩妥的做法。
class MarketDataSource:
def get_price(self, symbol):
raise NotImplementedError
策略模塊只依賴接口,而不關心數據是實時推送還是延時獲取。這樣在不同環境下切換數據源時,不需要修改策略邏輯本身,也更容易發現數據時序帶來的差異。
這種設計並不能消除時間差,但至少可以讓時間差顯性化。
一點工程層面的觀察
行情數據並不是中性的輸入。在交易系統中,它同時定義了策略所處的時間座標。
當策略運行在不同的時間假設之上,即使代碼完全一致,系統行為也可能產生明顯差異。這種差異往往不會通過調參自然消失,而需要在架構層面被正視。
很多回測與在線運行之間的不一致,其實都可以從這裏找到線索。