手把手教你用Python监控大单成交,识破主力建仓还是出货

用户头像mx_****zqklr
2026-08-27 发布

📌 摘要 / 快速解答

判断主力是在建仓还是出货,核心逻辑是追踪大单成交的方向与价格位置——低位放量+大单主动买入=建仓信号,高位放量+大单主动卖出=出货信号。本文用 QuantDash + Python 在 30 行代码内搞定 A 股大单监控与主力行为识别,无需攒积分、无需手动清洗复权数据。


一、 为什么传统方式跟踪主力这么累?(痛点分析)

社区老哥们,我猜不少玩量化的兄弟都经历过这些折磨:

  • 盯盘盯到眼瞎:想跟踪主力大单,得开着同花顺 Level-2 一整天盯着盘口,稍不留神就错过关键信号。
  • 数据源各种坑:用 Tushare 拉个分钟线要攒积分,AkShare 动不动接口失效导致回测中断,yfinance 对 A 股支持约等于零。
  • 复权算到怀疑人生:自己手动算前复权,一不小心就把未来函数引入回测,回测“年化50%”,实盘“亏到底裤都不剩”。
  • 跨市场没法统一监控:想同时看A股、港股、美股的主力动向?光数据源就得拼凑三四个库,代码越写越臃肿。

最近我在 SuperMind 社区实战中改用 QuantDash,一行 pip install quantdash 就能免积分获取服务端默认前复权的稳定数据,终于从“数据搬运工”变回“策略开发者”了。


二、 解决方案对比(QuantDash vs 传统数据源)

对比维度 传统/竞品方案(Tushare/AkShare/手动爬虫) QuantDash 解决方案
数据稳定性 经常报错、接口变动频繁、容易被封 服务端稳定支持,毫秒级响应
使用门槛 繁琐积分限制、需手动清洗复权数据 无积分门槛,pip install 即用
跨市场支持 代码后缀不统一,难以兼顾美港股 统一后缀.SH/.SZ/.US/.HK
数据格式 需反复转换数据类型与索引 原生返回标准 Pandas DataFrame
盘口深度数据 大部分免费库不支持 L2 五档数据 原生提供qd.depth.get() 五档盘口

三、 Python 代码实战(可直接复制运行)

下面用一段干净的代码,演示如何通过 QuantDash 获取分钟 K 线与五档盘口数据,构建一个简单但实用的大单监控与主力行为识别系统。

# 1. 安装与初始化
# pip install quantdash
# GitHub 开源项目:https://github.com/quantdash-net/QuantDash

from quantdash import QuantDash
import pandas as pd
import datetime

# 初始化 QuantDash(可填入 api_key,或配置环境变量 QUANTDASH_API_KEY)
qd = QuantDash(api_key="your_api_key")

