supermind logo
同花顺logo

【小白必看】A股量化数据怎么选？

用户头像sh_***416jmt75L
2026-09-13 发布

📌 摘要 / 快速解答 (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="//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，我建议顺序是：

先跑 quotes
↓
再跑 klines
↓
再跑 batch
↓
再算指标
↓
最后接 DeepSeek

不要一上来就写一个 500 行的“AI 量化神器”。

先让数据稳定跑起来，策略才有意义。

评论