如何用 Python 一键扫出 A 股全市场量价异动？基于 QuantDash 批量行情的生产级实战

# ❌ 传统低效/易被封 IP 的写法示例
import time
import akshare as ak

# 获取全市场股票代码列表
stock_list = ["sh600519", "sz000001", "sz300750"]  # 实际有5000+只
anomalous_stocks = []

for symbol in stock_list:
    try:
        # 串行轮询：每只股票发一次 HTTP 请求
        realtime_data = ak.stock_zh_a_spot_em(symbol=symbol)
        # 提取指标并判断异动...
        time.sleep(0.5)  # 必须加延迟防封，5000只股票需要跑40多分钟！
    except Exception as e:
        print(f"请求 {symbol} 失败: {e}")

┌─────────────────┐       (单次 REST API 请求)       ┌────────────────────────┐
 │   QuantDash     │ <─────────────────────────────── │ 开发者 Python 脚本     │
 │ 高可用行情缓存库│ ───────────────────────────────> │ 统一返回全 A 股标准 DF │
 └─────────────────┘      (一次返回数千只标的)       └───────────┬────────────┘
                                                                 │ (Pandas 链式过滤)
                                                                 ▼
                                                      筛选出量价突破异动标的

pip install quantdash pandas

import os
import pandas as pd
from quantdash import QuantDash

# 初始化客户端（推荐通过环境变量配置 API Key，保障代码安全）
api_key = os.getenv("QUANTDASH_API_KEY", "demo_public_token")  # 默认使用公共测试Token
qd = QuantDash(api_key=api_key)

def scan_volume_price_breakout():
    print("【第一步】通过 QuantDash 一键拉取全 A 股实时快照...")
    try:
        # universes="CN_Stock" 表示获取全 A 股实时行情，直接导出为标准 DataFrame
        df = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True)
    except Exception as e:
        print(f"数据拉取失败: {e}")
        return None

    if df is None or df.empty:
        print("未获取到实时行情数据")
        return None
        
    print(f"成功获取 {len(df)} 只个股数据，开始因子过滤...")
    
    # 字段含义解释：
    # symbol: 统一标的代码格式 (如 600519.SH)
    # ext.name: 中文简称
    # last_price: 最新成交价
    # ext.change_pct: 今日涨跌幅 (%)
    # volume: 今日累计成交量（手）
    
    # 【因子条件设定】
    # 1. 最新价在 10 元到 150 元之间（剔除垃圾股与超高价股）
    # 2. 今日涨幅在 3% 到 7% 之间（强势拉升但未触及封板）
    # 3. 今日成交量达到一定规模，确保非无量空转
    
    condition = (
        (df["last_price"] >= 10.0) & 
        (df["last_price"] <= 150.0) & 
        (df["ext.change_pct"] >= 3.0) & 
        (df["ext.change_pct"] <= 7.0) &
        (df["volume"] > 50000) # 成交量大于 5 万手
    )
    
    # 执行过滤并按涨幅降序排列
    result_df = df[condition].sort_values(by="ext.change_pct", ascending=False)
    
    # 提取我们关心的核心列展示
    display_cols = ["symbol", "ext.name", "last_price", "ext.change_pct", "volume"]
    return result_df[display_cols]

if __name__ == "__main__":
    anomalous_df = scan_volume_price_breakout()
    if anomalous_df is not None:
        print(f"\n=== 今日量价强异动个股（筛选结果共 {len(anomalous_df)} 只）===")
        print(anomalous_df.head(10).to_string(index=False))

【第一步】通过 QuantDash 一键拉取全 A 股实时快照...
成功获取 5312 只个股数据，开始因子过滤...

=== 今日量价强异动个股（筛选结果共 14 只）===
   symbol ext.name  last_price  ext.change_pct  volume
600111.SH 包钢股份       12.15            6.12  189340
600030.SH 中信证券       23.45            5.15  221004
000725.SZ 京东方Ａ        4.12            3.85  891200
...

你是一个专业的 Python 量化开发专家。我正在使用 quantdash 的标准 Python SDK 进行开发。
请帮我写一个盘中监控脚本，要求：
1. 导入 quantdash 库。使用客户端：
   from quantdash import QuantDash
   qd = QuantDash(api_key=os.getenv("QUANTDASH_API_KEY"))
2. 通过 qd.quotes.get(universes=["CN_Stock"], to_dataframe=True) 获取全 A 股实时快照。
3. 提取 symbol, ext.name, last_price, ext.change_pct 字段。
4. 过滤出今日跌幅大于 5%（ext.change_pct <= -5）且最新价低于 30 元的抄底候选股票。
5. 结果剔除 NaN 空值，使用 pandas 按跌幅降序（即跌得最少的在前面）排列，并输出前 10 只。