一文搞定美股盘前、盘后、夜盘数据：从 API 到实战代码

由bq46wiqq创建，最终由bq46wiqq更新于2026-04-23 07:51 被浏览 4 用户

特斯拉财报发布在收盘后，股价瞬间跳涨 12%。各大财经媒体弹窗推送，社交媒体一片沸腾。你打开量化系统一看——最新价格纹丝不动，还是收盘时的数字。

不是数据源没推，是你的系统没接。

美股实际上每天有约 16 个小时可以交易，盘前 04:00、盘后到 20:00、部分品种还有夜盘。财报、美联储声明、重大并购，九成以上在盘前或盘后发布。如果你的行情系统只覆盖 09:30–16:00，你看到的开盘价其实是别人已经完成定价之后的价格。

这篇文章解决一个具体问题：用 Python 把美股盘前、盘中、盘后三段数据都接进来，一个 API，一套代码，能跑的那种。

一、美股四个交易时段

都是美东时间（EST/EDT）：

先把时间表记下来，下面的代码和逻辑全围着这四段转。

二、用 API 确认市场的时段定义

不同 API 对"盘前盘后"的定义可能略有差异（有的只到 19:00），所以先用接口查一下：

curl -H "X-API-Key: YOUR_API_KEY" \
  "https://api.tickdb.ai/v1/market/trading-sessions?market=US"

返回结构大概是这样：

{
  "market": "US",
  "trading_sessions": [
    { "begin_time": "400",  "end_time": "930",  "trade_session": 1 },
    { "begin_time": "930",  "end_time": "1600", "trade_session": 0 },
    { "begin_time": "1600", "end_time": "2000", "trade_session": 2 }
  ]
}

trade_session 字段的含义：0 盘中 / 1 盘前 / 2 盘后 / 3 夜盘。

一个容易踩的坑：这个字段只在 /trading-sessions 接口里出现。WebSocket 推送的实时 ticker 不会把 trade_session 塞在每条消息里——美股 WebSocket 消息只有三个字段：symbol、last_price、timestamp。

这不是 API 的设计缺陷，这是正确的设计。每条推送少塞两个字段，高频场景下省下的带宽很可观。时段判定放在客户端做，毫秒不到。

三、客户端做时段映射

用 Python 写一个纯本地的时段判断函数，拿到 timestamp 就能立刻知道当前是哪个时段。

import pytz
from datetime import datetime

EASTERN = pytz.timezone("US/Eastern")  # 自动处理夏令时

def get_trade_session(timestamp_ms: int) -> int:
    """
    根据 UTC 毫秒时间戳返回美股交易时段
    0=盘中, 1=盘前, 2=盘后, 3=夜盘, -1=非交易时段
    """
    dt = datetime.fromtimestamp(timestamp_ms / 1000, tz=EASTERN)
    hour = dt.hour + dt.minute / 60.0

    if 9.5 <= hour < 16.0:
        return 0
    elif 4.0 <= hour < 9.5:
        return 1
    elif 16.0 <= hour < 20.0:
        return 2
    elif hour >= 20.0 or hour < 4.0:
        return 3
    return -1

关于时区：一定要用 pytz 的 aware datetime（带时区信息），不要用 datetime.utcnow() 这种无时区的对象。代码里混用有时区和无时区的 datetime，是金融系统里最经典的一类 bug——数据对得上价格对得上，就是时间差了几个小时。养成习惯：所有时间戳在内存里都存毫秒整数，要展示给人看的时候才转成带时区的 datetime。

四、WebSocket 订阅：盘前盘后一起来

订阅 TSLA 的实时 ticker，推送进来的每条都本地算出时段：

import asyncio
import json
import websockets

API_KEY = "YOUR_API_KEY"
WS_URL = f"wss://api.tickdb.ai/v1/realtime?api_key={API_KEY}"

SESSION_NAMES = {0: "盘中", 1: "盘前", 2: "盘后", 3: "夜盘", -1: "闭市"}

async def monitor():
    async with websockets.connect(WS_URL) as ws:
        # 订阅
        await ws.send(json.dumps({
            "cmd": "subscribe",
            "data": {"channel": "ticker", "symbols": ["TSLA.US"]}
        }))

        # 每秒发一次 ping，保活
        async def ping():
            while True:
                await ws.send(json.dumps({"cmd": "ping"}))
                await asyncio.sleep(1)
        asyncio.create_task(ping())

        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("cmd") != "ticker":
                continue

            d = msg["data"]
            session = get_trade_session(d["timestamp"])
            print(f"[{SESSION_NAMES[session]}] {d['symbol']}  价格: {d['last_price']}")

asyncio.run(monitor())

跑起来你会看到，从凌晨 4 点开始就有盘前数据推进来，一直到晚上 8 点盘后结束。

五、盘前异动监控：一个实战场景

"盘前涨幅超过 3% 触发报警"是很多开盘策略的前置条件。但美股 WebSocket 推送里不带涨跌幅，得自己算——需要一个昨收价。

拿昨收价最干净的方式是拉一根日 K 线：

import requests

