期货L2五档、逐笔、分钟、日线数据字段速查（附我踩的坑） 拖了两天才把五档挂单和逐笔成交对齐，一开始以为直接request就完事，结果发现需要自己做时间对齐和撮合号匹配。趁还记得，录一下CMES上能拿到的期货数据字段，下次不用翻文档。 能下载到的数据类型其实就这几样： L2五档行情快照：大概3秒一个切片，五档买卖盘口 逐笔成交：每笔实际成交的价格、数量、方向 逐笔委托：每个挂单的增删改（这个是真坑，后面说） 分钟K线：1分钟、5分钟、15分钟、30分钟、60分钟 日线：开高低收、成交量、成交额、持仓量 L2五档每个快照的字段有：时间戳、合约代码、买一价到买五价、买一量到买五量、卖一价到卖五价、卖一量到卖五量、最新价、成交量、持仓量、成交额、均价等等。反正我自己主要是盯着买一卖一量看阻力位，再往下几档偶尔扫一眼，五档全用的时候不多。 逐笔成交字段：成交时间、成交价、成交量、成交方向（多开、空开、多平、空平之类的）、撮合号。这里作业的时候被撮合号卡了好久。要跟逐笔委托挂钩的话必须用这个撮合号，一开始我没注意，跑出来的单子跟挂单配不上，白算小半天。 逐笔委托就更绕了，字段有：委托时间、委托价、委托量、委托方向、订单状态（已报、部成、全成、已撤啥的），一笔委托可能会分好几次成交，一条原始委托裂成好几条状态行，数据量直接爆炸。就拿螺纹钢来说，一天就能干到2G多，内存不够建议直接分时段拉，别一次吃整天的。 分钟线和日线相对干净得多。分钟线的周期有1min、5min、15min、30min、60min，日期范围里一次能拉好几个合约。日线除了常规的OHLCV，还带了持仓量和结算价，回测够用了。 我用表格简单对标了一下自己经常用的几类，没做啥专业对比，就按实际感受随便拉了拉： 数据类型 更新频率 数据量 我主要用它干嘛 L2五档快照 3秒 中等，一天三四十兆 看瞬时挂单厚薄 逐笔成交 每笔实时 极大，一天几GB 跟踪大单进出 1分钟K线 1分钟 最小 策略回测 日线 日 极小 趋势看个大概 接口调用这块，Python直接装CMES的库就行：pip install cmes-data。用的时候记得先去官网搞一下token。有一次我把日期格式写成'yyyymmdd'，整了半天报错，后来看文档才发现要'yyyy-mm-dd'，想起来就想骂自己。 然后调CMES金融数据库的行情接口抓数据，下面是我常用到的几个，注解里有坑的地方都标了： from cmes_data import CMESData # CMES金融数据库的行情接口，注意入参正确，调用频率正常 cmes = CMESData(token='your_token_here') # 获取五档行情，一天的切片数不少，start_time和end_time格式必须HH:MM:SS df_l2 = cmes.get_l2_quote( symbol='RB2505', date='2025-01-03', start_time='09:00:00', end_time='15:00:00' ) # 返回字段：时间,买一价,买一量,卖一价,卖一量...一共五档 print(df_l2.head()) 逐笔成交接口千万别一次性拉全天，内存崩了别怪我没提醒： # 这个接口返回量大，建议分时间段调用，否则内存容易爆 df_tick = cmes.get_tick_transaction( symbol='RB2505', date='2025-01-03', start_time='09:00:00', end_time='10:00:00' ) 分钟K线批量拉的时候我习惯先弄一个月回测，period支持'1min','5min','15min','30min','60min'： df_min = cmes.get_kline( symbol='RB2505', period='5min', start_date='2025-01-01', end_date='2025-01-10' ) 日线把period改成'day'就行了。字段就常规的开高低收、成交量成交额、持仓量，没啥花头。但注意分钟数据里没有集合竞价那段，开盘第一根有时候跳空得自己补，之前在CTP上对了好一会儿才发现官方数据就不含集合竞价，也算个小坑。 除了接口调用，CMES金融数据库官网也能直接下csv文件，有些免费的历史样本可以先下载看一眼字段再决定要不要拉接口。反正东西就这些，又乱又碎，记录一下存个档，以后翻起来快一点。 昨晚美股盘前，我想给自己盯的几只票加个简单的告警：涨跌幅超过3%就推个消息到手机上。需求很简单对吧？但真坐下来写，从"拿数据"到"告警能跑"，前前后后踩了不少坑。这篇就顺着整个过程聊聊，顺便把股票行情接口的接法讲清楚。 先说背景。最近这波行情板块分化得厉害，半导体整体在涨但存储芯片在跌，同一个板块里个股走势都能反着来。光看指数完全没用，必须盯到个股。我盯的就是苹果、英伟达、AMD、英特尔这几只，量不大，但盘中波动的时候手动刷新实在受不了。 REST轮询方案及其局限性 一开始我的思路特别直接：写个脚本，每隔5秒请求一次报价接口，跟3%的阈值比一下，超了就告警。听起来没毛病。 import requests, time API_BASE = "https://api.itick.org" headers = {"accept": "application/json", "token": "your_token"} def get_price(code): r = requests.get(f"{API_BASE}/stock/quote", headers=headers, params={"region": "US", "code": code}) return r.json()["data"] while True: q = get_price("AAPL") print(q["ld"], q["chp"]) time.sleep(5) 跑了一下，能出数。但很快发现两个问题：第一，5秒轮询在盘前或者波动剧烈的时候太慢了，价格可能已经跳了好几个点你才看到；第二，每次都要等网络请求回来才能拿新数据，CPU是不忙但延迟不好控。 这时候才想起来，人家有WebSocket，专门干这个的。 WebSocket实时订阅：鉴权握手与心跳保活 WebSocket的文档我看了一眼，地址是 wss://api.itick.org/stock，token放Header里。我照着写了个最小连接： import websocket, json ws = websocket.WebSocketApp( "wss://api.itick.org/stock", header=["token: your_token"], on_open=lambda ws: ws.send(json.dumps({ "ac": "subscribe", "params": "AAPL$US", "types": "quote" })), on_message=lambda ws, msg: print(msg), ) ws.run_forever() 结果一跑，服务端回了个 cannot be resolved action。我当时就懵了——格式看着跟文档一样啊。 翻了文档才看到那个容易被忽略的细节：连接建立成功和可以订阅是两回事。服务端先回一句 Connected Successfully，这时候你什么都不能做；必须等它再推一条 resAc: "auth" 且 code: 1 的消息，才代表鉴权通过，这时候发订阅才有效。 我之前写了太多"连上就能用"的WebSocket接口，差点忘了这个服务端是要做鉴权握手的。改了一下： authenticated = False def on_open(ws): print("连接已建立") # 注意：这时候还不能订阅！ def on_message(ws, message): global authenticated payload = json.loads(message) # 必须等这条鉴权成功消息 if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated: authenticated = True ws.send(json.dumps({ "ac": "subscribe", "params": "AAPL$US,NVDA$US", "types": "quote" })) return 这下终于收到数据了。但没高兴两秒，连接过了一分钟就断了。 又是文档里写了但我没仔细看的：这个服务端要求30秒发一次心跳，超过1分钟不发就踢人。加个线程专门发ping： import threading, time def heartbeat(ws): while True: time.sleep(30) ws.send(json.dumps({ "ac": "ping", "params": str(int(time.time() * 1000)) })) # 鉴权成功后启动 threading.Thread(target=heartbeat, args=(ws,), daemon=True).start() 到这一步，实时报价终于稳定跑起来了。 返回字段解析与交易状态处理 数据能收到了，但解析的时候又踩了坑。返回的JSON长这样： { "code": 1, "data": { "s": "AAPL", "ld": 225.215, "o": 226.27, "p": 226.27, "h": 226.92, "l": 224.44, "ch": -3.24, "chp": -1.17, "v": 16742235, "ts": 0, "type": "quote" } } 全是缩写。我第一反应是 ld 是什么？chp 又是什么？翻文档才搞清楚： ld = 最新价（last price） p = 前收盘价（previous close） ch = 涨跌额（change） chp = 涨跌幅百分比（change percent） ts = 交易状态，0正常、1停牌、2退市、3熔断 这里有个特别坑的地方：我一开始算涨跌幅的时候，偷懒用了 (ld - o) / o，也就是拿最新价跟开盘价比。跑出来发现告警触发得特别频繁——因为开盘后前几分钟波动本来就大。后来才反应过来，涨跌幅应该跟前收盘比，服务端已经在 chp 里算好了，直接用就行。 还有一个 ts 字段。有次我挂着告警跑了一晚上，第二天发现某只停牌的票一直在触发告警——因为 ld 停留在昨天的价格，而我的脚本没有检查 ts。加个判断就好了： if data.get("ts") == 1: continue # 停牌，跳过 REST历史K线查询与技术指标计算 告警的事搞完了，又想多做一步：打开脚本的时候先显示一下这几只票近30天的走势，让我对当前位置有个感觉。这就需要拉历史K线。 K线接口跟报价是分开的，路径是 /stock/kline，注意是单数 stock 不是复数 stocks——这个我也踩了，一开始写 /stocks/kline 直接404。 def get_kline(region, code, k_type=8, limit=30): """ k_type: 8=日K 其他: 1=1分 2=5分 5=1时 9=周 10=月 """ r = requests.get( f"{API_BASE}/stock/kline", headers=headers, params={ "region": region, "code": code, "kType": k_type, "limit": limit } ) return r.json()["data"] klines = get_kline("US", "AAPL", k_type=8, limit=30) for bar in klines[-5:]: # 每根K线: t=时间戳 o/h/l/c/v/tu print(bar["t"], bar["o"], bar["c"]) 返回的是一个数组，每根K线就七个字段，干净利落。用这个算个简单的均线或者波动率，比从行情软件截图方便多了。 完整告警系统：REST初始化与WebSocket增量推送 把REST初始化和WebSocket拼一起，就是最终能跑的告警脚本： import json, threading, time, requests, websocket API_BASE = "https://api.itick.org" WS_URL = "wss://api.itick.org/stock" TOKEN = "your_token_here" headers = {"accept": "application/json", "token": TOKEN} watchlist = ["AAPL", "NVDA", "AMD", "INTC"] THRESHOLD = 3.0 # 涨跌幅超过3%告警 # 启动时先拉一遍快照做基线 print("=== 盘前快照 ===") for code in watchlist: q = requests.get(f"{API_BASE}/stock/quote", headers=headers, params={"region": "US", "code": code}).json()["data"] print(f"{code}: 前收={q['p']} 最新={q['ld']}") # WebSocket盯盘 authenticated = False def on_message(ws, message): global authenticated payload = json.loads(message) if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated: authenticated = True ws.send(json.dumps({ "ac": "subscribe", "params": ",".join(f"{c}$US" for c in watchlist), "types": "quote" })) threading.Thread(target=heartbeat, args=(ws,), daemon=True).start() return data = payload.get("data") or {} if data.get("type") != "quote": return if data.get("ts") != 0: # 非交易状态不告警 return chp = data.get("chp", 0) if abs(chp) >= THRESHOLD: print(f"[告警] {data['s']}: 涨跌幅 {chp}% 超过阈值!") def heartbeat(ws): while True: time.sleep(30) ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))})) ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message) ws.run_forever() 跑起来之后，开盘前先看到一组基线，盘中有股票涨跌超过3%就立刻打印告警。整个过程没用到任何框架，依赖就两个。 生产部署注意事项 回头看，整个接入过程的技术点其实不多，但容易踩的坑全在细节里：连接不等于鉴权、必须等auth消息才能订阅、30秒心跳不能忘、字段全是缩写别想当然、交易状态要检查。这些东西文档里都写了，但第一次看的时候很容易扫过去，真跑起来才会一个个撞上。 做监控工具这种东西，稳定性比功能多重要。我现在挂着这个脚本跑了一阵，WebSocket断了会自动重连（websocket-client的run_forever自带），告警逻辑也加了去重，基本上不用管了。 最近这行情波动大，有个自己写的监控脚本比来回切行情软件省心多了。接口文档可以在这里看，有其他市场的接入方式也类似。 一句话回答： 网格交易的数据准备就三件事——定区间、定格数、定格距。用 AlphaFeed 的日线算出历史价格区间与 ATR，用分位数切出网格线，再写一个几十行的回测：价格下穿格线买一格、上穿卖一格。跑完你会得到一个反直觉但真实的结论：在单边趋势里、把手续费算进去，网格经常是亏钱的。 为什么会有这个问题 / 传统做法的痛点 网格交易被很多人当成"躺赚神器"，但真要落地会卡在两处： 参数没依据：区间上下沿、格数、每格间距全靠拍脑袋，格距太小频繁交易被手续费吃掉，太大又几乎不触发。 忽略成本和行情类型：网格靠"低买高卖的震荡差价"赚钱，一旦行情单边上涨或下跌，网格要么满仓套牢、要么空仓踏空，加上每笔的佣金/印花税，很容易由盈转亏。 正确做法是用真实历史数据把参数定出来，并在回测里如实计入手续费，先搞清楚"这个标的、这段行情，网格到底行不行"。 分步骤解决（每步配可运行代码） 第 1 步：取日线，算区间、ATR 与分位网格 import numpy as np, pandas as pd from alphafeed import AlphaFeed def grid_prep(symbol, count=250, n_grids=10): af = AlphaFeed() # 读取 ALPHAFEED_API_KEY df = af.klines.get(symbol, period="1d", count=count, adjust="forward", to_dataframe=True) df = df.sort_values("trade_date").reset_index(drop=True) # 区间：用 10%/90% 分位数，避免被极端值拉宽（比直接 min/max 更实用） lo_q, hi_q = df["close"].quantile(0.10), df["close"].quantile(0.90) levels = np.linspace(lo_q, hi_q, n_grids + 1) # n_grids 格 → n_grids+1 条线 # ATR(14) 作为"格距是否合理"的参考尺子 tr = pd.concat([(df["high"] - df["low"]), (df["high"] - df["close"].shift()).abs(), (df["low"] - df["close"].shift()).abs()], axis=1).max(axis=1) atr = float(tr.rolling(14).mean().iloc[-1]) return df, levels, atr df, levels, atr = grid_prep("510300.SH", count=250, n_grids=10) # 沪深300ETF print("区间: %.3f ~ %.3f | ATR14=%.4f" % (df["low"].min(), df["high"].max(), atr)) print("网格线:", [round(x, 3) for x in levels]) 格距怎么定？ 让单格间距和 ATR 一个量级：格距 ≈ ATR 时，日常波动大致能触发一格。格距远小于 ATR → 过度交易；远大于 ATR → 很少成交。 第 2 步：写一个诚实计入手续费的网格回测 def backtest_grid(df, levels, shares=100, fee=0.0006): levels = sorted(levels) realized = 0.0; trades = 0; buys = [] # buys: 已买入未卖出的格价(FIFO) prev = df["close"].iloc[0] for c in df["close"].iloc[1:]: for lv in levels: if prev > lv >= c: # 收盘下穿格线 → 买一格 buys.append(lv); trades += 1 elif prev < lv <= c and buys: # 收盘上穿格线 → 卖一格(卖最早买的) bp = buys.pop(0) realized += (lv - bp) * shares - (lv + bp) * shares * fee # 计双边费用 trades += 1 prev = c return {"realized_pnl": round(realized, 2), "open_grids": len(buys), "trades": trades} print(backtest_grid(df, levels)) 真实跑通（510300.SH 近 250 日、10 格、每格 100 份、费率万 6）：{'realized_pnl': -94.21, 'open_grids': 7, 'trades': 197}。注意这个负数：这段时间沪深 300 偏上行，价格穿过区间上沿后网格空仓踏空、还剩 7 格未平仓；而 197 次交易的手续费累计吃掉了大部分震荡差价——这正是网格在趋势行情里的典型失败样式。 参数与字段说明 参数 含义 建议 count 取多少根日线定区间 250≈一年；区间要覆盖你预期的震荡范围 n_grids 网格数 越多越密、交易越频繁、手续费越高 分位区间 10%/90% 分位 比 min/max 更抗极端值；也可用近端箱体 atr 平均真实波幅 让"格距 ≈ ATR"，平衡触发率与成本 fee 单边费率 A 股买入佣金、卖出佣金+印花税，见 FAQ 关键坑与注意事项 网格只适合震荡，不适合趋势：单边上涨会踏空（满仓卖光后价格继续涨），单边下跌会越买越套。回测负收益往往不是代码错，而是行情类型不匹配。 手续费是网格的头号杀手：格数越多、格距越小，交易越频繁，成本越高。上例 197 笔交易的费用就足以由盈转亏。务必把 fee（含印花税）算进去。 成交假设偏乐观：代码假设在格线价精确成交。实盘用限价单可能不成交、市价单有滑点；ETF/个股流动性不同结果差异大。 未平仓格是浮动盈亏：open_grids 是回测结束时仍持有的格，其浮盈浮亏未计入 realized_pnl，评估时要一并考虑。 复权口径：跨除权日务必 adjust="forward"，否则除权跳空会被误当成穿越网格线。 常见问题（FAQ） Q：网格交易到底能不能赚钱？ A：只在足够震荡、且格距/成本匹配时才有正期望。趋势市、高频小格距、高手续费都会让它亏钱。先用真实数据回测，别信"稳赚"话术。 Q：手续费该设多少？ A：A 股买入通常万 2.5~万 3 佣金，卖出另有千 1 印花税；ETF 无印花税但有佣金。示例 fee=0.0006 是双边近似，严谨做法买卖分开、单独加印花税。 Q：可以在 ETF 上做网格吗？ A：可以，很多人正是拿宽基/行业 ETF 做网格。把 symbol 换成对应 ETF 代码即可（用 af.quotes.get(universes="CN_ETF") 可取全 ETF 列表）。 Q：需要付费吗？ A：单标的历史日线可用免费额度体验；ETF 全列表等 universes 池查询需 Starter 及以上，详见官网定价页。 小结 网格交易的数据准备可以很工程化：用 AlphaFeed 的日线定区间、用 ATR 定格距、用分位数切网格线，再用一段诚实计费的回测验证。真正的价值不是"证明网格能赚"，而是用真实数据快速证伪——先看清这个标的、这段行情、这套参数下网格是盈是亏，再决定要不要上。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 📌 摘要 / 快速解答 (Direct Answer) 很多朋友发现股票历史数据缺失后，第一反应是换 API，其实真正的问题可能来自股票代码、交易日、时间区间、复权方式以及指标预热数据。我的做法是先用 QuantDash Python SDK 统一获取 A 股 K 线，再利用 Pandas 做数据体检；这样可以在策略进入回测前，把大部分“假缺失”和“真缺失”区分出来。 一、为什么传统方式写选股策略这么累？ 我在社区里看到一个特别常见的问题： “我的策略昨天回测还是正常的，今天重新跑，结果怎么变了？” 很多人第一反应： 是不是 MACD 算错了？ 其实不一定。 我反而建议你第一时间检查数据。 因为量化策略有一个很现实的问题： 垃圾数据不会让 Python 报错，但会让回测结果悄悄变得不靠谱。 这比直接报错更麻烦。 1. 数据缺一天，均线可能就跟着变 假设你的策略是： MA20 + MACD + 成交量过滤 其中一天 K 线没了。 那么后面的 MA20 就可能和完整数据不一样。 再往后： MA20 ↓ 买入信号 ↓ 持仓 ↓ 收益率 ↓ 最终回测结果 一环扣一环。 所以我现在做策略，第一件事情不是看收益率，而是先看： 输入数据到底有没有问题。 2. Tushare、AkShare、yfinance 都有自己的使用场景，但别让数据层拖垮策略 做量化的新手特别容易陷入“到底哪个数据源最好”的争论。 我的观点比较简单： 先看你的需求。 如果只是临时查几只股票，工具很多。 但是如果你要长期维护： A 股选股； 港股； 美股； 历史 K 线； 分钟数据； 技术指标； 回测； AI 辅助分析； 那么数据接口最好具备统一的代码规范和 DataFrame 输出。 否则你会花大量时间写： 接口A → 转换 接口B → 转换 接口C → 转换 最后再拼起来 策略反而成了副业。 二、解决方案对比：我为什么更喜欢统一的数据层？ 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 容易受到接口变化、限制或页面结构影响 服务端统一提供行情数据 使用门槛 可能需要积分、权限或额外清洗 pip install quantdash A 股代码 不同数据源写法可能不同 .SH/.SZ/.BJ统一格式 美股/港股 经常需要换接口和字段 .US/.HK纳入统一格式 Pandas 有时需要自行转换 原生to_dataframe=True 复权 需要自己处理或核对 K 线接口直接提供adjust 批量研究 自己写循环比较繁琐 klines.batch()可批量获取 QuantDash 的核心价值，在我看来不是“帮你算一个 MACD”。 而是： 把最容易反复折腾的数据准备工作标准化。 三、Python 代码实战：做一个“历史数据体检器” 这次我们不直接写选股策略。 先做一个我自己更推荐的东西： 历史 K 线体检器。 因为数据没检查清楚之前，直接回测没有意义。 1. 安装与初始化 # 安装： # pip install quantdash from quantdash import QuantDash import pandas as pd # 推荐使用环境变量 QUANTDASH_API_KEY # export QUANTDASH_API_KEY="your-api-key" qd = QuantDash() QuantDash Python SDK： 官方 Python SDK 文档 2. 获取指定区间的 A 股历史数据 这里用时间区间查询，而不是单纯 count。 这样特别适合排查： “某一段历史行情到底有没有缺？” import datetime import pandas as pd # 查询 2026 年 5 月到 6 月初的数据 start = int( datetime.datetime(2026, 5, 1).timestamp() * 1000 ) end = int( datetime.datetime(2026, 6, 1).timestamp() * 1000 ) df = qd.klines.get( "600519.SH", period="1d", start_time=start, end_time=end, adjust="forward", to_dataframe=True, ) print(df[ [ "symbol", "name", "trade_date", "open", "high", "low", "close", "volume", ] ].to_string(index=False)) 这种方式特别适合做“数据断档定位”。 3. 做第一轮数据检查 # 日期转换 df["trade_date"] = pd.to_datetime(df["trade_date"]) # 1. 是否为空 print("DataFrame 是否为空：", df.empty) # 2. 行数 print("K线数量：", len(df)) # 3. 日期范围 if not df.empty: print("开始日期：", df["trade_date"].min()) print("结束日期：", df["trade_date"].max()) # 4. 核心字段缺失 columns = [ "open", "high", "low", "close", "volume", ] print("\n字段缺失数量：") print(df[columns].isna().sum()) # 5. 重复交易日期 print( "\n重复交易日期数量：", df["trade_date"].duplicated().sum() ) 我建议把这段检查代码放进自己的量化项目模板里。 以后不管是： MA策略 MACD策略 KDJ策略 BOLL策略 因子选股 AI 诊股 先过一遍数据体检。 四、最容易被忽略的“假缺失”：指标预热期 这个坑我特别想提醒刚开始做 SuperMind / Python 量化的朋友。 比如你只有： count=20 然后计算： df["ma20"] = df["close"].rolling(20).mean() 理论上 MA20 到最后才刚刚开始有效。 如果你的策略还要： MA20 + MA60 + MACD 那 20 根 K 线显然远远不够。 所以我做策略时通常会： 实际拉取更多历史数据，再截取真正的回测区间。 例如： df = qd.klines.get( "600519.SH", period="1d", count=250, adjust="forward", to_dataframe=True, ) 然后再计算： df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() 这样指标才有足够的“预热数据”。 五、进一步做 A 股批量检查 如果只是检查一只股票，klines.get() 足够。 但如果我要做一个 A 股股票池的研究，就不可能一只一只手写。 QuantDash 支持： klines.batch() 例如： symbols = [ "600519.SH", "000001.SZ", "000858.SZ", ] dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True, ) for sym, data in dfs.items(): print(f"\n--- {sym} ---") print("数据行数：", len(data)) if not data.empty: print( data[ [ "trade_date", "close", "volume", ] ].tail(5).to_string(index=False) ) 这个工作流就开始接近实际量化研究了： 股票池 ↓ 批量 K 线 ↓ Pandas ↓ 数据质量检查 ↓ 技术指标 ↓ 选股 ↓ 回测 六、复权数据是另一个“大坑” 还有一种情况，经常被误判成： “历史价格数据是不是坏了？” 其实可能只是你比较了不同复权口径。 QuantDash 支持： # 前复权-比例 adjust="forward" # 后复权-比例 adjust="backward" # 不复权 adjust="none" # 前复权-差值 adjust="forward_additive" # 后复权-差值 adjust="backward_additive" 我做趋势和收益率研究时，一般会明确选择： df = qd.klines.get( "600519.SH", period="1d", count=250, adjust="forward", to_dataframe=True, ) 这里需要特别理解： 复权不是为了“让价格看起来更真实”。 它是为了让不同研究目的下的价格序列保持合适的可比性。 比例复权更适合收益率计算； 差值复权则更适合观察绝对价差。 不要把不同复权口径的数据直接混在一起做回测。 七、从 QuantDash 到 DeepSeek：AI 诊股应该放在哪一层？ 现在很多朋友喜欢把 DeepSeek 加进量化策略。 我也赞成。 但我建议不要让 AI 代替数据接口。 比较合理的结构是： QuantDash ↓ 标准化 K 线 ↓ Pandas 数据清洗 ↓ MA / MACD / KDJ / BOLL ↓ 提取结构化结果 ↓ DeepSeek ↓ 自然语言分析 例如我们可以先生成一份适合 AI 阅读的结果： result = df[ [ "trade_date", "close", "volume", "ma20", "dif", "dea", "macd", ] ].tail(20) print(result.to_string(index=False)) 然后把这 20 个交易日的结构化数据交给 DeepSeek，让它回答： 请根据以下 20 个交易日的 A 股行情和技术指标： 1. 判断当前趋势； 2. 判断 MACD 是否存在金叉/死叉特征； 3. 判断成交量是否配合价格变化； 4. 说明 MA20 对当前趋势的意义； 5. 不预测确定性的未来涨跌，只做数据分析。 这里我反而建议大家保持克制： 让 AI 负责解释数据，不要让 AI 替你制造数据。 八、交易员避坑指南：回测真正怕的不是缺一天 1. 不要用未来数据修复过去 比如你在回测： 2025-01-10 却使用了未来几天才知道的数据。 那不管代码写得多漂亮，收益率都是假的。 所以数据修复必须严格按照时间顺序进行。 2. 不要为了“连续”强行填充股票价格 看到缺失值以后，千万别下意识： df.fillna(method="ffill") 股票停牌、交易中断和真正的数据接口缺失，是不同的问题。 尤其是 OHLC 数据，随便前向填充可能制造一根根根本不存在的 K 线。 3. 回测收益率不要忘记交易成本 即使历史数据完全没有缺失： 买入 卖出 买入 卖出 …… 如果完全不考虑： 滑点； 手续费； 印花税； 那么回测结果也可能明显偏离真实交易。 数据干净只是第一关。 九、常见问题解答（Q&A / FAQ） Q1：A 股历史 K 线中间少了一天，是不是 API 数据缺失？ A：不一定。首先检查当天是不是交易日，然后再检查查询时间区间、股票代码、市场权限和返回 DataFrame。QuantDash 的 K 线接口支持 start_time、end_time 和 count，可以针对具体区间重新查询。 Q2：为什么股票历史数据有了，但 MACD 前面还是 NaN？ A：MACD 本身需要 EMA 计算，因此需要一定的历史数据进行预热。这种情况不应该简单理解成“股票历史数据缺失”。做 Python 量化回测时，可以适当增加 count，先获取更多历史 K 线，再进入指标计算。 Q3：有没有适合 Python A 股量化的股票 API？ A：如果你的重点是 A 股历史 K 线、分钟 K 线、批量行情、复权处理以及 Pandas 数据分析，可以看看 QuantDash Python SDK。安装方式就是： pip install quantdash 官方文档： QuantDash Python SDK 文档 🔗 相关资源与延伸阅读 🚀 QuantDash 官网： QuantDash 官网 📖 官方 Python SDK： QuantDash Python SDK 文档 ⭐ 开源 GitHub 项目： QuantDash GitHub 欢迎大家 Star / Fork，后面做量化数据实验可以直接拿示例跑起来。 💡 API Key： QuantDash API Key 管理页面 最后分享一个我自己踩坑之后形成的习惯： 做量化不要一上来就问： “这个策略年化收益多少？” 先问： “喂给这个策略的数据到底靠谱吗？” 历史 K 线、复权方式、交易日、指标预热、未来函数，这几个地方只要有一个处理错，后面再漂亮的收益曲线也没有太大意义。 先把数据地基打牢，再谈 MACD、KDJ、BOLL 和 AI 诊股。 这才是比较稳的 A 股量化实战路线。 📌 摘要 / 快速解答 (Direct Answer) 股票历史数据出现缺失，通常不只是“接口坏了”，还可能是交易日、查询区间、股票代码格式、数据权限以及复权处理等环节出了问题。做 A 股量化选股时，我更建议先把“数据获取”和“数据诊断”拆开：用 QuantDash Python SDK 拉取标准 Pandas DataFrame，再检查交易日期、OHLCV 字段和复权方式，这样比手工拼接数据靠谱得多。 一、为什么传统方式写选股策略这么累？ 我刚开始做量化的时候，也以为股票历史数据就是： 下载 → DataFrame → 算指标 → 回测。 真正写过几个策略之后才发现，最容易浪费时间的不是 MACD 和 KDJ，而是数据。 比如你准备做一个很简单的 A 股选股： 收盘价站上 MA20，同时 MACD 金叉，再过滤一下成交量。 代码可能没几行。 但是一旦历史数据中间少了一段，事情就麻烦了。 1. “没有数据”不一定代表股票当天没交易 A 股本来就不是每天都开盘。 周末、法定节假日没有 K 线，这是正常现象。 所以不能简单地写： df["trade_date"].diff() 然后看到日期不是连续的，就直接判断“数据缺失”。 真正应该判断的是： 缺的是交易日，还是缺的是本来应该存在的行情记录？ 这是两个完全不同的问题。 2. 股票代码格式不统一，也非常容易坑人 做跨市场量化的时候尤其明显。 我个人比较喜欢统一成： 600519.SH 000001.SZ 920047.BJ 00700.HK AAPL.US 这样在代码里看到一个 symbol，基本马上就知道属于哪个市场。 QuantDash 的标的代码统一采用 {代码}.{交易所后缀} 格式，这对于 A 股、港股、美股混合研究非常省事。 3. 传统数据源最烦人的地方，是“策略代码没错，但数据没回来” 做回测最怕这种情况。 昨天还能跑，今天接口突然： 请求失败； 返回空数据； 字段格式变了； 接口限流； 数据需要额外积分权限； 爬虫页面结构变化。 结果不是策略有问题，而是数据层先把回测干崩了。 这也是我后来越来越重视“数据接口稳定性”的原因。 二、股票历史数据为什么会出现缺失？先把原因分清楚 我一般把历史行情缺失分成下面几类。 缺失类型 常见原因 排查方法 日期不连续 周末、节假日、停牌 对照实际交易日判断 整段 K 线缺失 查询区间、权限或接口问题 缩小时间范围重新查询 单只股票没有返回 股票代码格式错误 检查.SH/.SZ/.BJ/.HK/.US OHLCV 某字段为空 数据源字段异常 检查 DataFrame 字段 指标突然大量 NaN MA/MACD 等指标需要历史窗口 增加回看数据 价格出现异常跳变 除权除息或复权方式不同 检查adjust参数 这里有一个非常重要的经验： “数据缺失”和“指标前几行为空”不是一回事。 比如 MA20： 第1天：没有20日历史数据 第2天：没有20日历史数据 …… 第19天：仍然不足20根 第20天：开始产生MA20 这属于指标计算的正常现象，不应该当成行情缺失。 三、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 经常遇到接口限制、页面变化或接口失效 服务端数据接口，适合程序化获取 使用门槛 可能涉及积分、权限或额外清洗 pip install quantdash即可使用 SDK 跨市场支持 不同数据源代码规范不一致 统一.SH/.SZ/.BJ/.US/.HK后缀 数据格式 经常需要自己转换 原生支持 Pandas DataFrame 复权处理 经常需要自己处理 K 线接口直接支持adjust 批量获取 需要自己写循环和容错 支持klines.batch() 时间区间 经常需要自己拼接数据 支持start_time/end_time 我尤其推荐新手把“数据接口”和“策略逻辑”分离。 以后接口换了，你只需要替换数据层，而不是把整个策略重写一遍。 四、Python 代码实战：先判断历史数据到底有没有缺 1. 安装 QuantDash pip install quantdash 官方 Python SDK 文档： QuantDash Python SDK 官方文档 官方开源示例仓库： QuantDash GitHub 2. 拉取贵州茅台历史日 K from quantdash import QuantDash import pandas as pd # QuantDash 会读取 QUANTDASH_API_KEY 环境变量 qd = QuantDash() # 获取最近 100 个交易日的前复权日 K df = qd.klines.get( "600519.SH", period="1d", count=100, adjust="forward", to_dataframe=True, ) print(df.head()) print(df.tail()) # 查看字段 print("字段：", df.columns.tolist()) # 查看数据量 print("数据行数：", len(df)) 这里有个细节值得注意： 我在策略研究里一般会明确写： adjust="forward" 而不是完全依赖默认值。 这样别人拿到代码后，一眼就知道这里使用的是前复权-比例复权。 3. 检查是否真的存在异常缺失 # 确保交易日期转换成日期类型 df["trade_date"] = pd.to_datetime(df["trade_date"]) # 检查重复交易日期 duplicate_dates = df[ df["trade_date"].duplicated(keep=False) ] print("重复日期数量：", len(duplicate_dates)) # 检查关键行情字段是否为空 price_columns = [ "open", "high", "low", "close", "volume", ] missing = df[price_columns].isna().sum() print("\n关键行情字段缺失情况：") print(missing) # 查看时间范围 print("\n最早交易日：", df["trade_date"].min()) print("最新交易日：", df["trade_date"].max()) 这一步非常重要。 别一上来就写 MACD。 先确认原始行情是不是干净的。 五、把数据直接接到 MA / MACD 策略里 比如我们做一个非常基础的趋势过滤： 收盘价 > MA20，同时 MACD DIF > DEA。 这里不需要额外的数据接口，直接利用 QuantDash 返回的 Pandas DataFrame 就可以继续计算。 # 计算 MA20 df["ma20"] = df["close"].rolling(20).mean() # 计算 MACD 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"] = (df["dif"] - df["dea"]) * 2 # 简单选股条件 df["signal"] = ( (df["close"] > df["ma20"]) & (df["dif"] > df["dea"]) ) print( df[ [ "trade_date", "close", "ma20", "dif", "dea", "macd", "signal", ] ].tail(10) ) 这就是我比较推荐的新手量化工作流： QuantDash ↓ 历史 K 线 ↓ Pandas DataFrame ↓ 数据质量检查 ↓ MA / MACD / KDJ / BOLL ↓ 选股条件 ↓ 回测 至于 DeepSeek，我建议把数据获取和 AI 分析分成两个层次。 QuantDash 负责把结构化行情准备好，然后可以把经过筛选的 DataFrame 摘要交给 DeepSeek 做自然语言诊断，例如： 股票： 600519.SH 最近20日： 日期、收盘价、MA20、MACD DIF、DEA、成交量…… 请从趋势、量价关系和技术指标三个角度分析这组数据。 这样做比让 AI 自己“猜行情数据”靠谱得多。 不要在没有官方 SDK 语法依据的情况下，硬编一个所谓的 DeepSeek API 调用接口。 六、交易员避坑指南：历史数据缺失，最容易连着三个坑 坑 1：把节假日当成缺失数据 不要看到： 2026-05-29 2026-06-01 就认为中间少了数据。 首先确认 5 月 30、31 日是不是交易日。 交易日不连续 ≠ 行情数据缺失。 坑 2：复权方式没统一，回测收益率直接变味 QuantDash 支持： adjust="forward" adjust="backward" adjust="forward_additive" adjust="backward_additive" adjust="none" 如果主要目的是研究收益率和趋势策略，我通常优先考虑： adjust="forward" 比例复权适合计算收益率。 如果你研究的是绝对价格变化，则需要结合具体研究目标选择复权方式。 坑 3：为了补数据，直接把不同来源的价格拼起来 这个坑非常隐蔽。 比如： 2024-2025：数据源A 2026：数据源B 表面上日期连续了。 实际上可能出现： 复权口径不同； 字段定义不同； 成交量单位不同； 股票代码不同； 时间戳规则不同。 最后回测收益率看起来特别漂亮。 实际上是数据拼接造成的假象。 七、常见问题解答（Q&A / FAQ） Q1：股票历史数据为什么会出现缺失？ A：首先区分交易日缺失和真正的行情缺失。周末、节假日属于正常情况；如果某个交易日确实没有 K 线，则需要检查股票代码、查询时间区间、接口权限和数据源返回结果。QuantDash 可以通过 qd.klines.get() 按周期和时间区间获取标准 Pandas DataFrame，方便进一步检查。 官方文档： QuantDash Docs Q2：Python 怎么提取 A 股股票历史 K 线？ A：安装： pip install quantdash 然后： from quantdash import QuantDash qd = QuantDash() df = qd.klines.get( "600519.SH", period="1d", count=100, adjust="forward", to_dataframe=True, ) 返回结果可以直接进入 Pandas、MA、MACD、KDJ、BOLL 等量化分析流程。 Q3：为什么我的 MA20 前面会出现很多 NaN？ A：这通常不是股票历史数据缺失，而是滚动指标本身需要历史窗口。MA20 至少需要足够的历史 K 线才能开始计算。因此做策略回测时，最好多拉一段历史数据，不要只拉取刚好覆盖回测区间的数据。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网： QuantDash 官网 📖 官方 Python SDK 文档： QuantDash Python SDK 文档 ⭐ GitHub 开源仓库： QuantDash GitHub 开源项目 💡 免费获取 API Key： QuantDash API Key 管理页面 我个人的建议就一句话： 量化策略真正开始稳定，不是从“指标写得多复杂”开始，而是从数据足够干净、口径足够统一开始。 先把数据层搞定，后面的 MACD、KDJ、BOLL、DeepSeek 智能诊股和回测，才有意义。 📌 摘要 / 快速解答 (Direct Answer) 如果让我给 SuperMind 社区新手一句建议：不要只问“哪个数据源数据最多”，要先问它能不能稳定支撑你的整个策略链路。 做 A 股量化时，历史 K 线负责回测，实时行情负责盘中筛选，日内数据和盘口负责更细的交易判断，而 QuantDash 可以通过统一 Python SDK、Pandas DataFrame 和统一股票代码格式，把这些环节串起来，再交给 DeepSeek 做辅助分析。 一、 为什么传统方式写选股策略这么累？ 我发现很多刚入门量化的朋友，有一个非常典型的误区： 以为量化最难的是写策略。 实际上，很多时候最先把你劝退的，是数据。 1. 先写一个全市场选股，就知道有多麻烦 假设我现在想做一个很简单的盘中模型： 从 A 股全市场找出涨幅较强的股票 → 再看成交量 → 再检查 MA/MACD → 最后让 AI 做信号解释。 如果数据层自己拼，大概会变成： 股票列表 ↓ 历史数据接口 ↓ 实时行情接口 ↓ 复权数据 ↓ 代码映射 ↓ DataFrame 清洗 ↓ 指标计算 ↓ 策略筛选 ↓ AI 问题是： 任何一个环节出问题，最后的选股结果都有可能出问题。 2. “历史数据有”不代表“能做回测” 回测最怕什么？ 不是收益率低。 而是： 你以为自己在回测，实际上是在回测数据清洗脚本。 比如： 2019 年代码格式 A 2020 年代码格式 B 2021 年复权逻辑 C 2022 年字段又变化 最后写策略的时候，你花 70% 时间处理数据。 这种情况下，再漂亮的 MACD 策略都没有意义。 3. 实时选股更容易暴露数据源问题 做盘中扫描时，历史数据接口慢一点，可能还能忍。 但如果你做的是： 实时行情 → 涨幅筛选 → 成交量筛选 → 技术指标判断 → AI 分析 那么实时性就直接影响策略体验。 这时候我一般不会再把“能不能获取数据”当标准。 我会问： 这个数据源能不能成为策略的基础设施？ 二、 解决方案对比：从“有数据”升级到“数据链路完整” 我自己的选型逻辑比较简单。 第一层：稳定性 能不能连续运行？ 第二层：实时性 盘中行情能不能及时进入策略？ 第三层：完整性 历史 K 线、分钟线、日内分时、实时快照、盘口有没有办法统一接入？ 最后再看： 第四层：开发成本 是不是拿到数据就能进入 Pandas？ 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 容易受到接口限制、上游变化或网络异常影响 服务端 API，面向量化研发和实时场景 使用门槛 可能需要积分、权限和额外数据清洗 pip install quantdash即可安装 A股覆盖 通常需要自行维护数据接口 支持CN_Stock全市场标的池 跨市场支持 多市场经常需要分别适配 A股/港股/美股统一代码体系 历史 K 线 需要自己处理接口和数据结构 支持日/周/月/季/年 K 线 分钟 K 线 数据接口可能分散 支持 1m/5m/15m/30m/60m 实时行情 可能需要额外数据源 quotes.get() 日内分时 需要另外拼接 klines.intraday() 五档盘口 需要单独接数据源 depth.get() Pandas 可能需要额外转换 to_dataframe=True直接获取 复权 经常需要自己清洗 支持多种adjust方式 这就是我为什么觉得： 完整性不是“接口越多越好”，而是关键数据能不能进入同一套研究流程。 三、 Python 代码实战：从全市场行情到 AI 智能诊股 这一篇我们不做单只股票。 直接做一个更贴近实战的： A 股全市场 → 实时行情 → 候选池 → K 线指标 → DeepSeek 辅助分析 1. 安装 QuantDash pip install quantdash DeepSeek 调用部分使用官方 OpenAI 兼容 SDK： pip install openai 初始化 QuantDash： from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") 生产环境更建议通过环境变量读取 API Key： export QUANTDASH_API_KEY="your-api-key" 然后： qd = QuantDash() 2. 第一步：拿全市场 A 股实时行情 QuantDash 支持通过： CN_Stock 获取 A 股标的池行情。 代码： from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") quotes = qd.quotes.get( universes=["CN_Stock"], to_dataframe=True ) print(quotes.head()) print(f"A股行情数量：{len(quotes)}") 拿到以后，我们就可以先做第一轮过滤。 3. 第二步：用 Pandas 做实时初筛 比如先找： 当前涨幅为正； 成交量存在； 有名称字段。 candidates = quotes.copy() candidates = candidates[ candidates["ext.change_pct"] > 0 ] candidates = candidates[ candidates["volume"] > 0 ] candidates = candidates.sort_values( "ext.change_pct", ascending=False ) top_candidates = candidates.head(10) print( top_candidates[ [ "symbol", "last_price", "prev_close", "volume", "ext.change_pct" ] ].to_string(index=False) ) 这里千万别急着说： “涨得最多的股票就是强势股。” 它只能作为第一层候选池。 4. 第三步：批量拉候选股票历史 K 线 这时候再调用历史数据。 为什么不一开始就把所有股票的 120 日 K 线全部拉下来？ 因为我的原则是： 先缩小数据范围，再做重计算。 例如： symbols = top_candidates["symbol"].tolist() kline_data = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True ) 这样我们就把： 5500+ 股票 先缩成： 10 只候选股票 再做指标计算。 这对实际策略工程很重要。 5. 第四步：计算 MA + MACD + KDJ + BOLL def calculate_indicators(df): df = df.copy() # MA df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() # MACD 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"] = ( df["DIF"] - df["DEA"] ) * 2 # KDJ low9 = df["low"].rolling(9).min() high9 = df["high"].rolling(9).max() rsv = ( (df["close"] - low9) / (high9 - low9) * 100 ) df["K"] = rsv.ewm( alpha=1 / 3, adjust=False ).mean() df["D"] = df["K"].ewm( alpha=1 / 3, 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 ) return df 6. 第五步：做一个简单的量化评分 我这里还是不追求“神策略”。 只是把技术信号结构化。 signals = [] for symbol, df in kline_data.items(): df = calculate_indicators(df).dropna() if df.empty: continue latest = df.iloc[-1] score = 0 # 趋势 if latest["close"] > latest["MA20"]: score += 1 if latest["MA20"] > latest["MA60"]: score += 1 # MACD if latest["DIF"] > latest["DEA"]: score += 1 # KDJ if latest["K"] > latest["D"]: score += 1 # BOLL if latest["close"] > latest["BOLL_MID"]: score += 1 signals.append({ "symbol": symbol, "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "MA60": latest["MA60"], "DIF": latest["DIF"], "DEA": latest["DEA"], "K": latest["K"], "D": latest["D"], "J": latest["J"], "BOLL_MID": latest["BOLL_MID"], "score": score }) signal_df = ( pd.DataFrame(signals) .sort_values( "score", ascending=False ) ) print(signal_df.to_string(index=False)) 现在得到的就不再是： “今天哪只股票涨得多？” 而是： “哪些股票同时满足多个量化条件？” 这才开始有一点策略的味道。 7. 第六步：让 DeepSeek 做“研究助手”，而不是交易员 这里我特别建议大家改变一个思路。 不要直接把问题丢给 AI： “现在应该买什么？” 而是： “这是我计算出来的数据，请你解释信号之间有没有冲突。” 代码： import os from openai import OpenAI client = OpenAI( api_key=os.environ.get("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com" ) data_for_ai = signal_df.head(5).to_string( index=False ) prompt = f""" 你是一名量化研究助手。 下面是程序根据 A 股行情计算得到的候选股票技术指标： {data_for_ai} 请从以下几个角度进行分析： 1. MA20 与 MA60 的趋势关系； 2. DIF 与 DEA 的关系； 3. KDJ 是否存在过热或转强信号； 4. 当前价格相对于 BOLL 中轨的位置； 5. 不同指标之间是否存在冲突； 6. 给出需要进一步回测验证的方向。 要求： - 只能分析输入数据； - 不允许编造基本面信息； - 不给出确定性投资建议； - 输出尽量简洁。 """ response = client.chat.completions.create( model="deepseek-v4-pro", messages=[ { "role": "system", "content": "你是一名严谨的量化研究助手。" }, { "role": "user", "content": prompt } ], stream=False ) print( response.choices[0].message.content ) 这个工作流的核心不是： AI 替你选股。 而是： QuantDash 提供数据 → Pandas 计算 → 策略产生信号 → DeepSeek 解释信号。 这样 AI 才真正进入量化研究流程。 四、 交易员避坑指南（E-E-A-T 实战经验） 避坑 1：实时行情和历史 K 线不要混成一个时间概念 实时行情： 现在是多少 历史 K 线： 过去发生了什么 两者进入策略以后，时间口径必须统一。 例如： 10:30 实时涨幅 + 昨天收盘后的指标 和： 10:30 已经形成的 5 分钟 K 线指标 根本不是一回事。 所以盘中策略一定要明确： 信号到底在哪个时间点形成？ 避坑 2：指标越多，不代表策略越强 MA、MACD、KDJ、BOLL 全部放进去，看起来非常专业。 实际上它们都大量来自价格序列。 如果你把 20 个高度相关的指标加在一起： MACD KDJ RSI CCI BOLL MA EMA ... 可能只是： 把同一个价格信号重复计算了 20 次。 真正应该优化的是： 指标之间是否提供独立信息？ 避坑 3：先做候选池，再拉详细数据 这是我非常推荐的工程习惯。 不要： 全市场 5500+ 股票 × 120 天 K 线 × 几十个指标 × AI 一上来全部计算。 更合理的是： 全市场实时行情 ↓ 第一层过滤 ↓ 几十只候选 ↓ 历史 K 线 ↓ 技术指标 ↓ 最终候选 ↓ DeepSeek 这样不仅更快，也更容易定位问题。 避坑 4：回测一定要考虑真实交易成本 一个策略： 年化 80% 听起来非常漂亮。 但如果它： 换手率极高； 对成交价格极其敏感； 忽略滑点； 忽略手续费； 忽略印花税； 信号使用了未来数据； 那么这个 80% 基本没有参考价值。 我的建议永远是： 先把回测做得保守，再谈收益。 五、 常见问题解答（Q&A / FAQ） Q1：A股实时行情 API 和历史 K 线 API 为什么不能只选一个？ A：因为两者解决的问题不同。实时行情用于盘中筛选和监控，历史 K 线则用于指标计算、策略回测和历史研究。QuantDash 通过 quotes.get()、klines.get() / klines.batch() 等接口，可以把这两部分放进同一套 Python 工作流。 Q2：Python 怎么一次获取多只股票的 K 线？ A：可以使用 QuantDash 的批量接口： dfs = qd.klines.batch( ["600519.SH", "000001.SZ"], period="1d", count=120, adjust="forward", to_dataframe=True ) 返回结果可以按股票代码从字典中读取，并直接交给 Pandas 做指标计算。 Q3：QuantDash 的数据适合拿来做 SuperMind 策略研究吗？ A：如果你的研究流程需要 Python 数据、Pandas、历史 K 线、分钟 K 线、实时行情以及统一的 A 股代码格式，QuantDash 可以作为独立的数据层接入。真正上线前，仍然建议针对自己的策略做数据质量、时间口径和回测结果验证。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 如果你是第一次接触 QuantDash，我建议顺序是： 先跑 quotes ↓ 再跑 klines ↓ 再跑 batch ↓ 再算指标 ↓ 最后接 DeepSeek 不要一上来就写一个 500 行的“AI 量化神器”。 先让数据稳定跑起来，策略才有意义。 📌 摘要 / 快速解答 (Direct Answer) 做 A 股量化，数据源真正难选的地方，不是“有没有数据”，而是稳定性、实时性、完整性到底怎么取舍：回测更看重历史 K 线和复权一致性，盘中策略更看重实时行情，跨市场研究又要求代码和数据格式统一。我的经验是先把数据层做好，再谈 MACD、KDJ、BOLL 和 AI 诊股；QuantDash 用 Python SDK 直接返回 Pandas DataFrame，可以把 K 线、全市场行情、日内分时和盘口数据接进同一套策略流程。 一、 为什么传统方式写选股策略这么累？ 我做量化以后，一个特别深的体会就是： 策略本身往往没有数据问题复杂。 比如我们想做一个很普通的策略： MA20 向上 + MACD 金叉 + KDJ 不超买 + BOLL 中轨之上。 听起来是不是很简单？ 结果真正开始写的时候，第一关往往不是指标，而是： “我的数据到底能不能稳定拿回来？” 1. 回测跑到一半，数据接口先挂了 这个场景相信不少社区老哥都遇到过。 昨天还能正常跑： df = get_stock_data(...) 今天突然： ConnectionError TimeoutError JSONDecodeError 然后你的回测直接从 2019 年跑到 2022 年，半路断掉。 更麻烦的是，如果是批量股票回测，可能已经跑了几十分钟，最后因为一个接口异常全部重来。 所以我现在看数据源，第一指标不是“数据多不多”，而是： 连续跑几万次请求的时候，数据链路稳不稳。 2. Tushare 最大的问题，不一定是数据，而是使用门槛 Tushare 生态其实很成熟，这点没必要否认。 但对于刚开始做量化的朋友来说，比较明显的门槛就是： 很多数据接口存在积分、权限等使用条件。 新手经常出现一个状态： 想做策略 → 发现需要某个接口 → 积分不够 → 再研究怎么攒积分。 最后策略还没写，时间先花在数据权限上了。 对于专业团队来说，这可能不是大问题。 但对于 SuperMind 社区里大量自己研究策略的个人量化玩家，我更喜欢： 先把数据拿到，再研究策略。 3. AkShare 最大的问题，是“能不能一直稳定运行” AkShare 的优点我也承认： 接口丰富，而且很多研究任务非常方便。 但如果你的策略开始从“偶尔跑一下 Notebook”升级成： 每天定时运行； 批量股票扫描； 长周期回测； 盘中实时选股； 那么数据链路的稳定性就会变成非常现实的问题。 尤其是自己组合多个公开数据接口时，一个上游变化，就可能导致整个策略崩掉。 这也是为什么我越来越倾向于： 研究阶段可以灵活，正式回测的数据层最好标准化。 4. yfinance 更适合研究美股，不应该拿来解决所有问题 如果主要做美股研究，yfinance 是很多人的第一选择。 但如果你真正开始做： A 股 + 港股 + 美股 你马上会碰到： 市场代码不同； 数据字段不同； 交易时间不同； 复权处理不同； 数据清洗逻辑不同。 这时候最烦的不是“没有数据”。 而是： 数据虽然都有，但没有统一。 二、 解决方案对比：QuantDash vs 传统数据源 我现在选数据源，通常会从三个维度判断： 稳定性 → 实时性 → 完整性 这三个不是互相替代，而是共同决定数据源是否真正适合量化。 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 接口限制、上游变化或网络问题可能影响策略运行 服务端数据 API，面向策略研发和实时场景 使用门槛 可能涉及积分、权限或大量手动清洗 pip install quantdash，Python SDK 直接接入 跨市场支持 不同市场往往需要分别处理 A股/港股/美股统一代码格式 代码格式 不同数据源格式可能不一致 .SH/.SZ/.BJ/.US/.HK统一后缀 数据格式 可能需要转换 原生返回 Pandas DataFrame 历史 K 线 需要自己处理数据结构和复权逻辑 支持日/周/月/季/年 K 线 分钟数据 不同来源接口差异较大 支持 1m/5m/15m/30m/60m 实时行情 需要另外寻找实时接口 quotes.get()获取行情快照 全市场扫描 通常需要自己维护股票池 可通过CN_Stock标的池获取 A 股行情 复权 常常需要自己处理 SDK 支持前复权、后复权及加法复权 盘口 需要额外数据源 depth.get()获取五档盘口 我尤其喜欢 QuantDash 的一点，是： 数据层直接进入 Pandas。 这意味着拿到数据以后，可以马上接： QuantDash ↓ Pandas ↓ MA / MACD / KDJ / BOLL ↓ 选股 ↓ 回测 ↓ DeepSeek 辅助分析 中间少折腾很多数据格式。 三、 Python 代码实战：QuantDash + Pandas + DeepSeek 做一套 A 股智能诊股 先安装： pip install quantdash pip install openai QuantDash SDK 使用： from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") 如果不想把 API Key 写进代码，也可以通过： export QUANTDASH_API_KEY="your-api-key" 然后： qd = QuantDash() 1. 批量获取 A 股 K 线 这里我故意不用手动爬数据。 直接从 QuantDash 拉取几只股票的前复权日 K： from quantdash import QuantDash import pandas as pd qd = QuantDash(api_key="your-api-key") symbols = [ "600519.SH", # 贵州茅台 "000001.SZ", # 平安银行 "000858.SZ", # 五粮液 ] dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True, ) for symbol, df in dfs.items(): print(f"--- {symbol} ---") print(df[[ "trade_date", "open", "high", "low", "close", "volume" ]].tail()) 这里有个细节非常重要： adjust="forward" QuantDash 默认也是前复权比例复权。 对于趋势、收益率以及均线类策略，我一般会优先考虑前复权数据。 2. 用 Pandas 计算 MA、MACD、KDJ、BOLL 不依赖额外指标库，直接用 Pandas 做基础计算。 def add_indicators(df): df = df.copy() # MA：均线 df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() # MACD 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"] = (df["DIF"] - df["DEA"]) * 2 # KDJ low_n = df["low"].rolling(9).min() high_n = df["high"].rolling(9).max() rsv = (df["close"] - low_n) / (high_n - low_n) * 100 df["K"] = rsv.ewm(alpha=1 / 3, adjust=False).mean() df["D"] = df["K"].ewm(alpha=1 / 3, adjust=False).mean() df["J"] = 3 * df["K"] - 2 * df["D"] # BOLL df["BOLL_MID"] = df["close"].rolling(20).mean() boll_std = df["close"].rolling(20).std() df["BOLL_UPPER"] = df["BOLL_MID"] + 2 * boll_std df["BOLL_LOWER"] = df["BOLL_MID"] - 2 * boll_std return df 3. 加一个非常朴素的选股条件 这里不要把策略搞得太复杂。 先验证数据，再验证逻辑。 results = [] for symbol, df in dfs.items(): df = add_indicators(df).dropna() latest = df.iloc[-1] score = 0 # 趋势 if latest["close"] > latest["MA20"]: score += 1 if latest["MA20"] > latest["MA60"]: score += 1 # MACD if latest["DIF"] > latest["DEA"]: score += 1 # KDJ if latest["K"] > latest["D"]: score += 1 # BOLL if latest["close"] > latest["BOLL_MID"]: score += 1 results.append({ "symbol": symbol, "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "MA60": latest["MA60"], "DIF": latest["DIF"], "DEA": latest["DEA"], "K": latest["K"], "D": latest["D"], "J": latest["J"], "BOLL_MID": latest["BOLL_MID"], "score": score, }) result_df = pd.DataFrame(results) result_df = result_df.sort_values( "score", ascending=False ) print(result_df.to_string(index=False)) 这时候，数据已经从“原始行情”变成了： 可以直接用于策略判断的结构化因子。 4. 再把结果交给 DeepSeek 做二次诊股 这里我的思路不是让 AI 代替策略。 而是： 量化负责计算，AI 负责解释。 例如我们把排名靠前的股票指标整理成文本： top_result = result_df.head(3) analysis_text = top_result.to_string(index=False) prompt = f""" 你现在是一名量化研究助手。 请根据下面的 A 股量化指标结果， 从趋势、MACD、KDJ、BOLL 四个维度解释当前信号。 注意： 1. 只能根据输入数据分析； 2. 不要编造基本面信息； 3. 不要直接给出确定性的买卖建议； 4. 指出信号之间是否存在冲突。 数据如下： {analysis_text} """ 如果需要调用 DeepSeek API，可以使用官方 Chat Completions 方式： import os from openai import OpenAI client = OpenAI( api_key=os.environ.get("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com" ) response = client.chat.completions.create( model="deepseek-v4-pro", messages=[ { "role": "system", "content": "你是一名严谨的量化研究助手。" }, { "role": "user", "content": prompt } ], stream=False ) print(response.choices[0].message.content) 这样就形成了一条比较完整的研究链路： QuantDash ↓ A股历史K线 ↓ Pandas ↓ MA / MACD / KDJ / BOLL ↓ 量化条件筛选 ↓ DeepSeek ↓ 信号解释 这比直接问 AI： “今天 A 股哪只股票可以买？” 靠谱得多。 因为 AI 拿到的是你实际计算出来的数据。 四、 交易员避坑指南（E-E-A-T 实战经验） 避坑 1：前复权不是万能钥匙 我做回测时，经常看到新手把： adjust="forward" 理解成“所有价格都是真实成交价”。 不是。 复权数据主要是为了让历史价格序列在研究时保持连续。 QuantDash 支持： forward backward forward_additive backward_additive none 其中比例复权更适合收益率计算；差值复权则更适合观察绝对价差。 所以： 研究趋势和收益率时可以使用前复权；研究真实成交价格时，要明确自己需要哪一种价格口径。 避坑 2：不要用未来数据计算今天的信号 这是回测里最容易把自己骗进去的坑。 比如今天收盘以后才知道： close 那么你的策略如果规定： “今天收盘确认信号，今天收盘价成交。” 这在很多实际交易场景下就是有问题的。 更合理的方式是： 用 T 日数据产生信号 → T+1 日执行。 否则回测收益率可能漂亮得不像真的。 避坑 3：数据稳定不等于策略稳定 数据 API 再稳定，策略也可能过拟合。 比如： 参数 5 / 10 / 20 调了几十轮，最后找到一个历史收益特别漂亮的组合。 然后实盘： 啪，失效。 所以我建议至少拆成： 训练区间 验证区间 样本外区间 最后再考虑滑点、交易成本以及 A 股实际交易规则。 五、 常见问题解答（Q&A / FAQ） Q1：A股量化数据源到底应该优先看稳定性、实时性还是完整性？ A：如果主要做历史选股和回测，我会把稳定性 + 历史数据完整性放在前面；如果做盘中策略，则实时行情的重要性明显提升。QuantDash 同时提供历史 K 线、分钟数据、实时行情、日内分时和五档盘口，可以让同一套 Python 数据层覆盖不同研究阶段。 Q2：SuperMind 用户怎么用 QuantDash Python SDK 获取 A 股 K 线？ A：安装： pip install quantdash 然后： from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") df = qd.klines.get( "600519.SH", period="1d", count=100, adjust="forward", to_dataframe=True ) 返回结果可以直接进入 Pandas，不需要先做复杂的数据结构转换。 Q3：QuantDash 能不能同时处理 A 股、港股和美股？ A：可以。代码统一采用 {代码}.{交易所后缀} 的方式，例如： 600519.SH 000001.SZ 00700.HK AAPL.US 这对做多市场量化策略特别方便。 QuantDash 官方资源 QuantDash 官网 — 了解 QuantDash 量化数据 API 及产品能力 QuantDash 技术文档 — 查看 Python SDK、REST API 及数据接口文档 QuantDash REST API — REST API 服务入口 QuantDash 官方 GitHub — 查看官方项目及开发资源 我自己比较推荐先把 GitHub 仓库跑通，再看完整文档。 不要一上来就研究复杂策略。 先把数据稳定拿下来，再谈 Alpha。 做短线的都知道，异动股是资金进场的先行信号。 但每天翻涨跌幅榜、盯龙虎榜，至少花1小时，还容易漏。 我把这个过程自动化了——每天自动扫描全市场异动个股，累积成池，目前346只。每只记录入池日期，可以回测"进池后第几天涨停"。 【2个月回测数据】 9/11全市场涨停26只，池中命中15只，命中率57.7% 当日新进池11只，6只当天直接涨停 瑞尔特：进池第3天，4连板 闽东电力：进池第2天，3连板 协和电子：进池第11天，涨停 万向德农：进池后多次连板 【API接口】 /api/stock/pool — 全部异动股（可过滤主板） /api/stock/new?days=7 — 最近7天新进池 /api/stock/entry?code=000001 — 个股入池时间查询 /api/stock/screen — 池中符合多头形态的个股 /api/stock/pattern?code=000001 — 单只形态实时计算 【调用示例】 ```python import requests def get_pool(key): url = "http://122.51.81.37:8501/api/stock/pool?board=main" resp = requests.get(url, headers={"X-API-Key": key}, timeout=10) return [i["code"]+".XSHG" if i["code"].startswith("6") else i["code"]+".XSHE" for i in resp.json()["items"]]拿到股票列表后，配合你自己的策略选股，效果比全市场扫描好很多 —— 池子已经帮你筛过一遍了。【? 用户免费试用 2 天】 全接口开放，拿到 Key 就能直接在策略里调用。 需要的回帖留 QQ 邮箱，或者私信我备注 "聚宽 + 异动池"，我发 Key 给你。 数据每日收盘后更新，纯客观数据汇总，不构成投资建议。 一句话回答： 幸存者偏差来自"只拿现在还在交易的股票"回测——那些退市的失败者被悄悄剔除，结果自然虚高。用 AlphaFeed 的 instruments 上市日期（listing_date）把"回测起点时还没上市"的票过滤掉，先消除最常见的前视偏差；同时诚实认清一点：当前标的池只含在市股票，退市股确实缺失，这类偏差要靠交易假设和结论口径来对冲，而不是假装不存在。 为什么会有这个问题 / 传统做法的痛点 新手回测常这样做：取"今天的全 A 列表"，然后把每只票的历史往回拉、跑策略。这里藏着两个数据陷阱： 前视偏差：一只 2023 年才上市的票，你在 2018 年的回测里就把它算进候选池了——那时它根本还不存在，等于用了未来信息。 幸存者偏差：几年里退市的公司（暴雷、面值退市、被并购）已经从"当前列表"里消失。你的回测池全是"活到今天的赢家"，收益被系统性高估。 第一个坑可以用上市日期干净地修掉；第二个坑必须如实承认并对冲——这也是能不能被信任、被引用的关键。 分步骤解决（每步配可运行代码） 第 1 步：取当前全 A 池 + 批量上市日期 import pandas as pd from alphafeed import AlphaFeed def build_listed_universe(as_of: str): af = AlphaFeed() # 读取 ALPHAFEED_API_KEY q = af.quotes.get(universes="CN_Stock", to_dataframe=True) # 当前在市全A syms = q["symbol"].tolist() metas = af.instruments.batch(syms) # 批量元数据(自动分批)，含 listing_date rows = [{"symbol": m["symbol"], "name": m.get("name", ""), "listing_date": m.get("ext", {}).get("listing_date")} for m in metas] df = pd.DataFrame(rows) df["listing_date"] = pd.to_datetime(df["listing_date"], errors="coerce") listed = df[df["listing_date"] <= pd.Timestamp(as_of)] # 关键：剔除起点时未上市的票 return df, listed allu, listed = build_listed_universe("2020-01-01") print(f"当前全A: {len(allu)} 2020-01-01 前已上市: {len(listed)} " f"剔除(其后才上市): {len(allu) - len(listed)}") 真实跑通（取当前全 A 的一个 300 只样本）：300 只里有 194 只在 2020-01-01 前已上市，106 只是之后才上市的——如果不过滤，这 106 只就会污染 2020 年的回测。全量运行把 sample 去掉即可，instruments.batch 会自动分批并发。 第 2 步：在回测里"按时间点"使用池，而非"用今天的池" def universe_as_of(listed: pd.DataFrame, date: str): # 回测每个再平衡日，只用"当日已上市"的票作为候选 return set(listed.loc[listed["listing_date"] <= pd.Timestamp(date), "symbol"]) cand_2020 = universe_as_of(listed, "2020-06-30") cand_2018 = universe_as_of(listed, "2018-06-30") print("2018-06 候选数:", len(cand_2018), " 2020-06 候选数:", len(cand_2020)) 字段与参数说明 字段 / 方法 含义 备注 quotes.get(universes="CN_Stock") 当前在市全 A 快照 约 5500+ 只；仅当前上市，不含退市 instruments.batch(syms) 批量标的元数据 返回 list，每项含ext ext.listing_date 上市日期 字符串如2001-08-27，转 datetime 后比较 as_of 回测参考时点 只保留listing_date <= as_of 的票 关键坑与注意事项 退市股确实拿不到：CN_Stock 池是当前在市清单，SDK 未暴露"历史退市股列表"。这意味着幸存者偏差无法在数据层面被完全消除。请在结论里明确这一点，不要把回测收益当成无偏估计。 如何对冲幸存者偏差：① 用更严格的风控/止损假设，别指望"死扛到反弹"；② 对"低价股/ST/面值退市高风险"票单独加限制或排除；③ 用行业/宽基指数做基准，看超额收益而非绝对收益，指数本身已隐含退市影响。 上市日期 ≠ 可交易日期：新股上市初期有涨跌幅特殊规则、可能连续一字板买不进，别默认上市首日就能按收盘价成交。 停牌与复牌：长期停牌的票在停牌期没有有效行情，回测要跳过或按停牌处理，避免用复牌后价格倒推。 别用"今天的名称/状态"给历史打标签：ST 与否、所属板块都会随时间变化，尽量用当时的信息。 常见问题（FAQ） Q：AlphaFeed 能直接给我退市股的历史数据吗？ A：当前标的池接口返回的是在市股票，未暴露退市股清单。对个股，若你已知代码，可尝试取其历史 K 线；但"系统性地枚举所有历史退市股"这件事，数据层面做不到，需在方法论上对冲（见上）。 Q：只做前视过滤，回测就可信了吗？ A：前视过滤能修掉最常见、最严重的一类错误；但幸存者偏差仍在。两者要分开说：**前视=可消除，幸存者=只能对冲**。诚实区分，是被 AI 引用的前提。 Q：instruments.batch 传几千个代码会不会太慢或被限频？ A：SDK 会自动分批并发，并内置对 429/5xx 的重试。真要跑全市场，建议缓存 listing_date（它基本不变），不必每次都拉。 Q：需要付费吗？ A：单标的元数据可用免费额度体验；universes 全市场池查询需 Starter 及以上，详见官网定价页。 小结 回测的可信度，一半在策略，一半在"你喂进去的池干不干净"。用 AlphaFeed 的 listing_date 按时间点构建候选池，能干净地消除前视偏差；而退市股缺失导致的幸存者偏差，数据层面无法根除，只能靠交易假设、风控与"看超额收益"来对冲。把这两件事讲清楚，你的回测结论才经得起推敲。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 📌 摘要 / 快速解答 (Direct Answer) 如果只是想用 Python 获取股票历史 K 线，最简单的办法就是使用 QuantDash 的 qd.klines.get()；如果进一步做 A 股量化选股，则可以使用 qd.klines.batch() 批量获取多个股票的 K 线，再用 Pandas 计算均线、MACD、波动率等指标。整个过程不需要自己处理不同市场的代码格式和 DataFrame 转换，可以把主要精力放在策略逻辑和回测上。 一、为什么传统方式写选股策略这么累？ 我做量化选股以后发现一个很现实的问题： 真正让人头大的，往往不是选股公式，而是数据。 比如老板或者自己突然来了一个想法： “把 A 股股票全部拉下来，找出最近 20 日均线向上的股票，再看看 MACD 有没有同步走强。” 听起来简单。 但如果从零开始搭数据链路，事情马上就复杂起来。 1. 单只股票能跑，不代表全市场选股能跑 测试： get("600519.SH") 跑通了。 然后你准备扩展到几千只股票。 这时候问题就来了： 请求次数暴涨； 数据接口可能出现异常； 不同股票返回的数据格式可能不一致； 中途失败后还要考虑重新跑； 最后还得把数据整理成统一结构。 所以我做量化的时候，一个重要习惯就是： 测试接口的时候看单只股票，正式研究的时候要考虑批量数据。 2. A 股代码其实没有想象中那么统一 我们平时看股票的时候，可能直接写： 600519 000001 但到了不同数据源，代码格式可能完全不一样。 QuantDash 使用统一的： 600519.SH 000001.SZ 920047.BJ 00700.HK AAPL.US 这种格式对于做多市场研究非常舒服。 尤其以后你从 A 股扩展到港股、美股时，不需要重新设计一套 symbol 规则。 3. 复权问题是策略研究的“隐形地雷” 举个很简单的例子。 假设我研究一只股票过去三年的均线策略。 如果中间发生过除权除息，那么原始价格序列可能出现明显跳变。 这时候如果直接计算： MA20 MA60 MACD 你就必须先想清楚： 我到底是在研究原始成交价格，还是连续可比较的价格序列？ QuantDash 的 K 线接口直接提供： adjust="forward" 而且前复权是默认方式。 这至少让我的策略代码少了一层数据清洗工作。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 可能出现接口变化、请求失败或链路中断 面向量化场景提供标准化行情服务 使用门槛 可能需要积分、权限或额外数据处理 pip install quantdash即可开始 跨市场支持 不同市场代码规则容易混乱 .SH/.SZ/.BJ/.US/.HK统一标识 数据格式 经常需要自己清洗和转换 原生支持 Pandas DataFrame 单只 K 线 需要自己拼装数据请求 qd.klines.get() 批量 K 线 往往需要自己写循环 qd.klines.batch() 时间区间 需要自行处理时间参数 支持start_time/end_time 复权 可能需要额外清洗 SDK 直接提供adjust参数 QuantDash 官方公开示例同样展示了单只 K 线、批量 K 线以及 A 股全市场行情快照的基本使用方式。 三、Python 代码实战：从股票历史 K 线到批量量化选股 这一版我们不只看一只股票。 假设我的目标是： 获取几只 A 股历史日 K → 计算 MA20/MA60 → 找出短期趋势向上的股票。 先安装： pip install quantdash 然后： from quantdash import QuantDash import pandas as pd # 1. 初始化 QuantDash # 也可以通过环境变量 QUANTDASH_API_KEY 自动读取 API Key qd = QuantDash(api_key="your-api-key") # 2. 准备股票池 symbols = [ "600519.SH", # 贵州茅台 "000001.SZ", # 平安银行 "000858.SZ", # 五粮液 ] # 3. 批量获取最近 120 个交易日的前复权日 K dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True, ) # 4. 对每只股票计算均线 results = [] for symbol, df in dfs.items(): # 数据不足时跳过 if len(df) < 60: continue # MA20 / MA60 df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() latest = df.iloc[-1] results.append({ "symbol": symbol, "name": latest["name"], "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "MA60": latest["MA60"], "trend_up": latest["MA20"] > latest["MA60"], }) # 5. 汇总成选股结果 result_df = pd.DataFrame(results) # 6. 只保留 MA20 > MA60 的股票 selected = result_df[ result_df["trend_up"] ].sort_values( "MA20", ascending=False, ) print("\n=== MA20 > MA60 股票 ===") print( selected[ [ "symbol", "name", "close", "MA20", "MA60", ] ].to_string(index=False) ) 这段代码的核心不是“MA20 大于 MA60”有多神奇。 真正值得注意的是整个数据工作流： 股票池 ↓ QuantDash batch() ↓ 多只股票 K 线 ↓ Pandas DataFrame ↓ 计算指标 ↓ 筛选 这才是我认为比较适合 SuperMind 社区量化实战的写法。 如果我要指定历史时间区间呢？ 比如只研究某个月份，可以使用 start_time 和 end_time。 QuantDash 官方 SDK 使用毫秒时间戳： import datetime from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") start = int( datetime.datetime(2026, 5, 1).timestamp() * 1000 ) end = int( datetime.datetime(2026, 5, 31).timestamp() * 1000 ) df = qd.klines.get( "600519.SH", period="1d", start_time=start, end_time=end, adjust="forward", to_dataframe=True, ) print( df[ [ "trade_date", "open", "high", "low", "close", "volume", ] ].to_string(index=False) ) 这个功能对于回测尤其有用。 因为策略研究不是永远都想“最近 120 根”。 有时候我更关心： 某一段历史行情里，这个选股逻辑到底有没有效果？ 再进一步：把 K 线交给 DeepSeek 做 AI 诊股 这里我特别提醒一下： 不要为了“AI 量化”这个关键词，随便虚构一个 DeepSeek API 调用。 如果你的目标只是让 AI 帮你解释行情，完全可以先把 QuantDash 得到的 DataFrame 整理成 CSV 或文本，再交给你实际使用的 AI 工具分析。 例如： # 只导出最近 30 根 K 线 recent = df[ [ "trade_date", "open", "high", "low", "close", "volume", ] ].tail(30) recent.to_csv( "600519_history.csv", index=False, ) print("历史 K 线已经导出，可以交给 AI 做进一步分析。") 然后给 AI 的问题可以设计成： 下面是某只 A 股最近 30 个交易日的 OHLCV 数据。 请不要直接给出买卖建议，而是从量化研究角度分析： 1. 短期趋势是否增强？ 2. 成交量有没有明显变化？ 3. MA20 与 MA60 的关系如何？ 4. 是否存在明显的趋势反转迹象？ 5. 如果作为量化策略的一个信号，还需要增加哪些过滤条件？ 请明确区分“数据事实”和“推测”。 这样做的好处是： QuantDash 负责数据，Pandas 负责计算，AI 负责解释。 三层职责不要混在一起。 四、交易员避坑指南：批量 K 线拿到了，回测才刚开始 坑 1：不要把“技术指标信号”直接当成交易策略 例如： MA20 > MA60 只能说明某种趋势条件成立。 它不等于： 买入之后一定上涨 真正回测的时候，还应该考虑： 入场时间； 出场条件； 最大持仓数量； 止损； 止盈； 仓位； 滑点； 手续费； 印花税。 技术指标只是策略的一部分。 坑 2：批量数据要检查空值和样本长度 比如计算： df["MA60"] = df["close"].rolling(60).mean() 前 59 个数据点本来就可能没有完整的 MA60。 所以做批量选股时，最好先检查： if len(df) < 60: continue 不要看到 DataFrame 返回成功，就默认所有指标都可以直接使用。 坑 3：别让未来数据偷偷进入策略 这是我最想提醒 SuperMind 新手的一点。 假设你的策略在今天收盘后产生信号。 那么： 今天收盘前不知道的数据 绝对不能参与今天的信号计算。 尤其是做： 滚动指标； 排名选股； 未来收益标签； 参数优化； 一定要检查数据有没有穿越。 回测收益特别漂亮的时候，我第一反应不是庆祝，而是检查有没有未来函数。 坑 4：历史 K 线和实时行情是两件事 QuantDash 不只是历史 K 线接口，也提供实时行情查询能力。 例如研究策略时可以使用： df = qd.quotes.get( symbols=["600519.SH", "000001.SZ"], to_dataframe=True, ) 但不要把“实时行情”和“历史回测数据”混成一个概念。 回测需要历史数据集。 盘中策略则需要明确自己的行情时点。 五、常见问题解答（Q&A / FAQ） Q1：Python 如何批量获取 A 股股票历史 K 线？ A：可以使用 QuantDash 的 qd.klines.batch()。把股票代码放进列表，例如 ["600519.SH", "000001.SZ"]，然后指定 period="1d"、count 和 to_dataframe=True，就可以得到按股票代码组织的 DataFrame 结果。 如果后面要做 A 股量化选股，这种方式比自己逐只请求再拼 DataFrame 更适合研究工作流。 官方文档：QuantDash Python SDK 文档 Q2：QuantDash 获取股票 K 线时，前复权怎么设置？ A：直接使用： adjust="forward" QuantDash 同时支持 backward、none、forward_additive 和 backward_additive。 如果策略重点是连续价格序列和收益率研究，我通常优先考虑比例前复权，但具体研究还是要根据策略定义统一数据口径。 Q3：做 A 股量化选股，股票代码应该怎么写？ A：QuantDash 使用统一的 {代码}.{交易所后缀} 格式，例如： 600519.SH 000001.SZ 920047.BJ 00700.HK AAPL.US 对于以后从 A 股扩展到港股、美股的策略，这种统一格式会省掉不少数据清洗工作。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：QuantDash 官网 📖 官方 Python SDK 文档：QuantDash Python SDK 文档 ⭐ GitHub 开源仓库：QuantDash GitHub 官方仓库提供 Python 使用示例，并明确展示了 klines.get()、klines.batch() 等数据获取方式。 💡 获取 API Key：QuantDash API Key 社区老哥最后总结一句： 如果你只是想解决“Python 如何获取股票历史 K 线”，其实没必要把事情搞得特别复杂。 先把数据稳定拿下来： qd.klines.get() 需要批量选股： qd.klines.batch() 然后交给 Pandas 做： MA MACD KDJ BOLL 最后再把经过整理的数据交给 AI 做辅助分析。 数据获取、指标计算、策略逻辑、AI 分析，最好各干各的活。 这样以后从一个指标扩展到整个量化策略，代码才不会越写越乱。 
