做港股量化，信噪比是决定策略成败的关键之一。我在构建实盘信号系统时，发现直接从WebSocket接入的原始成交数据里，混入了大量的非主动交易，如果不加甄别，因子计算和回测结果都会严重偏离。这里我就结合自己的实战经验，聊聊如何利用成交元数据，把自动对盘和碎股从真实成交中剥离出来。
量化场景下的痛点 量化模型对数据的纯净度要求很高。比如计算主动买入量、资金流向指标时，如果掺入了系统撮合产生的自动对盘，就会虚增交易活跃度，导致假信号。而碎股因为是零散交易，不反映主流资金意图，同样属于噪声。最直接的问题是：数据源通常不会直接给这些成交打上醒目的“噪声”标签，需要我们构建一套自动标记逻辑。手动处理不仅慢，而且无法应用于实盘级别的毫秒级决策。
关键判别字段 从WebSocket推送看，一笔成交记录核心字段如下：
| 字段 | 含义 |
|---|---|
| time | 成交时间 |
| price | 成交价格 |
| volume | 成交量 |
| trade_type | 类型标识（常不可靠） |
| match_id | 撮合编号（若有） |
实践中我提炼了三条量化规则：
- 成交量规则：港股整手多为100股，
volume模100非零或小于整手的，判定为碎股。 - 时间密集规则：对同一标的，如果在极短的时间窗口内出现多笔整手成交，且买卖方向高度对称，那大概率是系统自动对盘。
- 对手方规则：部分API会暴露买卖经纪商或账号信息。如果买方和卖方同时指向系统专用代码（例如“SYS”），直接归类为自动对盘。
实盘处理逻辑 我使用AllTick等提供的港股WebSocket数据，通过在线方式实现实时打标签。每笔成交推送到达时，立即按上述规则分类，并写入带有标签的时序队列，供下游因子计算和信号生成模块使用。代码示例如下：
from websocket import create_connection
import json
# 这里填你的 AllTick API Token
API_TOKEN = '你的API_TOKEN'
ws_url = f"wss://ws.alltick.co/stock?token={API_TOKEN}"
ws = create_connection(ws_url)
# 订阅港股 00700.HK 的成交数据
subscribe_msg = {
"action": "subscribe",
"symbol": "00700.HK",
"type": "transaction"
}
ws.send(json.dumps(subscribe_msg))
def check_auto_match(tick):
# 假设系统自动对盘的成交方为 "SYS"
return tick.get('buyer') == 'SYS' and tick.get('seller') == 'SYS'
while True:
data = ws.recv()
tick = json.loads(data)
volume = tick.get('volume', 0)
if volume < 100:
tick['tag'] = '碎股'
elif check_auto_match(tick):
tick['tag'] = '自动对盘'
else:
tick['tag'] = '普通成交'
print(tick['time'], tick['price'], tick['volume'], tick['tag'])
这个简单的处理逻辑，能让我在实盘中持续输出带标签的成交流，为后续策略提供高质量输入。
对分析工作的提升 有了清晰标签后，我的量化框架只用“普通成交”计算资金流向、量比和主动买卖差等指标，回测的夏普比率明显更稳健。同时，我能监测到自动对盘发生的频率，一旦系统对盘异常增多，可能预示市场深度不足，可作为风控因子之一。碎股的独立统计则帮助我评估零散投资者的行为，进一步丰富多因子体系。总之，从源头把数据洗干净，量化之路会顺畅很多。
