盘中筛选既要价格又要盘口？按需分层比一把抓更容易维护

┌─────────────────────────────────────┐
│           筛选调度层                  │
│  1. 调用行情层获取全池最新价格        │
│  2. 用本地条件过滤候选池              │
│  3. 调用盘口层查询候选池的深度        │
│  4. 合并结果供决策                   │
└──────┬──────────────────────┬───────┘
       │                      │
┌──────▼──────┐      ┌──────▼──────┐
│  行情层      │      │   盘口层    │
│  quotes.get  │      │  depth.get │
│  独立重试     │      │  独立重试   │
│  独立频率     │      │  独立频率   │
└──────────────┘      └─────────────┘

# 环境：Python 3.10，quantdash==0.1.0
import os
import time
from typing import List

from quantdash import QuantDash
import pandas as pd

qd = QuantDash(api_key=os.getenv("QUANTDASH_API_KEY"))

# ── 行情层 ──────────────────────────────
def fetch_quotes(symbols: List[str]) -> pd.DataFrame:
    """获取一批标的的实时行情，独立重试"""
    max_retries = 2
    for attempt in range(max_retries):
        try:
            df = qd.quotes.get(
                symbols=symbols,
                to_dataframe=True,
            )
            if df is not None and not df.empty:
                return df
        except Exception as e:
            if "429" in str(e):
                # 限流时按服务端指示等待
                time.sleep(5)
                continue
            raise
    return pd.DataFrame()


# ── 盘口层 ──────────────────────────────
def fetch_depths(symbols: List[str]) -> dict:
    """获取候选标的的五档盘口，独立重试"""
    if not symbols:
        return {}
    max_retries = 2
    for attempt in range(max_retries):
        try:
            result = qd.depth.batch(symbols)
            return result if result else {}
        except Exception as e:
            if "429" in str(e):
                time.sleep(3)
                continue
            raise
    return {}


# ── 筛选调度 ────────────────────────────
def screen_candidates(pool_symbols: List[str]) -> pd.DataFrame:
    """筛选流程：先行情后盘口，分层获取"""
    # 第1步：获取全池实时行情
    quotes_df = fetch_quotes(pool_symbols)
    if quotes_df.empty:
        return pd.DataFrame()

    # 第2步：本地过滤——涨跌幅 > 2% 且成交量大于均量
    candidates = quotes_df[
        (quotes_df["ext.change_pct"] > 0.02)
        & (quotes_df["volume"] > 1_000_000)
    ]

    if candidates.empty:
        return candidates

    # 第3步：只对候选标的查盘口
    candidate_symbols = candidates["symbol"].tolist()
    depths = fetch_depths(candidate_symbols)

    # 第4步：将盘口信息合并到结果中
    for sym in candidate_symbols:
        if sym in depths:
            d = depths[sym]
            candidates.loc[
                candidates["symbol"] == sym, "bid1_price"
            ] = d["bid_prices"][0] if d.get("bid_prices") else None
            candidates.loc[
                candidates["symbol"] == sym, "ask1_price"
            ] = d["ask_prices"][0] if d.get("ask_prices") else None

    return candidates


# 使用示例
if __name__ == "__main__":
    pool = ["600519.SH", "000001.SZ", "300750.SZ", "002594.SZ"]
    result = screen_candidates(pool)
    print(result[["symbol", "ext.last_price", "ext.change_pct", "bid1_price", "ask1_price"]])