美股有 4 个交易时段，为什么你的代码只接了 1 个
TickDB Research ·
第一次接美股 API，大概率只订了正盘（09:30–16:00）的数据。
这个决定让你每天有 8.5 小时是瞎的——盘前 4 小时、盘后 4 小时、夜盘另算。更隐蔽的问题是：70% 以上的美股财报在盘后发布，非农和 CPI 数据在盘前 08:30 发布。策略代码如果不知道当前处于哪个时段，它无法判断是否该执行——盘前流动性只有正盘的几分之一，市价单在这时候滑点会大到你怀疑人生。
一、trade_session：API 文档里被跳过的字段
美股 4 个时段对应一个标准枚举：
| 时段 | trade_session 值 | 美东时间 | 关键特征 |
|---|---|---|---|
| 盘前 | 1 | 04:00–09:30 | 流动性薄，机构试盘，重大新闻消化窗口 |
| 盘中 | 0 | 09:30–16:00 | 流动性最充沛，价差最小 |
| 盘后 | 2 | 16:00–20:00 | 财报集中发布，价格反应剧烈 |
| 夜盘 | 3 | 20:00–次日 04:00 | 亚太资金主导，部分品种支持 |
调用交易时段接口：
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 个真实工程坑
| 坑 | 具体表现 | 解法 |
|---|---|---|
| 时区陷阱 | API 返回美东时间，服务器跑 UTC，夏令时切换偏移 1 小时 | 用 zoneinfo 做 ET↔UTC 转换，自动识别夏令时 |
| 字段格式 | begin_time=400 是 4:00 AM，不是 400 毫秒 | 解析：hour = val // 100; minute = val % 100 |
| 夜盘跨午夜 | begin_time=2000, end_time=400，开始大于结束 | 判断用 OR：t >= 2000 or t < 400 |
| 缓存缺失 | 每次时段判断都调 REST 接口，高频策略下开销不可忽视 | 时段表缓存内存，只在日期切换或重连时刷新 |
| 数据源差异 | trade_session=3 夜盘并非所有数据源都支持 | 接入前确认覆盖范围，不要假设支持 |
三、代码：两版实现
主版本：直接调交易时段接口（推荐）
这是正确的工程做法——从接口拿权威时段定义，不自己硬编码时间边界：
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
两个版本的核心差异：
| 对比项 | 接口版 | yfinance 版 |
|---|---|---|
| 时段定义来源 | 接口动态获取，自动同步 | 本地硬编码，需手动维护 |
| 多市场支持 | 一个函数，market 参数切换 | 港股/A股需要换库 |
| 实时推送 | 支持 WebSocket | 只有 REST 轮询 |
| 适用场景 | 生产环境 | 个人项目 / 回测 |
四、按时段切换订阅的生产逻辑
时段感知不只是"知道现在几点"，更重要的是按时段调整策略行为：
# 生产版要点：按时段切换订阅列表
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（备选版）。