首页 > IT职业 > IT生活 > 三分钟用Python调用a股行情api获取实时股价
前阵子在整理自己的交易数据时,我总觉得每次开客户端刷股价就像在打转。尤其是同时盯着几只股票时,手得不停切换窗口,思路跟着断掉。我就想,如果能在后台用 Python 直接拿到实时价格,那效率肯定高得多。
一开始尝试网页抓取,刷新限制和延迟让我心累。后来我试着用 WebSocket,把行情数据直接拉到自己手里,每条价格像心跳一样跳动,让我慢慢感受到市场节奏。尤其在价格快速波动时,会有一种“市场参与感”,客户端完全体会不到。
整个过程只需要一个库:
pip install websocket-client
WebSocket 对高频行情特别适合,延迟低,稳定性也比轮询网页高。
我关注的第一只股票是 沪市 600000 (SH600000)。为了保证结构清晰,我把 WebSocket 连接和订阅逻辑分开。
import websocket
import json
import threading
import uuid
# 以 AllTick 的接口为例
WS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=你的Token"
SYMBOLS = ["600000.SH"] # 可以订阅多只股票
def on_message(ws, message):
msg = json.loads(message)
if msg.get("cmd_id") == 22998: # 推送的逐笔成交消息
for tick in msg["data"]:
print(f"{tick['code']} 成交价:{tick['price']} 成交量:{tick['volume']} 方向:{tick['trade_direction']}")
def on_open(ws):
# 构建订阅请求
req = {
"cmd_id": 22004,
"seq_id": 1,
"trace": str(uuid.uuid4()),
"data": {
"symbol_list": [{"code": code} for code in SYMBOLS]
}
}
ws.send(json.dumps(req))
def run_ws():
ws = websocket.WebSocketApp(
WS_URL,
on_open=on_open,
on_message=on_message
)
ws.run_forever()
if __name__ == "__main__":
t = threading.Thread(target=run_ws)
t.start()
我一般让它在后台跑,偶尔看看输出的价格跳动,发现实时行情比客户端刷新直观太多,也更能感受到波动节奏。
输出大概是这样:
600000.SH 成交价:5.23 成交量:300 方向:1 600000.SH 成交价:5.25 成交量:150 方向:2 600000.SH 成交价:5.22 成交量:200 方向:1
通过这种连续推送,我能直观感受到价格波动的节奏,平稳期关注趋势,波动期关注波幅。
想同时观察多只股票,只需在 SYMBOLS 中增加股票代码,订阅请求会一次性覆盖所有股票:
SYMBOLS = ["600000.SH", "000001.SZ", "000002.SZ"]
我试过五六只股票同时订阅,推送非常流畅。如果订阅数量上百,就需要考虑网络和消息处理能力,这也让我意识到实时行情不仅是接口问题,更涉及系统设计和数据流处理。
通过这个实验,我总结了几点体会:
稳定性 :WebSocket 断开或超时是常态,需要心跳和自动重连机制。
数据价值 :每条成交数据都是瞬时市场信息,可以统计、分析甚至触发提醒。
节奏感 :连续的价格跳动让我对短期波动形成直觉,体验远胜单纯看数字。
我把实时行情和简单分析结合起来,比如计算涨跌幅、绘制曲线或者触发提醒,比开客户端刷新要灵活得多,也更有参与感。
把实时股价拉到自己手里,不只是看数字,更像在感受市场的脉搏。整个实验从连接到处理推送,让我对数据流、处理效率和分析价值有了直观理解。自己搭建的小工具,让数据变得可操作、可分析,同时也让我意识到实时数据的连续性和节奏感往往比单纯的数值更有意义。用 AllTick API订阅实时行情,每次看到价格跳动,都能感受到微妙的“参与市场”的体验,这是客户端无法提供的,也让我在操作中慢慢形成对短期波动的直觉。
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70038549/viewspace-3115585/ ,如需转载,请注明出处,否则将追究法律责任。