别让数据延迟吃掉你的Alpha：浅谈港股高频策略的行情接入

做量化的朋友都知道，策略回测是一回事，上实盘又是另一回事。特别是在港股市场，没有涨跌停限制且支持T+0，盘口的瞬间价差稍纵即逝。最近在复盘一个轮动策略时，我发现滑点（Slippage）异常严重，排查一圈后发现，锅竟然在数据源的接收方式上。

从“盲人摸象”到“全景监控” 很多初级策略在获取行情时，习惯用While True循环去请求数据。这种做法在低频策略里勉强够用，但一旦涉及到Tick级的套利，你拿到的永远是“过去”的价格。痛点非常直观：

信息滞后：当你收到价格并发出指令时，盘口早就变了，导致由于价格不匹配而撤单或成交在不利价位。
并发限制：大部分商业接口对HTTP请求频率都有严格限制，想通过高频轮询来模拟实时流，很容易被服务商封IP。

实战解决方案 要解决这个问题，必须从底层改变数据交互逻辑。我们在实盘部署中（以Alltick提供的行情服务为例），采用了双轨制的数据接入方案。

1. 基础数据的按需获取 对于开盘前的持仓核对、资金查询，或者低频的分钟级策略，我们依然沿用HTTP协议。这种方式稳健可靠，适合处理静态数据。

import requests

# API URL（请根据实际情况替换）
api_url = 'https://api.alltick.co/v1/stock/realtime'
# 设置请求头
headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'your_api_key'  # 替换为实际的API密钥
}

# 请求参数
params = {
    'symbols': '00700.HK,00005.HK',  # 港股股票代码
    'fields': 'price,volume,change'  # 获取实时价格、成交量、涨跌幅等信息
}

# 发送GET请求
response = requests.get(api_url, headers=headers, params=params)

# 输出结果
if response.status_code == 200:
    print(f"实时行情数据：{response.json()}")
else:
    print(f"请求失败，状态码：{response.status_code}")

2. 核心策略的事件驱动 对于捕捉Alpha的核心逻辑，必须上WebSocket。通过在Python中维护一个Socket客户端，策略引擎变成了“事件驱动”模式——数据来了就触发计算，没有数据就挂起等待。这不仅降低了CPU占用，更重要的是保证了行情的鲜度。

import json
import websocket
from loguru import logger

class WebSocketClient:
    def __init__(self, api_url):
        self.api_url = api_url
        self.ws = None

    def connect(self):
        """建立WebSocket连接"""
        self.ws = websocket.WebSocketApp(
            self.api_url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
        )
        self.ws.run_forever()

    def on_message(self, ws, message):
        """接收消息并处理"""
        try:
            data = json.loads(message)
            logger.info(f"实时数据：{data}")
        except Exception as e:
            logger.error(f"处理消息失败：{e}")

    def on_error(self, ws, error):
        """错误回调"""
        logger.error(f"WebSocket错误：{error}")

    def on_close(self, ws, close_status_code, close_msg):
        """关闭连接的回调"""
        logger.info(f"WebSocket连接关闭，状态码：{close_status_code}")

# 使用示例
if __name__ == "__main__":
    ws_url = "wss://api.alltick.co/realtime-data"
    client = WebSocketClient(ws_url)
    client.connect()

避坑指南 在实操中，大家一定要注意WebSocket的心跳维护（Heartbeat）。港股午休时段长，如果没有心跳机制，连接很容易意外断开，导致下午开盘“踏空”。

别让数据延迟吃掉你的Alpha：浅谈港股高频策略的行情接入

评论