def get_prev_close(symbol: str, api_key: str) -> float:
    """用日 K 线接口取前一交易日收盘价"""
    resp = requests.get(
        "https://api.tickdb.ai/v1/market/kline",
        headers={"X-API-Key": api_key},
        params={"symbol": symbol, "interval": "1d", "limit": 2}
    )
    data = resp.json()
    klines = data.get("data", {}).get("klines", [])
    # 倒数第二根是前一交易日（最后一根是今天正在形成的）
    if len(klines) >= 2:
        return float(klines[-2]["close"])
    return float(klines[-1]["close"]) if klines else 0

把它和 WebSocket 组合起来，加上异动判断和 60 秒防重复报警：

import asyncio
import json
import requests
import websockets

API_KEY = "YOUR_API_KEY"
WS_URL = f"wss://api.tickdb.ai/v1/realtime?api_key={API_KEY}"
WATCH = ["TSLA.US", "NVDA.US", "AAPL.US", "META.US", "AMZN.US"]
THRESHOLD = 3.0  # 盘前涨跌幅触发阈值 %

async def premarket_alert():
    # 启动时拉一遍所有标的的昨收价
    prev_close = {s: get_prev_close(s, API_KEY) for s in WATCH}
    print(f"昨收价加载完成: {prev_close}")

    last_alert_ts = {}  # 防重复报警

    async with websockets.connect(WS_URL) as ws:
        await ws.send(json.dumps({
            "cmd": "subscribe",
            "data": {"channel": "ticker", "symbols": WATCH}
        }))

        async def ping():
            while True:
                await ws.send(json.dumps({"cmd": "ping"}))
                await asyncio.sleep(1)
        asyncio.create_task(ping())

        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("cmd") != "ticker":
                continue

            d = msg["data"]
            # 只看盘前
            if get_trade_session(d["timestamp"]) != 1:
                continue

            sym = d["symbol"]
            price = float(d["last_price"])
            prev = prev_close.get(sym, 0)
            if prev == 0:
                continue

            change = (price - prev) / prev * 100
            if abs(change) < THRESHOLD:
                continue

            # 同一标的 60 秒内不重复报警
            now = asyncio.get_event_loop().time()
            if now - last_alert_ts.get(sym, 0) < 60:
                continue
            last_alert_ts[sym] = now

            arrow = "↑" if change > 0 else "↓"
            print(f"🚨 盘前异动 {arrow} {sym}  {change:+.2f}%  现价: {price}  昨收: {prev}")

asyncio.run(premarket_alert())

这段代码在美东 04:00–09:30 运行（北京时间 17:00–22:30 左右，夏令时有差异），盘前任何标的动超过 3% 就打一条。你可以把 print 换成企业微信/钉钉 Webhook，变成真正的报警系统。

六、历史盘前数据：回测用

回测策略需要过去的盘前数据。用 REST K 线接口拉 1 分钟 K 线，然后客户端按时间戳过滤：

import pytz
import requests
from datetime import datetime, timedelta

EASTERN = pytz.timezone("US/Eastern")

def fetch_premarket_klines(symbol: str, days: int, api_key: str):
    """拉过去 N 天的 1 分钟 K 线，客户端过滤盘前段"""
    end = datetime.now(EASTERN)
    start = end - timedelta(days=days)

    resp = requests.get(
        "https://api.tickdb.ai/v1/market/kline",
        headers={"X-API-Key": api_key},
        params={
            "symbol": symbol,
            "interval": "1m",
            "start_time": int(start.astimezone(pytz.UTC).timestamp() * 1000),
            "end_time": int(end.astimezone(pytz.UTC).timestamp() * 1000),
            "limit": 1000
        }
    )
    klines = resp.json().get("data", {}).get("klines", [])

    # 客户端按美东时间筛出盘前段 04:00–09:30
    premarket = []
    for k in klines:
        dt = datetime.fromtimestamp(k["time"] / 1000, tz=EASTERN)
        hour = dt.hour + dt.minute / 60.0
        if 4.0 <= hour < 9.5:
            premarket.append(k)

    return premarket

bars = fetch_premarket_klines("AAPL.US", days=5, api_key=API_KEY)
print(f"盘前 1 分钟 K 线: {len(bars)} 根")

七、几个真实会踩的坑

1. 数字类型是字符串，不是 float API 返回的价格字段是字符串形式（避免浮点精度问题），比如 "last_price": "242.35"。用之前记得 float()，否则字符串比较会给你惊喜。

2. 夏令时切换那一周 每年 3 月中和 11 月初，美国切换夏令时。用 pytz.timezone("US/Eastern") 会自动处理；但如果你在代码里硬编码 UTC-5 或 UTC-4，切换那周数据必然错一小时。

3. 盘前流动性真的很薄 盘前报价跳动可能只是因为一笔几百股的成交，不代表趋势。阈值设得太敏感容易误报，建议结合成交量做二次过滤——比如要求触发时同时满足"当前分钟成交量 > 最近 5 根均值的 3 倍"。

4. WebSocket 会断，要自己处理重连 生产环境要把上面的 async with 包一层 while True + try/except，断开后休眠 3–5 秒重连。心跳丢失、网络抖动都会导致连接断开。

八、结尾

• 代码涉及的接口都在 TickDB API 文档：https://docs.tickdb.ai
• 完整示例代码：https://github.com/TickDB/tickdb-unified-realtime-marketdata-api
• API Key 在 https://tickdb.ai 注册后控制台生成

盘前盘后是美股一整天行情里信息密度最高的窗口。接进来的成本其实不高，但不接的机会成本会在某一天的财报季里被放大很多倍。