量化手记：A股Tick级数据实时清洗与策略对接实录

做过量化策略的朋友都知道，回测是一回事，实盘又是另一回事。很多策略在回测中表现优异，但一上实盘就因为“滑点”太大而失效。究其原因，往往是因为行情数据的获取速度不够快，导致信号触发滞后。

从盘口到信号 作为一名在一线摸爬滚打的数据分析师，我深知“快”字的价值。传统的 HTTP 轮询模式在行情剧烈波动时，往往会漏掉关键的成交明细。为了解决这个问题，我们需要通过 WebSocket 接入实时的 Tick 数据流。

数据流接入实战 我们需要的数据不仅仅是当前的最新价，更包括成交量、买卖盘口等深度信息。通过长连接订阅，我们可以像接自来水一样，源源不断地获取市场动态。这里以目前市面上数据质量较稳的 AllTick 接口格式为例，分享一下如何用 Python 搭建这一套即时行情接收系统。

策略端接入代码

import websocket
import json

# WebSocket 地址
ws_url = "wss://ws.alltick.co/realtime-stock"

# 订阅股票代码
stock_code = "SH600519"  # 贵州茅台

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        for item in data["data"]:
            print(f"股票: {item['s']}, 最新价: {item['p']}, 成交量: {item['v']}, 时间: {item['t']}")

def on_open(ws):
    sub_msg = json.dumps({"type": "subscribe", "symbol": stock_code})
    ws.send(sub_msg)
    print(f"已订阅 {stock_code} 实时行情")

def on_close(ws):
    print("连接已关闭")

ws = websocket.WebSocketApp(
    ws_url,
    on_open=on_open,
    on_message=on_message,
    on_close=on_close
)

ws.run_forever()

因子计算与信号生成 获取到原始 Tick 后，直接入库意义不大，关键在于实时计算因子。例如，我们需要实时监控某只股票的主动买入量。通过 Pandas，我们可以将推送过来的 JSON 数据瞬间转化为 DataFrame，并计算当下的资金流向。

import pandas as pd

df = pd.DataFrame(columns=["code", "price", "volume", "time"])

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        for item in data["data"]:
            df.loc[len(df)] = [item['s'], item['p'], item['v'], item['t']]
            print(df.tail(1))

实盘心得 这种基于 WebSocket 的架构，是搭建本地化量化终端的基础。它不仅能让你看到市场的每一次跳动，更能让你的策略逻辑在数据到达的瞬间完成计算与触发，这才是程序化交易的核心竞争力。

量化手记：A股Tick级数据实时清洗与策略对接实录

评论