在接入 A 股行情數據時,有一個經常被忽略但非常關鍵的細節:你拿到的是實時數據,還是延時數據?
我們在雪球、東方財富這類網站上看到的行情更新看起來很快,其實大多是延時數據。這在官網也會註明,比如“本頁面所示行情數據為 T+0 延時15分鐘,僅供參考”等。
延時行情對日常查看、趨勢判斷是足夠的,但如果你在做以下這些事情,就會有明顯影響:
- 高頻或低延遲策略開發
- 實盤交易系統對接
- 實時風控與盤口建模
- 多源行情同步與聚合分析
在這些場景中,延時15秒甚至幾秒鐘都可能導致策略響應偏離真實市場狀態。
相比之下,實時行情 API通過WebSocket 毫秒級推送最新市場數據,沒有中間緩存或延遲,可以更直接地反映市場當下的狀態。尤其是 Level 2 行情,還能提供更深的盤口信息,比如買賣五檔、委託隊列等。
這並不是説實時行情一定更好,關鍵在於你的應用是否真的需要它。如果你只是做中線選股或者參考日內波動,延時數據可能已經夠用了;但若你需要交易決策或建模依賴市場瞬時狀態,實時行情就變得必要了。
訂閲K線
import asyncio
import json
import websockets
# A股行情的websocket訂閲地址
WS_URL = "wss://data.infoway.io/ws?business=stock&apikey=yourApiKey"
# 請先在官網https://infoway.io 申請免費API key
async def connect_and_receive():
async with websockets.connect(WS_URL) as websocket:
# 發送初始化消息,這裏訂閲的是平安銀行(000001.SZ)的1分鐘K線數據
init_message = {
"code": 10004,
"trace": "423afec425004bd8a5e02e1ba5f9b2b0",
"data": {
"arr": [
{
"type": 1, # 1分鐘K線
"codes": "000001.SZ" # A股股票代碼,例如 平安銀行
}
]
}
}
await websocket.send(json.dumps(init_message))
async def send_ping():
while True:
await asyncio.sleep(30)
ping_message = {
"code": 10010,
"trace": "423afec425004bd8a5e02e1ba5f9b2b0"
}
await websocket.send(json.dumps(ping_message))
ping_task = asyncio.create_task(send_ping())
try:
while True:
message = await websocket.recv()
print(f"Message received: {message}")
except websockets.exceptions.ConnectionClosedOK:
print("Connection closed normally")
finally:
ping_task.cancel()
# 運行主函數
asyncio.run(connect_and_receive())
返回示例
{
"c": "11.25", // 收盤價
"h": "11.28", // 最高價
"l": "11.22", // 最低價
"o": "11.23", // 開盤價
"pca": "0.02", // 價格變化
"pfr": "0.18%", // 價格變化百分比
"s": "000001.SZ", // 股票代碼
"t": 1747550648097, // 時間戳
"ty": 1, // K線類型:1 表示1分鐘K線
"v": "258400", // 交易量(單位:股)
"vw": "11.2458" // 加權平均價格
}
訂閲盤口數據
import asyncio
import json
import websockets
# A股行情的websocket訂閲地址
WS_URL = "wss://data.infoway.io/ws?business=stock&apikey=yourApiKey"
# 請先在官網 https://infoway.io 申請免費API key
async def connect_and_receive():
async with websockets.connect(WS_URL) as websocket:
# 發送初始化消息,訂閲的是 A股平安銀行(000001.SZ)的盤口數據
init_message = {
"code": 10002, # 盤口訂閲的請求協議號
"trace": "423afec425004bd8a5e02e1ba5f9b2b0", # 可追溯ID(可隨機生成)
"data": {
"codes": "000001.SZ" # A股股票代碼,例如平安銀行
}
}
await websocket.send(json.dumps(init_message))
# 設置 ping 任務,保持連接
async def send_ping():
while True:
await asyncio.sleep(30)
ping_message = {
"code": 10010,
"trace": "423afec425004bd8a5e02e1ba5f9b2b0"
}
await websocket.send(json.dumps(ping_message))
# 啓動 ping 協程
ping_task = asyncio.create_task(send_ping())
try:
# 持續接收消息
while True:
message = await websocket.recv()
print(f"Message received: {message}")
except websockets.exceptions.ConnectionClosedOK:
print("Connection closed normally")
finally:
# 取消 ping 任務
ping_task.cancel()
# 運行主函數
asyncio.run(connect_and_receive())
返回示例
{
"a": [
[
"11.25", // 賣盤價格1
"11.26", // 賣盤價格2
"11.27", // 賣盤價格3
"11.28", // 賣盤價格4
"11.29" // 賣盤價格5
],
[
"23000", // 賣盤數量1
"18000", // 賣盤數量2
"9500", // 賣盤數量3
"6000", // 賣盤數量4
"7200" // 賣盤數量5
]
],
"b": [
[
"11.24", // 買盤價格1
"11.23", // 買盤價格2
"11.22", // 買盤價格3
"11.21", // 買盤價格4
"11.20" // 買盤價格5
],
[
"26000", // 買盤數量1
"20000", // 買盤數量2
"10000", // 買盤數量3
"7500", // 買盤數量4
"8500" // 買盤數量5
]
],
"s": "000001.SZ", // A股股票代碼(平安銀行)
"t": 1747553102161 // 時間戳
}