美股行情 API 中的夏令时陷阱、盘前盘后数据与市场熔断机制

很多做过 A 股或港股行情接口的开发者，在切换到美股数据时都会遇到同一批问题：行情接口明明通了，但开盘时间的判断每年总有几天是错的；K 线数据里出现了奇怪的"夜间 bar"，价格跳得很厉害却查不到任何公告；一只热门股突然不再推送 tick，不知道是行情中断了还是接口出了问题；回测框架在某些日期里计算出了不可能存在的策略收益……

这些坑有一个共同来源：美股有几个 A 股完全没有的市场机制，如果用处理 A 股的思路直接套，代码会在最不起眼的地方出错。

本文逐一把这些机制讲清楚，并给出配套的接口调用示例。

---

一、美股代码格式与交易时间全景

代码格式

美股代码统一使用 .US 后缀，不区分 NYSE（纽约证券交易所）和 NASDAQ（纳斯达克）。接口层面两个交易所的数据可以通过同一套接口获取：

股票	代码	所属交易所
苹果	AAPL.US	NASDAQ
微软	MSFT.US	NASDAQ
英伟达	NVDA.US	NASDAQ
特斯拉	TSLA.US	NASDAQ
谷歌（Alphabet A）	GOOGL.US	NASDAQ
亚马逊	AMZN.US	NASDAQ
Meta	META.US	NASDAQ
摩根大通	JPM.US	NYSE
伯克希尔哈撒韦 B	BRK.B.US	NYSE
埃克森美孚	XOM.US	NYSE

注意 BRK.B.US 这类带点的代码，在拼接 URL 时要注意 . 的转义问题（部分 HTTP 框架会把路径中的 . 特殊处理）。

交易时段全景

美股没有午休，交易日从早到晚可以分为四个时段：

时段	美东时间（ET）	北京时间（冬令时 UTC-5）	北京时间（夏令时 UTC-4）
盘前交易（Pre-market）	04:00–09:30	17:00–22:30 次日	16:00–21:30 次日
正常交易（Regular session）	09:30–16:00	22:30–次日 05:00	21:30–次日 04:00
盘后交易（After-hours）	16:00–20:00	次日 05:00–09:00	次日 04:00–08:00

这里有两个对我们国人来说非常陌生的东西：夏令时和盘前盘后交易。下面分别详细展开。

---

二、夏令时：每年两次让判断逻辑出错的陷阱

什么是夏令时

美国使用夏令时（Daylight Saving Time，DST）。每年 3 月第二个周日凌晨 2:00，时钟拨快一小时，进入 EDT（UTC-4）；11 月第一个周日凌晨 2:00，时钟拨回一小时，回到 EST（UTC-5）。

对中国开发者来说，这意味着：

• 冬令时（EST，UTC-5）：北京时间 22:30 是美股正常交易开盘，次日 05:00 收盘
• 夏令时（EDT，UTC-4）：北京时间 21:30 是美股正常交易开盘，次日 04:00 收盘

每年有两个日期，北京时间的开盘时间会突然提前或推后一小时。 如果你的代码写的是 if beijing_hour == 22 and beijing_minute == 30: market_open()，那么夏令时期间每天都会晚开一个小时。

2026 年的切换日期：

• 夏令时开始：2026 年 3 月 8 日（周日）凌晨 2:00 ET
• 夏令时结束：2026 年 11 月 1 日（周日）凌晨 2:00 ET

错误示范与正确做法

危险写法：硬编码 UTC 偏移量

from datetime import datetime, timezone, timedelta

# ❌ 错误：只在冬令时正确，夏令时时开盘判断会晚1小时
EST = timezone(timedelta(hours=-5))

def is_us_open_wrong() -> bool:
    now_est = datetime.now(EST)
    t = now_est.time()
    return (now_est.weekday() < 5
            and t.hour == 9 and t.minute >= 30
            or t.hour in range(10, 16))

正确做法：使用命名时区，让库自动处理 DST 切换

from datetime import datetime
import pytz   # pip install pytz

