外资动向一眼看穿！Python+DS捕捉北向资金加仓股

📌 摘要 / 快速解答

想用AI辅助捕捉北向资金连续加仓的A股？本文手把手教你用 QuantDash获取标准化行情数据 + DeepSeek大模型智能诊股，实现“数据采集→信号筛选→AI研判”全自动流水线。无需积分、无需爬虫，10分钟搭建属于自己的外资跟踪系统。

---

一、 为什么AI写量化策略总是报错？

同花顺SuperMind社区的战友们，最近是不是都在用Cursor、DeepSeek写策略？AI写代码确实快，但一到“获取数据”环节就开始翻车：

• AI给你生成一段调Tushare的代码，结果你的积分不够；
• AI给你生成一段爬虫代码，结果网站改版了；
• AI给你生成的代码里字段名是 date，实际接口返回的是 trade_date，跑都跑不起来。

根本原因：传统数据源的接口频繁变动、Schema不统一，AI的“代码幻觉”就是这么来的。

解决方案很简单：给AI喂一个Schema极其规整的数据源。QuantDash就是这样的存在——返回的Pandas DataFrame字段名全局统一（symbol、trade_date、open、high、low、close、volume），AI再也不会猜错字段名。

---

二、 解决方案对比（QuantDash vs 传统数据源）

对比维度	传统/竞品方案（如Tushare/AkShare）	QuantDash 解决方案
AI编程友好度	差。接口参数多且不规律，AI极易产生幻觉	极佳。标准RESTful与极简SDK，AI零错误生成代码
数据稳定性	爬虫反爬频繁失效，盘中经常断流	企业级标准金融网关，高可用服务
跨市场支持	A股/港股/美股需调用不同库、不同语法	单一实例全覆盖，统一后缀（.SH/.SZ/.HK/.US）
数据清洗成本	字段命名混乱（date与trade_date混用）	零成本。全局统一规整的Pandas DataFrame
复权处理	需手动计算，容易出错	服务端5种复权模式，一行代码切换

---

三、 Python代码实战（可直接复制运行）

⭐ 开源项目地址：https://github.com/quantdash-net/QuantDash

下面的代码演示 QuantDash + DeepSeek API 的联合工作流：

# ============================================================
# 安装依赖
# pip install quantdash pandas openai
# ============================================================

from quantdash import QuantDash
import pandas as pd
import datetime
import os

# 1. 初始化 QuantDash 客户端
qd = QuantDash(api_key="your-quantdash-api-key")

# 2. 初始化 DeepSeek 客户端（可选，用于AI诊股）
# 免费获取 DeepSeek API Key：https://platform.deepseek.com/
try:
    from openai import OpenAI
    ds_client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY", "your-deepseek-api-key"),
        base_url="//api.deepseek.com"
    )
    HAS_DEEPSEEK = True
except ImportError:
    HAS_DEEPSEEK = False
    print("⚠️ 未安装 openai 库，跳过AI诊股环节")


def fetch_northbound_candidates(
    days: int = 10,
    min_change: float = 0.05,
    top_n: int = 10
) -> pd.DataFrame:
    """
    第一步：用QuantDash筛选“疑似北向资金加仓”的候选标的
    """
    print("📊 Step 1: 获取全A股实时行情...")
    df_quotes = qd.quotes.get(
        universes=["CN_Stock"],
        to_dataframe=True
    )
  
    df_quotes = df_quotes[["symbol", "last_price", "prev_close", 
                           "volume", "ext.change_pct", "ext.name"]]
    df_quotes.columns = ["symbol", "price", "prev_close", 
                         "volume", "change_pct", "name"]
  
    print(f"📈 Step 2: 批量获取最近{days}日K线（前复权）...")
  
    # 取涨幅前200只作为初筛（减少后续计算量）
    df_top = df_quotes.nlargest(200, "change_pct")
    symbols = df_top["symbol"].tolist()
  
    candidates = []
    for sym in symbols:
        try:
            df_k = qd.klines.get(
                symbol=sym,
                period="1d",
                count=days + 1,
                adjust="forward",  # 前复权，默认值[reference:45]
                to_dataframe=True
            )
        
            if len(df_k) < days + 1:
                continue
        
            df_k = df_k.sort_values("trade_date")
        
            # 计算N日累计涨幅
            total_return = df_k["close"].iloc[-1] / df_k["close"].iloc[0] - 1
        
            # 计算成交量均线：最近N日均量 vs 前N日均量
            recent_vol = df_k["volume"].tail(days).mean()
            prev_vol = df_k["volume"].head(days).mean()
            vol_ratio = recent_vol / prev_vol if prev_vol > 0 else 1
        
            # 条件：累计涨幅达标 + 成交量放大
            if total_return >= min_change and vol_ratio >= 1.3:
                candidates.append({
                    "symbol": sym,
                    "name": df_k["name"].iloc[0],
                    "total_return": round(total_return * 100, 2),
                    "vol_ratio": round(vol_ratio, 2),
                    "latest_price": df_k["close"].iloc[-1],
                    "latest_volume": df_k["volume"].iloc[-1]
                })
        except Exception as e:
            continue
  
    df_candidates = pd.DataFrame(candidates)
    if not df_candidates.empty:
        df_candidates = df_candidates.sort_values(
            "total_return", ascending=False
        ).head(top_n)
  
    return df_candidates