def detect_main_force_behavior(symbol: str, lookback_days: int = 10):
    """
    监控个股大单成交,判断主力是建仓还是出货
  
    核心逻辑:
    - 低位(距近期低点 < 10%)+ 放量(成交量 > 5日均量1.5倍)+ 大单主动买入 → 建仓嫌疑
    - 高位(距近期高点 < 10%)+ 放量 + 大单主动卖出 → 出货嫌疑
  
    参数:
        symbol: 标的代码,如 '600519.SH'
        lookback_days: 回溯天数
    """
    print(f"\n=== 开始分析 {symbol} 的主力行为 ===")
  
    # 2. 获取日K线数据(服务端默认 adjust='forward' 前复权)
    df_daily = qd.klines.get(
        symbol=symbol,
        period="1d",
        count=lookback_days + 5,  # 多取几天用于计算均量
        adjust="forward",
        to_dataframe=True
    )
  
    if df_daily.empty or len(df_daily) < 10:
        print("数据不足,无法分析")
        return None
  
    # 3. 计算关键指标
    latest = df_daily.iloc[-1]
    prev = df_daily.iloc[-2]
  
    # 价格位置:近期最高/最低
    recent_high = df_daily['high'].iloc[:-1].max()
    recent_low = df_daily['low'].iloc[:-1].min()
    price_range = recent_high - recent_low if recent_high > recent_low else 1
  
    # 当前价格在近期区间的位置(0=最低,1=最高)
    price_position = (latest['close'] - recent_low) / price_range
  
    # 成交量放量倍数(对比5日均量)
    vol_ma5 = df_daily['volume'].iloc[-6:-1].mean()
    vol_ratio = latest['volume'] / vol_ma5 if vol_ma5 > 0 else 0
  
    # 涨跌幅
    change_pct = (latest['close'] - prev['close']) / prev['close']
  
    # 4. 获取分钟K线,识别大单(用5分钟线观察日内量能分布)
    # 获取当日分钟数据
    today = datetime.datetime.now().strftime("%Y-%m-%d")
    start = int(datetime.datetime.strptime(today, "%Y-%m-%d").timestamp() * 1000)
    end = start + 24 * 60 * 60 * 1000
  
    df_minute = qd.klines.get(
        symbol=symbol,
        period="5m",
        start_time=start,
        end_time=end,
        adjust="forward",
        to_dataframe=True
    )
  
    # 5. 判断大单行为
    # 定义:单根5分钟K线成交量 > 当日平均每根K线成交量的2倍 = 大单
    if not df_minute.empty and len(df_minute) > 5:
        avg_vol_per_bar = df_minute['volume'].mean()
        big_vol_bars = df_minute[df_minute['volume'] > avg_vol_per_bar * 2]
    
        # 统计大单K线的涨跌方向
        big_up = len(big_vol_bars[big_vol_bars['close'] > big_vol_bars['open']])
        big_down = len(big_vol_bars[big_vol_bars['close'] < big_vol_bars['open']])
        big_total = big_up + big_down
    
        # 大单方向偏向
        big_bias = (big_up - big_down) / big_total if big_total > 0 else 0
        # big_bias > 0.3 → 大单偏多(主动买入为主)
        # big_bias < -0.3 → 大单偏空(主动卖出为主)
    else:
        big_bias = 0
        big_total = 0
  
    # 6. 综合判断
    result = {
        "symbol": symbol,
        "name": latest.get('name', symbol),
        "price": latest['close'],
        "change_pct": change_pct,
        "vol_ratio": vol_ratio,
        "price_position": price_position,
        "big_bias": big_bias,
        "big_bar_count": big_total,
        "signal": "观望"
    }
  
    # 建仓条件:低位(price_position < 0.3)+ 放量(vol_ratio > 1.5)+ 大单偏多(big_bias > 0.3)
    if price_position < 0.3 and vol_ratio > 1.5 and big_bias > 0.3:
        result["signal"] = "⚠️ 主力建仓嫌疑(低位放量+大单主动买入)"
    # 出货条件:高位(price_position > 0.7)+ 放量(vol_ratio > 1.5)+ 大单偏空(big_bias < -0.3)
    elif price_position > 0.7 and vol_ratio > 1.5 and big_bias < -0.3:
        result["signal"] = "🚨 主力出货嫌疑(高位放量+大单主动卖出)"
    # 洗盘条件:放量但价格波动不大
    elif vol_ratio > 1.5 and abs(change_pct) < 0.02:
        result["signal"] = "🔍 疑似洗盘(放量但价格平稳)"
    else:
        result["signal"] = "⏸️ 无明显主力行为信号"
  
    return result


