做美股行情工具或者量化策略的人，几乎都会纠结同一个问题：实时行情到底该用HTTP轮询还是WebSocket推送？网上的说法五花八门，有人说不轮询就是不懂行情，有人说WebSocket才是正经解法。与其听别人讲，不如自己写代码测一遍——把两种方式接到同一个数据源上，用统一的口径量一量延迟、请求量、资源占用，结论自然就出来了。
![美股实时行情接入.jpg](https://developer.qcloudimg.com/http-save/yehe-2926320/67728ba5f3bf21f806e1dc0f10ba18d7.jpg)

这篇就是一次完整的测量过程记录：怎么设计测试、REST侧怎么测、WebSocket侧怎么测、结果怎么对比、最后什么场景该选哪个。

## 为什么延迟这事值得认真测

很多做美股的朋友一开始觉得"延迟差个一两秒无所谓"。这取决于你的策略：

- 日频选股：收盘后拿数据，延迟完全不敏感
- 分钟级监控：一两秒延迟可以接受，但要稳定
- 盘口、事件驱动、套利类策略：延迟直接决定策略能不能成立

另外延迟不是只看"快不快"，还要看**抖不抖**。一个接口平均100毫秒但偶尔跳到5秒，对策略的伤害比稳定200毫秒还大。所以测量的时候，别只记平均值，最大值和分布也要记。

## 测试设计：先定好口径再动手

延迟测试最容易出问题的地方是**口径不统一**。测REST和测WebSocket，如果"延迟"的定义不一样，对比就没有意义。

我建议统一用这个口径：


延迟 = 数据到达本地的时间 - 行情数据自身的生成时间戳


两种方式都对比这个差值。REST响应里有最新成交时间戳 `t`（毫秒），WebSocket推送的报价消息里也有 `t`。本地收到数据时记一个 `time.time() * 1000`，减去 `t`，就是这次数据的"端到端延迟"。

测量样本至少跑几百条，同时记录三组数：最小值、平均值、P95。只看平均会掩盖抖动，P95才能反映最差情况有多差。

还有一个公平性问题：**两边要订阅同一批标的、在同一个时段测**。美股盘中和盘后的推送频率完全不一样，跨时段对比出来的数字没有意义。

## REST轮询怎么测

REST这边思路很简单：循环调用报价接口，每收到一次响应就计算一次延迟，连续跑N次，把延迟序列存下来。

```python
import time
import requests

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

def measure_rest_polling(symbol="AAPL", region="US", rounds=200):
    latencies = []
    for _ in range(rounds):
        # 记录请求发出前的时间（毫秒）
        t0 = time.time() * 1000
        resp = requests.get(
            f"{API_BASE}/stock/quote",
            headers=headers,
            params={"region": region, "code": symbol}
        )
        data = resp.json()["data"]
        # 行情时间戳 t 是毫秒
        latency = t0 - data["t"]
        latencies.append(latency)
        time.sleep(1)  # 模拟真实轮询间隔
    return latencies

lats = measure_rest_polling()
lats.sort()
n = len(lats)
print(f"样本数: {n}")
print(f"最小: {lats[0]:.0f}ms, 平均: {sum(lats)/n:.0f}ms, P95: {lats[int(n*0.95)]:.0f}ms")
```

这里有个细节值得注意：`t0` 记的是"请求发出前"，但请求从发出到响应回来还有一个网络往返时间。严格说，REST轮询拿到的价格是**上一次成交**的价格，等你处理完这笔数据，行情可能已经又走了一段。所以REST的"有效延迟"天然比WebSocket大——它多了一个完整的请求-响应周期，哪怕你每秒刷一次也一样。

另外轮询频率是有上限的。一秒一刷，一小时就是3600个请求，一整天开着，请求量非常可观。对个人开发者来说，这既是成本问题，也是被限流的问题。

## WebSocket推送怎么测

WebSocket这边，数据是服务端主动推过来的，测量逻辑变成：每收到一条报价消息，就用消息里的 `t` 和本地接收时间算延迟。

```python
import json
import threading
import time
import websocket

WS_URL = "wss://api.itick.org/stock"
TOKEN = "your_token_here"
authenticated = False
latencies = []

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",
            "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":
        now = time.time() * 1000
        latency = now - data["t"]
        latencies.append(latency)

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()
```

跑一段时间，比如等几百条消息进来，然后离线分析 `latencies`：

```python
lats = sorted(latencies)
n = len(lats)
print(f"样本数: {n}")
print(f"最小: {lats[0]:.0f}ms, 平均: {sum(lats)/n:.0f}ms, P95: {lats[int(n*0.95)]:.0f}ms")
```

WebSocket的延迟有两个特点：一是**推送频率跟着行情走**——盘口活跃的时候消息多，清淡的时候消息少，不像轮询那样固定节奏；二是**连接建立后只占一个长连接**，不管订阅多少个标的，请求量都是恒定的，这是它跟轮询最本质的区别。

## 对比维度，不只是延迟

把两边都测完，你会得到两组数字。但选择方案不能只看延迟，我建议用四个维度一起看：

| 维度 | REST轮询 | WebSocket推送 |
|---|---|---|
| 延迟 | 多一个请求-响应周期，端到端延迟偏高 | 服务端主动推送，延迟低且稳定 |
| 请求量 | 与轮询频率成正比，长时间运行成本高 | 连接建立后恒定，不随标的数量增长 |
| 资源占用 | 每次请求都要建连、HTTP开销 | 一个长连接常驻，需要维护心跳 |
| 复杂度 | 简单，一个requests循环就够 | 需要处理鉴权、心跳、断线重连 |

结论其实很清晰：**低频、短时、脚本化场景用REST，高频、长时、多标场景用WebSocket**。日频任务、一次性拉快照、测试脚本，REST足够；盘中盯盘、事件驱动、跨标的实时监控，走WebSocket。

## 实测之外的几个坑

测量过程中顺便把几个老坑也验证了一遍，写下来免得大家再踩：

**鉴权时序。** WebSocket连接建立不等于可以订阅，必须等 `resAc:"auth"` 且 `code:1` 再发subscribe。我见过直接连上就订阅的代码，数据一条都收不到，程序还不报错。

**心跳不能省。** 文档要求每30秒发一次心跳，不发的话长连接会被服务端断开。跑测量脚本一挂几小时，心跳漏了，后半段数据就全没了。

**时间戳单位。** 报价和K线的时间戳都是毫秒，直接用 `time.time()`（秒）去减会得到一个巨大的负数。统一用毫秒口径，或者写个工具函数转换，别在业务代码里随手除。

**交易状态要判断。** 美股盘前、盘中、盘后的行情特征差别很大，`ts` 字段标识交易状态。做延迟测试最好选正常交易时段，不然测出来的分布毫无参考价值。

## 小结

"轮询还是推送"这个问题，答案不在网上，在你的测量脚本里。把口径定好、两边各跑一遍、四维度对比，你的场景该用哪个自然就清楚了。对大多数做美股行情的个人开发者来说，最终方案通常是：REST拿快照和历史数据，WebSocket做盘中实时监控，两者各司其职。






做美股行情工具或者量化策略的人，几乎都会纠结同一个问题：实时行情到底该用HTTP轮询还是WebSocket推送？网上的说法五花八门，有人说不轮询就是不懂行情，有人说WebSocket才是正经解法。与其听别人讲，不如自己写代码测一遍——把两种方式接到同一个数据源上，用统一的口径量一量延迟、请求量、资源占用，结论自然就出来了。

Python美股实时行情接入：REST轮询与WebSocket推送的延迟，自己动手测一次

本文通过实测对比REST轮询与WebSocket推送在美股行情接入中的表现，统一以“数据到达本地时间 - 行情生成时间戳”为延迟口径，量化比较延迟、请求量、资源占用与复杂度。结果表明：轮询适合低频、短时任务，但延迟高、成本大；WebSocket延迟更低、稳定且请求恒定，适用于高频、长时监控。结合场景选择，建议日频用轮询，盘中实时监控用推送，两者互补。实测还揭示了鉴权时序、心跳维护、时间戳单位等关键

编程语言

美股行情工具开发中，HTTP轮询与WebSocket推送的实时性对比实测：统一采用数据时间戳到本地的端到端延迟口径，分别测REST与WS的延迟、P95、资源占用。结果显示WebSocket延迟更低更稳定，REST简单但请求量大。最终建议：低频脚本用REST，高频实时监控用WebSocket，需注意鉴权时序、心跳及时间戳单位。

往返时间

代码解释

WebSocket

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

美股行情工具开发中，HTTP轮询与WebSocket推送的实时性对比实测：统一采用数据时间戳到本地的端到端延迟口径，分别测REST与WS的延迟、P95、资源占用。结果显示WebSocket延迟更低更稳定，REST简单但请求量大。最终建议：低频脚本用REST，高频实时监控用WebSocket，需注意鉴权时序、心跳及时间戳...

Python美股实时行情接入：REST轮询与WebSocket推送的延迟，自己动手测一次-腾讯云开发者社区-腾讯云

Python美股实时行情接入：REST轮询与WebSocket推送的延迟，自己动手测一次

Python美股实时行情接入：REST轮询与WebSocket推送的延迟，自己动手测一次

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