Python A股实时行情API接入：沪深股票数据与WebSocket推送实战

做国内量化或者行情工具的开发者，A股数据肯定是绕不开的。但接A股数据的时候有个问题：A股的交易规则跟美股、港股都不一样——涨跌停、T+1、集合竞价、午休，这些规则在API数据里是怎么体现的？

这篇从A股交易规则出发，讲一下怎么用Python接A股实时行情，以及这些特殊规则在返回数据里怎么对应。如果你在找一个稳定的股票数据接口或者实时行情API，这篇应该能帮你省掉不少踩坑的时间。

为什么需要专业的A股数据接口

做量化策略或者行情面板的时候，数据是最基础的。一开始我试过爬财经网站的页面，但稳定性太差了——页面结构一改就挂，而且延迟很高，做实时监控根本没法用。后来才转用专业的金融数据API，至少数据格式稳定、接口响应快，不用每天盯着爬虫脚本挂没挂。

选接口的时候主要看几点：一是数据覆盖全不全，沪深两市的股票能不能都拿到；二是实时性够不够，做日内策略延迟太高就没意义了；三是历史K线数据够不够全，回测需要长时间的历史数据。最后找了一圈试了几个接口，留下来一直在用的是itick，主要是接口设计比较干净，文档也写得清楚。

A股交易规则与数据字段的对应

先把A股的交易规则理清楚，这样看返回数据的时候才知道每个字段是什么意思。

交易时间：上午930，下午1300。中午有1.5小时休市。

涨跌停：主板股票±10%，创业板/科创板±20%，ST股票±5%。涨停的时候买盘挂单很多，卖盘很少；跌停反过来。

T+1：当天买的股票第二天才能卖，但API返回的成交量和成交额是实时的。

集合竞价：开盘前925是集合竞价时间，这时候也有成交数据。

这些规则在API返回数据里怎么体现呢？看几个关键字段：

• ts：交易状态，0=正常交易，1=停牌，2=退市，3=熔断。停牌的时候没有成交。

• ld：最新价。涨停的时候这个价就是涨停价。

• chp：涨跌幅百分比。±10%就是涨停/跌停了。

• v：当日累计成交量。

• tu：当日累计成交额。

REST拉取A股报价

先从REST开始，确认数据格式。做行情面板或者批量拉快照的时候，REST接口最方便。

import requests

API_BASE = "https://api.itick.org"

TOKEN = "your_token_here"

headers = {"accept": "application/json", "token": TOKEN}

# 拉A股实时报价

def  get_a_share_quote(code):

resp = requests.get(

f"{API_BASE}/stock/quote",

headers=headers,

params={

"region": "CN",

"code": code

}

)

return resp.json()["data"]

# 拉几只A股大盘股

stocks = ["600519", "000001", "300750"]

for code in stocks:

q = get_a_share_quote(code)

sign = "+"  if q["chp"] >= 0  else  ""

print(f"{code}: {q['ld']} ({sign}{q['chp']}%)")

返回字段跟其他市场基本一致：ld最新价、chp涨跌幅、o/h/l开高低、v成交量。做技术分析或者行情看板这些字段够用了。

注意：A股代码是6位数字，传参数的时候要传字符串形式（”600519”），不要传数字。一开始我传了数字类型，结果接口返回参数错误，折腾了半天才反应过来。

WebSocket订阅A股

做实时监控或者盘中告警的时候，轮询REST接口效率太低，这时候就得用WebSocket实时推送了。

WebSocket订阅格式跟美股一样，是两个参数：代码$市场。

import json

import threading

import time

import websocket

WS_URL = "wss://api.itick.org/stock"

TOKEN = "your_token_here"

authenticated = False

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

# 先等鉴权成功

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

# A股订阅格式：代码$市场

ws.send(json.dumps({

"ac": "subscribe",

"params": "600519$CN,000001$CN",

"types": "quote"

}))

return

if payload.get("resAc") == "pong":

return

data = payload.get("data") or {}

if data.get("type") == "quote":

sign = "+"  if data["chp"] >= 0  else  ""

print(f"{data['s']}: {data['ld']} ({sign}{data['chp']}%)")