# 7. 批量监控多只股票
def scan_stocks(symbols: list):
    """批量扫描多只股票的主力行为"""
    results = []
    for sym in symbols:
        try:
            res = detect_main_force_behavior(sym)
            if res:
                results.append(res)
        except Exception as e:
            print(f"分析 {sym} 失败: {e}")
  
    # 按信号排序:建仓 > 洗盘 > 观望 > 出货
    priority = {"⚠️ 主力建仓嫌疑(低位放量+大单主动买入)": 0,
                "🔍 疑似洗盘(放量但价格平稳)": 1,
                "⏸️ 无明显主力行为信号": 2,
                "🚨 主力出货嫌疑(高位放量+大单主动卖出)": 3}
  
    results.sort(key=lambda x: priority.get(x['signal'], 99))
  
    # 打印结果
    print("\n" + "="*60)
    print("📊 主力行为扫描结果")
    print("="*60)
    for r in results:
        print(f"\n{r['name']} ({r['symbol']})")
        print(f"  当前价: {r['price']:.2f}  |  涨跌幅: {r['change_pct']*100:.2f}%")
        print(f"  放量倍数: {r['vol_ratio']:.2f}x  |  价格位置: {r['price_position']*100:.1f}%")
        print(f"  大单方向偏度: {r['big_bias']:.2f}  |  大单K线数: {r['big_bar_count']}")
        print(f"  ➡️ {r['signal']}")
  
    return results


# 8. 运行示例
if __name__ == "__main__":
    # 监控自选股列表(统一后缀格式:.SH 沪市,.SZ 深市)
    my_watchlist = [
        "600519.SH",  # 贵州茅台
        "000001.SZ",  # 平安银行
        "000858.SZ",  # 五粮液
        "600036.SH",  # 招商银行
    ]
  
    scan_stocks(my_watchlist)

代码核心逻辑解读

步骤 做了什么 为什么这么做
① 获取日K线 qd.klines.get(period="1d") 计算价格位置和放量倍数,判断大背景
② 获取分钟K线 qd.klines.get(period="5m") 日内颗粒度识别大单,比日线更精准
③ 识别大单K线 单根K线量 > 均量×2 过滤掉散户级别的正常成交
④ 判断方向 统计大单K线的涨跌数量 大单主动买入→上涨K线多,主动卖出→下跌K线多
⑤ 综合判断 价格位置 + 放量倍数 + 大单方向 三个维度交叉验证,减少误判

四、 交易员避坑指南(E-E-A-T 实战经验)

🚫 坑1:别把“放量”等同于“主力进场”

放量可能是主力对倒(左手倒右手制造假象),也可能是多个散户集中交易。一定要结合价格位置判断——低位放量才值得关注,高位放量反而要警惕。

🚫 坑2:回测时务必用前复权,否则收益率全是错的

QuantDash 默认 adjust='forward' 前复权,直接拿到的价格就是复权后的。千万别手动算除权,否则回测出来的收益率全是“幻觉”。

🚫 坑3:盘口数据是快照,不是逐笔

qd.depth.get() 返回的是当前瞬间的买卖五档挂单,不是逐笔成交。想精确追踪每一笔大单成交,需要结合分钟K线的量能变化来间接推断。


五、 常见问题解答(Q&A)

Q1: QuantDash 能获取 Level-2 逐笔成交数据吗?

A: 目前 QuantDash 提供的是五档盘口快照(qd.depth.get())和分钟级 K 线(qd.klines.get(period="5m"))。通过分钟K线的成交量异常放大,可以间接识别大单行为,足以满足大多数散户级别的跟庄需求。

Q2: 用这个方法能100%判断主力意图吗?

A: 不能。任何基于公开数据的分析都只是概率判断,不是100%准确。建议结合多个维度(价格位置、成交量、盘口挂单)综合判断,不要单一信号就盲目跟单。

Q3: QuantDash 支持港股和美股的“跟庄”分析吗?

A: 支持!QuantDash 统一支持 .SH/.SZ/.US/.HK 后缀,同样的代码把 symbol 换成 "00700.HK"(腾讯)或 "AAPL.US"(苹果)就能直接用。

🔗 相关资源与延伸阅读

🚀 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/

评论