美股有 4 个交易时段，为什么你的代码只接了 1 个

第一次接美股 API，大概率只订了正盘（09:30–16:00）的数据。

这个决定让你每天有 8.5 小时是瞎的——盘前 4 小时、盘后 4 小时、夜盘另算。更隐蔽的问题是：70% 以上的美股财报在盘后发布，非农和 CPI 数据在盘前 08:30 发布。策略代码如果不知道当前处于哪个时段，它无法判断是否该执行——盘前流动性只有正盘的几分之一，市价单在这时候滑点会大到你怀疑人生。

一、trade_session：API 文档里被跳过的字段

美股 4 个时段对应一个标准枚举：

调用交易时段接口：

GET /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},
    {"begin_time": 2000, "end_time": 400,  "trade_session": 3}
  ]
}

begin_time 和 end_time 的格式是 hhmm，不是毫秒——400 是 4:00 AM ET，930 是 9:30 AM ET。这是新手最容易踩的第一个坑，把它当时间戳处理，时段判断直接全错。

二、5 个真实工程坑

三、代码：两版实现

主版本：直接调交易时段接口（推荐）

这是正确的工程做法——从接口拿权威时段定义，不自己硬编码时间边界：

import requests
from datetime import datetime
from zoneinfo import ZoneInfo

API_KEY = "YOUR_API_KEY"   # 替换为你的 API Key
BASE_URL = "https://api.tickdb.ai"
EASTERN = ZoneInfo("America/New_York")

# 缓存时段表（避免每次判断都调接口）
_session_cache: dict = {}


def fetch_sessions(market: str = "US") -> list:
    """从接口获取交易时段定义，结果缓存到内存"""
    global _session_cache
    today = datetime.now(EASTERN).date().isoformat()
    cache_key = f"{market}_{today}"

    if cache_key in _session_cache:
        return _session_cache[cache_key]

    resp = requests.get(
        f"{BASE_URL}/v1/market/trading-sessions",
        params={"market": market},
        headers={"X-API-Key": API_KEY},
        timeout=10
    )
    resp.raise_for_status()

    sessions = resp.json()["trading_sessions"]
    _session_cache[cache_key] = sessions
    return sessions


def get_current_session(market: str = "US") -> int:
    """
    返回当前 trade_session 枚举值：
    0=盘中, 1=盘前, 2=盘后, 3=夜盘, -1=休市
    """
    sessions = fetch_sessions(market)
    now_et = datetime.now(EASTERN)
    t = now_et.hour * 100 + now_et.minute

    for s in sessions:
        begin = s["begin_time"]
        end = s["end_time"]
        session_type = s["trade_session"]

        if begin > end:
            # 跨午夜时段（如夜盘 2000–0400）
            if t >= begin or t < end:
                return session_type
        else:
            if begin <= t < end:
                return session_type
    return -1


if __name__ == "__main__":
    names = {0: "盘中", 1: "盘前", 2: "盘后", 3: "夜盘", -1: "休市"}
    session = get_current_session("US")
    print(f"当前美股时段: {names[session]}")

    # 查港股时段（含午休）
    session_hk = get_current_session("HK")
    print(f"当前港股时段: {names[session_hk]}")

备选版本：yfinance（个人项目 / 本地回测）

yfinance 没有 trade_session 字段，需要自己根据时间戳映射：

import yfinance as yf
from zoneinfo import ZoneInfo

EASTERN = ZoneInfo("America/New_York")

# 本地时段定义（手动维护，无法自动同步数据源更新）
LOCAL_SESSIONS = [
    (400,  930,  1),   # 盘前
    (930,  1600, 0),   # 盘中
    (1600, 2000, 2),   # 盘后
    (2000, 2400, 3),   # 夜盘前半
    (0,    400,  3),   # 夜盘后半（跨午夜）
]


def get_premarket_yf(symbol: str = "AAPL", days: int = 5):
    """用 yfinance 拉取盘前数据（个人项目够用，生产环境注意限频）"""
    ticker = yf.Ticker(symbol)
    # prepost=True 是关键，不加只有正盘数据
    df = ticker.history(period=f"{days + 2}d", interval="1m", prepost=True)

    if df.empty:
        return df

    # 必须先转为美东时间再过滤，否则夏令时会偏移
    df_et = df.tz_convert(EASTERN)

    # inclusive="left" 确保不包含 09:30（正盘第一分钟）
    premarket = df_et.between_time("04:00", "09:30", inclusive="left")
    return premarket

两个版本的核心差异：

四、按时段切换订阅的生产逻辑

时段感知不只是"知道现在几点"，更重要的是按时段调整策略行为：

# 生产版要点：按时段切换订阅列表
CORE_SYMBOLS = ["AAPL.US", "TSLA.US", "NVDA.US"]   # 核心 50 只（示例用 3 只）
FULL_SYMBOLS  = ["AAPL.US", "TSLA.US", "NVDA.US", "MSFT.US", "GOOGL.US"]  # 全量


def get_symbols_by_session(session: int) -> list:
    """按时段返回应订阅的标的列表"""
    if session == 1:    # 盘前：流动性薄，只订核心品种
        return CORE_SYMBOLS[:50]
    elif session == 0:  # 盘中：全量
        return FULL_SYMBOLS
    else:               # 盘后/夜盘：精简
        return CORE_SYMBOLS[:20]


async def on_reconnect():
    """WebSocket 断线重连后，重新判断时段，重新决定订阅配置"""
    session = get_current_session("US")
    symbols = get_symbols_by_session(session)
    await resubscribe(symbols)
    # 重连后不能假设时段未变——跨交易日或夏令时切换都会改变边界

工程预警：fetch_sessions() 不要每次都调 REST——时段定义不会分钟级变化。缓存到内存，只在日期切换或 WebSocket 重连时刷新，日常查询走内存，O(1) 开销。

五、为什么多市场会让这件事变复杂

如果策略同时监控美股盘前、A 股早盘集合竞价、港股 Pre-open，三套时段规则完全不同：

• 美股：4 个时段，夜盘跨午夜
• 港股：有午休（12:00–13:00），Pre-open 是独立时段
• A 股：有集合竞价（09:15–09:25）、连续竞价、尾盘集合竞价

自己维护这三套映射逻辑，代码里是三套 if/else、三套时区处理、三套字段解析。每次数据源调整时段边界，三个地方都要改。

TickDB 的 trade_session 接口对三个市场返回同一套枚举——market 参数传 US、HK 或 CN，返回结构一致，差异封装在 API 层。这把多市场时段问题从"业务逻辑"降级为"基础设施"，策略代码只需要关心 trade_session 的值，不关心底层各市场怎么定义的。

有了美股 API ≠ 有了盘前数据。时段部分往往是文档里被跳过的章节——不是因为不重要，是因为正盘能跑通的代码已经够忙的了，时段判断看起来是"以后再说"的事，直到漏掉一次财报盘后异动。

你的策略跑在哪个时段上？有没有因为没接盘前数据错过信号？评论区聊聊。

本文数据来自 TickDB。代码依赖：Python 3.9+（zoneinfo 内置）、requests、yfinance（备选版）。