ET = pytz.timezone("America/New_York")   # 自动处理 EST/EDT 切换

def us_session_status() -> str:
    """返回当前美股市场状态（自动适配夏令时）。"""
    now = datetime.now(ET)
    if now.weekday() >= 5:      # 周六/周日
        return "休市"
    h, m = now.hour, now.minute
    minutes = h * 60 + m
    if   240 <= minutes < 570:  return "盘前"      # 04:00–09:30
    elif 570 <= minutes < 960:  return "正常交易"   # 09:30–16:00
    elif 960 <= minutes < 1200: return "盘后"      # 16:00–20:00
    else:                       return "休市"

print(us_session_status())

标准库 zoneinfo（Python 3.9+）是 pytz 的现代替代，无需额外安装：

from datetime import datetime
from zoneinfo import ZoneInfo

ET = ZoneInfo("America/New_York")

now_et = datetime.now(ET)
print(f"当前美东时间：{now_et.strftime('%Y-%m-%d %H:%M %Z')}")
# 输出示例（夏令时）：2026-06-23 09:47 EDT
# 输出示例（冬令时）：2026-01-15 14:32 EST

一句话结论：任何涉及美股交易时间判断的代码，都用 America/New_York 时区，永远不要写 UTC-5 或 UTC-4。

---

三、行情接口实战

美股（NYSE + NASDAQ）使用 /stock/ 接口路径，与 A 股、港股共用同一套接口体系，通过代码后缀 .US 区分。

3.1 实时成交明细

import requests

API_KEY = "YOUR_API_KEY"
BASE    = "https://data.infoway.io"

# Mag 7 —— 七大科技巨头
codes = "AAPL.US,MSFT.US,NVDA.US,TSLA.US,GOOGL.US,AMZN.US,META.US"

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']:<12} ${tick['p']:>10}  成交量={tick['v']:>10}股  {direction}")

输出示例：

AAPL.US      $  213.45  成交量=   1284300股  ↑买
MSFT.US      $  430.12  成交量=    892400股  ↓卖
NVDA.US      $  134.88  成交量=   2341500股  ↑买
TSLA.US      $  248.30  成交量=   4128200股  ↑买
GOOGL.US     $  179.64  成交量=   1073600股  —
AMZN.US      $  224.91  成交量=   1487300股  ↓卖
META.US      $  582.37  成交量=    641800股  ↑买

注意：美股 v 字段是股数（shares），没有"手"的概念。美股最小交易单位是 1 股，部分平台支持分数股（fractional shares），但接口层面通常以整股报量。

返回字段：

字段	说明
s	股票代码
t	成交时间戳（毫秒，UTC）
p	最新成交价（美元）
v	成交量（股）
vw	成交额（美元）
td	方向：0=默认，1=主买，2=主卖

3.2 实时盘口深度

resp = requests.get(
    f"{BASE}/stock/batch_depth/NVDA.US",
    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"  ${float(price):>8.2f}   {vol:>10} 股")

spread = round(float(asks[0][0]) - float(bids[0][0]), 2)
print(f"\n  价差：${spread:.2f}\n")

print("买盘（bid）")
for price, vol in bids[:5]:
    print(f"  ${float(price):>8.2f}   {vol:>10} 股")

3.3 历史 K 线下载

美股 K 线同样通过 POST 请求获取，支持按时间戳翻页向前查询：

import requests, json, time as _time