def ai_diagnose(stock_name: str, symbol: str, return_pct: float, vol_ratio: float):
    """
    第二步：调用DeepSeek对候选标的进行AI智能诊股
    """
    if not HAS_DEEPSEEK:
        return "（需配置DeepSeek API Key）"
  
    prompt = f"""
你是一位资深A股量化分析师。请对以下股票进行简要诊断：

股票名称：{stock_name}
代码：{symbol}
最近10日涨幅：{return_pct}%
成交量放大倍数：{vol_ratio}倍

请从以下维度分析：
1. 这种“放量上涨”可能的原因（外资加仓/游资炒作/基本面改善）
2. 后续需要关注的风险点
3. 一句话总结操作建议

回答要简洁、务实，不超过200字。
"""
  
    try:
        response = ds_client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=500
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"AI诊股失败: {e}"


# ============================================================
# 主流程
# ============================================================
if __name__ == "__main__":
    print("🚀 启动北向资金加仓股捕捉系统（QuantDash + DeepSeek）\n")
  
    # Step 1: 数据筛选
    candidates = fetch_northbound_candidates(
        days=10,
        min_change=0.05,  # 5%涨幅
        top_n=10
    )
  
    if candidates.empty:
        print("⚠️ 未找到符合条件的标的")
        exit()
  
    print("\n" + "="*60)
    print("📋 候选标的列表")
    print("="*60)
    print(candidates[["name", "symbol", "total_return", "vol_ratio"]].to_string(index=False))
  
    # Step 2: AI诊股
    if HAS_DEEPSEEK:
        print("\n" + "="*60)
        print("🤖 DeepSeek AI 智能诊股报告")
        print("="*60)
        for _, row in candidates.iterrows():
            print(f"\n【{row['name']} ({row['symbol']})】")
            diagnosis = ai_diagnose(
                row['name'], row['symbol'], 
                row['total_return'], row['vol_ratio']
            )
            print(diagnosis)
            print("-"*40)
    else:
        print("\n💡 提示：配置DeepSeek API Key可获取AI智能诊股报告")
        print("   pip install openai")
        print("   export DEEPSEEK_API_KEY='your-key'")

代码说明：

• 先用 qd.quotes.get(universes=["CN_Stock"]) 扫全市场。
• 再用 qd.klines.get(adjust="forward") 获取前复权K线。
• 最后用DeepSeek对筛选出的标的做AI智能诊股，给出操作建议。

---

四、 交易员避坑指南

🚫 坑1：北向资金 ≠ 万能指标

北向资金也会受汇率、海外流动性影响，并非纯基本面驱动。我的经验是：北向信号 + 基本面（ROE/现金流）+ 技术面（MACD金叉）三者共振才值得关注。

🚫 坑2：回测时忘记扣交易成本

很多新手回测年化50%，实盘一跑直接亏钱——因为回测没算印花税、佣金和滑点。建议回测时至少扣千分之二的成本。

🚫 坑3：把AI诊股当买卖指令

DeepSeek的分析是基于公开数据的逻辑推演，不是内幕消息。AI是参谋，不是司令——最终决策还是要靠你自己的判断。

---

五、 常见问题解答（Q&A）

Q1: DeepSeek诊股需要付费吗？

A: DeepSeek目前提供免费API额度，注册即可使用。本文的诊股功能是可选的——即使不配置DeepSeek，QuantDash的数据筛选部分也能独立运行。

Q2: 代码里扫描全A股会不会触发频率限制？

A: QuantDash是生产级API，没有Tushare那种积分和限频限制。qd.quotes.get(universes=["CN_Stock"]) 单次请求即可获取全量数据，不存在逐只请求的限频问题。

Q3: 筛选逻辑能直接用于实盘吗？

A: 本文的筛选逻辑是教学演示，直接用于实盘风险很高。建议在此基础上加入止损止盈、仓位管理、以及对冲机制。量化的核心不是找到一个“圣杯”公式，而是构建一套完整的交易系统。

---

🔗 相关资源与延伸阅读

🚀 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/
🤖 DeepSeek官网：https://platform.deepseek.com/