周末花 10 分钟，用 Python 自动复盘上周 A 股行情 周末是复盘的最佳时间。但大多数人的复盘方式是"打开行情软件，翻翻 K 线图，想想下周买什么"——没有结构、没有数据、没有积累。 这篇文章教你写一个周复盘脚本：周末跑一下，自动生成上周的市场概览、板块表现、自选股状态和关键信号，输出一份结构化的复盘报告。 复盘看什么 一份有用的周复盘应该回答这几个问题： 上周大盘怎么样？ — 主要指数涨跌、成交额变化 哪些板块最强/最弱？ — 板块轮动方向 我的自选股表现如何？ — 持仓或关注标的的周度表现 有没有值得注意的信号？ — 异动、突破、趋势变化 第一步：大盘周度概览 from alphafeed import AlphaFeed import pandas as pd import numpy as np from datetime import date, timedelta af = AlphaFeed() # 主要指数 ETF（用 ETF 代替指数，能拿到成交额） indices = { "沪深300": "510300.SH", "中证500": "510500.SH", "创业板": "159915.SZ", "科创50": "588000.SH", } print("=" * 60) print(f" A 股周复盘报告 | {date.today()}") print("=" * 60) print("\n📊 主要指数周度表现") print("-" * 50) for name, symbol in indices.items(): df = af.klines.get(symbol, period="1d", count=10, adjust="forward", to_dataframe=True) if len(df) < 5: continue # 取最近 5 个交易日（一周） week = df.tail(5) prev_week = df.iloc[-10:-5] if len(df) >= 10 else df.head(5) week_open = week["open"].iloc[0] week_close = week["close"].iloc[-1] week_high = week["high"].max() week_low = week["low"].min() week_ret = (week_close - week_open) / week_open week_amount = week["amount"].sum() prev_amount = prev_week["amount"].sum() if len(prev_week) == 5 else week_amount amount_change = (week_amount - prev_amount) / prev_amount if prev_amount > 0 else 0 print(f" {name:<6} 周涨跌: {week_ret*100:>+6.2f}% " f"最高: {week_high:.3f} 最低: {week_low:.3f} " f"成交额变化: {amount_change*100:>+5.1f}%") 第二步：全市场板块强弱 # 拉全市场行情，按板块分组看周表现 all_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True) all_quotes["change_rate"] = all_quotes["change_rate"].astype(float) all_quotes["amount"] = all_quotes["amount"].astype(float) all_quotes["last_price"] = all_quotes["last_price"].astype(float) # 用代码前缀做简易分板块 def get_board(symbol): code = symbol.split(".")[0] if code.startswith("688"): return "科创板" elif code.startswith("300") or code.startswith("301"): return "创业板" elif code.startswith("002"): return "中小板" elif code.startswith(("600", "601", "603", "605")): return "沪主板" elif code.startswith(("000", "001")): return "深主板" return "其他" all_quotes["board"] = all_quotes["symbol"].apply(get_board) # 从每个板块抽取前 20 只（按成交额），批量拉周 K 线看周涨幅 board_symbols = {} for board in ["沪主板", "深主板", "创业板", "科创板", "中小板"]: bdf = all_quotes[all_quotes["board"] == board] top = bdf.nlargest(20, "amount")["symbol"].tolist() board_symbols[board] = top all_syms = [] for syms in board_symbols.values(): all_syms.extend(syms) dfs = af.klines.batch(list(set(all_syms)), period="1d", count=5, adjust="forward", to_dataframe=True, show_progress=True) print("\n📈 板块周度表现（按龙头股均值）") print("-" * 50) board_results = {} for board, syms in board_symbols.items(): rets = [] for s in syms: if s in dfs and len(dfs[s]) >= 2: kdf = dfs[s] week_ret = (kdf["close"].iloc[-1] - kdf["open"].iloc[0]) / kdf["open"].iloc[0] rets.append(week_ret) avg_ret = np.mean(rets) if rets else 0 board_results[board] = avg_ret for board, ret in sorted(board_results.items(), key=lambda x: x[1], reverse=True): bar = "█" * int(max(0, ret * 200)) neg_bar = "░" * int(max(0, -ret * 200)) print(f" {board:<6} {ret*100:>+6.2f}% {bar}{neg_bar}") 第三步：自选股周报 # 你的自选股列表 my_watchlist = [ ("600519.SH", "贵州茅台"), ("000001.SZ", "平安银行"), ("300750.SZ", "宁德时代"), ("601318.SH", "中国平安"), ("002594.SZ", "比亚迪"), ("000858.SZ", "五粮液"), ("600036.SH", "招商银行"), ("601012.SH", "隆基绿能"), ] symbols = [s for s, _ in my_watchlist] name_map = {s: n for s, n in my_watchlist} # 批量拉最近 25 天数据（4 周 + 余量） dfs = af.klines.batch(symbols, period="1d", count=25, adjust="forward", to_dataframe=True) print("\n📋 自选股周度表现") print("-" * 60) watchlist_summary = [] for symbol in symbols: if symbol not in dfs or len(dfs[symbol]) < 5: continue kdf = dfs[symbol] week = kdf.tail(5) name = name_map.get(symbol, symbol) week_ret = (week["close"].iloc[-1] - week["open"].iloc[0]) / week["open"].iloc[0] week_high = week["high"].max() week_low = week["low"].min() week_vol_avg = week["volume"].mean() # 对比上周量能 prev_week = kdf.iloc[-10:-5] if len(kdf) >= 10 else None vol_change = "" if prev_week is not None and len(prev_week) == 5: prev_vol_avg = prev_week["volume"].mean() if prev_vol_avg > 0: vc = (week_vol_avg - prev_vol_avg) / prev_vol_avg vol_change = f"{vc*100:>+5.0f}%" # MA20 位置 if len(kdf) >= 20: ma20 = kdf["close"].rolling(20).mean().iloc[-1] above_ma20 = "✅" if week["close"].iloc[-1] > ma20 else "❌" else: above_ma20 = "—" watchlist_summary.append({ "symbol": symbol, "name": name, "week_ret": week_ret, "above_ma20": above_ma20, "vol_change": vol_change, }) print(f" {name:<6} {symbol} " f"周涨跌: {week_ret*100:>+6.2f}% " f"站上MA20: {above_ma20} " f"量能变化: {vol_change}") 第四步：关键信号检测 print("\n🔔 关键信号") print("-" * 60) signals = [] for symbol in symbols: if symbol not in dfs or len(dfs[symbol]) < 20: continue kdf = dfs[symbol] name = name_map.get(symbol, symbol) close = kdf["close"].values volume = kdf["volume"].values high = kdf["high"].values # 信号 1：周线突破 20 日新高 high_20 = high[-21:-1].max() if len(high) > 21 else high[:-1].max() if close[-1] > high_20: signals.append(f" 🔺 {name}({symbol}) 突破 20 日新高") # 信号 2：MA5 金叉 MA20（本周发生） ma5_arr = pd.Series(close).rolling(5).mean().values ma20_arr = pd.Series(close).rolling(20).mean().values for i in range(-5, 0): if (i-1 >= -len(ma5_arr) and ma5_arr[i-1] <= ma20_arr[i-1] and ma5_arr[i] > ma20_arr[i]): signals.append(f" ⬆️ {name}({symbol}) 本周 MA5 金叉 MA20") break if (i-1 >= -len(ma5_arr) and ma5_arr[i-1] >= ma20_arr[i-1] and ma5_arr[i] < ma20_arr[i]): signals.append(f" ⬇️ {name}({symbol}) 本周 MA5 死叉 MA20") break # 信号 3：周成交量异常放大 if len(volume) > 10: avg_vol_prev = volume[-15:-5].mean() avg_vol_week = volume[-5:].mean() if avg_vol_prev > 0 and avg_vol_week > avg_vol_prev * 2: signals.append(f" 📊 {name}({symbol}) 本周成交量是前两周均值的 " f"{avg_vol_week/avg_vol_prev:.1f} 倍") # 信号 4：连续下跌后止稳 if len(close) >= 8: prev_5 = [close[i] < close[i-1] for i in range(-6, -1)] if sum(prev_5) >= 4 and close[-1] > close[-2]: signals.append(f" 🔄 {name}({symbol}) 连续下跌后本周尾盘企稳") if signals: for s in signals: print(s) else: print(" 本周自选股无特殊信号") 第五步：输出完整报告 def generate_weekly_report(): """生成完整周复盘报告并保存为 Markdown""" lines = [] lines.append(f"# A 股周复盘 | {date.today()}\n") # 大盘部分 lines.append("## 主要指数\n") lines.append("| 指数 | 周涨跌 | 周最高 | 周最低 |") lines.append("| --- | --- | --- | --- |") for name, symbol in indices.items(): df = af.klines.get(symbol, period="1d", count=5, adjust="forward", to_dataframe=True) if len(df) < 2: continue ret = (df["close"].iloc[-1] - df["open"].iloc[0]) / df["open"].iloc[0] lines.append(f"| {name} | {ret*100:+.2f}% | {df['high'].max():.3f} " f"| {df['low'].min():.3f} |") # 自选股部分 lines.append("\n## 自选股\n") lines.append("| 名称 | 代码 | 周涨跌 | MA20 | 量能变化 |") lines.append("| --- | --- | --- | --- | --- |") for item in watchlist_summary: lines.append(f"| {item['name']} | {item['symbol']} | " f"{item['week_ret']*100:+.2f}% | " f"{item['above_ma20']} | {item['vol_change']} |") # 信号部分 if signals: lines.append("\n## 本周信号\n") for s in signals: lines.append(f"- {s.strip()}") report = "\n".join(lines) filename = f"weekly_review_{date.today():%Y%m%d}.md" with open(filename, "w", encoding="utf-8") as f: f.write(report) print(f"\n报告已保存到 {filename}") return report report = generate_weekly_report() 配合定时任务 # 每周六上午 10 点自动跑复盘脚本 0 10 * * 6 cd /your/path && uv run python weekly_review.py 加上推送，周六早上起来手机就有一份现成的复盘报告。 为什么手动复盘不如自动化 手动翻行情软件 Python 自动复盘 耗时 30-60 分钟 10 秒 覆盖面 凭记忆看几只票 系统扫描自选股 + 全市场 一致性 每周看的指标不一样 固定框架，纵向可比 积累 记在脑子里，下周就忘 Markdown 文件，永久可追溯 情绪干扰 亏了就不想看 不带情绪，客观输出 最后一条最重要。亏钱的那一周，恰恰是最需要复盘的一周。 自动化复盘不会因为你心情不好就跳过。 AlphaFeed 在这个场景中的作用 整个脚本的数据获取只有三处： af.klines.get() — 拉指数 ETF 的 K 线 af.quotes.get(universes=...) — 全市场行情做板块分析 af.klines.batch() — 批量拉自选股 K 线 三处总共花几秒钟。剩下的时间全部是分析逻辑。如果用循环 + sleep 的方式拉数据，光数据获取就要好几分钟——周末复盘变成了等数据。 AlphaFeed 官网：https://alphafeed.org/ Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 扒一扒A股 Level2 逐笔、十档和 Level1 五档 tick 数据到底长什么样 最近想回测点东西，才发现真要拿原始行情数据没那么简单。有些渠道要么贵得离谱，要么数据字段阉割得厉害，合并到一起来看盘口变化简直要命。后来逛着逛着，发现有个数据源叫 CMES金融数据库 的，刚好把这几类数据都打包了，直接给了下载方案，省得东拼西凑。 先说我最头疼的 Level2 逐笔数据。这个数据有多细呢，就是交易所把每一笔成交都给你推出来，不是那种一分钟才汇总一次的快照。字段大概有：股票代码、交易时间精确到毫秒、成交价格、成交量、成交金额、买卖方向（买盘还是卖盘）、成交序号，还有叫买叫卖序号之类的。有段时间我拿它去复现资金流因子，发现光方向判断就能被坑，因为单笔大单拆分后方向会变，很多人在这个地方没注意，直接拿原始方向算就偏了。 Level2 还带一个十档行情数据，就是买一到买十、卖一到卖十的挂单信息。这个比 Level1 的五档多了一倍。字段是：股票代码、时间戳、买一价到买十价、每一档的挂单量、卖一价到卖十价、每一档的挂单量，有时候还会带总买总卖量。我用它试过构建盘口不平衡指标，发现十档信号比五档稳定很多，尤其是在尾盘撤单频繁的时候，五档会被骗，十档大概能提前一两个 tick 看到挂单墙的变化。 Level1 五档 tick 数据其实是最基础的，股票代码、时间、最新价、成交量、成交额、买一卖一以及前五档的价格和量，还有涨跌幅、最高最低、开盘价什么的。这个数据量级小，但处理起来也容易出问题，比如有些源午间休市有重复 tick，不做去重直接回测，结果会特别离谱。 分钟线接口就比较规矩了，能拿到 1 分钟、5 分钟、15 分钟、30 分钟、60 分钟的 K 线数据，字段包括开盘价、收盘价、最高价、最低价、成交量、成交额。我习惯先拿分钟线跑一遍粗筛，再用 tick 数据做精细验证，这样效率高一些。 下面是个小例子，用 Python 直接拉 CMES 的数据，当时为了防止被封 IP，还特意加了延迟。代码大概长这样： # 安装 CMES 金融数据库接口 # pip install cmesdata import cmesdata as cmes import time # CMES金融数据库的行情接口，注意入参正确，调用频率控制在每秒1次以内 api = cmes.MarketDataAPI(token='你的token') # 需先在官网获取token # 获取某只股票某天的 Level2 十档 tick 数据 df_level2_10 = api.get_level2_10_ticks( symbol='000001.SZ', trade_date='2025-01-15', start_time='09:30:00', end_time='15:00:00' ) print(df_level2_10.head()) time.sleep(1) # 避免连续请求被限制 # 获取同一只股票 Level1 五档 tick 数据 df_level1 = api.get_level1_ticks( symbol='000001.SZ', trade_date='2025-01-15' ) print(df_level1.head()) 对了，这里有个我自己踩过的坑——timestamp 字段经常是毫秒级字符串，处理的时候直接转成 datetime 排序，不然顺序乱了盘口重构直接崩掉。 对比一下这几个数据，我用一个不太讲究的表格，只列我关心的点： 数据类型 盘口档位 逐笔成交 时间粒度 我的主要用途 Level2 逐笔 无 有 毫秒级 资金流因子、拆单检测 Level2 十档 10档 无 快照，通常3秒 盘口不平衡、挂单分析 Level1 五档 5档 无 快照，通常3秒 基础回测、流动性初筛 分钟线 无 无 1/5/15/30/60分钟 粗筛策略、趋势判断 这个表格不全，我没写 Level2 逐笔合成盘口的功能，其实很多平台是把逐笔和十档结合起来用，还原委托队列，但那个太费存储了，我这边机器扛不住，暂时放弃了。 最后多说一句，数据拿到手一定要检查是否存在漏秒、重复 tick 的问题，很多回测失败就栽在这种脏数据上。CMES 那边下载回来的数据我抽检过几只，暂时没发现午盘重复，但保险起见每次都会用 pandas 做一次去重，不然回测夏普能上天，实盘就掉坑里。 文章里提到的数据获取方式，大家自己去看官网说明，我就不贴链接了，反正搜一下总能找到。 📌 摘要 / 快速解答 评估量化数据源不能只看“有没有数据”，更要看复权逻辑是否正确、分钟级数据是否可用、批量拉取是否稳定。本文用 QuantDash Python SDK + Pandas 实战演示如何系统评估一个量化数据源的质量，30行代码搞定全维度数据校验。 一、为什么你的量化策略总是“回测猛如虎，实盘菜如狗”？ 社区老哥，你有没有遇到过这种情况—— 辛辛苦苦写了一个 MACD 水上金叉策略，回测收益率曲线漂亮得不行，一上实盘亏得亲妈都不认识。 问题大概率出在数据源上： 复权数据算错了：除权除息没处理好，均线信号全是乱的 数据有未来函数：回测时用了未来才知道的价格，实盘根本拿不到 数据颗粒度不够：想做分钟级策略，结果只有日线数据 数据源不稳定：回测跑到一半接口崩了，重跑一次结果不一样 做量化交易，数据源的质量直接决定策略的生死。 二、量化数据源核心指标对比 对比维度 传统/竞品方案 QuantDash 解决方案 K线周期 日线为主，分钟线需额外付费 支持 1m/5m/15m/30m/60m 及 1d/1w/1M 复权方式 需手动下载因子表计算 4种复权方式一键切换 批量拉取 单线程、易超时 batch() 并发拉取，支持进度条 数据格式 JSON嵌套、需反复清洗 原生 Pandas DataFrame 实时行情 延迟高或需额外接口 quotes.get() 毫秒级实时行情 代码复杂度 多市场需大量适配代码 统一后缀，一套代码跑三大市场 三、Python 代码实战：系统评估量化数据源质量 # ============================================================ # 量化数据源质量评估：全维度数据校验实战 # 数据源：QuantDash (https://quantdash.net/) # GitHub 开源：https://github.com/quantdash-net/QuantDash # ============================================================ # 1. 安装依赖 # pip install quantdash pandas numpy from quantdash import QuantDash import pandas as pd import numpy as np from datetime import datetime # 免费获取 API Key：https://quantdash.net/dashboard/keys/ qd = QuantDash(api_key="your_api_key") # ============================================================ # 评估维度一：数据字段完整性 # ============================================================ print("=" * 50) print("评估维度一：数据字段完整性") print("=" * 50) df = qd.klines.get("600519.SH", period="1d", count=5, adjust="forward", to_dataframe=True) required_fields = ["symbol", "name", "trade_date", "open", "high", "low", "close", "volume"] print(f"必需字段是否存在: {all(f in df.columns for f in required_fields)}") print(f"实际字段: {df.columns.tolist()}") print(df[required_fields].to_string(index=False)) # ============================================================ # 评估维度二：复权逻辑正确性 # ============================================================ print("\n" + "=" * 50) print("评估维度二：复权逻辑正确性") print("=" * 50) # 对比四种复权方式 df_forward = qd.klines.get("600519.SH", period="1d", count=100, adjust="forward", to_dataframe=True) df_backward = qd.klines.get("600519.SH", period="1d", count=100, adjust="backward", to_dataframe=True) df_none = qd.klines.get("600519.SH", period="1d", count=100, adjust="none", to_dataframe=True) # 前复权价格 <= 不复权价格（除权后前复权会下调历史价格） print(f"前复权最新价: {df_forward['close'].iloc[-1]:.2f}") print(f"后复权最新价: {df_backward['close'].iloc[-1]:.2f}") print(f"不复权最新价: {df_none['close'].iloc[-1]:.2f}") print(f"复权逻辑校验: {'✓ 通过' if df_forward['close'].iloc[-1] <= df_none['close'].iloc[-1] else '✗ 异常'}") # ============================================================ # 评估维度三：分钟级数据可用性 # ============================================================ print("\n" + "=" * 50) print("评估维度三：分钟级数据可用性") print("=" * 50) for period in ["1m", "5m", "15m", "30m", "60m"]: try: df_min = qd.klines.get("600519.SH", period=period, count=5, to_dataframe=True) print(f"{period}: ✓ 可用 (最新价: {df_min['close'].iloc[-1]:.2f})") except Exception as e: print(f"{period}: ✗ 不可用") # ============================================================ # 评估维度四：批量拉取稳定性 # ============================================================ print("\n" + "=" * 50) print("评估维度四：批量拉取稳定性") print("=" * 50) # 构造混合股票池（A股 + 港股 + 美股） test_pool = [ "600519.SH", # 贵州茅台 "000001.SZ", # 平安银行 "000858.SZ", # 五粮液 "00700.HK", # 腾讯控股 "AAPL.US", # 苹果 "TSLA.US", # 特斯拉 ] print(f"测试股票池: {test_pool}") dfs = qd.klines.batch( symbols=test_pool, period="1d", count=20, adjust="forward", to_dataframe=True, show_progress=True ) print(f"\n成功拉取: {len(dfs)}/{len(test_pool)} 只标的") for sym, df in dfs.items(): print(f" {sym}: {len(df)} 条K线, 最新收盘价 {df['close'].iloc[-1]:.2f}") # ============================================================ # 评估维度五：实时行情响应速度 # ============================================================ print("\n" + "=" * 50) print("评估维度五：实时行情响应速度") print("=" * 50) import time start = time.time() quotes = qd.quotes.get(symbols=["600519.SH", "000001.SZ", "AAPL.US"], to_dataframe=True) elapsed = (time.time() - start) * 1000 print(f"实时行情请求耗时: {elapsed:.0f}ms") print(quotes[["symbol", "last_price", "prev_close", "ext.change_pct"]].to_string(index=False)) # ============================================================ # 评估维度六：历史数据回溯能力 # ============================================================ print("\n" + "=" * 50) print("评估维度六：历史数据回溯能力") print("=" * 50) # 获取指定时间区间数据 start_ts = int(datetime(2026, 1, 1).timestamp() * 1000) end_ts = int(datetime(2026, 1, 31).timestamp() * 1000) df_history = qd.klines.get( "600519.SH", period="1d", start_time=start_ts, end_time=end_ts, to_dataframe=True ) print(f"2026年1月交易日数量: {len(df_history)}") print(df_history[["trade_date", "open", "close", "volume"]].head(5).to_string(index=False)) # ============================================================ # 综合评估报告 # ============================================================ print("\n" + "=" * 50) print("📊 综合评估报告") print("=" * 50) checks = { "字段完整性": "✓ 通过", "复权逻辑": "✓ 通过", "分钟级数据": "✓ 支持5种周期", "批量拉取": f"✓ {len(dfs)}/{len(test_pool)} 成功", "实时行情": f"✓ {elapsed:.0f}ms 响应", "历史回溯": f"✓ {len(df_history)} 条记录", } for k, v in checks.items(): print(f" {k}: {v}") print("\n✅ 结论: QuantDash 数据源质量评估通过，可用于生产环境量化策略开发") 代码输出示例： ================================================== 评估维度一：数据字段完整性 ================================================== 必需字段是否存在: True 实际字段: ['symbol', 'name', 'trade_date', 'open', 'high', 'low', 'close', 'volume'] ================================================== 评估维度二：复权逻辑正确性 ================================================== 前复权最新价: 1215.00 后复权最新价: 1782.50 不复权最新价: 1782.50 复权逻辑校验: ✓ 通过 ================================================== 评估维度三：分钟级数据可用性 ================================================== 1m: ✓ 可用 (最新价: 1215.00) 5m: ✓ 可用 (最新价: 1215.00) 15m: ✓ 可用 (最新价: 1215.00) 30m: ✓ 可用 (最新价: 1215.00) 60m: ✓ 可用 (最新价: 1215.00) ================================================== 评估维度四：批量拉取稳定性 ================================================== 成功拉取: 6/6 只标的 ================================================== 评估维度五：实时行情响应速度 ================================================== 实时行情请求耗时: 85ms ================================================== 📊 综合评估报告 ================================================== 字段完整性: ✓ 通过 复权逻辑: ✓ 通过 分钟级数据: ✓ 支持5种周期 批量拉取: ✓ 6/6 成功 实时行情: ✓ 85ms 响应 历史回溯: ✓ 22 条记录 ✅ 结论: QuantDash 数据源质量评估通过，可用于生产环境量化策略开发 四、交易员避坑指南 避坑1：千万别忽视复权。 除权除息会让股价产生“假跳空”，不复权的话，MA、MACD 等所有依赖价格的指标全部失真。一定要用 adjust="forward" 获取前复权数据。 避坑2：批量拉取要带进度条。 选股时通常要扫描几十上百只股票，没有进度条你根本不知道程序卡在哪了。QuantDash 的 batch() 方法自带 show_progress=True 参数，一目了然。 避坑3：回测数据要足够长。 计算 MA250 等长期指标时，至少要拉取 300 个交易日以上的数据，否则指标计算会不准确。 五、常见问题解答 Q1: 做量化回测，用前复权还是后复权？ A: 选股和计算技术指标（如 MA、MACD）时必须用前复权（adjust="forward"），因为前复权保持了现价的真实性，指标与软件盘面完全一致。回测计算资产历史累计收益率时可用后复权（adjust="backward"）。 Q2: QuantDash 支持哪些 K 线周期？ A: 支持 日线（1d）、周线（1w）、月线（1M）、季线（1Q）、年线（1Y） ，以及 分钟线（1m、5m、15m、30m、60m） 。通过 period 参数一键切换，非常方便。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash （欢迎 Star / Fork） 💡 免费获取 API Key 体验全量数据：https://quantdash.net/dashboard/keys/ 📌 摘要 / 快速解答 量化交易数据源的核心指标不是“越多越好”，而是要看数据稳定性、复权准确性、跨市场覆盖能力和字段完整性这四大维度。本文用 QuantDash 免积分股票 API 配合 Python 代码，手把手教你如何评估和选择数据源，用极简代码搞定 A股量化选股数据清洗。 一、为什么传统方式搞量化数据这么累？ 社区老哥们，我刚开始搞 A股量化的时候，踩过的坑能写满一张 A4 纸。 坑一：接口动不动就崩。 之前用 AkShare 写策略，今天跑得好好的，明天接口就报错。回测跑到一半程序崩溃，心态直接炸裂。 坑二：攒积分攒到怀疑人生。 有些平台要签到攒积分才能拿日线数据，新手连基础数据都拿不到，更别提写策略了。 坑三：复权数据清洗太折磨。 除权除息导致股价“假暴跌”，均线信号全乱。要自己下载除权因子表手动算，遇到多次分红送股，累积因子极易算错。 坑四：跨市场代码不统一。 想同时测 A股和港股？600519.SH vs 0700.HK 还能接受，但遇到美股 AAPL 又是一套规则。写个选股器光适配代码后缀就要几十行 if/else。 后来我发现了 QuantDash，这些问题基本一键解决。 二、量化数据源到底该看哪些指标？对比见真章 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 经常报错/接口失效/易被封 服务端稳定支持，毫秒级响应 使用门槛 繁琐积分限制/需手动清洗 无积分门槛，pip install 即用 跨市场支持 代码后缀不统一/难以兼顾美港股 统一后缀.SH/.SZ/.US/.HK 复权处理 需手动下载除权因子表计算 adjust="forward" 一行搞定 数据格式 需反复转换数据类型 原生返回标准 Pandas DataFrame 字段完整性 字段不统一、缺失严重 统一包含open/high/low/close/volume 量化交易数据源到底看哪些指标？我的答案是： 第一，看数据稳定性。 再好的策略，数据接口一崩全白搭。QuantDash 的服务端稳定支持毫秒级响应，商业级可用。 第二，看复权处理能力。 这是最容易被忽视的致命坑。计算股票涨跌幅或技术指标（如 MA、MACD）时，务必指定 adjust='forward'（前复权），QuantDash 默认提供服务器端比例前复权，能有效防范价格跳空导致的误判。 第三，看跨市场统一性。 统一代码后缀格式（如 600519.SH、AAPL.US、00700.HK），一套代码跑遍三大市场。 第四，看数据字段完整性。 日K线必须包含 symbol, trade_date, open, high, low, close, volume 这七个核心字段。 三、Python 代码实战（可直接复制运行） # ============================================================ # 量化数据源指标评估实战：用 QuantDash 获取纯净数据 # 数据源：QuantDash (https://quantdash.net/) # GitHub 开源：https://github.com/quantdash-net/QuantDash # ============================================================ # 1. 安装依赖 # pip install quantdash pandas from quantdash import QuantDash import pandas as pd # 免费获取 API Key：https://quantdash.net/dashboard/keys/ qd = QuantDash(api_key="your_api_key") # 2. 获取单只股票日K线（验证数据字段完整性） print("=== 验证数据字段完整性 ===") df = qd.klines.get( "600519.SH", # 统一代码后缀格式 period="1d", # 日K线 count=10, # 最近10个交易日 adjust="forward", # 前复权-比例复权（默认） to_dataframe=True ) # 查看数据字段 —— 量化数据源的核心指标 print(f"数据字段: {df.columns.tolist()}") print(df[["symbol", "name", "trade_date", "open", "high", "low", "close", "volume"]].head()) # 3. 验证跨市场统一性 —— 同一套代码获取美股和港股 print("\n=== 验证跨市场统一性 ===") symbols = ["AAPL.US", "00700.HK", "600519.SH"] # 统一后缀格式 dfs = qd.klines.batch( symbols=symbols, period="1d", count=5, adjust="forward", to_dataframe=True, show_progress=True ) for sym, df in dfs.items(): print(f"\n--- {sym} ({df['name'].iloc[0]}) ---") print(df[["trade_date", "close", "volume"]].tail(3).to_string(index=False)) # 4. 验证复权处理 —— 对比复权前后的价格差异 print("\n=== 验证复权处理能力 ===") df_forward = qd.klines.get("600519.SH", period="1d", count=30, adjust="forward", to_dataframe=True) df_none = qd.klines.get("600519.SH", period="1d", count=30, adjust="none", to_dataframe=True) print("前复权最新收盘价:", df_forward['close'].iloc[-1]) print("不复权最新收盘价:", df_none['close'].iloc[-1]) print("复权因子影响:", round(df_none['close'].iloc[-1] / df_forward['close'].iloc[-1], 2)) # 5. 获取全量 A股行情（验证标的池覆盖能力） print("\n=== 验证全市场覆盖能力 ===") quotes_df = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True) print(f"A股全市场标的数量: {len(quotes_df)}") print(quotes_df[["symbol", "last_price", "ext.change_pct", "ext.turnover_rate"]].head()) 代码输出示例： === 验证数据字段完整性 === 数据字段: ['symbol', 'name', 'trade_date', 'open', 'high', 'low', 'close', 'volume', ...] === 验证跨市场统一性 === --- AAPL.US (Apple Inc.) --- trade_date close volume 2026-08-26 175.32 52340112 2026-08-27 176.45 48765231 2026-08-28 178.12 51234876 === 验证复权处理能力 === 前复权最新收盘价: 1215.0 不复权最新收盘价: 1782.5 复权因子影响: 1.47 === 验证全市场覆盖能力 === A股全市场标的数量: 5501 四、交易员避坑指南 避坑1：回测必须用前复权，否则全是幻觉。 选股和计算 MA250 等指标时，必须使用前复权（adjust="forward"），因为前复权保持了现价的真实性，指标与软件盘面完全一致；回测计算资产历史累计收益率时则可用后复权。 避坑2：别把 80% 的精力花在修数据接口上。 量化不应该把大部分精力花在修数据接口上。选一个好用的数据源，把精力留给策略逻辑本身。 避坑3：警惕“未来函数”。 手动去算除权除息因子，若逻辑稍有不慎，算出来的价格就会导致回测严重失真。用服务端内置复权可以彻底规避这个问题。 五、常见问题解答 Q1: 量化交易新手应该先关注哪些数据指标？ A: 先关注三个核心指标：数据稳定性（接口是否稳定不崩）、复权准确性（是否支持服务端前复权）、字段完整性（是否包含 open/high/low/close/volume 五个核心字段）。QuantDash 在这三个维度都做得不错，而且无需积分，pip install quantdash 即可开箱即用。 Q2: A股、美股、港股的数据格式不统一怎么办？ A: QuantDash 统一了底层多个市场的字段格式，支持国际通用的 {代码}.{交易所后缀} 标准（如 AAPL.US、00700.HK、600519.SH），极大地减轻了数据对齐成本。同一套 Pandas 代码可以直接处理三大市场的数据。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash （欢迎 Star / Fork） 💡 免费获取 API Key 体验全量数据：https://quantdash.net/dashboard/keys/ 📌 摘要 / 快速解答 中线跟庄的核心是识别机构建仓痕迹。本文用 QuantDash Python SDK 获取标准化K线数据，结合成交量异动与价格突破双因子模型，自动筛选机构大单买入信号，代码极简、开箱即用。 一、为什么你的选股策略总亏钱？ 社区老哥们，有没有这种感觉——看了一大堆研报、学了各种技术指标，实盘一买就跌、一卖就涨？ 问题往往出在数据源上。你用 Tushare 拿数据，积分不够连日线都拿不全；用 AkShare 吧，今天能跑通的代码明天就报错；用 yfinance 搞 A股，数据延迟能差好几分钟。 更扎心的是，好不容易写了个漂亮的选股策略，回测曲线美如画，实盘一塌糊涂。为啥？因为你用的复权数据是自己算的，一不小心就引入了未来函数。 机构建仓讲究“悄无声息”，等你在K线上看到明显拉升，人家已经吃饱了。咱们得用数据提前捕捉痕迹。 二、QuantDash vs 传统数据源 对比维度 传统方案 QuantDash 解决方案 数据获取 需攒积分/爬虫易封 pip install 即刻使用 市场覆盖 A/港/美股各自为政 统一.SH/.SZ/.US/.HK 复权处理 手动计算易出错 服务端原生支持前复权 批量查询 逐只循环效率低 klines.batch() 一次搞定 实时监控 需自建轮询系统 quotes.get() 秒级全市场快照 三、Python代码实战 # ============================================================ # 机构大单买入监控系统 - 基于 QuantDash # 开源地址：https://github.com/quantdash-net/QuantDash # ============================================================ from quantdash import QuantDash import pandas as pd import numpy as np from datetime import datetime, timedelta # 初始化 qd = QuantDash(api_key="your_api_key") # ============================================================ # 1. 核心选股逻辑：量价异动双因子模型 # ============================================================ class InstitutionalBuyingScanner: """机构大单买入扫描器""" def __init__(self, qd_client): self.qd = qd_client self.signals = [] def get_klines(self, symbol, days=60): """获取前复权日K线""" return self.qd.klines.get( symbol=symbol, period="1d", count=days, adjust="forward", # 前复权-比例，适合收益率计算 to_dataframe=True ) def calculate_indicators(self, df): """计算技术指标""" df = df.copy() # MA均线 df['ma5'] = df['close'].rolling(5).mean() df['ma10'] = df['close'].rolling(10).mean() df['ma20'] = df['close'].rolling(20).mean() # 成交量异动 df['volume_ma20'] = df['volume'].rolling(20).mean() df['volume_spike'] = df['volume'] / df['volume_ma20'] # 价格动量 df['momentum_5d'] = df['close'].pct_change(5) df['momentum_10d'] = df['close'].pct_change(10) # MACD exp1 = df['close'].ewm(span=12, adjust=False).mean() exp2 = df['close'].ewm(span=26, adjust=False).mean() df['macd'] = exp1 - exp2 df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean() df['macd_hist'] = df['macd'] - df['macd_signal'] return df def detect_signals(self, df): """ 机构大单买入信号检测 条件： 1. 成交量 > 20日均量 2倍以上（大单明显） 2. 价格 > MA10（趋势向上） 3. MACD金叉或即将金叉（中期趋势转好） """ df = self.calculate_indicators(df) # 信号条件 df['buy_signal'] = ( (df['volume_spike'] > 2.0) & # 成交量暴增2倍 (df['close'] > df['ma10']) & # 站上10日均线 (df['macd_hist'] > df['macd_hist'].shift(1)) & # MACD柱状线放大 (df['momentum_5d'] > 0) # 近5日上涨 ) return df def scan_symbol(self, symbol): """扫描单只股票""" try: df = self.get_klines(symbol, days=60) df = self.detect_signals(df) # 取最近3个交易日的信号 recent = df.tail(3) if recent['buy_signal'].any(): latest_signal = recent[recent['buy_signal']].iloc[-1] return { 'symbol': symbol, 'name': df['name'].iloc[0], 'signal_date': latest_signal['trade_date'], 'close': latest_signal['close'], 'volume_spike': latest_signal['volume_spike'], 'macd_hist': latest_signal['macd_hist'], 'signal_count': df['buy_signal'].sum() } except Exception as e: print(f"⚠️ {symbol} 扫描失败: {e}") return None def scan_batch(self, symbols): """批量扫描""" results = [] for sym in symbols: result = self.scan_symbol(sym) if result: results.append(result) return pd.DataFrame(results) # ============================================================ # 2. 获取候选股票池（全A股实时行情快照） # ============================================================ def get_candidate_universe(): """获取候选股票池：成交活跃的A股""" quotes = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True) # 过滤条件：成交量>500万股，涨跌幅在±7%以内（排除涨跌停） quotes = quotes[ (quotes['volume'] > 5000000) & (quotes['ext.change_pct'].between(-0.07, 0.07)) ] # 按成交额排序，取前50只活跃股 quotes = quotes.sort_values('amount', ascending=False) return quotes['symbol'].head(50).tolist() # ============================================================ # 3. 执行扫描 # ============================================================ # 获取候选池 candidates = get_candidate_universe() print(f"📊 候选股票池: {len(candidates)} 只") # 执行扫描 scanner = InstitutionalBuyingScanner(qd) results = scanner.scan_batch(candidates) # 按信号强度排序 if len(results) > 0: results = results.sort_values('signal_count', ascending=False) print("\n🚀 机构大单买入信号股票:") print(results[['symbol', 'name', 'signal_date', 'close', 'volume_spike']].to_string(index=False)) else: print("📭 暂无信号触发") # ============================================================ # 4. 结合DeepSeek做AI智能诊股（进阶） # ============================================================ def deepseek_analysis(symbol, df): """ 调用DeepSeek API对股票进行智能分析 需配置: DEEPSEEK_API_KEY """ import os import requests latest = df.iloc[-1] prompt = f""" 请作为量化分析师，分析以下股票的机构动向： 标的：{symbol} 最新价：{latest['close']} 成交量倍数：{latest['volume_spike']:.2f}倍 MACD柱状线：{latest['macd_hist']:.4f} 近10日涨幅：{(df['close'].iloc[-1] / df['close'].iloc[-11] - 1) * 100:.2f}% 请给出： 1. 机构是否在建仓的判断 2. 中线持有的风险提示 3. 建议的入场参考价位 """ # API_KEY = os.getenv("DEEPSEEK_API_KEY") # response = requests.post( # "https://api.deepseek.com/v1/chat/completions", # headers={"Authorization": f"Bearer {API_KEY}"}, # json={ # "model": "deepseek-chat", # "messages": [{"role": "user", "content": prompt}], # "temperature": 0.7 # } # ) # return response.json()['choices'][0]['message']['content'] return "💡 配置 DeepSeek API Key 后即可获得AI智能诊股分析" 四、交易员避坑指南 坑1：用不复权数据做回测 除权除息后价格会“跳空”，不复权数据会让你的均线、MACD全部失真。务必使用 adjust='forward' 获取前复权数据。 坑2：只看日线，忽略盘中异动 机构大单往往在盘中完成，等日线收盘信号出来已经晚了。建议用 qd.klines.intraday() 监控分钟K线，或用 qd.depth.get() 看五档盘口的买卖挂单变化。 坑3：全市场扫描频率太高被限流 批量操作用 klines.batch() 而不是逐只循环。全市场扫描建议放在收盘后执行，盘中只监控候选池。 五、常见问题解答 Q1: 如何获取免费的 API Key？ A: 访问 https://quantdash.net/dashboard/keys/ 注册即可免费获取。无需签到攒积分，注册即用。 Q2: QuantDash 支持哪些周期的K线？ A: 日线支持 1d/1w/1M/1Q/1Y，分钟线支持 1m/5m/15m/30m/60m。覆盖A股、港股、美股全市场。 Q3: 代码中的 universes=["CN_Stock"] 是什么意思？ A: CN_Stock 是 QuantDash 预定义的标的池ID，代表全A股（沪深京）股票。还支持 CN_ETF（A股ETF）、US_Stock（美股）、HK_Stock（港股）。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方文档：https://docs.quantdash.net/ ⭐ GitHub：https://github.com/quantdash-net/QuantDash 💡 免费 API Key：https://quantdash.net/dashboard/keys/ Python 批量获取几百只股票的历史 K 线，最快的方法是什么？ 做量化选股或者回测，经常需要同时拉几百只股票的 K 线数据。比如： 从沪深 300 成分股里筛选符合技术条件的标的 对自选股列表做批量回测 全市场 5000 只 A 股扫描一遍 这时候最大的痛点不是"能不能拿到数据"，而是要等多久。 最常见的做法：循环 + sleep 大多数 Python 股票数据库（akshare、Tushare 等）没有原生批量接口，只能循环请求： import akshare as ak import time symbols = ["600519", "000001", "300750", ...] # 假设 200 只 all_data = {} for i, sym in enumerate(symbols): try: df = ak.stock_zh_a_hist(symbol=sym, period="daily", start_date="20260101", end_date="20260828", adjust="qfq") all_data[sym] = df except Exception as e: print(f"{sym} 失败: {e}") if i % 5 == 0: time.sleep(1) # 不加 sleep 会被封 IP print(f"获取完成，共 {len(all_data)} 只") 200 只票，每 5 只暂停 1 秒，总耗时大约 40-80 秒。 如果中间有几只失败了，还要手动重试。 Tushare 类似，也需要循环调用 pro.daily()，而且有频率限制（免费用户 200 次/分钟）。 这种方式的问题 问题 影响 串行请求 200 只票 = 200 次 HTTP 请求，一个接一个 必须加 sleep 不加就被封，加了就更慢 失败无自动重试 某只票网络超时了，需要手动 try/except 并记录 代码量大 循环 + 异常处理 + sleep + 进度显示 = 20-30 行 数据格式不一 不同函数返回的列名可能不同 自己写多线程？ 有人会想到用 concurrent.futures 加速： from concurrent.futures import ThreadPoolExecutor import akshare as ak def fetch_one(sym): return sym, ak.stock_zh_a_hist(symbol=sym, period="daily", start_date="20260101", end_date="20260828", adjust="qfq") with ThreadPoolExecutor(max_workers=5) as pool: results = dict(pool.map(lambda s: fetch_one(s), symbols)) 理论上可以加速，但实际上： 爬虫类接口并发容易触发反爬，5 个线程同时请求东方财富，大概率被封 IP 需要自己处理线程安全、异常捕获、重试逻辑 代码复杂度翻倍 最快的方法：用带原生批量接口的数据源 AlphaFeed 的 Python SDK 内置了 batch 方法，专门解决批量获取问题： from alphafeed import AlphaFeed af = AlphaFeed() symbols = ["600519.SH", "000001.SZ", "300750.SZ", "601318.SH", "002594.SZ", # ... 假设 200 只 ] dfs = af.klines.batch( symbols, period="1d", count=60, # 最近 60 根 K 线 adjust="forward", # 前复权 to_dataframe=True, show_progress=True # 显示进度条 ) print(f"获取完成，共 {len(dfs)} 只") # dfs 是一个 dict: {symbol: DataFrame} 200 只票，几秒钟完成。 batch 内部做了什么 你不需要关心细节，但了解一下有助于理解为什么它快： 自动分块 — 把 200 个 symbol 分成若干小批次 多线程并发 — 默认 5 个线程同时请求（是官方 API 并发，不是爬虫并发，不会被封） 自动重试 — 某个请求因网络波动失败，自动重试 3 次（只对网络错误、超时、5xx 状态码重试） 结果合并 — 全部完成后返回一个 {symbol: DataFrame} 的字典 你要写的代码：3 行。 循环、sleep、try/except、线程池——全部不用写。 实际对比：200 只票的 K 线获取 指标 akshare 循环 Tushare 循环 自写多线程 AlphaFeed batch 代码行数 ~20 行 ~20 行 ~30 行 3 行 耗时（200 只） 40-80 秒 30-60 秒 15-30 秒 3-5 秒 需要 sleep ✅ ✅ ⚠️ 看接口 ❌ 失败重试 手动写 手动写 手动写 内置 进度显示 手动写 手动写 手动写 内置 show_progress 被封风险 高 中 高 无 500 只、1000 只、5000 只呢？ 差距只会更大： # 全市场 A 股扫描：先拿代码列表，再批量拉 K 线 # 第一步：获取全市场标的 all_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True) symbols = all_quotes["symbol"].tolist() print(f"全市场共 {len(symbols)} 只") # 第二步：批量拉最近 20 天日 K 线（全部） dfs = af.klines.batch( symbols, period="1d", count=20, adjust="forward", to_dataframe=True, show_progress=True ) print(f"成功获取 {len(dfs)} 只") 5000 只 A 股全拉一遍，AlphaFeed batch 大约 30-60 秒完成。循环 + sleep 的方式？你可以自己算：5000 × 0.5 秒 = 至少 40 分钟。 不只是 K 线——所有数据类型都支持 batch # 批量获取五档盘口 depths = af.depth.batch(symbols_50) # 批量获取日内分钟线 intraday = af.klines.intraday_batch(symbols_50, period="1m", to_dataframe=True) # 批量获取标的信息 instruments = af.instruments.batch(symbols_50) 同样的模式：传一个列表进去，自动并发 + 重试，返回结果。 拿到数据之后能干什么 批量 K 线的典型用途： 批量计算技术指标 import pandas as pd for symbol, df in dfs.items(): if len(df) < 20: continue df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() df["vol_ratio"] = df["volume"] / df["volume"].rolling(10).mean() 批量选股 selected = [] for symbol, df in dfs.items(): if len(df) < 20: continue latest = df.iloc[-1] ma20 = df["close"].rolling(20).mean().iloc[-1] if (latest["close"] > ma20 and latest["volume"] > df["volume"].iloc[-11:-1].mean() * 2): selected.append(symbol) print(f"符合条件：{len(selected)} 只") 批量回测一个策略 results = {} for symbol, df in dfs.items(): if len(df) < 20: continue df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() df["signal"] = (df["ma5"] > df["ma20"]).astype(int) df["ret"] = df["close"].pct_change() * df["signal"].shift(1) results[symbol] = df["ret"].sum() results_df = pd.Series(results).sort_values(ascending=False) print("策略收益 Top 10：") print(results_df.head(10)) 数据获取只花几秒，剩下的时间全部用在分析上。 这才是正确的时间分配。 怎么开始 pip install alphafeed 注册 AlphaFeed 账号，免费版就可以使用 klines.batch 获取 A 股日 K 线。 把你现在循环 + sleep 的代码替换成 batch 调用，感受一下差距。 AlphaFeed 官网：https://alphafeed.org/ Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart batch 接口文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 期权分钟级历史行情数据 之前想做期权波动率曲面，光是找分钟线数据就让我头大。Wind太贵，聚宽那些只有日线，有些接口还动不动就断。后来在CMES金融数据库的下载页翻了翻，发现它直接把期权分钟K线数据开放出来了，当时觉得省事儿不少，就顺手整理了一份字段说明，给有同样需求的朋友避避坑。 这个数据覆盖了上交所的ETF期权和股指期权，按合约给分钟线。我主要下50ETF期权合约，一合约一个文件，csv格式，直接扔进pandas就能跑。说说里面都有啥。 下面是我自己常看的几个字段，不是全部，但做回测和计算希腊字母足够用了： 字段 含义 实际使用要注意的 trade_time 分钟线的时间戳 格式是YYYY-MM-DD HH:MM:00，整分钟对齐 open 开盘价 该分钟第一笔成交，如果没成交就沿用上一分钟收盘 high 最高价 分钟内的最高成交价，偶尔有插针 low 最低价 同理 close 收盘价 分钟最后一笔成交 volume 成交量 单位是张，注意区分累计和区间，这里给的是分钟内的增量 open_interest 持仓量 这一分钟结束时持仓，有些平台只给日末，这个有分钟序列 iv 隐含波动率 基于BS反推的，用的时候注意无风险利率和分红，有些深度虚值合约IV会缺失 delta delta 模型计算值，默认参数可以在文档里查 gamma gamma 同上 vega vega 同上 theta theta 同上 希腊字母这块，我还发现一个点：数据里给的数值是基于收盘价计算的，如果你要算盘中实时的，得自己用成交价重新算一遍，不然会有点滞后。不过做回测问题不大。 下载方式有两种，一种是网页直接下载压缩包，另一种是调接口。我习惯用python脚本批量拉，因为合约多，手动下太累。接口文档在CMES金融数据库的vip页面能找到，这里贴一下我用的代码片段，免安装的坑。 # 安装，我用的python3.9 # pip install cmesdata -i https://pypi.org/simple # CMES金融数据库的行情接口，注意入参正确，调用频率正常 from cmesdata import CMESClient client = CMESClient() # 获取一张50ETF认购期权合约的分钟数据，起始日期自己改 df = client.get_option_minute( symbol='510050C2306M03000', start='2023-06-01', end='2023-06-02' ) print(df.head()) 这个接口返回的df就是pandas的DataFrame，直接带上面那些字段。调用频率我自己试下来，一秒一次基本不会触发限制，千万别并发太多，不然会返回空数据，好像有一层较弱的反爬。 数据质量方面，主力合约和平值附近的合约分钟线基本完整，但有些远月或者深度虚值合约在盘初几分钟会缺，应该是当时没有成交。我一般都会用前值填充一下，反正影响不大。时间跨度，我看50ETF期权最早能到2015年，股指期权是2019年，做历史回测够用了。 说个让我比较意外的地方，它持仓量数据是分钟级的，这个好多平台都不给，只有日末。用分钟持仓量结合价格变化，能看出一些日内多空博弈的痕迹，挺好玩的。不过提醒一下，这个持仓量是交易所发布的数据，盘中更新可能会有几秒延迟，别拿来做高频。 噢对了，文件名是合约代码+日期，自己写个循环遍历合约列表就能批量下。合约列表也可以在CMES的页面搜到，不用自己维护。 大概就这些，懒得总结，有需要的自己去翻吧。 前言 A 股短线市场有一句共识：开盘定全天，竞价见主力。绝大多数散户仅简单看高开幅度选股，忽略竞价时段分段规则、量价共振、板块联动、历史竞价数据校验四大核心维度，频繁踩入诱多撤单、无量虚高、孤立个股三大陷阱。9:15-9:25 集合竞价是全天唯一无真实成交、仅通过挂单博弈暴露主力真实意图的窗口，也是散户与量化资金每日第一波筛选强势股的核心战场。 本文结合 XTick 竞价实时 / 历史数据接口，提供的全市场竞价明细、分时挂单、板块竞价统计数据，从竞价底层规则、核心量化指标、标准化选股流程、风险避雷、资金回测标准五个模块，搭建一套完整、原创、可实盘落地的集合竞价选股策略，搭配图表梳理核心逻辑，全程总结实操口诀与筛查清单。 一、先吃透竞价时间分段规则：分清虚假信号与真实资金 集合竞价 10 分钟分为三个完全不同的交易时段，不同时段挂单、撤单规则差异直接决定数据参考价值，这是竞价选股的底层前提，下表清晰区分各时段用途与选股取舍标准： 时间段 交易规则 主力行为特征 选股参考价值 实操建议 9:15-9:20 可挂单、可全额撤单 试盘诱多，挂巨量涨停单后快速撤单，制造虚假强势 极低，仅作情绪观察，不纳入选股条件 不以此阶段价格、封单作为判断依据 9:20-9:25 可挂单、禁止撤单 真实资金博弈，所有挂单将参与 9:25 最终撮合 极高，选股核心观测窗口 重点跟踪价格曲线、匹配量、未匹配挂单变化 9:25 整 一次性撮合，生成开盘价 确定当日基准成交价、竞价总成交额、量比 最高，最终筛选决策节点 9:25-9:305 分钟完成全部选股筛选 9:25-9:30 静默委托期，不更新竞价行情 资金预埋下单，无行情变动 无行情参考，仅规划开盘买卖点 复盘竞价板块热度，准备分时验证 竞价倒V诱多分时图（9:15拉板9:20撤单跳水） 核心总结 1 条：所有选股判断，只采信 9:20 后不可撤单时段数据，9:15-9:20 行情全部忽略。借助 XTick 竞价实时接口可持续拉取 9:20-9:25 逐秒挂单数据，批量监控全市场个股竞价走势，避免手动翻页遗漏主线热点标的。 二、竞价选股四大核心量化指标（数据接口可直接提取） 通过 XTick 竞价数据接口可一键获取全市场个股竞价涨幅、竞价成交额、量比、竞价换手率、委比、封单量六大字段，其中四大指标形成共振，是强势股的硬性筛选门槛，分维度拆解标准： 1. 竞价涨幅：2.8%-3.5% 黄金区间（风险收益最优） ＜2.8%：小幅高开，无主力主动进攻，多为散户零散挂单，日内翻绿概率超 60%，放弃； 2.8%-3.5%：温和抢筹区间，抛压充分消化，做多意愿明确，回测当日收红概率 87.2%，首选； 3.5%-7%：强高开，必须搭配巨量竞价成交额 + 主线板块联动，否则极易高开低走； ＞7%：超高开，仅主线龙头 + 重大实质性利好可参与，其余全部规避，主力出货高发区。 2. 竞价量能双指标：量比 + 竞价成交额（杜绝无量虚涨） 量能是区分 “真抢筹” 与 “假拉升” 的核心标尺，两个指标必须同步达标： 竞价量比≥5：代表盘前 5 分钟成交活跃度远超近 5 日平均，资金集中进场；量比 3-5 为备选，＜3 直接剔除； 竞价成交额门槛（按流通市值分层）： 小盘股（流通市值＜50 亿）：竞价成交额≥800 万； 中盘股（50-200 亿）：竞价成交额≥1500 万； 大盘权重（＞200 亿）：竞价成交额≥3000 万。 补充校验：竞价成交额占前一日总成交额比例≥3%，说明资金关注度显著提升。 3. 竞价换手率：区分筹码交换充分度 小盘股竞价换手率≥0.8%，中盘股≥0.5%；换手率过低代表场内筹码惜售 / 场外资金不愿进场，无上涨动力。XTick 竞价历史接口可导出近 30 日个股竞价换手率均值，对比当日数值判断是否异常放量。 4. 板块联动度：单一强势个股无参与价值 竞价选股最容易被忽略的核心条件：个股必须归属当日主线热点板块，同板块至少 3 只个股同步满足高开放量标准，形成板块共振。孤立高开个股，哪怕单指标全部达标，次日上涨概率不足 35%。 健康阶梯式竞价曲线（9:20后稳步抬升） 核心指标共振总结表 表格 指标 强势股硬性标准 预警风险标准 竞价涨幅 2.8%-3.5% ＞7% 无量、＜2% 量比 ≥5 ＜3 竞价成交额 匹配流通市值分层门槛 不足门槛 50% 板块联动 板块≥3 只个股竞价走强 板块仅 1 只个股高开 三、标准化 4 步竞价选股实操流程（9:25-9:30 完成） 依托 XTick 竞价全量数据接口（code=all 参数一键拉取全市场竞价数据），建立机械化筛选流程，规避主观情绪化选股，步骤清晰可执行： 第一步：基础池过滤（批量接口一次性剔除垃圾标的） 通过接口过滤条件批量筛选，一次性排除不合格个股： 剔除 ST、*ST、停牌个股； 剔除流通市值＜20 亿（流动性不足）、＞500 亿（爆发力弱）标的； 剔除 9:20-9:25 竞价价格波动超 1.2%、走势倒 V 跳水个股； 剔除存在减持、问询函、业绩暴雷等当日利空公告个股。 第二步：四大核心指标定量筛查 在过滤后的基础池中，严格匹配共振条件： 竞价涨幅 2.8%-3.5% + 量比≥5 + 竞价成交额达标 + 竞价换手率达标，剩余标的为初选池，通常全市场每日仅 10-20 只。 第三步：板块热度二次精选 调用概念板块成分股接口，匹配初选池个股所属行业 / 概念板块，统计板块竞价强势个股数量，只保留板块前排人气标的，单板块仅留存 1-2 只最优个股，压缩至 3-5 只核心观察池。 第四步：K 线趋势终选（盘前快速校验） 剔除跌破 20 日均线、处于高位连续加速 3 板以上、前期放量套牢平台个股；优先选择低位突破、回踩均线企稳的标的，最终锁定 1-2 只当日竞价选股目标。 四、三大竞价诱多陷阱（高频亏损点，接口数据可提前识别） 结合历史竞价回测数据，90% 竞价高开低走亏损案例，均源于以下三类盘面信号，通过 XTick 历史竞价接口复盘可快速识别特征： 陷阱 1：9:15 虚假巨量封单，9:20 集体撤单 盘面特征：9:15-9:19 涨停挂单几十亿，9:20 撤单后价格直线回落，竞价最终小幅高开，量比极低。 数据特征：9:15 封单 / 9:25 最终封单比值＞10 倍，挂单规模断崖式下跌。 避雷规则：9:20 后封单持续萎缩个股，直接放弃。 陷阱 2：量价完全背离，高开无量 盘面特征：竞价涨幅 5% 以上，但竞价成交额不足门槛一半，量比＜2，仅少量资金拉高价格，无资金承接。 避雷规则：涨幅越高，对量能要求越高，超高开无量标的一律不参与。 陷阱 3：板块孤立个股，无题材支撑 盘面特征：全市场仅该股竞价强势，所属板块其余个股全部低开平走，无板块资金抱团。 底层逻辑：主力仅针对个股自救，无市场合力，开盘冲高后缺乏跟风盘，极易跳水。 竞价选股核心逻辑示意图 五、配套交易纪律：选股 + 买卖 + 止损完整闭环 竞价选股只是筛选标的，配套严格仓位、止盈止损纪律才能实现长期正向收益，总结标准化交易规则： 仓位控制：大盘放量上涨环境仓位不超过 6 成；大盘缩量调整环境仓位压缩至 2 成以内，单只个股仓位不超过总资金 30%； 买入时机：9:30 开盘后股价站稳分时均价线、回踩均价快速拉升时分批介入；不直接在 9:25 集合竞价挂单追高； 止盈标准：日内涨幅 5% 减仓一半，冲击涨停持有至次日；当日无法涨停，冲高 8% 无条件清仓； 止损铁律：开盘后跌破竞价成交价 5%、持续 3 分钟无法收回均价线，立刻止损离场，不格局扛单。 六、竞价数据接口赋能策略优化（量化升级方案） 手动看盘存在数据不全、统计效率低的短板，XTick 全套竞价接口可实现策略自动化迭代，适配个人复盘与量化程序开发： 竞价实时接口（黄金 / 至尊权限）：9:15-9:25 持续拉取逐秒竞价明细，批量监控全市场标的，实时筛选符合指标个股； 竞价历史接口（至尊权限）：导出近 3 年每日全市场竞价数据，对选股指标做回测，调整涨幅、量比阈值适配不同市场行情； 增量更新接口：盘后一次性拉取当日完整竞价数据库，批量复盘当日选股标的走势，优化板块筛选、趋势过滤条件； 概念板块接口：自动统计每日各概念板块竞价强势个股数量，快速锁定主线题材，替代手动翻板块。 七、全文核心总结（实操口诀 + 清单） 竞价选股实操口诀 915 行情全作废，920 之后看真伪； 高开三点黄金位，量比过五才有力； 成交额按市值配，板块抱团才有戏； 倒 V 无量全避雷，跌破竞价果断退。 每日竞价选股 10 秒自检清单 ✅ 仅参考 9:20-9:25 不可撤单数据 ✅ 竞价涨幅锁定 2.8%-3.5% ✅ 量比≥5，竞价成交额匹配市值门槛 ✅ 同板块至少 3 只个股同步强势 ✅ 无高位加速、无当日利空公告 ✅ 股价站稳 20 日均线，趋势向上 ❌ 9:15 巨量撤单、倒 V 走势直接剔除 ❌ 高开超 7% 且无量，放弃参与 ❌ 板块独苗个股，不纳入观察池 结语 集合竞价选股不是单纯的抓涨停投机手法，而是一套基于交易规则、量化数据、市场合力的短线筛选体系。普通投资者可先用软件手动套用 4 步筛选流程建立盘感，有量化需求可依托 XTick 竞价数据接口实现全自动化选股与历史回测。策略核心不在于追求每日抓板，而是通过严格的定量条件过滤高风险标的，在风险收益比最优的竞价区间捕捉主线资金动向，配合标准化止盈止损，实现长期稳定的短线收益。同时需牢记：竞价数据仅为开盘预判工具，最终交易决策仍需结合开盘后分时承接、大盘环境动态调整，不可单一依靠盘前竞价信号盲目入场。 美股Level2 Tick数据，到底能扒到哪些字段 之前做美股盘口分析，想找一份完整的逐笔成交和十档订单薄，结果不是被收割买会员，就是数据缺胳膊少腿。后来实在没辙，顺着一些量化群里的讨论，在CMES金融数据库上扒到了他们打包好的历史数据，干脆把里面的字段全拎出来，免得下次再绕弯路。 先说一下这里面给的几类数据包，不是那种只给你一个“时间、价格、成交量”就完事的简化版，而是直接给到交易所原始结构的tick数据。主要分三块： 逐笔成交记录 十档订单薄快照 分钟级OHLCV汇总 下面这张表是我从他们接口文档里扒出来的，截取的是逐笔成交和订单薄里比较关键的字段，不是全部，但你真正用到的也就这些。 字段名 说明 出现在哪种数据里 ticker 股票代码，比如AAPL 三种都有 date 交易日期 三种都有 time 精确到毫秒的时间戳 逐笔成交、订单薄 price 成交价格 逐笔成交 size 成交股数 逐笔成交 exchange 成交所在交易所代码 逐笔成交 sale_condition 成交条件，比如@、F、I之类 逐笔成交 bid_price_1 ~ bid_price_10 买一到买十的价格 订单薄 ask_price_1 ~ ask_price_10 卖一到卖十的价格 订单薄 bid_size_1 ~ bid_size_10 各档挂单量 订单薄 ask_size_1 ~ ask_size_10 各档挂单量 订单薄 open, high, low, close 分钟K线四价 分钟汇总 volume 分钟成交量 分钟汇总 vwap 分钟成交量加权均价 分钟汇总 订单薄数据看着很全，但实际用的时候踩过一个坑：有些冷门票的十档挂单经常是空的，或者只有一两档，别一上来就默认所有股票都有十档深度，这个锅得自己背。 逐笔成交里的sale_condition字段很多人会忽略，但这个东西在判断是主动买还是主动卖，或者是不是暗池成交的时候特别有用。比如条件"A"和"@ 不同，直接影响到你算出来的买卖力道，这个细节我当时也是折腾了挺久才搞明白。 分钟数据看着简单，但它的vwap是直接算好的，比自己用逐笔重算要省事不少，而且和交易所公布的偏差很小，做日内策略复盘基本够用。 如果你也想把这堆数据拉下来自己玩，他们提供了Python的库，装起来很简单。下面这段代码我实际跑通过，接口调用的时候注意频率，别一上来就死命循环，会触发限流。 # 安装CMES金融数据库的行情接口 pip install cmes from cmes import DataClient # 用你自己的token初始化，token在官网个人中心获取 client = DataClient(token='your_token') # 拉取美股逐笔成交，symbol和date必须填，调用频率正常就好 df_trade = client.get_tick_trade(symbol='TSLA', date='2024-03-14') print(df_trade.head()) # 拉取十档订单薄，同样注意入参不要错 df_orderbook = client.get_orderbook(symbol='TSLA', date='2024-03-14') print(df_orderbook.head()) # 分钟数据可以直接拿，返回的是DataFrame df_min = client.get_minute_bar(symbol='TSLA', date='2024-03-14') print(df_min.head()) 别问为什么全是TSLA和AAPL，因为这两流动性最好，数据量也大，经常一跑就是几百万行，内存不够的记得先切日期。 有一点要提醒，这些数据是历史tick，不是实时流，所以做不了实盘交易，只能做历史回测或者研究。但好处是数据不用自己存，直接按天拉，省去整理数据库的麻烦。 另外，下载的时候如果遇到网络波动导致超时，把日期范围缩小一点再试，或者加个重试机制，这种小坑只能自己踩过才知道。 差不多就这些。数据字段就摆在那，具体怎么用，看你是想挖因子还是做微观结构，反正别在找数据这一步上浪费太多时间，毕竟分析才是大头。 导言：分时图——市场的“心电图” “为什么我一买就跌，一卖就涨？”这是无数投资者在复盘时最感无力的控诉。在波谲云诡的二级市场，情绪往往是交易的大敌。如果你想穿透表象看清本质，就必须学会解读分时图。 如果把股票看作一个生命体，分时图就是它的“心电图”。它不仅实时记录着股价的每一次律动，更通过分时线、分量和均价线这三个核心窗口，呈视出主力资金的脉搏。高水平的分析者，还会辅以“委托单”观察买卖意愿的虚实，并从“分时成交”的细节中寻找支撑与阻力的证据。看懂分时图，你才能从“凭感觉交易”跨越到“看证据执行”。 核心概念：均价线——股价的“中枢神经” 在实战分析中，均价线的地位至高无上。它是当日开盘至今所有成交总金额除以成交总股数所得的平均价格连线。 “均价线就像股价的中枢神经，反映了市场参与者的平均持仓成本。” 当股价位于均价线上方，说明多方占优；反之，则空方控盘。均价线不仅是波动的平衡点，更是我们判断主力意图的定盘星。 信号一：穿线放量，资金进场的号角 当股价在低位横盘后，突然放量上穿均价线，这绝非偶然的随机波动，而是场外大资金进场的明确号角。这种“量价齐升”的瞬间，往往意味着主力开始打破当天的平衡。如果穿线后股价能稳健地运行在均价线上方，通常暗示主力正在建仓。对于短线高手而言，这便是精准的“上车”时机，后续大概率会迎来一波快速拉升。 信号二：量价共振，优雅的“同步上行” 最健康的上升形态，是股价与均价线呈现出一种“量价共振”的协调感。股价阶梯式上扬，均价线也跟随平稳上移。此时的成交量特征表现为“涨时放量、跌时缩量”。这种形态说明多头力量充沛且有序，是短线操作中最理想的持筹环境。 信号三：警惕高开低走的“诱多”陷阱 这是一种极具杀伤力的走势：早盘受利好刺激大幅高开，甚至瞬间冲高，随后成交量异常放大却无法维持股价，转而一路跳水砸穿均价线。这是典型的主力利用利好消息“诱多”并趁机获利离场的骗局。高开时的放量往往是动能衰竭的信号。 “这种‘啪啪打脸’的走势如果看不懂，这时候再不跑，你就是‘韭菜本菜’。” 信号四：回踩不破，主力在“扎营托底” 当股价在均价线上方运行，回落触碰均价线却始终不破，并随即放量反弹时，说明均价线处有大资金在进行强力的“托底”。在技术分析逻辑中，我们强调“确认”的力量。资深交易者通常会耐心等待股价对均价线的第三次测试。一旦三次测试均能守住，则证明主力已在底部“扎营”完毕，准备发起总攻冲锋。 信号五：梯形上涨，主力边洗边拉 如果你看到股价走势如同一级一级的楼梯，每段拉升后都伴随着一段窄幅的横盘整理，那么你很可能遇上了主力“边洗边拉”的套路。这种形态的核心逻辑是通过横盘清洗获利盘和浮筹，为更高目标的拉升减压。识别该信号的要点是：股价始终运行在均价线上方，且每次向上突破“台阶”时都伴随成交量的释放。这种走势的终点往往是“一口气干到天花板”。 信号六：巨量冲高后的蓄势横盘 早盘股价放巨量猛烈冲高，随后成交量呈“断崖式”骤减，股价进入长时间的缩量横盘。此时不必急于获利了结，这往往是主力在拉升一段后的“情绪测试”。只要股价能守住均价线且不出现反转信号，说明拉升并非虚火，而是在蓄势等待下一次的突破。 信号七：缠绕不清时的“资金效率”观望 当股价与均价线像“耳机线”一样反复纠缠、反复穿插时，说明多空双方陷入了极度的胶着状态。从资本效率的角度看，这种形态代表市场缺乏主攻方向，属于低胜率区域。高手的策略是持币观望，因为在“夹心饼干”行情中穿梭不仅容易亏损，更会消耗大量的机会成本。唯有等突破方向明确后，才是出手的良机。 信号八：量能分布的“控盘真相” 这是高手最青睐的洗盘识别信号：股价在均价线上方时放量，而被打压至均价线下方时成交量立即萎缩。这种“线上放量、线下缩量”的特征，揭示了主力的真实态度——向下砸盘只是为了清场跟风盘，并无真正出货意愿。线下缩量正说明了主力依然牢牢“控盘”，筹码锁定良好。这不仅不该卖出，反而是极佳的介入点。 结语：从交易冲动到标准程序（SOP）的跃迁 学习这八大分时信号，其终极目的不是为了寻找“致富秘籍”，而是为了建立一套标准化的交易操作流程（SOP）。在股市中生存，技术是表象，纪律是骨架。 请将这些信号视为你的“交易军令状”：符合条件则动，不符合条件则等。当你能克制住本能的冲动，只依据分时图上的证据进行决策时，财富自然会向你靠拢。 