def fetch_us_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(
        f"{BASE}/stock/v2/batch_kline",
        headers={"apiKey": 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_one_year(symbol: str) -> list[dict]:
    """下载近一年日 K 线数据。"""
    all_bars: list[dict] = []
    cursor: int | None  = None
    import time
    one_year_ago = int(time.time()) - 365 * 86400

    while True:
        batch = fetch_us_klines(symbol, kline_type=8, count=500, until_ts=cursor)
        if not batch:
            break
        all_bars.extend(batch)
        oldest = int(batch[-1]["t"])
        if oldest < one_year_ago or (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_one_year("NVDA.US")
print(f"英伟达日K共 {len(bars)} 根 | 最新收盘：${bars[-1]['c']}")

klineType 参数说明：1=1分钟，2=5分钟，3=15分钟，4=30分钟，5=1小时，8=日K，9=周K，10=月K。（详情见官方文档）

---

四、盘前盘后数据：A 股开发者没见过的扩展交易时段

美股独有一个机制：除了 09:30–16:00 ET 的正常交易时段，股票还可以在盘前（04:00–09:30）和盘后（16:00–20:00）交易。这两段时间合称"延伸交易时段"（Extended Hours Trading）。

为什么这件事对数据处理很重要

财报几乎都发布在盘后。美国上市公司发布季度财报的惯例是在正常交易结束后的盘后（或下一个交易日开盘前），这样可以避免财报数据在正常交易中引发的极端波动。

结果就是：一只股票今天 16:00 收盘是 200 美元，盘后财报发布，利润超预期 30%，到今天 18:00 盘后交易价格已经涨到 240 美元，明天 09:30 的正常开盘价大概率也是 240 美元附近。如果你的 K 线数据只包含正常交易时段，你会看到今天 200 → 明天开盘 240，中间没有任何过渡过程——这个跳空完全是正常的盘后行情导致的，而不是真空跳跃。

盘前盘后行情的数据特征

与正常交易时段相比，延伸时段有几个固有特征：

特征	说明
流动性差	参与者少，订单薄，价差（ask-bid spread）通常是正常时段的 5–20 倍
价格波动大	小额订单就能推动价格大幅移动，技术形态参考价值低
期权市场关闭	衍生品不参与，盘前/盘后的价格缺乏期权 delta 对冲的平滑作用
执行成本高	实际成交价格可能与行情显示价格偏差明显

如何区分正常时段与延伸时段的 tick

拿到一条成交明细时，可以通过时间戳判断它属于哪个时段：

from datetime import datetime
from zoneinfo import ZoneInfo

ET = ZoneInfo("America/New_York")

def classify_tick(ts_ms: int) -> str:
    """判断 tick 属于哪个交易时段（毫秒时间戳，UTC）。"""
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=ET)
    if dt.weekday() >= 5:
        return "非交易日"
    h, m = dt.hour, dt.minute
    minutes = h * 60 + m
    if   240 <= minutes < 570:  return "盘前"
    elif 570 <= minutes < 960:  return "正常交易"
    elif 960 <= minutes < 1200: return "盘后"
    else:                       return "非交易时段"

# 使用示例
for tick in resp.json()["data"]:
    session = classify_tick(tick["t"])
    print(f"{tick['s']}  ${tick['p']}  时段={session}")

这个分类在聚合日内数据时非常有用：如果你在构建 OHLCV 日 bar，应当明确决定是否将盘前/盘后数据并入当天的 high/low，还是只统计正常时段的极值。两种做法都有合理场景，但必须显式选择，不能混用。

---

五、美股没有涨跌停，但有两类熔断机制

这是最容易被 A 股开发者忽略的一个设计差异。

个股层面：没有固定涨跌停，但有"交易暂停"

美股个股没有 ±10% 或 ±20% 的每日涨跌幅限制。理论上一只股票可以在一天内跌去 90%（极端情况如公司破产消息）或涨数倍（如被收购时的溢价）。

但这不意味着完全没有保护机制。SEC 实施了一套叫 LULD（Limit Up-Limit Down） 的规则：如果一只标普 500 成分股在 5 分钟内价格移动超过 5%，或者非成分股超过 10%/20%，交易所会触发 15 秒的"暂停竞价"窗口，价格被限制在一个区间内，超出区间的订单不得成交，超时后恢复。

从行情数据角度来看，LULD 暂停期间，对应股票的 tick 推送会短暂中断，但不会像 A 股涨跌停那样整个交易日锁死。通常 15–60 秒后恢复正常推送。

市场层面：S&P 500 指数熔断

当 S&P 500 指数单日跌幅达到以下阈值时，全市场（NYSE + NASDAQ）暂停交易：

触发条件	暂停时长	触发限制
下跌 7%（Level 1）	15 分钟	仅在 15:25 ET 之前可触发
下跌 13%（Level 2）	15 分钟	仅在 15:25 ET 之前可触发
下跌 20%（Level 3）	当日收盘	任何时间均可触发

历史上最近一次触发是 2020 年 3 月新冠疫情初期，在 8 个交易日内连续触发了 4 次 Level 1 熔断。

对数据消费者的影响：市场级熔断触发时，你订阅的所有股票 tick 会同时停止推送。判断是市场级熔断还是自己的连接断了，可以通过两点：① 所有订阅的标的同时静默（连接断开只会影响你的客户端，不会让多个不相关的股票同时停止）；② 熔断发生时间通常伴随着消息面（可以结合时间戳判断是否处于熔断可能发生的条件下）。

---

六、WebSocket 实时订阅：Mag 7 实时监控

以下是一个完整的美股 WebSocket 客户端，订阅 Mag 7 的实时成交和 1 分钟 K 线，包含断线重连和心跳保活：

import asyncio, json, uuid, logging
from datetime import datetime
from zoneinfo import ZoneInfo
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("us-stock-ws")

ET     = ZoneInfo("America/New_York")
MAG7   = "AAPL.US,MSFT.US,NVDA.US,TSLA.US,GOOGL.US,AMZN.US,META.US"


def us_session_status() -> str:
    now = datetime.now(ET)
    if now.weekday() >= 5:
        return "休市"
    m = now.hour * 60 + now.minute
    if   240 <= m < 570:  return "盘前"
    elif 570 <= m < 960:  return "正常交易"
    elif 960 <= m < 1200: return "盘后"
    return "休市"


class USStockClient:

    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:
        # 订阅实时成交
        await self._send({"code": 10000, "trace": self._trace(),
                          "data": {"codes": MAG7}})
        await asyncio.sleep(5)
        # 订阅 1 分钟 K 线
        await self._send({"code": 10006, "trace": self._trace(),
                          "data": {"arr": [{"type": 1, "codes": MAG7}]}})
        log.info("订阅完成 | 当前时段：%s | ET时间：%s",
                 us_session_status(),
                 datetime.now(ET).strftime("%H:%M %Z"))

    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"), "?")
            session   = us_session_status()
            prefix    = "【盘前】" if session == "盘前" else ("【盘后】" if session == "盘后" else "")
            log.info("%s成交 %-12s $%-10s vol=%-10s %s",
                     prefix, data.get("s"), data.get("p"), data.get("v"), direction)

        elif code == 10008: # K 线推送
            pct = data.get("pfr", "0%")
            log.info("K线  %-12s 开=$%-8s 高=$%-8s 低=$%-8s 收=$%-8s 涨跌=%s",
                     data.get("s"), data.get("o"), data.get("h"),
                     data.get("l"), data.get("c"), pct)

        elif code in (10001, 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 | ET：%s",
                     us_session_status(), datetime.now(ET).strftime("%H:%M %Z"))
            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:
            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 后重连... | 当前时段：%s", backoff, us_session_status())
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)

    def stop(self) -> None:
        self.running = False


async def main():
    client = USStockClient(api_key="YOUR_API_KEY")
    await client.start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        log.info("已停止")

注意 _dispatch 里对盘前/盘后时段的标注。如果你的系统对延伸时段的数据有不同处理逻辑（比如不触发价格预警，或者单独存储），可以在这里插入判断。

---

七、财报日行情：从 tick 数据识别暴涨暴跌时机

美股财报季（通常是每个季度末后约 3–6 周）是最活跃的盘后行情时段。以下是几个从行情数据本身识别财报日异常的方法：

成交量突增检测

财报发布后盘后成交量通常是平日盘后的 10–50 倍。一个简单的异常检测：

from collections import deque

class VolumeAnomalyDetector:
    """基于滑动窗口的成交量异常检测器。"""

    def __init__(self, window: int = 20, threshold: float = 5.0):
        self.window    = window       # 历史窗口大小（分钟 K 数量）
        self.threshold = threshold    # 突增倍数阈值
        self._history: dict[str, deque] = {}

    def feed(self, symbol: str, volume: int) -> bool:
        """
        传入最新分钟 K 的成交量，返回 True 表示检测到异常放量。
        """
        if symbol not in self._history:
            self._history[symbol] = deque(maxlen=self.window)
        hist = self._history[symbol]

        if len(hist) >= 5:   # 至少有 5 根历史 K 才开始检测
            avg = sum(hist) / len(hist)
            if avg > 0 and volume > avg * self.threshold:
                return True

        hist.append(volume)
        return False


detector = VolumeAnomalyDetector(window=20, threshold=5.0)

# 在 WebSocket 的 K 线推送处理中使用：
# if code == 10008:
#     sym = data.get("s")
#     vol = int(data.get("v", 0))
#     if detector.feed(sym, vol):
#         log.warning("⚠ 成交量异常放量 %s，当前 vol=%d（可能是财报或重大事件）", sym, vol)

盘后价格跳空检测

收盘后进入盘后时段，如果最新 tick 价格与正常收盘价偏差超过阈值，可以认为发生了盘后重大事件：

def check_afterhours_gap(close_price: float, latest_price: float,
                          threshold_pct: float = 3.0) -> str:
    pct_change = (latest_price - close_price) / close_price * 100
    if pct_change >= threshold_pct:
        return f"盘后上涨 {pct_change:.1f}%，疑似利好财报"
    elif pct_change <= -threshold_pct:
        return f"盘后下跌 {pct_change:.1f}%，疑似利空财报"
    return "盘后价格平稳"

# 示例
print(check_afterhours_gap(200.0, 240.0))   # → 盘后上涨 20.0%，疑似利好财报
print(check_afterhours_gap(200.0, 176.0))   # → 盘后下跌 12.0%，疑似利空财报

---

八、常见问题与错误速查

现象	原因	处理方式
开盘时间判断每年有几天不对	用了固定 UTC 偏移量，没有考虑夏令时切换	改用 America/New_York 命名时区
K 线里出现"夜间 bar"，价格波动异常	盘前/盘后 tick 被拼入了日 bar	过滤掉延伸时段的数据，或单独处理
某只股票 tick 突然停止几十秒	LULD 触发了个股交易暂停	属正常现象，静默后会自动恢复，不要触发重连
所有订阅标的同时停止推送超过 15 分钟	可能触发了市场级熔断（Level 1/2）	检查 S&P 500 当日跌幅；熔断结束后自动恢复
历史价格某天出现大幅跳空	股票分拆（如 NVDA 曾做过 10:1 分拆）导致复权价跳变	使用前复权数据进行历史对比，原始价格不适合跨分拆日计算涨跌幅
接口返回 ret: 508	代码格式错误或股票已退市/更名	检查是否带 .US 后缀；部分股票更名后代码变化（如 Facebook → META.US）
接口返回 ret: 429	请求频率超出当前套餐限制	降低轮询频率，或改用 WebSocket 订阅代替轮询
WebSocket 在 60 秒内自动断连	未发送心跳包	每 30 秒发送一次 {"code": 10010, ...} 心跳

---

小结

美股行情数据的坑，几乎都来自三个地方：

• 夏令时：用 America/New_York 命名时区，永远不要写 UTC-5 或 UTC-4
• 延伸时段：盘前盘后是真实交易，不是噪声，财报行情大概率就在这里
• 无涨跌停 + 双层熔断：个股理论上无限制（但有 LULD 暂停），市场级有 7%/13%/20% 三档熔断

接口层面和 A 股、港股基本一致，换个代码后缀（.US）和接口路径（/stock/）就能接通。真正需要额外处理的是时区和时段逻辑——把这部分做对，剩下的事情不会比 A 股更复杂。