做量化最痛苦的时刻,不是策略没思路,而是回测收益率200%,实盘却在亏钱。最近在优化一个港股T+0策略时,我再次深刻体会到“数据颗粒度”对交易结果的毁灭性影响。
很多交易者习惯了券商提供的快照数据(Snapshot),往往是3秒甚至5秒刷新一次。这对于看日线的长线投资没问题,但对于量化策略,这几秒钟的真空期就是黑箱。港股的盘口变化极快,必须使用基于 WebSocket 的实时 Tick 数据流。
为什么轮询在实盘中行不通? 当你用 requests.get 去请求数据时,你不仅面临着网络RTT延迟,还时刻悬着一把“API限流”的达摩克利斯之剑。一旦请求过快被封IP,整个策略就瞎了。而长连接模式下,服务器有新成交就推给你,延迟通常控制在毫秒级。
在实战部署中,数据的“无感切换”也非常重要。很多团队在从A股拓展到港美股时,会发现代码要重写,因为数据格式变了。这时候,选择那些能够提供跨市场统一数据格式的源头(比如类似 AllTick API 这种将多市场数据结构化的服务)就显得很有前瞻性,它能确保你的策略逻辑不用因为换了个市场就推倒重来。
这就好比无论开法拉利还是拖拉机,方向盘的操作逻辑得是一样的。
分享一段我在实盘环境中使用的基础订阅脚本,剥离了复杂的风控逻辑,只保留最核心的行情获取部分:
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
if "data" in data:
tick = data["data"]
price = tick.get("last_price")
ts = tick.get("timestamp")
print(f"price={price}, time={ts}")
def on_open(ws):
subscribe_msg = {
"cmd": "subscribe",
"args": {
"symbol": "HKEX:HSI",
"type": "tick"
}
}
ws.send(json.dumps(subscribe_msg))
if __name__ == "__main__":
ws = websocket.WebSocketApp(
"wss://stream.alltick.co",
on_open=on_open,
on_message=on_message
)
ws.run_forever()
拿到了这些高频的 last_price 和 timestamp,你才能精准地计算盘口压力,判断资金流向。记住,在量化战场上,比别人快0.1秒看到价格,就是你的护城河。


