昨晚美股盘前，我想给自己盯的几只票加个简单的告警：涨跌幅超过3%就推个消息到手机上。需求很简单对吧？但真坐下来写，从"拿数据"到"告警能跑"，前前后后踩了不少坑。这篇就顺着整个过程聊聊，顺便把股票行情接口的接法讲清楚。
![股票实时价格告警系统.jpg](https://developer.qcloudimg.com/http-save/yehe-2926320/38bf02e57ee5470d309d0842c74c377e.jpg)

先说背景。最近这波行情板块分化得厉害，半导体整体在涨但存储芯片在跌，同一个板块里个股走势都能反着来。光看指数完全没用，必须盯到个股。我盯的就是苹果、英伟达、AMD、英特尔这几只，量不大，但盘中波动的时候手动刷新实在受不了。

## REST轮询方案及其局限性

一开始我的思路特别直接：写个脚本，每隔5秒请求一次报价接口，跟3%的阈值比一下，超了就告警。听起来没毛病。

```python
import requests, time

API_BASE = "https://api.itick.org"
headers = {"accept": "application/json", "token": "your_token"}

def get_price(code):
    r = requests.get(f"{API_BASE}/stock/quote",
                     headers=headers,
                     params={"region": "US", "code": code})
    return r.json()["data"]

while True:
    q = get_price("AAPL")
    print(q["ld"], q["chp"])
    time.sleep(5)
```

跑了一下，能出数。但很快发现两个问题：第一，5秒轮询在盘前或者波动剧烈的时候太慢了，价格可能已经跳了好几个点你才看到；第二，每次都要等网络请求回来才能拿新数据，CPU是不忙但延迟不好控。

这时候才想起来，人家有WebSocket，专门干这个的。

## WebSocket实时订阅：鉴权握手与心跳保活

WebSocket的文档我看了一眼，token放Header里。我照着写了个最小连接：

```python
import websocket, json

ws = websocket.WebSocketApp(
    "wss://api.itick.org/stock",
    header=["token: your_token"],
    on_open=lambda ws: ws.send(json.dumps({
        "ac": "subscribe",
        "params": "AAPL$US",
        "types": "quote"
    })),
    on_message=lambda ws, msg: print(msg),
)
ws.run_forever()
```

结果一跑，服务端回了个 `cannot be resolved action`。我当时就懵了——格式看着跟文档一样啊。

翻了文档才看到那个容易被忽略的细节：连接建立成功和可以订阅是两回事。服务端先回一句 `Connected Successfully`，这时候你什么都不能做；必须等它再推一条 `resAc: "auth"` 且 `code: 1` 的消息，才代表鉴权通过，这时候发订阅才有效。

我之前写了太多"连上就能用"的WebSocket接口，差点忘了这个服务端是要做鉴权握手的。改了一下：

```python
authenticated = False

def on_open(ws):
    print("连接已建立")  # 注意：这时候还不能订阅！

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
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "AAPL$US,NVDA$US",
            "types": "quote"
        }))
        return
```

这下终于收到数据了。但没高兴两秒，连接过了一分钟就断了。

又是文档里写了但我没仔细看的：这个服务端要求30秒发一次心跳，超过1分钟不发就踢人。加个线程专门发ping：

```python
import threading, time

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }))

# 鉴权成功后启动
threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
```

到这一步，实时报价终于稳定跑起来了。

## 返回字段解析与交易状态处理

数据能收到了，但解析的时候又踩了坑。返回的JSON长这样：

```json
{
  "code": 1,
  "data": {
    "s": "AAPL",
    "ld": 225.215,
    "o": 226.27,
    "p": 226.27,
    "h": 226.92,
    "l": 224.44,
    "ch": -3.24,
    "chp": -1.17,
    "v": 16742235,
    "ts": 0,
    "type": "quote"
  }
}
```

全是缩写。我第一反应是 `ld` 是什么？`chp` 又是什么？翻文档才搞清楚：

- `ld` = 最新价（last price）
- `p` = 前收盘价（previous close）
- `ch` = 涨跌额（change）
- `chp` = 涨跌幅百分比（change percent）
- `ts` = 交易状态，0正常、1停牌、2退市、3熔断

这里有个特别坑的地方：我一开始算涨跌幅的时候，偷懒用了 `(ld - o) / o`，也就是拿最新价跟开盘价比。跑出来发现告警触发得特别频繁——因为开盘后前几分钟波动本来就大。后来才反应过来，涨跌幅应该跟前收盘比，服务端已经在 `chp` 里算好了，直接用就行。

还有一个 `ts` 字段。有次我挂着告警跑了一晚上，第二天发现某只停牌的票一直在触发告警——因为 `ld` 停留在昨天的价格，而我的脚本没有检查 `ts`。加个判断就好了：

```python
if data.get("ts") == 1:
    continue  # 停牌，跳过
```

## REST历史K线查询与技术指标计算

告警的事搞完了，又想多做一步：打开脚本的时候先显示一下这几只票近30天的走势，让我对当前位置有个感觉。这就需要拉历史K线。

K线接口跟报价是分开的，路径是 `/stock/kline`，注意是单数 `stock` 不是复数 `stocks`——这个我也踩了，一开始写 `/stocks/kline` 直接404。

```python
def get_kline(region, code, k_type=8, limit=30):
    """
    k_type: 8=日K
    其他: 1=1分 2=5分 5=1时 9=周 10=月
    """
    r = requests.get(
        f"{API_BASE}/stock/kline",
        headers=headers,
        params={
            "region": region,
            "code": code,
            "kType": k_type,
            "limit": limit
        }
    )
    return r.json()["data"]

klines = get_kline("US", "AAPL", k_type=8, limit=30)
for bar in klines[-5:]:
    # 每根K线: t=时间戳 o/h/l/c/v/tu
    print(bar["t"], bar["o"], bar["c"])
```

返回的是一个数组，每根K线就七个字段，干净利落。用这个算个简单的均线或者波动率，比从行情软件截图方便多了。

## 完整告警系统：REST初始化与WebSocket增量推送

把REST初始化和WebSocket拼一起，就是最终能跑的告警脚本：

```python
import json, threading, time, 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}

watchlist = ["AAPL", "NVDA", "AMD", "INTC"]
THRESHOLD = 3.0  # 涨跌幅超过3%告警

# 启动时先拉一遍快照做基线
print("=== 盘前快照 ===")
for code in watchlist:
    q = requests.get(f"{API_BASE}/stock/quote",
                      headers=headers,
                      params={"region": "US", "code": code}).json()["data"]
    print(f"{code}: 前收={q['p']} 最新={q['ld']}")

# 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
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": ",".join(f"{c}$US" for c in watchlist),
            "types": "quote"
        }))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        return

    data = payload.get("data") or {}
    if data.get("type") != "quote":
        return
    if data.get("ts") != 0:  # 非交易状态不告警
        return

    chp = data.get("chp", 0)
    if abs(chp) >= THRESHOLD:
        print(f"[告警] {data['s']}: 涨跌幅 {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()
```

跑起来之后，开盘前先看到一组基线，盘中有股票涨跌超过3%就立刻打印告警。整个过程没用到任何框架，依赖就两个。

## 生产部署注意事项

回头看，整个接入过程的技术点其实不多，但容易踩的坑全在细节里：连接不等于鉴权、必须等auth消息才能订阅、30秒心跳不能忘、字段全是缩写别想当然、交易状态要检查。这些东西文档里都写了，但第一次看的时候很容易扫过去，真跑起来才会一个个撞上。

做监控工具这种东西，稳定性比功能多重要。我现在挂着这个脚本跑了一阵，WebSocket断了会自动重连（websocket-client的run_forever自带），告警逻辑也加了去重，基本上不用管了。

最近这行情波动大，有个自己写的监控脚本比来回切行情软件省心多了。[gitHub文档](https://github.com/itick-org)，有其他市场的接入方式也类似。





昨晚美股盘前，我想给自己盯的几只票加个简单的告警：涨跌幅超过3%就推个消息到手机上。需求很简单对吧？但真坐下来写，从"拿数据"到"告警能跑"，前前后后踩了不少坑。这篇就顺着整个过程聊聊，顺便把股票行情接口的接法讲清楚。

Python股票实时价格告警系统：WebSocket订阅与REST快照实战

本文分享了构建美股实时价格告警系统的完整过程：从初始的REST轮询因延迟高、效率低而受限，转向使用WebSocket实现低延迟订阅。重点解决了鉴权握手、心跳保活、数据字段解析（如chp涨跌幅、ts交易状态）等关键问题，并结合历史K线接口完成盘前快照分析。最终实现基于前收盘价计算涨跌幅、过滤停牌股的稳定告警系统，兼顾实时性与可靠性。

人工智能

盘前实时监控苹果、英伟达等个股涨跌幅超3%自动告警，详解REST轮询局限与WebSocket鉴权、心跳保活、字段解析等关键踩坑点，附完整Python脚本及K线查询方法，适合快速搭建稳定盯盘工具。

代码解释

WebSocket

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Python股票实时价格告警系统：WebSocket订阅与REST快照实战-腾讯云开发者社区-腾讯云

Python股票实时价格告警系统：WebSocket订阅与REST快照实战

Python股票实时价格告警系统：WebSocket订阅与REST快照实战

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