📌 摘要 / 快速解答
中线跟庄的核心是识别机构建仓痕迹。本文用 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(
# "//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/