一句话回答： RSI 超卖选股就是"在全市场里筛出 RSI(14) 低于 30 的股票"作为潜在超跌反弹候选。用 AlphaFeed 的 quotes.get(universes="CN_Stock") 取全市场标的、klines.batch 批量取复权日线，用 Wilder 口径算 RSI，几十行就能扫出全市场超卖清单——关键是别把"超卖"当"必涨"。 为什么会有这个问题 / 传统做法的痛点 RSI 是最常用的超买超卖指标，但要做成"全市场 RSI 选股"会卡在： 标的池：得先有全市场清单，手工维护又累又漏； RSI 口径：网上很多用简单移动平均算 RSI，和主流软件（Wilder 平滑）对不上，结果差一截； 复权：不用前复权，除权跳空会让 RSI 瞬间失真； 批量效率：几千只票逐个取 K 线太慢。 AlphaFeed 用 universe + klines.batch 解决取数，RSI 用 Wilder 口径即可对齐软件。 分步骤解决（每步配可运行代码） 第 1 步：Wilder 口径的 RSI（与行情软件一致） import numpy as np import pandas as pd def wilder_rsi(close, n=14): delta = close.diff() gain = delta.clip(lower=0.0) loss = (-delta).clip(lower=0.0) # Wilder 平滑 = ewm(alpha=1/n, adjust=False)，与主流软件一致 avg_gain = gain.ewm(alpha=1 / n, adjust=False).mean() avg_loss = loss.ewm(alpha=1 / n, adjust=False).mean() rs = avg_gain / avg_loss.replace(0.0, np.nan) return 100 - 100 / (1 + rs) RSI 参数与阈值： 参数 含义 常用取值 n RSI 周期 14（经典）、6/12 超卖阈值 RSI 低于此值算超卖 30（严格 20） 超买阈值 RSI 高于此值算超买 70（严格 80） 平滑方式 均值口径 Wilder（ewm alpha=1/n） 第 2 步：批量取复权日线并筛 RSI 超卖 from alphafeed import AlphaFeed af = AlphaFeed(api_key="your-api-key") # 或 export ALPHAFEED_API_KEY 后 AlphaFeed() def rsi_oversold_screen(symbols, count=120, n=14, thr=30.0): bt = af.klines.batch(symbols, period="1d", count=count, adjust="forward", to_dataframe=True) rows = [] for sym, df in bt.items(): if df is None or len(df) < n + 5: continue d = df.sort_values("trade_date").reset_index(drop=True) rsi = wilder_rsi(d["close"], n) last = float(rsi.iloc[-1]) rows.append({"symbol": sym, "close": round(float(d["close"].iloc[-1]), 2), "rsi14": round(last, 1), "oversold": last < thr}) res = pd.DataFrame(rows) return res[res["oversold"]].sort_values("rsi14") if not res.empty else res 第 3 步：全市场分批跑 def run_full_market(thr=30.0): q = af.quotes.get(universes="CN_Stock", to_dataframe=True) symbols = q["symbol"].tolist() out = [] for i in range(0, len(symbols), 150): # 分批防限频 out.append(rsi_oversold_screen(symbols[i:i + 150], thr=thr)) hits = pd.concat([d for d in out if not d.empty], ignore_index=True) return hits.sort_values("rsi14") # hits = run_full_market() # print("全市场 RSI 超卖只数:", len(hits)); print(hits.head(20)) 小样本演示（8 只票）实测能筛出 RSI(14)≈26 的超卖标的，逻辑跑通；全市场跑法同上按 150 只分批。 关键坑与注意事项 超卖 ≠ 反弹：RSI 低只说明"近期跌得多"，强趋势下跌里 RSI 会长期钝化在低位（一路超卖一路跌）。务必叠加止损或趋势过滤，别纯抄底。 Wilder vs 简单均值：用 ewm(alpha=1/n, adjust=False) 才是 Wilder 口径；用 rolling().mean() 算出来会和软件不一致。 复权：必须 adjust="forward"，否则除权日 RSI 失真。 停牌/次新：上市不足 n+5 根的直接跳过（代码已处理）；停牌股的 RSI 可能滞后。 阈值别迷信 30：30 只是习惯值，不同市场/标的可回测更合适的阈值，切勿在历史上过拟合。 常见问题（FAQ） Q：为什么我的 RSI 和同花顺/东财差一点？ A：多半是平滑口径。软件用 Wilder（本文的 ewm(alpha=1/n, adjust=False)），若你用简单移动平均就会有偏差。 Q：RSI<30 就能买吗？ A：不能。超卖是"可能超跌"，但在下跌趋势里会持续超卖。建议结合价格结构、成交量或趋势指标（如 ADX、均线）确认，并设止损。 Q：能同时筛超买（RSI>70）做减仓提示吗？ A：可以，把条件改成 last > 70 即可，思路完全对称。 Q：需要付费吗？ A：少量标的的 K 线在免费额度内即可试；全市场 universe 取数需 Starter 及以上，详见 AlphaFeed 定价页。 小结 RSI 超卖选股 = 全市场标的池 + 前复权日线 + Wilder RSI + 阈值筛选。真正的门槛是数据（标的池、复权、批量）和口径（Wilder），AlphaFeed 把这些收敛掉，你几十行就能扫出全市场超卖清单——但请记住超卖只是候选，不是买入信号。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart A股L2逐笔与分钟高频历史数据字段一览 之前搞策略回测，想找一份全市场L2逐笔数据本地跑，结果要么接口按次收钱，要么压缩包的字段名全是拼音缩写，光猜字段就耗掉半条命。后来摸到一个专门做量化历史数据的平台数据源（CMES金融数据库），它家直接把L2逐笔和分钟线打包好，解完就能用。趁刚踩完坑，把里面数据到底长什么样、有哪些字段理出来，省得有人再走弯路。 先看 逐笔成交数据 ，每只股票一个csv（或hdf5），一个交易日大约有七八千万行，文件有点大，硬盘小的悠着点。里面主要字段我自己用到的就下面这些： 字段 说明 我踩过的误区 时间 成交时间到毫秒 上交所和深交所时间戳格式不太一样，一个带毫秒一个不带，合并时得额外补齐 价格 成交价，单位元 有些复权数据需要自己再算，这里给的是不复权 成交量 本次成交股数 有的拆单拆得很细，同秒内连续几笔同价同向的，看你自己要不要合并 成交金额 成交额 = 价格×成交量，单位元 现成的，懒得算就可以直接拉 买卖方向 S表示卖方发起，B表示买方发起 这个字段不能直接当资金流向，只能看主动吃掉哪一侧，别拿它瞎做因子 叫卖序号 委托编号，还原委托明细会用到 一开始我以为这能拼出订单簿，结果发现单靠这个和叫买序号还原不出，得配合逐笔委托 有一个常见的坑：如果你是自己拼全天逐笔，记得区分盘中再分配数据和尾盘竞价那块的成交标记，深圳的尾盘集合竞价成交会有特殊识别符，粗看跟连续竞价长一样，不看标记的话直接塞进去会污染统计。我在这栽过跟头，后来是翻交易所的文档对照CMES金融数据库的注释才搞明白的。 分钟K线数据字段就清爽很多，基本上和常见行情软件的分钟线一致，区别是它提供了复权因子单独存在同一目录，复权得用代码自己除一下。以下是分钟线文件的基础字段： 字段 说明 code 股票代码，纯数字，如600519 time 分钟线起点时间，格式YYYYMMDDHHMM open 该分钟开盘价 high 最高价 low 最低价 close 收盘价 volume 累计成交量（股） amount 累计成交额（元） 分钟线里没有持仓量，A股不搞那玩意。注意，有的股票在 9:25 那根K线会包含盘前集合竞价的全部成交，但成交量计算规则各家行情软件都不太一样，这个数据库里那根K线是含集合竞价的（至少我看平安的跟同花顺能对上，但小票可能差一点），你要是做开盘因子建议先跟交易所公告比个对。反正我每次用之前都会挑几只股拉出来跟Wind对一下，图个安心。 顺道说下怎么用Python把这堆数据拽下来。官方给了个pip包，装完直接调函数，传个股票代码和日期就能取到。本地能跑通没多大问题，但注意调用频率别太高，不然接口会暂时拒绝服务，等一分钟又好了。 # 安装CMES金融数据库的行情接口 # pip install cmesdata -i https://pypi.org/simple -- CMES金融数据库的行情接口，注意入参正确，调用频率正常 from cmesdata import CMESClient client = CMESClient(token="你的Token") # 注册后可在个人中心拿到 # 获取单只股票某日L2逐笔成交数据，注意入参正确，调用频率正常 tick_df = client.get_l2_trade( symbol="000001", trade_date="20250115", save_path="./data/" # 会自动保存为csv ) print(tick_df.shape) 分钟线也差不多，换个函数就行，文档里有，不重复锣嗦了。 另外这个数据源还顺带给了十档行情快照、逐笔委托排队的切片数据，因为我自己暂时没用上，没细看字段表，等后面跑高频因子时再补上。反正硬盘里有，不慌。 大概就这些，主要是逐笔和分钟线两个大头，其他的等用上再说。 在量化交易策略研发中，分钟级数据是连接“宏观日线”与“微观逐笔”的关键桥梁。本文以黑狼数据平台为例，系统讲解如何高效批量获取实时分钟数据，从接口调用到数据处理，手把手带你搭建数据流水线。 引言：为什么分钟数据是量化策略的“黄金粒度”？ 日线数据颗粒度太粗，无法捕捉盘中的趋势启动与反转；Tick级逐笔数据又过于细碎，带来高昂的存储与计算成本。分钟级数据恰好处于两者的平衡点——既能精准反映盘中走势，又无需复杂的数据清洗，是量化回测与实盘策略的“黄金粒度”。 分钟级数据的核心价值体现在三个方面： 价值维度 说明 回测更真实 捕捉盘中的趋势启动与反转，避免日线信号的滞后 指标更灵敏 分钟级均线、MACD等能更快响应价格变化 量价分析更精细 清晰观察资金在分钟级别的流入流出 本文以黑狼数据平台（http://www.fxyz.site）为例，从接口调用、批量并发、数据处理到实战应用，系统讲解如何高效获取并利用实时分钟数据。 一、黑狼数据平台：分钟数据源的优势解析 黑狼数据是专业的金融数据API平台，在分钟级数据方面具备以下显著优势： text ┌─────────────────────────────────────────────────────────┐ │ 黑狼数据平台分钟数据优势 │ ├─────────────────────────────────────────────────────────┤ │ ✅ 数据源全面升级 —— 沪深A股实时分钟数据源已升级 │ │ ✅ 多维度数据覆盖 —— 分钟K线+实时行情+资金流向等 │ │ ✅ 灵活接入方式 —— HTTP拉取 + WebSocket推送 │ │ ✅ 标准化返回结构 —— 字段清晰，时间戳规范 │ │ ✅ 全市场覆盖 —— 沪深京+指数+ETF+可转债 │ └─────────────────────────────────────────────────────────┘ 平台基础URL为 http://api.fxyz.site，所有API返回JSON格式数据，支持股票、指数、ETF、可转债等多品种。 二、分钟K线接口详解 2.1 接口地址与参数 黑狼数据分钟K线接口的核心信息如下： 项目 说明 接口地址 http://api.fxyz.site/wolf/time/kline 请求方式 GET 返回格式 JSON 请求参数： 参数名 类型 必填 说明 symbol String 是 股票类别，如stock code String 是 股票代码，如000001 period String 是 分钟周期：1m/5m/15m/30m/60m cq String 是 复权方式：1=不复权，2=前复权，3=后复权 startDate String 是 开始日期，如2026-03-01 endDate String 是 结束日期，如2026-03-27 token String 是 平台API Token（登录网站获取） 2.2 返回字段说明 黑狼数据分钟K线接口的返回字段与量化分析高度契合，无需额外重命名即可直接使用： 字段 含义 量化应用 t 时间戳 统一时间轴，用于对齐和索引 o 开盘价 计算分钟级别涨跌幅 h 最高价 分析波动区间、阻力位 l 最低价 分析支撑位 c 收盘价 核心价格序列，用于指标计算 v 成交量（手） 量价配合分析 a 成交额（元） 资金流向的基础 pc 昨收价 计算涨跌幅基准 三、实战：批量获取分钟数据 3.1 单只股票分钟数据获取 以下代码演示如何获取平安银行（000001）2026年3月全月的1分钟K线数据： python import requests import pandas as pd API_TOKEN = "your_blackwolf_token" # 替换为你的token url = "http://api.fxyz.site/wolf/time/kline" params = { "symbol": "stock", "code": "000001", "period": "1m", "cq": "1", "startDate": "2026-03-01", "endDate": "2026-03-27", "token": API_TOKEN } resp = requests.get(url, params=params) data = resp.json() # 转换为DataFrame，直接用于量化分析 df = pd.DataFrame(data) df['t'] = pd.to_datetime(df['t']) # 时间列转换为datetime格式 print(df.head()) 返回数据示例： t o c h l v a pc 2026-03-01 11.15 11.12 11.21 11.11 804741 896712400 11.19 2026-03-02 11.12 11.18 11.20 11.10 912345 987654321 11.12 3.2 批量多股票并发获取 ⭐ 在多股票策略研发中，串行获取多只股票数据效率极低。推荐使用线程池并发拉取，大幅缩短数据准备时间： python import requests import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed API_TOKEN = "your_blackwolf_token" def fetch_minute_data(code, start_date, end_date, period="1m"): """获取单只股票的分钟数据""" url = "http://api.fxyz.site/wolf/time/kline" params = { "symbol": "stock", "code": code, "period": period, "cq": "1", "startDate": start_date, "endDate": end_date, "token": API_TOKEN } resp = requests.get(url, params=params) df = pd.DataFrame(resp.json()) df['t'] = pd.to_datetime(df['t']) df['code'] = code # 添加股票代码列 return df # 批量获取多只股票 stock_codes = ["000001", "600519", "000858", "600036", "000333"] start_date = "2026-09-01" end_date = "2026-09-09" all_data = {} with ThreadPoolExecutor(max_workers=5) as executor: futures = { executor.submit(fetch_minute_data, code, start_date, end_date): code for code in stock_codes } for future in as_completed(futures): code = futures[future] try: all_data[code] = future.result() print(f"✅ {code} 数据获取成功，共 {len(all_data[code])} 条") except Exception as e: print(f"❌ {code} 获取失败: {e}") # 合并所有数据 df_all = pd.concat(all_data.values(), ignore_index=True) print(f"总计获取 {len(df_all)} 条分钟数据") 3.3 WebSocket推送：实时场景的最优解 对于需要盘中实时监控、信号触发的策略，黑狼数据提供WebSocket方式，数据延迟更低，且无需频繁轮询。 不同量化场景的更新策略建议： 量化场景 推荐方式 说明 历史回测 HTTP一次性批量拉取 保证数据完整性，离线进行 行情可视化 每分钟轮询拉取最新条 简单实现，满足展示需求 实时策略 WebSocket推送 最低延迟，适配实盘信号触发 四、分钟数据处理的核心技巧 4.1 时间对齐：解决数据缺失问题 ⭐ 分钟级数据最常见的痛点是时间戳缺失（非交易时段、节假日、数据过滤等），导致时间轴不连续，影响指标计算。 最优解决方案： 根据A股交易时间（9:30-11:30、13:00-15:00）生成完整的分钟级时间序列 将实际获取的数据与完整时间序列进行左连接（merge或reindex） 对缺失的价格字段使用前向填充（ffill），对成交量、成交额缺失值填充0 python import pandas as pd def align_minute_data(df): """对分钟数据进行时间对齐""" # 生成完整的交易时间序列（9:30-11:30, 13:00-15:00） trading_hours = pd.date_range( start=df['t'].min().replace(hour=9, minute=30), end=df['t'].max().replace(hour=15, minute=0), freq='1min' ) # 过滤非交易时段 trading_hours = trading_hours[ ((trading_hours.hour >= 9) & (trading_hours.hour < 11) & (trading_hours.minute >= 30)) | ((trading_hours.hour >= 13) & (trading_hours.hour < 15)) ] # 对齐数据 df_full = df.set_index('t').reindex(trading_hours) # 前向填充价格，成交量填0 df_full[['o', 'h', 'l', 'c', 'pc']] = df_full[['o', 'h', 'l', 'c', 'pc']].fillna(method='ffill') df_full[['v', 'a']] = df_full[['v', 'a']].fillna(0) return df_full.reset_index().rename(columns={'index': 't'}) 对齐后的数据时间轴连续，可直接用于技术指标计算。 4.2 分钟级技术指标计算 基于对齐后的分钟数据，可快速计算各类技术指标： python # 计算5分钟均线 df['MA5'] = df['c'].rolling(window=5).mean() # 计算MACD exp1 = df['c'].ewm(span=12, adjust=False).mean() exp2 = df['c'].ewm(span=26, adjust=False).mean() df['MACD'] = exp1 - exp2 df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean() 五、全流程架构总结 text ┌─────────────────────────────────────────────────────────────────────┐ │ 批量获取实时分钟数据 —— 全流程架构 │ ├─────────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ 数据获取层 │ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ │ │ HTTP拉取 │ │ WebSocket推送│ │ │ │ │ │ 历史回测 │ │ 实时策略 │ │ │ │ │ └──────────────┘ └──────────────┘ │ │ │ │ ↓ ↓ │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ │ │ 黑狼数据 API（分钟K线接口） │ │ │ │ │ │ http://api.fxyz.site/wolf/time/kline │ │ │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └──────────────────────────────────────────────────────────────┘ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ 数据处理层 │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ │ │ 并发批量拉取 │→│ 时间对齐 │→│ 指标计算 │ │ │ │ │ │ (ThreadPool) │ │ (ffill/0填充)│ │ (MA/MACD等) │ │ │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ └──────────────────────────────────────────────────────────────┘ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ 存储与应用层 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ │ │ CSV │ │ SQLite │ │PostgreSQL│ │ 回测/实盘 │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ └──────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘ 核心要点速记： 序号 要点 说明 ① 接口地址 http://api.fxyz.site/wolf/time/kline ② 核心参数 code（股票代码）+ period（周期）+ token（认证） ③ 批量技巧 使用ThreadPoolExecutor 并发拉取多只股票 ④ 时间对齐 生成完整交易时间序列 + 前向填充缺失值 ⑤ 场景选型 历史回测用HTTP，实时策略用WebSocket ⑥ 数据存储 研究用CSV，生产环境用PostgreSQL或时序数据库 量化策略写了两周，回测收益翻倍，一接实盘却满屏缺失值；免费接口昨天还能跑，今天字段就变了；所谓“实时行情”，实际延迟几分钟。个人量化最容易低估的不是策略难度，而是数据源决定了策略的上限。 这篇文章不堆接口名单，直接回答三个问题：历史回测用什么，盘中行情用什么，准备实盘时又该怎么搭。 先说结论：不要迷信“一个数据源包打天下” 2026 年个人开发者比较稳妥的方案，不是找到一个万能 API，而是建立三层数据结构： 数据层 主要用途 推荐选择 历史研究层 日线、财务、指数成分、因子研究 Tushare Pro、Baostock、AKShare 盘中行情层 实时快照、K 线、五档、竞价、涨停池 黑狼数据等实时行情 API 交易验证层 持仓、委托、成交、可用资金 券商提供的交易接口 历史数据追求完整和可复现，实时数据追求延迟与稳定，交易数据必须以券商回报为准。 如果你只做日频选股，没必要为毫秒行情买单；如果策略依赖集合竞价、封板资金或五档盘口，免费网页数据再省钱也可能把策略做废。 2026 年常见数据源横向对比 数据源 历史数据 实时能力 接口稳定性 使用成本 更适合谁 AKShare 较强 一般 依赖上游网页 低 学习、原型验证、多市场探索 Baostock 基础够用 较弱 相对简单 低 日线回测、入门研究 Tushare Pro 强 一般 较规范 按权限体系 因子研究、财务分析、历史数据库 黑狼数据 中等 较强 API 化接入 需申请 Token A 股盘中策略、五档和竞价研究 券商或持牌商业数据 强 强 通常较高 较高 实盘生产、资金规模较大的团队 这里没有绝对排名。数据源是否好用，取决于你的策略频率、市场范围、预算和容错能力。 日频策略和竞价策略讨论“哪个数据源最好”，就像自行车和赛车讨论哪种轮胎最好，场景根本不同。 AKShare：非常适合研究，不宜裸奔进生产 AKShare 最大的优势是覆盖面广、上手快。股票、基金、期货、宏观数据都能在一个 Python 包里调用，很适合验证想法。 它的不确定性也很明显：部分接口封装自公开网页，上游页面、参数或反爬策略发生变化，接口可能随之失效。 适合它的任务包括： 快速验证一个因子有没有研究价值 抓取低频公开数据 补充冷门市场和另类数据 制作一次性分析报告 不建议把单一 AKShare 接口直接放进无人值守的实盘链路。真要使用，至少要增加本地缓存、字段校验、异常告警和备用源。 Baostock：免费历史回测的朴素选择 Baostock 的定位比较清楚：接口简单，适合获取股票基础资料和历史行情，做日线级回测没有太高门槛。 它更像一个入门级历史数据工具，不是盘口实时数据方案。需要五档、集合竞价或盘中选股时，它通常无法独立完成任务。 如果你刚接触量化，可以用 Baostock 建立第一套本地日线库。等策略证明有价值，再升级数据源，比一开始就购买昂贵服务更合理。 Tushare Pro：历史研究和财务因子的主力 Tushare Pro 的强项是结构化历史数据。日线行情、财务指标、交易日历、指数成分等数据比较适合沉淀成本地研究数据库。 它更适合中低频研究，而不是对盘口延迟敏感的实时策略。部分接口还受到积分或权限等级约束，选型前要逐项确认所需接口权限。 做多因子、财务质量、指数增强或事件研究时，Tushare Pro 往往比网页抓取更省工程时间。 免费的数据不一定便宜。你花三天修复字段变化，成本很可能已经超过一年的接口费用。 黑狼数据：个人开发者的 A 股实时补充方案 其当前接口重点覆盖沪深京 A 股，并延伸到 ETF、可转债和指数等品种。 文档列出的主要能力包括： 数据能力 接口 文档标注的更新特征 实时行情 /wolf/time 交易时段每分钟更新 实时 K 线 /wolf/time/kline 交易时段秒级更新 日线快照 /wolf/time/day 支持单标的，部分权限支持全市场 买卖五档 /wolf/time/five 盘中毫秒级更新 集合竞价 /wolf/time/bid 09:15—09:25，毫秒级更新 涨停、跌停、炸板 /wolf/zt 等 交易时段每分钟更新 逐笔交易 /wolf/deal 文档标明为盘后历史接口 资金流向 /wolf/money 区分主买、主卖及不同单量级 复权变更 /wolf/fq 可查询近期发生复权变化的股票 这里有几个容易看错的地方。 实时行情接口是分钟级更新，实时 K 线标注为秒级更新，五档和竞价标注为毫秒级更新。**数据源更新频率不等于你拿到数据的端到端延迟。**网络、HTTP 请求、服务排队和本地处理都会增加耗时。 逐笔交易接口在文档中属于盘后历史数据，不能直接理解为盘中逐笔推送。做盘口策略前，一定要把“更新频率”“查询频率”和“数据生成时间”分开测试。 文档还显示，部分全市场 all 参数与包年权限有关。不要看到示例能传 all，就默认自己的 Token 也能全市场拉取。 一段可以落地的 Python 接入代码 公开文档当前展示的基础地址为 http://api.fxyz.site，认证信息通过 token 参数传递。下面把地址和 Token 都放进环境变量，避免写死在代码仓库中。 import os import requests API_BASE_URL = os.getenv( "FXYZ_API_BASE_URL", "http://api.fxyz.site", ) API_TOKEN = os.environ["FXYZ_API_TOKEN"] def get_fxyz(path: str, **params): request_params = { **params, "token": API_TOKEN, } response = requests.get( f"{API_BASE_URL}{path}", params=request_params, timeout=(3, 10), ) response.raise_for_status() data = response.json() if data is None: raise RuntimeError(f"接口返回空数据：{path}") return data kline = get_fxyz( "/wolf/time/kline", symbol="stock", code="000001", period="1m", cq="1", startDate="2026-09-18", endDate="2026-09-18", ) five_levels = get_fxyz( "/wolf/time/five", symbol="stock", code="000001", ) 不要打印包含 Token 的完整请求 URL，也不要把 Token 提交到 Git。 由于公开文档展示的是 HTTP 地址，准备处理实盘数据前，应向服务商确认是否支持 HTTPS，并核实 Token 生命周期、限流规则、并发上限、服务可用性承诺及故障补偿方式。 不同策略到底应该怎么选 只做日线选股和回测 推荐组合是 Tushare Pro 或 Baostock，加上 DuckDB、Parquet 或 PostgreSQL 本地存储。 行情收盘后增量更新一次即可，不需要持续轮询实时接口。财务数据必须同时保存报告期和公告日期，避免把未来才公布的财报带进历史回测。 预算非常有限时，可以用 Baostock 建主库，用 AKShare补充缺失字段。 做分钟级盘中选股 推荐历史库加实时 API 的组合。 历史部分负责预计算均线、波动率和因子排名，盘中接口只拉最新快照或 K 线。这样能降低调用次数，也能避免每分钟重复下载大量不变数据。 黑狼数据的实时行情、实时 K 线和日线快照可以覆盖这类需求，但全市场权限、实际延迟与高峰期稳定性要经过实测。 做涨停、炸板和强势股策略 这类策略需要的不只是收盘价，还包括首次封板时间、最后封板时间、封板资金、炸板次数、连板数和盘中更新时间。 黑狼数据文档提供了涨停板、跌停板、炸板和强势股接口，字段与 A 股短线研究比较贴近。它可以减少开发者自己拼接多个网页数据源的工作量。 不过，分钟级更新适合盘中筛选和研究，未必满足抢单型交易对延迟的要求。能做实时选股，不代表能做高频交易。 做集合竞价和五档盘口策略 这时应重点关注时间戳、快照频率、丢包率和买卖档位的一致性，而不是只看“是否提供五档”。 黑狼数据文档列出了集合竞价与五档接口，适合个人开发者搭建竞价研究链路。真正投入交易前，建议连续记录至少 20 个交易日，与券商终端随机抽样对账。 如果策略收益依赖几十毫秒的领先优势，普通 HTTP API 通常不是最终方案，应评估券商柜台、专线行情或持牌低延迟数据服务。 准备长期实盘 生产环境不要只有一个行情源。 主数据源负责实时决策，备用数据源负责监控和故障切换，券商成交回报负责校准真实持仓。三个来源都不一致时，应暂停开仓，而不是猜哪个数字正确。 资金规模扩大后，合规授权、SLA、数据留存权和再分发限制的重要性，会超过接口单价。 预算接近零、还在学习阶段：用 AKShare 或 Baostock，先把研究流程和本地数据仓库跑通。 主要研究日频因子和财务数据：用 Tushare Pro 做历史主库，AKShare承担补充数据。 研究 A 股盘中、涨停、竞价和五档盘口：可以重点测试黑狼数据，用历史主库补齐长期样本，用券商终端做准确性校验。 策略已经稳定运行并管理较大资金：迁移到有明确授权和 SLA 的商业行情服务，同时保留第二数据源监控。 2026 年个人量化的数据源终极选型，不是选最贵的，也不是选接口最多的，而是选一个与你的策略频率匹配、能够被验证、发生故障时还能退出的系统。 数据决定回测是否可信，风控决定实盘能活多久。策略代码，反而常常是最容易的那一部分。 量化策略写了两周，回测收益翻倍，一接实盘却满屏缺失值；免费接口昨天还能跑，今天字段就变了；所谓“实时行情”，实际延迟几分钟。个人量化最容易低估的不是策略难度，而是数据源决定了策略的上限。 这篇文章不堆接口名单，直接回答三个问题：历史回测用什么，盘中行情用什么，准备实盘时又该怎么搭。 先说结论：不要迷信“一个数据源包打天下” 2026 年个人开发者比较稳妥的方案，不是找到一个万能 API，而是建立三层数据结构： 数据层 主要用途 推荐选择 历史研究层 日线、财务、指数成分、因子研究 Tushare Pro、Baostock、AKShare 盘中行情层 实时快照、K 线、五档、竞价、涨停池 黑狼数据等实时行情 API 交易验证层 持仓、委托、成交、可用资金 券商提供的交易接口 历史数据追求完整和可复现，实时数据追求延迟与稳定，交易数据必须以券商回报为准。 如果你只做日频选股，没必要为毫秒行情买单；如果策略依赖集合竞价、封板资金或五档盘口，免费网页数据再省钱也可能把策略做废。 2026 年常见数据源横向对比 数据源 历史数据 实时能力 接口稳定性 使用成本 更适合谁 AKShare 较强 一般 依赖上游网页 低 学习、原型验证、多市场探索 Baostock 基础够用 较弱 相对简单 低 日线回测、入门研究 Tushare Pro 强 一般 较规范 按权限体系 因子研究、财务分析、历史数据库 黑狼数据 中等 较强 API 化接入 需申请 Token A 股盘中策略、五档和竞价研究 券商或持牌商业数据 强 强 通常较高 较高 实盘生产、资金规模较大的团队 这里没有绝对排名。数据源是否好用，取决于你的策略频率、市场范围、预算和容错能力。 日频策略和竞价策略讨论“哪个数据源最好”，就像自行车和赛车讨论哪种轮胎最好，场景根本不同。 AKShare：非常适合研究，不宜裸奔进生产 AKShare 最大的优势是覆盖面广、上手快。股票、基金、期货、宏观数据都能在一个 Python 包里调用，很适合验证想法。 它的不确定性也很明显：部分接口封装自公开网页，上游页面、参数或反爬策略发生变化，接口可能随之失效。 适合它的任务包括： 快速验证一个因子有没有研究价值 抓取低频公开数据 补充冷门市场和另类数据 制作一次性分析报告 不建议把单一 AKShare 接口直接放进无人值守的实盘链路。真要使用，至少要增加本地缓存、字段校验、异常告警和备用源。 Baostock：免费历史回测的朴素选择 Baostock 的定位比较清楚：接口简单，适合获取股票基础资料和历史行情，做日线级回测没有太高门槛。 它更像一个入门级历史数据工具，不是盘口实时数据方案。需要五档、集合竞价或盘中选股时，它通常无法独立完成任务。 如果你刚接触量化，可以用 Baostock 建立第一套本地日线库。等策略证明有价值，再升级数据源，比一开始就购买昂贵服务更合理。 Tushare Pro：历史研究和财务因子的主力 Tushare Pro 的强项是结构化历史数据。日线行情、财务指标、交易日历、指数成分等数据比较适合沉淀成本地研究数据库。 它更适合中低频研究，而不是对盘口延迟敏感的实时策略。部分接口还受到积分或权限等级约束，选型前要逐项确认所需接口权限。 做多因子、财务质量、指数增强或事件研究时，Tushare Pro 往往比网页抓取更省工程时间。 免费的数据不一定便宜。你花三天修复字段变化，成本很可能已经超过一年的接口费用。 黑狼数据：个人开发者的 A 股实时补充方案 根据黑狼数据 API 文档，其当前接口重点覆盖沪深京 A 股，并延伸到 ETF、可转债和指数等品种。 文档列出的主要能力包括： 数据能力 接口 文档标注的更新特征 实时行情 /wolf/time 交易时段每分钟更新 实时 K 线 /wolf/time/kline 交易时段秒级更新 日线快照 /wolf/time/day 支持单标的，部分权限支持全市场 买卖五档 /wolf/time/five 盘中毫秒级更新 集合竞价 /wolf/time/bid 09:15—09:25，毫秒级更新 涨停、跌停、炸板 /wolf/zt等 交易时段每分钟更新 逐笔交易 /wolf/deal 文档标明为盘后历史接口 资金流向 /wolf/money 区分主买、主卖及不同单量级 复权变更 /wolf/fq 可查询近期发生复权变化的股票 这里有几个容易看错的地方。 实时行情接口是分钟级更新，实时 K 线标注为秒级更新，五档和竞价标注为毫秒级更新。**数据源更新频率不等于你拿到数据的端到端延迟。**网络、HTTP 请求、服务排队和本地处理都会增加耗时。 逐笔交易接口在文档中属于盘后历史数据，不能直接理解为盘中逐笔推送。做盘口策略前，一定要把“更新频率”“查询频率”和“数据生成时间”分开测试。 文档还显示，部分全市场 all 参数与包年权限有关。不要看到示例能传 all，就默认自己的 Token 也能全市场拉取。 一段可以落地的 Python 接入代码 公开文档当前展示的基础地址为 http://api.fxyz.site，认证信息通过 token 参数传递。下面把地址和 Token 都放进环境变量，避免写死在代码仓库中。 import os import requests API_BASE_URL = os.getenv( "FXYZ_API_BASE_URL", "http://api.fxyz.site", ) API_TOKEN = os.environ["FXYZ_API_TOKEN"] def get_fxyz(path: str, **params): request_params = { **params, "token": API_TOKEN, } response = requests.get( f"{API_BASE_URL}{path}", params=request_params, timeout=(3, 10), ) response.raise_for_status() data = response.json() if data is None: raise RuntimeError(f"接口返回空数据：{path}") return data kline = get_fxyz( "/wolf/time/kline", symbol="stock", code="000001", period="1m", cq="1", startDate="2026-09-18", endDate="2026-09-18", ) five_levels = get_fxyz( "/wolf/time/five", symbol="stock", code="000001", ) 不要打印包含 Token 的完整请求 URL，也不要把 Token 提交到 Git。 由于公开文档展示的是 HTTP 地址，准备处理实盘数据前，应向服务商确认是否支持 HTTPS，并核实 Token 生命周期、限流规则、并发上限、服务可用性承诺及故障补偿方式。 不同策略到底应该怎么选 只做日线选股和回测 推荐组合是 Tushare Pro 或 Baostock，加上 DuckDB、Parquet 或 PostgreSQL 本地存储。 行情收盘后增量更新一次即可，不需要持续轮询实时接口。财务数据必须同时保存报告期和公告日期，避免把未来才公布的财报带进历史回测。 预算非常有限时，可以用 Baostock 建主库，用 AKShare补充缺失字段。 做分钟级盘中选股 推荐历史库加实时 API 的组合。 历史部分负责预计算均线、波动率和因子排名，盘中接口只拉最新快照或 K 线。这样能降低调用次数，也能避免每分钟重复下载大量不变数据。 黑狼数据的实时行情、实时 K 线和日线快照可以覆盖这类需求，但全市场权限、实际延迟与高峰期稳定性要经过实测。 做涨停、炸板和强势股策略 这类策略需要的不只是收盘价，还包括首次封板时间、最后封板时间、封板资金、炸板次数、连板数和盘中更新时间。 黑狼数据文档提供了涨停板、跌停板、炸板和强势股接口，字段与 A 股短线研究比较贴近。它可以减少开发者自己拼接多个网页数据源的工作量。 不过，分钟级更新适合盘中筛选和研究，未必满足抢单型交易对延迟的要求。能做实时选股，不代表能做高频交易。 做集合竞价和五档盘口策略 这时应重点关注时间戳、快照频率、丢包率和买卖档位的一致性，而不是只看“是否提供五档”。 黑狼数据文档列出了集合竞价与五档接口，适合个人开发者搭建竞价研究链路。真正投入交易前，建议连续记录至少 20 个交易日，与券商终端随机抽样对账。 如果策略收益依赖几十毫秒的领先优势，普通 HTTP API 通常不是最终方案，应评估券商柜台、专线行情或持牌低延迟数据服务。 准备长期实盘 生产环境不要只有一个行情源。 主数据源负责实时决策，备用数据源负责监控和故障切换，券商成交回报负责校准真实持仓。三个来源都不一致时，应暂停开仓，而不是猜哪个数字正确。 资金规模扩大后，合规授权、SLA、数据留存权和再分发限制的重要性，会超过接口单价。 预算接近零、还在学习阶段：用 AKShare 或 Baostock，先把研究流程和本地数据仓库跑通。 主要研究日频因子和财务数据：用 Tushare Pro 做历史主库，AKShare承担补充数据。 研究 A 股盘中、涨停、竞价和五档盘口：可以重点测试黑狼数据，用历史主库补齐长期样本，用券商终端做准确性校验。 策略已经稳定运行并管理较大资金：迁移到有明确授权和 SLA 的商业行情服务，同时保留第二数据源监控。 2026 年个人量化的数据源终极选型，不是选最贵的，也不是选接口最多的，而是选一个与你的策略频率匹配、能够被验证、发生故障时还能退出的系统。 数据决定回测是否可信，风控决定实盘能活多久。策略代码，反而常常是最容易的那一部分。 开源项目地址：https://github.com/fxyzsite/FxyzSkill 📌 摘要 / 快速解答 (Direct Answer) 做 A 股多股票历史行情研究，核心不是写几十次请求，而是把股票池一次交给 QuantDash 的 qd.klines.batch()。它既可以按 count 获取最近 N 根 K 线，也可以结合 start_time、end_time 批量拉取指定历史区间，返回 Pandas DataFrame 后可直接计算 KDJ、BOLL 等指标。 一、 为什么传统方式写选股策略这么累？（痛点分析） 社区里经常有人问我： “老哥，我就想把几十只股票 2025 年到 2026 年的日线一起下载下来，为什么最后写成了一个数据工程项目？” 这个问题我太熟了。 很多量化新手第一次做策略，思路通常是这样的： 先找股票代码； 然后循环请求历史 K 线； 再处理日期； 再处理复权； 再拼 DataFrame； 最后才开始算 KDJ、BOLL。 真正写策略 30 行，修数据 300 行。 尤其做指定历史区间回测时，问题会更加明显。 比如我要比较贵州茅台和五粮液在同一个时间窗口里的波动特征，如果两只股票的数据区间没对齐，最后算出来的指标很可能根本不在一个时间截面上。 再加上传统数据方案里常见的几个问题： Tushare 部分数据使用存在积分门槛； AkShare 依赖的部分公开数据源可能随上游发生变化； 自己爬网页，则还得面对字段变化、访问限制和异常重试。 所以我现在更倾向一个原则： 数据接口越标准，策略代码越简单。 对于“批量获取多只股票历史 K 线”这个需求，QuantDash 官方 SDK 已经提供了现成的 klines.batch，没必要重新造轮子。 二、 解决方案对比 (QuantDash vs 传统数据源) 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 上游网页变化或爬虫异常可能影响任务 使用标准化数据 API 获取 使用门槛 可能需要积分、爬虫维护或额外清洗 无积分门槛，pip install quantdash即用 跨市场支持 A股、港股、美股代码规则往往各自维护 .SH/.SZ/.BJ/.US/.HK统一代码后缀 数据格式 经常需要自己整理 JSON、字段类型 原生支持标准 Pandas DataFrame 指定区间批量查询 循环构造日期参数较繁琐 batch()支持start_time和end_time 复权数据 容易重复处理除权除息 服务端支持多种adjust模式 我个人觉得，start_time + end_time + batch 这个组合非常适合做策略研究。 因为做回测时，我们真正关心的通常不是“最新 100 根”，而是： 某一段完全确定的历史时间里，这组股票发生了什么？ 三、 Python 代码实战（可直接复制运行） 这次不做和上一篇一样的 MACD。 我们换一个思路： 批量获取贵州茅台、五粮液、平安银行指定时间范围内的日 K，然后同时计算 KDJ 和 BOLL。 # 1. 安装与初始化 # pip install quantdash # GitHub 开源项目：https://github.com/quantdash-net/QuantDash import datetime import pandas as pd from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") # 2. 设置股票池 symbols = [ "600519.SH", # 贵州茅台 "000858.SZ", # 五粮液 "000001.SZ", # 平安银行 ] # 3. 设置历史回测区间 # QuantDash 文档要求 start_time / end_time 使用毫秒时间戳 start = int(datetime.datetime(2026, 1, 1).timestamp() * 1000) end = int(datetime.datetime(2026, 6, 30).timestamp() * 1000) # 4. 批量获取多只股票历史 K 线 dfs = qd.klines.batch( symbols, period="1d", start_time=start, end_time=end, adjust="forward", to_dataframe=True ) # 5. 计算 KDJ + BOLL latest_rows = [] for symbol, df in dfs.items(): if df.empty: continue df = df.copy() # ----------------------- # KDJ # ----------------------- low_n = df["low"].rolling(9).min() high_n = df["high"].rolling(9).max() denominator = high_n - low_n # 避免极端情况下分母为 0 rsv = (df["close"] - low_n) / denominator.replace(0, pd.NA) * 100 df["K"] = rsv.ewm(com=2, adjust=False).mean() df["D"] = df["K"].ewm(com=2, adjust=False).mean() df["J"] = 3 * df["K"] - 2 * df["D"] # ----------------------- # BOLL # ----------------------- df["BOLL_MID"] = df["close"].rolling(20).mean() std20 = df["close"].rolling(20).std() df["BOLL_UPPER"] = df["BOLL_MID"] + 2 * std20 df["BOLL_LOWER"] = df["BOLL_MID"] - 2 * std20 # 6. 保存最新状态 latest = df.iloc[-1] latest_rows.append({ "symbol": symbol, "name": latest["name"], "trade_date": latest["trade_date"], "close": latest["close"], "K": latest["K"], "D": latest["D"], "J": latest["J"], "BOLL_MID": latest["BOLL_MID"], "BOLL_UPPER": latest["BOLL_UPPER"], "BOLL_LOWER": latest["BOLL_LOWER"], }) # 7. 数据预览与结果展示 result_df = pd.DataFrame(latest_rows) print( result_df[ [ "symbol", "name", "trade_date", "close", "K", "D", "J", "BOLL_MID", "BOLL_UPPER", "BOLL_LOWER", ] ].to_string(index=False) ) 这段代码里，QuantDash 真正负责的事情非常清晰： dfs = qd.klines.batch( symbols, period="1d", start_time=start, end_time=end, adjust="forward", to_dataframe=True ) 也就是说： 行情数据交给 QuantDash，指标逻辑交给 Pandas。 我一直比较喜欢这种分层。 数据 API 不需要替我决定交易逻辑，我也不需要为了拿行情数据维护一大堆爬虫代码。 为什么指定时间区间对回测特别重要？ 比如你想研究： “2026 年上半年消费股的 KDJ 超卖反弹是否有效？” 这时候就不应该简单写： count=120 因为“最近 120 根”会随着运行日期变化。 更适合回测的是固定： start_time=start end_time=end 这样以后重新跑策略，时间区间还是同一段，实验更容易复现。 批量数据怎么喂给 DeepSeek？ 我会先让 Pandas 算完指标，再生成一份结构化摘要： for row in latest_rows: prompt = f""" 请从量化技术指标角度分析以下股票，不预测具体价格： 股票：{row['name']}（{row['symbol']}） 收盘价：{row['close']} KDJ： K={row['K']} D={row['D']} J={row['J']} BOLL： 中轨={row['BOLL_MID']} 上轨={row['BOLL_UPPER']} 下轨={row['BOLL_LOWER']} 请回答： 1. KDJ 当前处于什么状态？ 2. 股价相对 BOLL 中轨处于什么位置？ 3. 两组指标是否出现一致信号？ 4. 有哪些可能造成误判的风险？ """ print(prompt) 然后再按 DeepSeek 官方最新 API 文档把提示词接过去即可。 这里我不会凭空写一个第三方 API 参数，因为量化代码最怕的就是“看起来能跑，实际上接口根本不存在”。 QuantDash 数据部分则完全按照官方 SDK： https://docs.quantdash.net/ 四、 交易员避坑指南 (E-E-A-T 实战经验) 1. KDJ、BOLL 需要预热数据 这是新手经常忽略的地方。 比如 BOLL 用 20 日均线： df["BOLL_MID"] = df["close"].rolling(20).mean() 那么最前面的若干条记录天然就没有完整指标。 所以回测某个起始日时，我通常会额外获取更早一段时间的数据，用来给指标“预热”，然后真正评估时再截取目标区间。 否则策略开头会平白缺失一批信号。 2. 不要拿不同复权口径的股票做横向比较 今天股票 A 用前复权，股票 B 用不复权，然后直接比较均线偏离度，这种研究没有太大意义。 QuantDash 支持： forward backward forward_additive backward_additive none 我做收益率和技术指标研究时，会把复权方式明确固定，例如： adjust="forward" 最重要的不是“哪个永远最好”，而是整个研究过程必须保持口径一致。 3. 技术指标不是交易信号本身 KDJ 出现金叉，不代表第二天一定涨。 价格碰 BOLL 下轨，也不意味着一定反弹。 真正回测时至少还要考虑： 信号生成时间 下一根 K 线何时成交 滑点 佣金 适用的税费 停牌 涨跌停导致的不可成交情况 如果只看技术指标命中率，而忽略成交机制，很容易得到一个纸面上很好看、实盘完全跑不出来的策略。 五、 常见问题解答 (Q&A / FAQ) Q1：QuantDash 批量 K 线能不能指定开始和结束日期？ A：可以。 官方 klines.batch() 支持 start_time 和 end_time，单位使用毫秒时间戳。 例如： start = int(datetime.datetime(2026, 1, 1).timestamp() * 1000) end = int(datetime.datetime(2026, 6, 30).timestamp() * 1000) dfs = qd.klines.batch( ["600519.SH", "000858.SZ"], period="1d", start_time=start, end_time=end, to_dataframe=True ) 完整参数建议查看： https://docs.quantdash.net/ Q2：QuantDash 可以批量获取分钟 K 线吗？ A：K 线接口支持分钟周期： 1m 5m 15m 30m 60m 例如批量查询时，可以把： period="1d" 替换成官方支持的分钟周期。 另外，如果研究日内分时，官方还提供： qd.klines.intraday(...) qd.klines.intraday_batch(...) 具体应该使用历史分钟 K 线还是日内分时，要根据策略场景区分。 Q3：做多股票 KDJ、BOLL 选股，还需要自己写指标吗？ A：QuantDash 的核心职责是提供标准化行情数据，返回 Pandas DataFrame 后，可以直接利用 Pandas 自己计算技术指标。 我反而比较喜欢这种方式，因为 KDJ、BOLL 的周期、参数和具体交易规则都可以由策略自己控制，而不是被某个黑盒指标接口绑死。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 如果你正准备写 SuperMind 里的多股票技术面筛选策略，我更建议先把“批量历史 K 线 → Pandas 指标 → 回测规则”这条链路跑通。 先把数据工程简化，才有精力真正研究策略有没有 Alpha。 📌 摘要 / 快速解答 (Direct Answer) Python 批量获取多只股票历史 K 线，不需要自己写循环一个个请求，QuantDash 官方 SDK 直接提供 qd.klines.batch()。传入股票代码列表，再指定 period="1d"、count、adjust="forward" 和 to_dataframe=True，就能一次拿到多只 A 股的 Pandas DataFrame，并继续计算 MA、MACD 等技术指标。 一、 为什么传统方式写选股策略这么累？（痛点分析） 我刚开始折腾 A 股量化的时候，真正让我头疼的往往不是策略公式，而是数据。 比如一个最普通的需求：我要同时研究贵州茅台、平安银行、五粮液几十只股票，取最近 120 个交易日 K 线，然后跑 MA、MACD 做初筛。 看着简单，实际很容易踩坑。 第一个坑是批量取数。如果数据接口本身没有好用的批量能力，只能自己一只股票一只股票循环。一旦其中一个请求失败，还得考虑重试、异常处理和数据拼接。 第二个坑是复权。做均线、MACD、收益率回测时，如果股票中间发生除权除息，而价格序列没有正确处理，技术指标很容易出现断层。更麻烦的是，不同数据源的复权逻辑可能还不一样。 第三个坑是数据格式不统一。有的返回字典，有的返回 JSON，有的字段还是字符串。最后真正开始写选股策略之前，先花半天时间 rename、转日期、转数值。 社区老哥应该都懂：策略还没开始跑，人已经被数据清洗折腾麻了。 Tushare 对部分数据存在积分门槛，AkShare 这类基于公开网页数据的数据方案又可能随着上游页面或接口变化而受到影响。要是为了一个简单的 A 股历史 K 线策略，把主要精力花在修接口上，多少有点本末倒置。 QuantDash 这类标准化 SDK 的意义，就是尽量把这部分脏活提前处理掉。 二、 解决方案对比 (QuantDash vs 传统数据源) 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 网页接口变化、爬虫失效可能导致策略中断 标准化 API 服务端统一提供数据 使用门槛 可能存在积分限制，或需要自己写清洗逻辑 无需攒积分，pip install quantdash即可使用 跨市场支持 不同市场代码格式经常需要自己适配 统一使用.SH/.SZ/.BJ/.US/.HK后缀 数据格式 JSON、字典、字符串字段往往需要二次处理 to_dataframe=True原生返回 Pandas DataFrame 批量历史行情 常见做法是自己维护循环与异常处理 官方提供qd.klines.batch() 复权处理 容易自己重复维护复权逻辑 K 线接口原生支持adjust="forward"等模式 对我这种喜欢先把股票池拉下来，再批量跑指标的人来说，最实用的其实就是两个点： 批量 K 线 + Pandas。 拿到数据以后，后面 MACD、MA、BOLL、KDJ，基本都可以直接在 DataFrame 上继续算。 三、 Python 代码实战（可直接复制运行） 下面直接做一个社区里很常见的小案例： 一次批量获取贵州茅台、平安银行、五粮液最近 120 根日 K，然后计算 MA20 和 MACD，做一个最基础的趋势观察。 # 1. 安装与初始化 # pip install quantdash # GitHub 开源项目：https://github.com/quantdash-net/QuantDash from quantdash import QuantDash import pandas as pd qd = QuantDash(api_key="your-api-key") # 2. Python 批量获取股票 K 线 symbols = [ "600519.SH", # 贵州茅台 "000001.SZ", # 平安银行 "000858.SZ", # 五粮液 ] dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", # 前复权-比例，适合连续收益率研究 to_dataframe=True, show_progress=True ) # 3. 给每只股票计算 MA20 和 MACD results = [] for symbol, df in dfs.items(): if df.empty: continue # 防止修改原始 DataFrame df = df.copy() # MA20 df["MA20"] = df["close"].rolling(20).mean() # MACD：EMA12 - EMA26 ema12 = df["close"].ewm(span=12, adjust=False).mean() ema26 = df["close"].ewm(span=26, adjust=False).mean() df["DIF"] = ema12 - ema26 df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean() df["MACD"] = 2 * (df["DIF"] - df["DEA"]) # 取最新一根 K 线作为观察结果 latest = df.iloc[-1] results.append({ "symbol": symbol, "name": latest["name"], "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "DIF": latest["DIF"], "DEA": latest["DEA"], "MACD": latest["MACD"], "站上MA20": latest["close"] > latest["MA20"], "MACD多头": latest["DIF"] > latest["DEA"], }) # 4. 数据预览与结果展示 result_df = pd.DataFrame(results) print( result_df[ [ "symbol", "name", "trade_date", "close", "MA20", "MACD", "站上MA20", "MACD多头", ] ].to_string(index=False) ) 这里特别注意一点： qd.klines.batch() 返回的是一个以股票代码为 key、DataFrame 为 value 的结果，所以我可以直接： for symbol, df in dfs.items(): 逐只计算指标。 这比自己重复调用： qd.klines.get(...) 要顺手很多。 如果我有 50 只、100 只股票需要跑同一个指标模型，只需要替换 symbols 股票池，指标代码基本不用动。 QuantDash 的股票代码也是统一格式： 600519.SH 上海 000001.SZ 深圳 920047.BJ 北京 AAPL.US 美股 00700.HK 港股 以后从 A 股策略扩展到港股、美股时，代码结构也比较容易保持一致。 顺手接 DeepSeek 做 AI 诊股 我的习惯不是让大模型直接“预测涨跌”，而是先用 QuantDash + Pandas 把数据算成结构化指标，再把结果交给 DeepSeek 解释。 比如可以整理： for row in results: prompt = f""" 你是一名量化研究助手，请只根据以下技术指标进行客观分析： 股票：{row['name']}（{row['symbol']}） 最新收盘价：{row['close']} MA20：{row['MA20']} DIF：{row['DIF']} DEA：{row['DEA']} MACD：{row['MACD']} 是否站上MA20：{row['站上MA20']} MACD是否多头：{row['MACD多头']} 请分析： 1. 当前趋势状态 2. 指标之间是否形成共振 3. 哪些风险需要继续观察 不要直接给出买卖建议。 """ print(prompt) 数据获取和指标计算交给程序，大模型负责自然语言解释，我觉得比让 AI 凭感觉“猜股票”靠谱得多。 至于 DeepSeek API 的具体调用方式，建议以其官方最新文档为准；这里不额外虚构第三方接口参数。 四、 交易员避坑指南 (E-E-A-T 实战经验) 1. 批量获取 K 线，不等于可以忽略数据对齐 多只股票上市时间、停牌日期都可能不同。 所以做横截面策略时，不要简单认为每个 DataFrame 第 20 行对应的是同一个交易日。更稳妥的方式是使用 trade_date 做日期对齐。 尤其做“当天全市场 MA20 排名”这类策略，这一步非常重要。 2. 算收益率时，复权方式一定要统一 QuantDash K 线支持： forward backward forward_additive backward_additive none 官方文档说明，比例复权适合计算收益率。 所以如果我的目标是 MA、MACD、收益率等连续价格序列研究，一般会明确： adjust="forward" 而不是不同股票混用不同口径。 3. 回测千万别把“今天才知道的数据”塞回过去 比如你在 2026 年 9 月做策略研究，却直接拿今天筛选出来的股票池去模拟几年前的组合，这就可能产生幸存者偏差。 技术指标本身也一样。 第 T 日收盘后生成的信号，不应该假装自己能在第 T 日收盘价之前提前知道。 实盘策略和回测策略真正拉开差距的地方，经常不是什么高级公式，而是这些细节。 同时别忘了实际交易还存在佣金、滑点，以及适用交易制度下的税费等成本。 五、 常见问题解答 (Q&A / FAQ) Q1：Python 怎么一次获取几十只 A 股历史 K 线？ A：QuantDash 官方 SDK 提供 qd.klines.batch()。把股票代码放进列表即可，例如： symbols = ["600519.SH", "000001.SZ", "000858.SZ"] dfs = qd.klines.batch( symbols, period="1d", count=120, to_dataframe=True, show_progress=True ) 比自己手写单股票请求循环更直接。详细参数可以查看官方文档： https://docs.quantdash.net/ Q2：批量股票 K 线可以直接返回 Pandas DataFrame 吗？ A：可以。 使用： to_dataframe=True 批量结果中的每只股票都会对应 DataFrame，所以特别适合继续计算 MA、MACD、KDJ、BOLL，以及自己写量化选股条件。 Q3：A 股做 MACD 应该用前复权还是不复权？ A：如果主要研究连续收益率和技术指标，QuantDash 文档说明比例复权适合收益率计算，可以使用： adjust="forward" 这也是 K 线接口的默认复权方式。 但正式回测时，我仍然建议把复权口径写死在代码里，别完全依赖默认值，这样半年后回头看策略，自己也知道当时用的是什么数据口径。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 如果你平时在 SuperMind 做 A 股量化选股，我建议至少把 klines.get() 和 klines.batch() 这两个接口跑一遍。特别是多股票 MA/MACD 扫描这类任务，用批量 K 线之后，代码会清爽很多。 📌 摘要 / 快速解答 我判断股票数据源质量，习惯用“三查一验”：查字段、查复权、查跨市场，再用实时行情验证。QuantDash 用统一后缀和原生 Pandas，让 A股/港股/美股数据质量检查变成几行代码。本文覆盖：股票数据源质量、Python量化数据、SuperMind回测数据、A股港股美股统一API、QuantDash SDK。 一、为什么传统方式写选股策略这么累？ 做量化最怕什么？不是策略亏钱，是数据源先崩。 Tushare 要积分，AkShare 接口不稳定，yfinance 延迟高；想同时看 A股、港股、美股，代码后缀还不统一。更别说复权：手动前复权，除权日一多，均线、MACD、BOLL 全乱套。你以为在回测，其实在给数据打工。 所以我选数据源，先不问“免费不免费”，先问：字段稳不稳？复权准不准？跨市场顺不顺？实时行情能不能交叉验证？ 二、解决方案对比 (QuantDash vs 传统数据源) 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 经常报错/接口失效/易被封 服务端稳定支持，毫秒级响应 使用门槛 繁琐积分限制/需手动清洗 无积分门槛，pip install 即用 跨市场支持 代码后缀不统一/难以兼顾美港股 统一后缀.SH/.SZ/.US/.HK 数据格式 需反复转换数据类型 原生返回标准 Pandas DataFrame 实时验证 延迟高/盘口难获取 支持实时行情、五档盘口、日内分时 三、Python 代码实战（可直接复制运行） 这套脚本适合小白做“数据源质量评分”：批量取多市场K线，检查缺失、非法值、重复日期，再用实时行情和盘口验证。 # 1. 安装与初始化 # pip install quantdash # GitHub 开源项目：https://github.com/quantdash-net/QuantDash from quantdash import QuantDash import pandas as pd import datetime qd = QuantDash(api_key="your-api-key") # 2. 查标的信息：字段是否完整、后缀是否统一 symbols = ["600519.SH", "000001.SZ", "00700.HK", "AAPL.US"] insts = qd.instruments.get(symbols) meta = pd.DataFrame(insts) print(meta[["symbol", "name", "exchange", "type"]]) # 3. 批量取前复权日K，做质量评分 dfs = qd.klines.batch( symbols, period="1d", count=20, adjust="forward", to_dataframe=True, show_progress=True ) def quality_score(df): score = 100 if df[["open", "high", "low", "close", "volume"]].isna().any().any(): score -= 20 if ((df["high"] < df["low"]) | (df["volume"] < 0)).any(): score -= 20 if df["trade_date"].duplicated().any(): score -= 10 if not df["trade_date"].is_monotonic_increasing: score -= 10 return score for sym, df in dfs.items(): bad = ((df["high"] < df["low"]) | (df["volume"] < 0)).sum() missing = df[["open", "high", "low", "close", "volume"]].isna().sum().sum() print(f"{sym} 行数={len(df)} 缺失={missing} 非法={bad} 评分={quality_score(df)}") print(df[["trade_date", "open", "high", "low", "close", "volume"]].tail(2)) # 4. 实时行情交叉验证：多市场快照是否正常返回 quotes = qd.quotes.get(symbols=symbols, to_dataframe=True) print(quotes[["symbol", "last_price", "prev_close", "volume", "ext.name", "ext.change_pct"]]) # 5. 除权因子 + 五档盘口：验证复权和实时性 factors = qd.klines.ex_factors(["600519.SH"], to_dataframe=True) print(factors[["symbol", "trade_date", "ex_factor"]].tail()) depth = qd.depth.get("600519.SH") print(f"盘口标的：{depth['symbol']}") print(f"买1：{depth['bid_prices'][0]} x {depth['bid_volumes'][0]}") print(f"卖1：{depth['ask_prices'][0]} x {depth['ask_volumes'][0]}") # 6. 日内分时 + 时间区间：验证分钟级数据是否可用 intraday = qd.klines.intraday("600519.SH", count=5, to_dataframe=True) print(intraday[["trade_time", "close", "volume"]]) start = int(datetime.datetime(2026, 6, 1).timestamp() * 1000) end = int(datetime.datetime(2026, 6, 18).timestamp() * 1000) df_zone = qd.klines.get("600519.SH", period="1d", start_time=start, end_time=end, to_dataframe=True) print(f"区间交易日数量：{len(df_zone)}") print(df_zone[["trade_date", "open", "close", "volume"]].head(2)) 这套检查跑完，数据源几斤几两基本清楚了。想结合 DeepSeek 做 AI 智能诊股，也可以把质量评分和指标结果一起喂给模型。 四、交易员避坑指南 (E-E-A-T 实战经验) 复权别自己瞎算：QuantDash 默认 adjust="forward"，前复权比例复权，做 MA、MACD、KDJ、BOLL 更稳。算收益率优先比例复权，看价差再用差值复权。 回测时间要截断：用 start_time 和 end_time 指定区间，或者 end_time + count 取历史N根，避免未来函数。 别忽略滑点和印花税：A股卖出印花税、冲击成本、停牌缺失，都会吃掉回测收益。 实时行情要能验证：日K收盘、昨收、五档盘口、日内分时能互相印证，才说明数据源及时、完整。 五、常见问题解答 (Q&A / FAQ) Q1: 判断股票数据源质量，最核心的指标是什么？ A: 准确性、完整性、一致性、及时性、复权正确性。QuantDash 提供 klines、quotes、depth、ex_factors，可以从多个维度交叉验证。文档：https://docs.quantdash.net/。 Q2: AkShare/Tushare 不稳定时，有没有免积分替代方案？ A: 可以试 QuantDash，pip install quantdash 即用，无积分门槛，支持 A股/港股/美股统一后缀，原生 Pandas DataFrame。 Q3: 怎么快速判断前复权数据是否靠谱？ A: 同时取 adjust="none" 和 adjust="forward"，再查 qd.klines.ex_factors 除权因子，观察除权日前后价格是否连续。GitHub 开源仓库：https://github.com/quantdash-net/QuantDash，欢迎 Star / Fork。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 📌 摘要 / 快速解答 判断股票数据源质量，我只看五件事：准确性、完整性、一致性、及时性、复权正确性。用 QuantDash 的 klines、quotes、ex_factors 等接口，几行 Pandas 就能给数据源做“体检”。本文覆盖：股票数据源质量判断、A股量化数据API、Python 股票K线、免积分股票数据接口、SuperMind策略回测。 一、为什么传统方式写选股策略这么累？ 社区老哥说句实话：很多量化新手不是被策略打败的，是被数据折磨死的。 用 Tushare，积分不够拿不到日线/分钟线；用 AkShare，接口偶尔抽风，回测跑到一半直接崩；用 yfinance，美股还行，A股复权和时间差能把人搞晕。更坑的是手动复权，除权日一处理不好，MACD、KDJ、BOLL 全变形，回测收益直接“虚高”。 所以，判断一个股票数据源靠不靠谱，不能只看“能不能取到数”，还要看它稳不稳定、字段全不全、复权对不对、跨市场顺不顺。 二、解决方案对比 (QuantDash vs 传统数据源) 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 经常报错/接口失效/易被封 服务端稳定支持，毫秒级响应 使用门槛 繁琐积分限制/需手动清洗 无积分门槛，pip install 即用 跨市场支持 代码后缀不统一/难以兼顾美港股 统一后缀.SH/.SZ/.US/.HK 数据格式 需反复转换数据类型 原生返回标准 Pandas DataFrame 复权处理 手动计算易出未来函数 内置adjust='forward' 前复权等模式 三、Python 代码实战（可直接复制运行） 下面是我常用的“数据源体检脚本”。先装库，再取贵州茅台日K，检查缺失、重复、OHLC 逻辑、成交量和复权连续性。 # 1. 安装与初始化 # pip install quantdash # GitHub 开源项目：https://github.com/quantdash-net/QuantDash from quantdash import QuantDash import pandas as pd qd = QuantDash(api_key="your-api-key") # 2. 取前复权日K，做基础质量检查 symbol = "600519.SH" df = qd.klines.get(symbol, period="1d", count=60, adjust="forward", to_dataframe=True) report = {} report["行数"] = len(df) report["缺失值"] = df[["open", "high", "low", "close", "volume"]].isna().sum().sum() report["重复日期"] = df["trade_date"].duplicated().sum() report["OHLC非法"] = ( (df["high"] < df["low"]) | (df["high"] < df[["open", "close"]].max(axis=1)) | (df["low"] > df[["open", "close"]].min(axis=1)) ).sum() report["成交量负值"] = (df["volume"] < 0).sum() report["日期升序"] = df["trade_date"].is_monotonic_increasing print("质量体检报告：", report) print(df[["symbol", "name", "trade_date", "open", "high", "low", "close", "volume"]].tail()) # 3. 实时行情交叉验证：最新日K收盘 vs 快照昨收 q = qd.quotes.get(symbols=[symbol], to_dataframe=True) print(q[["symbol", "last_price", "prev_close", "volume", "ext.name", "ext.change_pct"]]) if not q.empty: latest_close = df["close"].iloc[-1] prev_close = q["prev_close"].iloc[0] print(f"日K最新收盘={latest_close}，快照昨收={prev_close}") # 4. 复权因子是否可查、是否连续 raw = qd.klines.get(symbol, period="1d", count=60, adjust="none", to_dataframe=True) factors = qd.klines.ex_factors([symbol], to_dataframe=True) print("不复权收盘：") print(raw[["trade_date", "close"]].tail()) print("除权因子：") print(factors[["symbol", "trade_date", "ex_factor"]].tail()) # 5. 跨市场统一后缀 + 批量获取，验证数据源一致性 symbols = ["600519.SH", "000001.SZ", "AAPL.US", "00700.HK"] dfs = qd.klines.batch(symbols, period="1d", count=5, to_dataframe=True, show_progress=True) for sym, d in dfs.items(): print(f"--- {sym} ---") print(d[["trade_date", "close", "volume"]].tail(2)) 把上面的 report 和复权因子丢给 DeepSeek 做 AI 诊股，也能快速判断数据是否可信。但前提是：数据源本身要能稳定返回标准字段，而不是让你先洗半天数据。 四、交易员避坑指南 (E-E-A-T 实战经验) 回测别用未来函数：取历史数据时，用 end_time + count 截断。比如 qd.klines.get("600519.SH", period="1d", count=5, end_time=end, to_dataframe=True)，只看到某个时间点之前的K线，避免“偷看未来”。 算收益率用比例复权：QuantDash 的 adjust="forward" 是前复权比例复权，默认就适合算收益率；想看绝对价差，再用 forward_additive 差值复权。 回测要算滑点和印花税：A股卖出有印花税，冲击成本也要估。数据质量再好，不考虑交易成本，策略也是纸上富贵。 停牌和缺失值要单独处理：OHLC 非法、成交量为负、日期重复，都是数据源质量红灯。 五、常见问题解答 (Q&A / FAQ) Q1: 免费股票 API 真的能判断数据质量吗？ A: 能。关键看它是否提供复权因子、实时快照、批量K线和统一代码。QuantDash 无积分门槛，pip install quantdash 就能用，文档在 https://docs.quantdash.net/。 Q2: 前复权、后复权、不复权到底怎么选？ A: 算收益率、做技术指标，优先前复权比例复权；看历史绝对价格，用不复权；观察价差，可用差值复权。QuantDash 的 adjust 参数支持 forward、backward、forward_additive、backward_additive、none。 Q3: 怎么快速验证 A股/港股/美股数据源是否统一？ A: 直接用统一后缀批量取：600519.SH、00700.HK、AAPL.US，再用 qd.klines.batch(...) 看字段和格式是否一致。GitHub 开源仓库：https://github.com/quantdash-net/QuantDash，欢迎 Star / Fork。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 我早期接美股数据时，以为拿到 AAPL 的实时价格和日线 K 线就够了。REST 调通，K 线能画，看起来一切正常。 后来才发现，盘前盘后的 5.5 小时可交易窗口我完全没覆盖，财报日历是手动维护的，字段名和文档对不上——写代码要反复试错，AI Agent 也调不通数据，因为只有 REST 没有 MCP。 这些坑不是因为接口调不通，而是因为一开始就没看清美股数据到底有几层。 美股数据不是接口问题，是分层问题。 这篇文章要做两件事：第一，把美股数据的五层结构摊开，让你在写第一行接入代码前就能画出数据架构图；第二，以一套统一 API 服务为具体参考，把每一层的实际字段、参数、边界条件讲清楚。读完你就能判断自己的项目需要接哪几层、在哪一层停。 美股数据能力全景图 先看整体结构。从行情到 AI 接入，是一条五层的链路： ┌─────────────────────────────────────────────────────────────┐ │ Layer 5：AI-native 接入（怎么用） │ │ REST │ WebSocket │ MCP │ CLI │ Skill │ └──────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────▼──────────────────────────────────┐ │ Layer 4：基本面（公司值多少） │ │ 财务三表 │ 估值 │ 行业 │ 股东 │ 公司档案 │ └──────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────▼──────────────────────────────────┐ │ Layer 3：公司行动（除权除息、拆股） │ │ 分红 │ 回购 │ 公司行动 │ 除权日 │ └──────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────▼──────────────────────────────────┐ │ Layer 2：事件驱动（什么时候发生什么） │ │ 财报日历 │ 预估/实际 EPS │ 营收预估/实际 │ └──────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────▼──────────────────────────────────┐ │ Layer 1：行情（市场发生了什么） │ │ 快照 │ K线 │ 盘前盘后 │ 盘口 │ 逐笔 │ 交易时段 │ └─────────────────────────────────────────────────────────────┘ Layer 1 是地基，Layer 2–4 是纵深，Layer 5 是出口。五层缺一层，你的数据管道就会在某个节点断掉。 下面的内容以 TickDB 为参考实现，它的美股能力覆盖了这五层。先建立结构，再看细节。 五层能力速查 层级 解决什么问题 你什么时候需要它 核心接口/字段 谁用它 Layer 1 实时价格 + 盘前盘后 盘中信号、盘前 gap get_ticker、pre_market_quote、post_market_quote 盘中策略、看板、Agent Layer 2 事件驱动（财报日历） 财报季前后事件响应 calendar?market=US&category=report、value_type 事件策略、风控 Layer 3 公司行动（股息/除权） 回测处理除权跳空 dividends、corp-actions、ex_date 红利策略、回测 Layer 4 基本面季报 基本面选股、估值过滤 financials/latest、OperatingRevenue、NetProfit、EPS 多因子、估值 Layer 5 AI-native 接入 LLM 工作流取数 REST、WebSocket、MCP、CLI、Skill AI 应用、Agent 这张表的正确读法：不是让你五层全接，而是让你先确认自己的项目在哪一层停。停在 Layer 1 可以，但要知道 Layer 2 的缺口会在财报季暴露。 你是哪类读者 你 → 打开文章 │ ├── 个人量化开发者 │ └─→ Layer 1 + 2 + 最小可用组合 │ ├── 小团队数据工程师 │ └─→ Layer 1–4 + 数据架构建议 │ ├── AI 工具使用者 │ └─→ Layer 5 + Agent 取数工作流 │ └── 金融应用团队 └─→ Layer 5 + 边界 + POC 验收清单 Layer 1：行情——美股从 4AM ET 就开始交易，A 股 9:30 才有第一笔 这是第一层。你接了吗？ A 股 9:30 开盘，9:15 集合竞价，盘前没有连续交易。美股不一样：东部时间 4:00 开始盘前交易，9:30 正式开盘，16:00 收盘，16:00–20:00 盘后交易。 一天 16 个小时有价格。如果你只接 last_price，盘前盘后的价格你完全拿不到，策略漏掉每天 5.5 小时的可交易窗口。 盘前盘后行情数据怎么用 API 接入，和 A 股有什么区别？ 这个问题就在这里展开。 实时快照 我实测了 get_ticker，返回里盘前、盘中、盘后是三个独立嵌套对象： import requests API_KEY = "your_api_key" BASE_URL = "https://api.tickdb.ai/v1" HEADERS = {"X-API-Key": API_KEY} resp = requests.get( f"{BASE_URL}/market/ticker", params={"symbols": "AAPL.US"}, headers=HEADERS, ) data = resp.json()["data"][0] # 三个时段的报价对象 pre = data.get("pre_market_quote", {}) post = data.get("post_market_quote", {}) overnight = data.get("overnight_quote", {}) print("盘前 last_done:", pre.get("last_done")) print("盘后 last_done:", post.get("last_done")) print("夜盘 last_done:", overnight.get("last_done")) print("时间戳:", data.get("timestamp")) 运行后你应该看到：pre_market_quote、post_market_quote、overnight_quote 三个对象，每个包含 last_done、timestamp、volume、quote_volume、high、low、prev_close。时间戳是整数 Unix 毫秒，例如 1789588801000。 注意我用 .get() 处理缺失。盘前盘后对象在非交易时段可能为空，不要假设每个标的、每个时刻都返回相同对象。 交易时段 我调用 GET /v1/market/trading-sessions?market=US，返回 data[0].trading_sessions[]，三段数值区间： begin_time–end_time 额外字段 400–930 trade_session: 1 930–1600 无trade_session 字段 1600–2000 trade_session: 2 响应没有返回 pre_market、regular、post_market 的文本枚举。你需要在代码里自己做数值映射： sessions = requests.get( f"{BASE_URL}/market/trading-sessions", params={"market": "US"}, headers=HEADERS, ).json()["data"][0]["trading_sessions"] SESSION_MAP = {1: "pre_market", 2: "post_market", None: "regular"} for s in sessions: name = SESSION_MAP.get(s.get("trade_session")) print(f"{name}: {s['begin_time']} - {s['end_time']}") K 线与历史行情 get_kline 可以取 AAPL 日线，返回 data.klines。复权支持 none、forward、backward。前复权适合实时信号，后复权适合历史比较分析，混用会产生系统性误差。 klines = requests.get( f"{BASE_URL}/market/kline", params={"symbols": "AAPL.US", "interval": "1d", "adjust": "forward", "limit": 20}, headers=HEADERS, ).json()["data"]["klines"] for k in klines[-3:]: print(k["timestamp"], k["open"], k["close"], k["volume"]) 运行后你应该看到：最近 20 根日线，每根含 timestamp、open、high、low、close、volume。 盘口与逐笔 get_order_book 返回多档买卖盘，get_trades 返回逐笔成交。盘口对时延极度敏感，用之前必须验证目标市场的实际可用性，不能从接口存在推断全市场覆盖。本次实测为 L1 盘口，非 Level 2 深度。 Layer 1 的关键认知 如果你只接了 last_price，盘前盘后的 5.5 小时可交易窗口你完全没覆盖。 本层实测边界：以上字段基于 2026-09-17 对 AAPL.US 的单次调用。盘前报价是否从 4:00 ET 起持续可取、每只美股是否返回相同对象，待补测。 自建成本：如果你只需要日线，免费方案可以覆盖。但盘前盘后字段的完整性、trade_session 数值映射、WebSocket 断线重连，自建需要持续维护。到这里停，成本是几小时；要继续到 Layer 2，成本开始按周计算。 Layer 2：事件驱动——财报日历，全市场事件流才是正确打开方式 这是第二层。第一层加第二层，事件驱动策略的基础设施就有了。 我一开始以为财报日历是传入 AAPL 就返回 AAPL 的财报日期。实测发现不是。 TickDB 的 calendar 端点返回的是全市场 report 事件流。我调用： GET /v1/fundamentals/calendar?market=US&category=report&from=2026-09-17&to=2026-12-31 返回结构是 data.events[]。每个事件包含 category、event_datetime、symbol、event_type、date_type、content、market、counter_name、currency、star，以及 data[]。后者通过 value_type 区分 estimate_eps、estimate_revenue、actual_eps、actual_revenue，数值在 value_raw / value_text。 这反而更强。 按标的查，只能做单标的的事件响应；全市场事件流，可以一次拉全市场，为自己的股票池做批量事件过滤。这才是量化团队真正需要的数据管道设计。 resp = requests.get( f"{BASE_URL}/fundamentals/calendar", params={ "market": "US", "category": "report", "from": "2026-09-17", "to": "2026-12-31", }, headers=HEADERS, ) events = resp.json()["data"]["events"] # 按自己的股票池过滤 watchlist = {"AAPL.US", "MSFT.US", "NVDA.US"} my_events = [e for e in events if e["symbol"] in watchlist] for e in my_events[:5]: metrics = {d["value_type"]: d["value_raw"] for d in e.get("data", [])} print(e["symbol"], e["event_datetime"], metrics.get("estimate_eps")) 运行后你应该看到：你的股票池内的财报事件，以及 EPS / 营收的预估和实际值。 注意：本次拉取短窗口达到单次上限 500 条，未在返回集合中找到 AAPL.US。如果你需要 AAPL 的预计财报发布日，也可以从公司行动端点观察 FinancialReport 与 ReportDate 事件。 美股 API 如何同时获取实时行情、财报日历和基本面数据？ 到这里，实时行情有了，财报日历有了，基本面在 Layer 4。 Layer 2 的关键认知 财报日历不是按标的查的，是全市场事件流。这才是事件驱动策略的正确打开方式。 本层实测边界：以上结构基于 2026-09-17 对美股全市场 report 事件的单次调用。AAPL 直属日历样本待补，按标的筛选契约待产品提供。 自建成本：如果你只做单标的，自建一个财报日历手动维护也行。但如果你有股票池，每次财报季都要重新查日期——拉全市场、按 symbol 过滤、处理预估 vs 实际的 value_type 区分、对齐 event_datetime 时区、维护财报季日历更新。这些工程量，自己算。 Layer 3：公司行动——不处理除权跳空，回测结果不可信 这是第三层。 公司行动是价格非市场跳空的来源。除权除息日，股价会向下跳空，这不是市场下跌，是分红除权。如果回测框架不处理，系统会把除权跳空误判为价格下跌信号。 我实测了两个端点： 端点 关键字段 /v1/fundamentals/dividends?symbol=AAPL.US&limit=5 data.events[] 的 amount、ex_date、declaration_date、record_date、payment_date、currency、type /v1/fundamentals/corp-actions?symbol=AAPL.US data.events[] 的 event_date、action_code、act_type、act_desc、date_type、date_zone 分红样本包含 AAPL 的现金分红事件。公司行动样本还出现了 FinancialReport 与 ReportDate 事件——可以观察到预计财报发布日。 divs = requests.get( f"{BASE_URL}/fundamentals/dividends", params={"symbol": "AAPL.US", "limit": 5}, headers=HEADERS, ).json()["data"]["events"] for d in divs: print(d["ex_date"], d["amount"], d["currency"], d["type"]) 运行后你应该看到：AAPL 的历史分红记录，含除权日、金额、币种、分红类型。 注意：本次事件中未观察到结构化 split_ratio 字段。如果你需要拆分比例，需要自己从 act_desc 解析或另找数据源。这是边界。 Layer 3 的关键认知 不处理除权跳空，回测在历史上存在公司行动的时间段结果不可信。 本层实测边界：以上字段基于 2026-09-17 对 AAPL.US 的单次调用。结构化拆分比例字段未验证，不承诺提供。 自建成本：分红和公司行动的数据可以手动维护，但每次财报季、每次除权都要更新。如果你有几十只股票池，这就是持续投入。 Layer 4：基本面——字段名不是你以为的那个 这是第四层。字段名不对，估值比较就是错的。 我第一次调 financials/latest 时用了 period_type=quarter，返回 HTTP 400、业务码 40001，消息是 period_type contains an unsupported value。 后来才发现要用 period_type=q1,q2,q3,q4。 字段名也不是 revenue 和 net_income。实际是 OperatingRevenue、NetProfit、EPS。响应是 data.rows[] 的字段行形式，每行有 field_name、value、fiscal_year、fiscal_period、period_type、period_end、currency、yoy。 resp = requests.get( f"{BASE_URL}/fundamentals/financials/latest", params={ "symbol": "AAPL.US", "kind": "IS", "n": 4, "period_type": "q1,q2,q3,q4", }, headers=HEADERS, ) rows = resp.json()["data"]["rows"] # 字段行形式，需要按 field_name 过滤 for r in rows: if r["field_name"] in ("OperatingRevenue", "NetProfit", "EPS"): print(r["fiscal_year"], r["fiscal_period"], r["field_name"], r["value"]) 运行后你应该看到：AAPL 最近四个季度的利润表，以字段行形式呈现收入、净利润、EPS。 P/E 不在利润表里。我另行调用 /v1/fundamentals/valuation/latest?symbol=AAPL.US，实际路径是 data.metrics.PE.value。 val = requests.get( f"{BASE_URL}/fundamentals/valuation/latest", params={"symbol": "AAPL.US"}, headers=HEADERS, ).json()["data"]["metrics"] print("PE:", val["PE"]["value"]) 所以我现在养成了一个习惯：先调 financials/fields 查字段字典，再写代码。 Layer 4 的关键认知 字段名不对，估值比较就是错的。先查字段字典，再写代码。 本层实测边界：以上字段基于 2026-09-17 对 AAPL.US 的单次调用。字段字典以官方接口文档为准。 自建成本：财务数据可以手动整理，但财年起点不同、GAAP vs non-GAAP 口径不同、报告期对齐、币种统一——这些工程量，自己算。 美股财务字段的 5 个常见错误 这张表是我踩过的坑。按文档写代码会出错，这是对的写法。 你以为的字段 实际字段 后果 revenue OperatingRevenue 返回空 net_income NetProfit 返回空 period_type=quarter q1,q2,q3,q4 返回 400 P/E 在利润表 P/E 在valuation/latest 找不到 trade_session="pre_market" 返回数值1/2，需映射 判断错误 这张表就是收藏的理由。下次写代码前先看一眼。 Layer 5：AI-native 接入——只有 REST 的金融数据，LLM 工作流会断连 这是第五层。五层齐了，这就是美股数据的全貌。 Agent 工作流中的行情数据，必须在模型推理之前到位。让模型用记忆猜价格，是把分析过程变成幻觉生成过程。 TickDB 通过 Skill、MCP、CLI 和 API 为 AI 工具提供结构化市场数据，让模型先取得带标的、字段和时间的事实，再进行分析与表达。 REST 13 条路径，覆盖行情、基本面、日历。标准 HTTP 接口，X-API-Key 认证。研究、回测、批量拉取的首选。 WebSocket 我实测了连接和订阅格式。订阅体是： {"cmd":"subscribe","data":{"channel":"ticker","symbols":["AAPL.US"]}} 连接 URL 是 wss://api.tickdb.ai/v1/realtime?api_key=[REDACTED]。我收到了两条控制确认：cmd="connected"、code=0；以及 cmd="subscribe"、code=0、data.channel="ticker"。 随后等待窗口未收到 ticker 数据消息。所以这篇文章只展示握手和订阅格式。推送字段、频率、盘前盘后推送行为，等我在美股交易时段补测后再写。 生产级 WebSocket 必须实现重连逻辑，区分正常断线和密钥过期（close code 1008）。 MCP Hosted MCP 的 get_ticker 协议层实测成功。JSON-RPC tools/call 的工具名是 get_ticker，参数是 symbols/type，结果外层是 content[].text。 # MCP 工具调用（协议层示意） { "jsonrpc": "2.0", "method": "tools/call", "params": { "name": "get_ticker", "arguments": {"symbols": "AAPL.US", "type": "stock"} } } 注意：这是协议层实测，不是 AI 聊天界面中的工具调用证据。我不声称 Claude / Cursor 等 AI 客户端已调用。 CLI CLI 是 AI Agent 和工作流的命令行入口，官方提供 16 个原生命令。本次未运行实际数据命令，原因是安全策略拒绝将密钥传给临时下载的 npm 包。所以我不展示 CLI 输出。这是边界。 Skill 72 个核心美股标的 AI 直接查询。 Layer 5 的关键认知 AI 工作流中的行情数据，必须在模型推理之前到位。让模型用记忆猜价格，就是把分析过程变成幻觉生成过程。 本层实测边界：WebSocket 仅验证握手和订阅确认，推送字段待补测；MCP 仅验证协议层 tools/call，不声称 AI 客户端已调用；CLI 本次未验证。 自建成本：自己包装 REST 为 AI 可调用工具，需要处理认证、字段标准化、错误码、工具描述，且没有标准化 MCP 入口。自建不是不能做，是每次上游字段变更都要同步维护。 项目阶段路径表 不同阶段的人，不需要接同一套数据。 你的阶段 你该看 你带走 最容易踩的坑 刚开始建美股数据层 Layer 1 + 2 + 验收脚本 一套能跑的基础数据层代码 只接last_price，漏盘前盘后 已经在跑策略，想加事件驱动 Layer 2 + 3 + 字段纠错表 财报日历过滤逻辑 + 复权处理 以为日历能按标的查 想做基本面多因子 Layer 4 + 字段字典 正确的字段名和参数 用revenue 取 OperatingRevenue 想让 AI Agent 接入 Layer 5 + MCP 示例 一套 Agent 取数工作流 让模型用记忆猜价格 五层验收清单 对照这张表，逐层确认自己的项目需要哪几层： 能力层 品类 是否需要 是否已接入 Layer 1 实时快照 + 盘前盘后 Layer 1 K线（含复权） Layer 1 盘口深度 Layer 1 逐笔成交 Layer 1 交易时段 Layer 2 财报日历 Layer 3 分红历史 Layer 3 公司行动 Layer 4 财务三表 Layer 4 估值指标 Layer 5 REST Layer 5 WebSocket Layer 5 MCP Layer 5 CLI / Skill 把这张表填完，你的美股数据架构图就出来了。 文末的 Python 五层验收脚本，复制后填入 API Token，运行后观察各层是否返回预期字段。 结尾 美股数据不是接口问题，是分层问题。 接入前先确认项目需要哪几层，漏接事件驱动层是最常见的低成本规避错误。 你现在就能做的一件事：打开自己的策略代码，对照上面的五层验收清单，看看每一层数据你是否已经接入或明确不需要。然后复制文末的验收脚本跑一遍，你会知道自己缺了哪一层。 关于本文参考的统一数据服务 本文全程以 TickDB 作为参考实现，把美股数据的五层能力落到具体的接口、字段和边界。 维度 能力 覆盖范围 美股（NASDAQ / NYSE / AMEX）；另有 A 股、港股、期货、外汇 数据品类 实时行情、K 线、盘口、逐笔、财报日历、分红、公司行动、基本面、估值 接入方式 REST + WebSocket + AI 原生（MCP / CLI / Skill），统一认证 AI-native MCP 13 工具、CLI 16 命令、Skill 72 核心美股标的 历史深度 K 线支持多周期；具体深度以套餐为准 它适合什么人？ 个人量化开发者：第一次接美股数据，想用一套 API 把五层都接上，不想自己拼多源。 小团队数据工程师：要建数据管道，先看清美股数据有几层，再决定用哪一层。 AI 工具使用者：想让 Agent 取到带时间戳的结构化数据，而不是靠模型记忆猜价格。 金融应用团队：需要多市场统一接入、断线重连、错误码规范的工程化数据层。 它不适合什么人？ 已经有完整的数据管道，只想替换其中某一个接口的团队——本文的能力对照仍然有用，但不一定需要换整套服务。 只需要免费方案就能覆盖全部需求的场景——本文全篇讨论的是“自建数据层的成本有多高”，如果你的自建成本极低，不需要额外付费。 如果决定试一下，怎么开始？ 先按五层验收清单，列出自己策略实际需要的数据层。然后从最小可用组合入手——多数个人量化开发者只需要 Layer 1（K 线含复权 + 实时快照 + 盘前盘后）+ Layer 2（财报日历）。跑通之后再决定是否扩展到 Layer 3–5。 附：五层验收脚本 """ US-FULL-01 五层验收脚本 运行前填入你的 API Key """ import requests API_KEY = "your_api_key" BASE_URL = "https://api.tickdb.ai/v1" HEADERS = {"X-API-Key": API_KEY} def check_layer_1(): """Layer 1: 实时行情 + 盘前盘后""" try: data = requests.get( f"{BASE_URL}/market/ticker", params={"symbols": "AAPL.US"}, headers=HEADERS, ).json()["data"][0] has_pre = "pre_market_quote" in data has_post = "post_market_quote" in data print(f"Layer 1: {'PASS' if has_pre and has_post else 'FAIL'}") return has_pre and has_post except Exception as e: print(f"Layer 1: FAIL - {e}") return False def check_layer_2(): """Layer 2: 财报日历""" try: events = requests.get( f"{BASE_URL}/fundamentals/calendar", params={"market": "US", "category": "report", "from": "2026-09-17", "to": "2026-12-31"}, headers=HEADERS, ).json()["data"]["events"] print(f"Layer 2: {'PASS' if len(events) > 0 else 'FAIL'} ({len(events)} events)") return len(events) > 0 except Exception as e: print(f"Layer 2: FAIL - {e}") return False def check_layer_3(): """Layer 3: 公司行动与股息""" try: events = requests.get( f"{BASE_URL}/fundamentals/dividends", params={"symbol": "AAPL.US", "limit": 5}, headers=HEADERS, ).json()["data"]["events"] print(f"Layer 3: {'PASS' if len(events) > 0 else 'FAIL'} ({len(events)} events)") return len(events) > 0 except Exception as e: print(f"Layer 3: FAIL - {e}") return False def check_layer_4(): """Layer 4: 基本面季报""" try: rows = requests.get( f"{BASE_URL}/fundamentals/financials/latest", params={"symbol": "AAPL.US", "kind": "IS", "n": 4, "period_type": "q1,q2,q3,q4"}, headers=HEADERS, ).json()["data"]["rows"] fields = {r["field_name"] for r in rows} ok = "OperatingRevenue" in fields and "NetProfit" in fields print(f"Layer 4: {'PASS' if ok else 'FAIL'}") if not ok: print(" 提示：检查字段名是否为 OperatingRevenue/NetProfit，" "参数是否为 q1,q2,q3,q4") return ok except Exception as e: print(f"Layer 4: FAIL - {e}") print(" 提示：period_type 不能用 quarter，要用 q1,q2,q3,q4") return False def check_layer_5(): """Layer 5: AI-native 接入（MCP 协议层）""" print("Layer 5: 需单独配置 X-TickDB-Key 进行 MCP 测试") print(" 参考：JSON-RPC tools/call，工具名 get_ticker，参数 symbols/type") return None if __name__ == "__main__": print("=" * 50) print("美股数据五层验收") print("=" * 50) results = { "Layer 1": check_layer_1(), "Layer 2": check_layer_2(), "Layer 3": check_layer_3(), "Layer 4": check_layer_4(), "Layer 5": check_layer_5(), } print("=" * 50) passed = sum(1 for v in results.values() if v is True) print(f"通过: {passed}/4 (Layer 5 需单独测试)") print("对照五层结构，确认你的项目需要哪几层。") 附录 A：接口示例 端点 功能 /market/ticker 单标的/批量行情快照 /market/kline 历史 K 线 /market/kline/latest 最新 K 线 /market/kline/ex-factors 复权因子 /market/intraday 分时数据 /market/depth 盘口深度 /market/trades 逐笔成交 /market/trades/vwap VWAP /market/trade-days 交易日历 /market/trading-sessions 交易时段 /market/stock-info 标的基础信息 /market/intervals/kline 可用 K 线周期 /fundamentals/calendar 财经日历（财报、分红、拆股、IPO 等） /fundamentals/dividends 分红历史 /fundamentals/corp-actions 公司行动 /fundamentals/financials/latest 最新财务 /fundamentals/financials/fields 财务字段字典 /fundamentals/valuation/latest 估值快照 /fundamentals/valuation/ts 估值时序 /fundamentals/profile 公司档案 /fundamentals/industry/peers 同业公司 /realtime WebSocket 实时订阅 附录 B：错误码速查 HTTP 业务码 含义 400 40001 period_type 参数不支持（用 q1,q2,q3,q4，不是 quarter） 400 2001 复权参数不支持 401 1005 API Key 过期 403 3009 接口未开放 403 3010 市场未开放 404 40404 上游无数据 404 40405 查询条件无有效业务数据 422 5006 复权基础数据不可用 429 — 请求频率超限 503 5005 复权因子不可用 WS close 1008 — 密钥过期 附录 C：常见问题 Q：财报日历为什么不能按标的查？ A：TickDB 的 calendar 端点返回全市场事件流，通过 symbol 字段过滤。这反而更适合做股票池的批量事件过滤。如果你需要 AAPL 的预计财报发布日，可以从公司行动端点观察 FinancialReport 与 ReportDate 事件。 Q：字段名为什么和文档不一样？ A：我实测时发现 revenue 返回空，实际字段是 OperatingRevenue；net_income 实际是 NetProfit。建议先调 financials/fields 查字段字典，再写代码。 Q：WebSocket 推送字段为什么没写？ A：我实测了连接和订阅确认，但在等待窗口未收到 ticker 推送。推送字段、频率、盘前盘后推送行为，等美股交易时段补测后再写。 Q：CLI 为什么没写实测？ A：本次未运行 CLI 实际数据命令，原因是安全策略拒绝将密钥传给临时下载的 npm 包。CLI 是 AI Agent 和工作流的命令行入口，官方提供 16 个原生命令。 Q：MCP 示例能在 Claude / Cursor 里直接用吗？ A：MCP 协议层 tools/call 实测成功，但我不声称 Claude / Cursor 等 AI 客户端已调用。实际接入需要配置 X-TickDB-Key。 附录 D：实测证据索引 证据编号 验证内容 样本 日期 EV-US-FULL-01-01 get_ticker 返回盘前盘后字段 AAPL.US 2026-09-17 EV-US-FULL-01-02 calendar 返回全市场 report 事件 US 2026-09-17 EV-US-FULL-01-05 WebSocket 连接与订阅确认 AAPL.US 2026-09-17 EV-US-FULL-01-07 dividends 返回分红事件 AAPL.US 2026-09-17 EV-US-FULL-01-08 corp-actions 返回公司行动事件 AAPL.US 2026-09-17 EV-US-FULL-01-13 financials/latest 返回季度利润表 AAPL.US 2026-09-17 EV-US-FULL-01-11 valuation/latest 返回 PE 路径 AAPL.US 2026-09-17 EV-US-FULL-01-14 MCPget_ticker 协议层成功 AAPL.US 2026-09-17 文中实测数据来自 2026-09-17 的 API 调用，样本标的为 AAPL.US。字段名、参数和接口行为以官方接口文档为准。WebSocket 推送字段、CLI 命令输出、AAPL 直属财报日历样本待补测。单次调用只证明本次密钥、样本、参数和时间；不证明持续可用性、完整覆盖、性能或投资结论。 
