很多开发者第一次接 A 股行情数据的时候，以为和接美股差不多——无非是换个接口地址，改下代码。然后就会发现：分钟 K 线中间有两小时的空洞，历史价格在某一天突然跳了 30%，WebSocket 每天下午一点才重新有数据，策略回测的信号永远慢一天……
这些坑几乎每个人都会踩。本文从 A 股市场结构出发，结合实际接口代码，把这些容易被忽略的细节讲清楚。
一、A 股代码规则：不查文档就能判断板块
A 股有四大交易场所：上海证券交易所（SSE/上交所）、深圳证券交易所（SZSE/深交所）、北京证券交易所（BSE/北交所），以及面向上市公司但不单独开市的全国股转系统（新三板）。
股票代码后缀：上交所用 .SH，深交所用 .SZ，北交所用 .BJ。
代码前缀本身就包含板块信息，不需要调接口就能判断：
| 代码前缀 | 交易所 | 板块 |
|---|---|---|
60xxxx | 上交所 | 沪市主板（A 股） |
688xxx | 上交所 | 科创板（注册制，±20% 涨跌幅） |
689xxx | 上交所 | 科创板 CDR |
000xxx / 001xxx | 深交所 | 深市主板 |
002xxx / 003xxx | 深交所 | 原中小板（2021 年并入主板） |
300xxx / 301xxx | 深交所 | 创业板（注册制，±20% 涨跌幅） |
83xxxx / 87xxxx | 北交所 | 北交所正式上市 |
43xxxx | 北交所 | 北交所精选层 |
ST 股（特别处理）不通过前缀识别，而是通过股票名称中的 ST 标记，代码不变但涨跌幅限制收窄为 ±5%。
用 Python 写一个快速识别函数：
def classify_a_share(symbol: str) -> dict:
"""
从股票代码识别板块信息，无需调用接口。
symbol 格式：600519.SH / 300750.SZ / 688599.SH
"""
code = symbol.split(".")[0]
suffix = symbol.split(".")[-1] if "." in symbol else ""
if code.startswith("6") and not code.startswith("688") and not code.startswith("689"):
board, limit = "沪市主板", 0.10
elif code.startswith("688") or code.startswith("689"):
board, limit = "科创板", 0.20
elif code.startswith("000") or code.startswith("001") or code.startswith("002") or code.startswith("003"):
board, limit = "深市主板", 0.10
elif code.startswith("300") or code.startswith("301"):
board, limit = "创业板", 0.20
elif code.startswith("83") or code.startswith("87") or code.startswith("43"):
board, limit = "北交所", 0.30
else:
board, limit = "未知", 0.10
return {"code": code, "suffix": suffix, "board": board, "daily_limit": limit}
print(classify_a_share("600519.SH")) # 茅台 → 沪市主板, ±10%
print(classify_a_share("300750.SZ")) # 宁德时代 → 创业板, ±20%
print(classify_a_share("688599.SH")) # 天合光能 → 科创板, ±20%
二、交易时间：比想象中更复杂
两段式 + 收盘竞价
A 股不是简单的 9:30–15:00 连续交易，实际分为四个时间段：
| 阶段 | 时间 | 说明 |
|---|---|---|
| 开盘集合竞价 | 9:15–9:25 | 价格发现，产生当日开盘价；9:20–9:25 不允许撤单 |
| 连续竞价（上午） | 9:30–11:30 | 正常撮合交易 |
| 午休 | 11:30–13:00 | 完全暂停，长达 1.5 小时 |
| 连续竞价（下午） | 13:00–14:57 | 正常撮合交易 |
| 收盘集合竞价 | 14:57–15:00 | 产生收盘价；期间订单可撤，15:00 最终定价 |
这里有几个开发者经常忽略的细节：
1. API 返回的交易时间段是 13:00-14:57，不是 15:00。最后三分钟的收盘竞价产生的成交数据会并入 14:57 这根分钟 K，或以独立方式处理——具体取决于数据源。如果你在 14:58 调用实时成交接口，拿到的 t 时间戳可能仍然是 14:57 的。
2. 分钟 K 线中会有 1.5 小时的空洞（11:30–13:00）。如果你在 backtest 框架里用等间距索引，这段空洞会被错误地填充为"价格没有变化"，导致均线等指标的计算偏差。正确做法是用交易日历过滤出有效的分钟 bar。
3. 开盘集合竞价（9:15–9:25）期间的行情数据性质不同：这段时间内没有真实成交，只有委托的"虚拟"价格。如果你的 WebSocket 在 9:15 之后才订阅，可能会收到竞价阶段的推送——不要把它当作连续竞价的交易数据处理。
三、行情接口实战：成交、盘口、K 线
A 股（沪市 + 深市）使用 /stock/ 接口路径，与美股、港股共用同一套接口体系，通过股票代码后缀区分市场。
3.1 实时成交明细
import requests
API_KEY = "YOUR_API_KEY"
BASE = "https://data.infoway.io"
# A 股代码格式：600519.SH（沪市）或 000858.SZ（深市）
codes = "600519.SH,000858.SZ,300750.SZ,688599.SH"
resp = requests.get(
f"{BASE}/stock/batch_trade/{codes}",
headers={"apiKey": API_KEY, "Accept": "application/json"}
)
for tick in resp.json()["data"]:
direction = {0: "—", 1: "↑买", 2: "↓卖"}.get(tick["td"], "?")
print(f"{tick['s']:<15} 价格={tick['p']:>8} 成交量={tick['v']:>10}手 方向={direction}")
注意：返回的 v 字段是股数（股），不是手数。A 股 1 手 = 100 股，显示时记得换算：
shares = int(tick["v"])
lots = shares // 100
print(f"成交 {lots} 手 ({shares} 股)")
返回字段：
| 字段 | 说明 |
|---|---|
s | 股票代码（含后缀） |
t | 成交时间戳（毫秒，UTC） |
p | 最新成交价（元，精度 0.01） |
v | 成交量（股，非手） |
vw | 成交额（元） |
td | 方向：0=默认，1=主买，2=主卖 |
3.2 实时盘口
resp = requests.get(
f"{BASE}/stock/batch_depth/600519.SH",
headers={"apiKey": API_KEY, "Accept": "application/json"}
)
book = resp.json()["data"][0]
asks = list(zip(book["a"][0], book["a"][1]))
bids = list(zip(book["b"][0], book["b"][1]))
print("卖盘（ask）")
for price, vol in asks[:5]:
print(f" {price:>8} 元 {int(vol)//100:>6} 手")
spread = float(asks[0][0]) - float(bids[0][0])
print(f"\n 价差：{spread:.2f} 元\n")
print("买盘（bid）")
for price, vol in bids[:5]:
print(f" {price:>8} 元 {int(vol)//100:>6} 手")
a[0] 和 a[1] 是两个平行数组，分别是卖盘价格和卖盘量，通过下标对应（index 0 是最优卖一）；b[0] / b[1] 同理对应买盘。
四、历史 K 线：三个容易踩的坑
坑一：分钟 K 线有时间空洞
A 股每天有效的分钟 bar 只有 241 根（9:30–11:30 共 120 根，13:00–14:57 共 117 根，加上集合竞价/收盘竞价各 1-2 根）。如果你按等间距索引处理，11:30 到 13:00 之间会产生 90 个"假的"空 bar。
建议在数据清洗阶段就过滤掉非交易时间的 bar：
from datetime import time
def is_valid_cn_bar(ts_seconds: int) -> bool:
"""判断秒时间戳（UTC）是否落在 A 股交易时段内（转为北京时间 UTC+8）。"""
from datetime import datetime, timezone, timedelta
CST = timezone(timedelta(hours=8))
t = datetime.fromtimestamp(ts_seconds, tz=CST).time()
morning = time(9, 30) <= t < time(11, 30)
afternoon = time(13, 0) <= t < time(15, 0)
return morning or afternoon
坑二：不做复权处理，历史价格会突然跳变
A 股上市公司送股、分红、配股频繁，除权后股价会出现跳变。如果你直接用未复权的历史价格计算均线或动量指标，会在除权日前后产生虚假信号。
以贵州茅台（600519.SH）为例：2020 年每股分红 17.02 元，除权日前后原始价格会出现约 1.3% 的向下跳变——这不是市场行情，而是数学事件。
前复权（backward adjusted）是回测中最常用的方式：把历史所有价格乘以复权因子，使价格序列连续。获取复权因子：
import requests
def get_adjustment_factors(symbol: str, begin_day: str, end_day: str) -> list[dict]:
resp = requests.get(
"https://data.infoway.io/common/basic/symbols/adjustment_factors",
params={"symbol": symbol, "market": "CN",
"beginDay": begin_day, "endDay": end_day},
headers={"apiKey": "YOUR_API_KEY"}
)
return resp.json().get("data", [])
factors = get_adjustment_factors("600519.SH", "20200101", "20261231")
# 返回每个交易日的 forward_factor（前复权因子）
# 应用复权因子到 K 线数据
factor_map = {f["trade_date"]: f["forward_factor"] for f in factors}
def apply_forward_adj(candle: dict, trade_date: str) -> dict:
factor = factor_map.get(trade_date, 1.0)
return {
**candle,
"o": str(round(float(candle["o"]) * factor, 4)),
"h": str(round(float(candle["h"]) * factor, 4)),
"l": str(round(float(candle["l"]) * factor, 4)),
"c": str(round(float(candle["c"]) * factor, 4)),
}
坑三：涨跌停日的 K 线形态异常
当股票涨停（或跌停）后，连续竞价阶段成交量极少甚至为零，价格全天锁死在涨停价。这会产生上影线/下影线极短或为零的特殊 K 线形态。如果你的策略对影线有假设（比如"长上影线意味着压力"），需要特别处理涨跌停日的 bar。
下面是完整的历史 K 线下载函数，带翻页：
import requests, json, time as _time
def fetch_cn_klines(symbol: str, kline_type: int = 8,
count: int = 500, until_ts: int | None = None) -> list[dict]:
payload = {"klineType": kline_type, "klineNum": count, "codes": symbol}
if until_ts:
payload["timestamp"] = until_ts
r = requests.post(
"https://data.infoway.io/stock/v2/batch_kline",
headers={"apiKey": "YOUR_API_KEY", "Content-Type": "application/json"},
data=json.dumps(payload)
)
r.raise_for_status()
for item in r.json().get("data", []):
if item["s"] == symbol:
return item.get("respList", [])
return []
def download_full_history(symbol: str) -> list[dict]:
"""翻页下载全部日 K 线历史。"""
all_bars: list[dict] = []
cursor: int | None = None
while True:
batch = fetch_cn_klines(symbol, kline_type=8, count=500, until_ts=cursor)
if not batch:
break
all_bars.extend(batch)
oldest = int(batch[-1]["t"])
if cursor and oldest >= cursor:
break
cursor = oldest
_time.sleep(0.3)
all_bars.sort(key=lambda b: int(b["t"]))
return all_bars
bars = download_full_history("600519.SH")
print(f"茅台日K共 {len(bars)} 根，最早 {bars[0]['t']}，最新 {bars[-1]['c']} 元")
K 线字段说明：
| 字段 | 说明 |
|---|---|
t | K 线开盘时间（Unix 秒，UTC） |
o / h / l / c | 开 / 高 / 低 / 收（元） |
v | 成交量（股，注意换算成手） |
vw | 成交额（元） |
pc | 较前一根 K 的涨跌幅（%） |
pca | 较前一根 K 的涨跌额（元） |
五、涨跌停识别算法
A 股涨跌停规则因板块不同而异，总结如下：
| 板块 | 涨跌幅限制 | 备注 |
|---|---|---|
| 沪/深主板 | ±10% | ST 股收窄为 ±5% |
| 创业板（注册制） | ±20% | 2020 年改革后 |
| 科创板 | ±20% | 上市前 5 日无限制 |
| 北交所 | ±30% | — |
| 退市整理期 | ±10%（仅下跌） | 实际上可以上涨但不超过 10% |
涨跌停价格计算（四舍五入至分，即 0.01 元精度）：
import math
def calc_limit_prices(prev_close: float, board: str,
is_st: bool = False) -> tuple[float, float]:
"""
返回 (涨停价, 跌停价)。
board 参考接口返回的 board 字段：
SHMainConnect / SHMainNonConnect / SZMainConnect / SZMainNonConnect
SHSTAR / SZGEMConnect / SZGEMNonConnect
"""
if is_st:
pct = 0.05
elif board in ("SHSTAR", "SZGEMConnect", "SZGEMNonConnect"):
pct = 0.20
else:
pct = 0.10
# 交易所规则：涨停价向下取整到分，跌停价向上取整到分
up = math.floor(prev_close * (1 + pct) * 100) / 100
down = math.ceil(prev_close * (1 - pct) * 100) / 100
return up, down
# 示例：茅台前收 1800.00 元，主板 ±10%
up, down = calc_limit_prices(1800.00, "SHMainConnect")
print(f"涨停价：{up:.2f} 跌停价：{down:.2f}")
# → 涨停价：1980.00 跌停价：1620.00
从实时成交数据判断是否触板：
def is_limit(current_price: float, prev_close: float,
board: str, is_st: bool = False) -> str:
up, down = calc_limit_prices(prev_close, board, is_st)
if abs(current_price - up) < 0.005:
return "涨停"
if abs(current_price - down) < 0.005:
return "跌停"
return "正常"
一个重要的细节：涨停板不等于无成交。创业板/科创板实行"涨停后仍可撤单"机制，股票触及 20% 涨停后，挂单方可以撤单，导致涨停打开再封回的"开板"行为。判断是否"封板"需要结合盘口数据：如果卖一挂单量为零，说明已封板。
六、WebSocket 实时行情订阅
A 股 WebSocket 接入地址与美股、港股相同：
wss://data.infoway.io/ws?business=stock&apikey=YOUR_API_KEY
以下是一个处理 A 股特殊场景的完整 Python 客户端，重点处理了午休期间的静默状态和涨跌停日志：
import asyncio, json, uuid, logging
from datetime import datetime, time, timezone, timedelta
from typing import Optional
import websockets
from websockets.exceptions import ConnectionClosed
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("a-share-ws")
CST = timezone(timedelta(hours=8))
def cn_market_status() -> str:
"""返回当前 A 股市场状态。"""
now = datetime.now(CST)
if now.weekday() >= 5:
return "休市"
t = now.time()
if time(9, 15) <= t < time(9, 30):
return "集合竞价"
if time(9, 30) <= t < time(11, 30):
return "上午交易"
if time(11, 30) <= t < time(13, 0):
return "午休"
if time(13, 0) <= t < time(14, 57):
return "下午交易"
if time(14, 57) <= t < time(15, 0):
return "收盘竞价"
return "休市"
# 监控的股票列表（示例：茅台、宁德时代、中芯国际、比亚迪）
SYMBOLS = "600519.SH,300750.SZ,688981.SH,002594.SZ"
class AShareClient:
def __init__(self, api_key: str):
self.url = f"wss://data.infoway.io/ws?business=stock&apikey={api_key}"
self.ws: Optional[websockets.WebSocketClientProtocol] = None
self.running = True
self._hb_task: Optional[asyncio.Task] = None
def _trace(self) -> str:
return str(uuid.uuid4())
async def _send(self, msg: dict) -> None:
if self.ws:
await self.ws.send(json.dumps(msg))
async def _subscribe(self) -> None:
# 订阅实时成交（代码 10000）
await self._send({"code": 10000, "trace": self._trace(),
"data": {"codes": SYMBOLS}})
await asyncio.sleep(5)
# 订阅盘口（代码 10003）
await self._send({"code": 10003, "trace": self._trace(),
"data": {"codes": SYMBOLS}})
await asyncio.sleep(5)
# 订阅 1 分钟 K 线（代码 10006）
await self._send({"code": 10006, "trace": self._trace(),
"data": {"arr": [{"type": 1, "codes": SYMBOLS}]}})
log.info("订阅完成 | 市场状态：%s", cn_market_status())
def _start_heartbeat(self) -> None:
if self._hb_task and not self._hb_task.done():
self._hb_task.cancel()
async def beat():
while True:
await asyncio.sleep(30)
if not self.ws or self.ws.close_code is not None:
break
try:
await self._send({"code": 10010, "trace": self._trace()})
except Exception:
break
self._hb_task = asyncio.create_task(beat())
def _dispatch(self, raw: str) -> None:
try:
msg = json.loads(raw)
except json.JSONDecodeError:
return
code = msg.get("code")
data = msg.get("data", {})
if code == 10002: # 成交推送
direction = {0: "—", 1: "↑主买", 2: "↓主卖"}.get(data.get("td"), "?")
vol_lots = int(data.get("v", 0)) // 100
log.info("成交 %-14s 价格=%-9s 成交量=%-6d手 %s",
data.get("s"), data.get("p"), vol_lots, direction)
elif code == 10005: # 盘口推送
best_bid = data.get("b", [[None]])[0][0]
best_ask = data.get("a", [[None]])[0][0]
if best_bid and best_ask:
spread = round(float(best_ask) - float(best_bid), 2)
log.info("盘口 %-14s 买一=%-9s 卖一=%-9s 价差=%.2f",
data.get("s"), best_bid, best_ask, spread)
elif code == 10008: # K 线推送
pct = data.get("pfr", "0%")
log.info("K线 %-14s 开=%-7s 高=%-7s 低=%-7s 收=%-7s 涨跌=%s",
data.get("s"), data.get("o"), data.get("h"),
data.get("l"), data.get("c"), pct)
elif code in (10001, 10004, 10007):
log.info("订阅确认 code=%s", code)
async def _connect_once(self) -> None:
async with websockets.connect(self.url) as ws:
self.ws = ws
log.info("已连接 | 市场状态：%s", cn_market_status())
await self._subscribe()
self._start_heartbeat()
try:
async for message in ws:
self._dispatch(message)
finally:
if self._hb_task:
self._hb_task.cancel()
self.ws = None
async def start(self) -> None:
backoff = 5
while self.running:
status = cn_market_status()
# 午休和休市期间暂停重连，避免无意义的断线重连
if status in ("午休", "休市"):
log.info("当前状态：%s，暂停连接（60s后再检查）", status)
await asyncio.sleep(60)
continue
try:
await self._connect_once()
backoff = 5
except ConnectionClosed as e:
log.warning("连接断开：%s", e)
except Exception as e:
log.error("连接异常：%s", e)
if not self.running:
break
log.info("%ds 后重连...", backoff)
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 60)
def stop(self) -> None:
self.running = False
async def main():
client = AShareClient(api_key="YOUR_API_KEY")
await client.start()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
log.info("已停止")
七、T+1 制度：量化开发的隐藏约束
A 股实行 T+1 交易制度：当天买入的股票，最早次日才能卖出。这与美股（T+0，可当天高抛低吸）和港股（T+0 日内交易可做）完全不同。
T+1 对策略开发的影响：
1. 日内信号无法当日平仓：如果你的策略在 14:00 产生卖出信号，但这只股票是当天 9:30 买入的，无法执行。回测框架必须在信号逻辑中加入"当日买入"标记。
2. 滑点模型不同：A 股买入委托通常在涨停位挂单等待成交，而美股的 market order 几乎即时成交。回测中使用过于乐观的成交假设，实盘会出现较大偏差。
3. 融券 T+0 是例外：通过融券（borrowing）买入的股票可以当日卖出，但融券需要开通两融账户，普通投资者通常不适用。
一个简单的 T+1 约束检查函数，可以嵌入回测框架：
from collections import defaultdict
from datetime import date
class T1PositionTracker:
"""跟踪 A 股 T+1 约束：今天买入的不能今天卖出。"""
def __init__(self):
self.positions: dict[str, int] = defaultdict(int) # symbol → 可卖数量（股）
self.today_bought: dict[str, int] = defaultdict(int) # 今日新买，次日才能卖
def buy(self, symbol: str, shares: int) -> None:
self.today_bought[symbol] += shares
def sell(self, symbol: str, shares: int) -> bool:
available = self.positions[symbol]
if shares > available:
print(f"⚠ {symbol} 可卖 {available} 股，不足 {shares} 股（T+1 限制）")
return False
self.positions[symbol] -= shares
return True
def next_day(self) -> None:
"""每日收盘后调用，将今日买入转为可卖持仓。"""
for symbol, shares in self.today_bought.items():
self.positions[symbol] += shares
self.today_bought.clear()
八、开发常见错误速查
| 错误码 | 说明 | 处理建议 |
|---|---|---|
ret: 508 | 股票代码不存在或已退市 | 检查代码格式是否含 .SH / .SZ 后缀 |
ret: 429 | 请求频率超限 | 降低轮询频率，或升级套餐 |
ret: 503 | K 线查询数量超出限制 | 单次最多 500 根，多标的同查时每标的仅返回 2 根 |
ret: 401 | API Key 认证失败 | 检查请求头是否设置 apiKey 而非 Authorization |
WS 错误 513 | WebSocket 心跳超时 | 每 30 秒发送一次心跳（code: 10010），超 60 秒断连 |
WS 错误 501 | 订阅频率超过每分钟 60 次 | 订阅请求加间隔，不要在连接建立后立即批量订阅 |
小结
A 股行情接口的核心难点不在接口本身，而在于市场机制与数据之间的对应关系：
- 代码规则决定涨跌幅上限，不需要接口就能判断
- 分钟 K 线有 1.5 小时午休空洞，回测框架需要过滤
- 复权因子是历史回测的前提，直接用原始价格会产生虚假信号
- 涨跌停计算因板块不同有 ±5% / ±10% / ±20% / ±30% 的差异
- T+1 制度意味着日内买入当日不可卖出，回测框架需要显式建模
- WebSocket 在午休期间流量会中断，需要区分"市场暂停"与"连接断开"