A股订阅是两个参数（600519$CN），跟美股一样，不是印度那种三个参数。做实时行情推送的时候，WebSocket比轮询REST效率高很多——不需要每秒发一次请求，服务端有数据就主动推过来。

K线数据获取

做技术分析或者策略回测的时候，需要历史K线数据。K线接口跟其他市场一样，只是region传 CN。

def  get_a_share_kline(code, k_type=8, limit=100):

"""

k_type: 1=1分钟 2=5分钟 5=1小时 8=日K 9=周K 10=月K

"""

resp = requests.get(

f"{API_BASE}/stock/kline",

headers=headers,

params={

"region": "CN",

"code": code,

"kType": k_type,

"limit": limit

}

)

return resp.json()["data"]

# 拉贵州茅台近100根日K

daily = get_a_share_kline("600519", k_type=8, limit=100)

closes = [bar["c"] for bar in daily]

# 算20日均线

ma20 = sum(closes[-20:]) / 20

print(f"600519最新收盘: {closes[-1]}, 20日均线: {ma20:.2f}")

K线返回标准OHLCV：t时间戳、o/h/l/c开高低收、v成交量。做技术分析直接用这些字段就行，不用自己再清洗数据。

A股特有的注意事项

交易时间有午休。 A股中午休市1.5小时（1100），这个时间段没有成交数据。做实时监控的时候要注意，别把午休时间当成流动性枯竭——我一开始做告警的时候，午休期间一直没数据还以为接口挂了，后来才反应过来是休市了。

涨跌停判断。 主板股票涨跌幅超过±10%就是涨停/跌停，创业板/科创板是±20%。做策略的时候要判断当前是否涨停，涨停的时候买不进去。可以通过chp字段判断，如果涨跌幅接近10%或者20%就要小心了。

代码是6位数字。 A股代码是6位数字（600519、000001），跟港股的5位数字（00700）、日本的4位数字（7203）都不一样。一开始我传了”700”，结果返回空数据——那是港股腾讯的代码，A股没有700这个代码。

region是CN。 A股市场region传 CN，不是 SH 或 SZ。

集合竞价时间。 开盘前925是集合竞价，这时候也有成交数据，但价格波动很大。做策略的时候要注意区分集合竞价和连续竞价，别把集合竞价的成交量当成正常交易的量。

完整脚本：A股行情监控

把前面几块拼起来，最终的完整脚本：

import json, time, threading, requests, websocket

API_BASE = "https://api.itick.org"

WS_URL = "wss://api.itick.org/stock"

TOKEN = "your_token_here"

headers = {"accept": "application/json", "token": TOKEN}

# A股监控列表

watchlist = ["600519", "000001", "300750"]

# 启动时拉快照

print("=== A股快照 ===")

for code in watchlist:

r = requests.get(f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "CN", "code": code}).json()["data"]

sign = "+"  if r["chp"] >= 0  else  ""

print(f" {code}: {r['ld']} ({sign}{r['chp']}%)")

# WebSocket实时推送

authenticated = False

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

params = ",".join(f"{c}$CN"  for c in watchlist)

ws.send(json.dumps({

"ac": "subscribe",

"params": params,

"types": "quote"

}))

threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()

return

if payload.get("resAc") == "pong":

return

data = payload.get("data") or {}

if data.get("type") == "quote":

sign = "+"  if data["chp"] >= 0  else  ""

print(f"[实时] {data['s']}: {data['ld']} ({sign}{data['chp']}%)")

def  heartbeat(ws):

while  True:

time.sleep(30)

ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message)

ws.run_forever()

这个脚本跑起来之后，开盘前先看到一组A股的快照，盘中实时更新价格和涨跌幅。做行情面板或者盘中告警的话，这个基础上再扩展就行。

小结

接A股数据的技术框架跟美股差不多——REST做快照和历史数据，WebSocket做实时推送。主要区别在交易规则（涨跌停、T+1、午休）和代码格式（6位数字）。这些细节不踩一遍坑很容易忽略。

参考文档：docs.itick.org/websocket/stocks

GitHub：https://github.com/itick-org/

本作品采用《CC 协议》，转载必须注明作者和本文链接