supermind logo
同花顺logo

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

用户头像Fxdund
2026-10-12 发布

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

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

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

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

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

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

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

交易时间：上午9:30-11:30，下午13:00-15:00。中午有1.5小时休市。

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

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

集合竞价：开盘前9:15-9:25是集合竞价时间，这时候也有成交数据。

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

  • ts：交易状态，0=正常交易，1=停牌，2=退市，3=熔断。停牌的时候没有成交。
  • ld：最新价。涨停的时候这个价就是涨停价。
  • chp：涨跌幅百分比。±10%就是涨停/跌停了。
  • v：当日累计成交量。
  • tu：当日累计成交额。

REST拉取A股报价

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

import requests

API_BASE = "//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小时（11:30-13:00），这个时间段没有成交数据。做实时监控的时候要注意，别把午休时间当成流动性枯竭——我一开始做告警的时候，午休期间一直没数据还以为接口挂了，后来才反应过来是休市了。

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

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

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

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

完整脚本：A股行情监控

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

import json, time, threading, requests, websocket

API_BASE = "//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位数字）。这些细节不踩一遍坑很容易忽略。

参考文档：https://docs.itick.org/websocket/stocks
GitHub：https://github.com/itick-org/

评论