如何利用 QuantDash API 与 Python 玩转高频五档盘口与全市场实时监控？

pip install quantdash pandas

import os
import time
from quantdash import QuantDash

# 初始化客户端
api_key = os.getenv("QUANTDASH_API_KEY", "your_api_key_here")
qd = QuantDash(api_key=api_key)

def calculate_obi_5(symbol: str):
    """
    获取指定标的的五档盘口数据，并计算 OBI
    """
    try:
        # 获取深度盘口数据
        depth = qd.depth.get(symbol)
        
        if not depth:
            print(f"未能获取 {symbol} 深度数据")
            return None
        
        # 提取买五和卖五的挂单量
        # QuantDash 返回的 depth 数据通常包含 bid_sizes 和 ask_sizes 列表
        bid_sizes = depth.get("bid_sizes", [])
        ask_sizes = depth.get("ask_sizes", [])
        
        if not bid_sizes or not ask_sizes:
            print(f"{symbol} 盘口挂单数据不完整")
            return None
            
        total_bid_volume = sum(bid_sizes[:5])
        total_ask_volume = sum(ask_sizes[:5])
        
        if (total_bid_volume + total_ask_volume) == 0:
            return 0.0
            
        # 计算 OBI
        obi = (total_bid_volume - total_ask_volume) / (total_bid_volume + total_ask_volume)
        
        # 提取买一价与卖一价
        bid_prices = depth.get("bid_prices", [])
        ask_prices = depth.get("ask_prices", [])
        bid_1 = bid_prices[0] if bid_prices else 0.0
        ask_1 = ask_prices[0] if ask_prices else 0.0
        
        return {
            "symbol": symbol,
            "bid_1": bid_1,
            "ask_1": ask_1,
            "total_bid_vol": total_bid_volume,
            "total_ask_vol": total_ask_volume,
            "obi": round(obi, 4)
        }
        
    except Exception as e:
        print(f"计算 OBI 失败: {e}")
        return None

# 模拟一个每 5 秒轮询一次的监控循环
if __name__ == "__main__":
    monitor_symbol = "AAPL.US"
    print(f"开始监控 {monitor_symbol} 盘口不平衡度...")
    
    try:
        for _ in range(5):  # 示例仅运行5次
            metrics = calculate_obi_5(monitor_symbol)
            if metrics:
                print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
                      f"标的: {metrics['symbol']} | "
                      f"买一/卖一: {metrics['bid_1']}/{metrics['ask_1']} | "
                      f"买五总量: {metrics['total_bid_vol']} | "
                      f"卖五总量: {metrics['total_ask_vol']} | "
                      f"OBI: {metrics['obi']}")
            time.sleep(5)
    except KeyboardInterrupt:
        print("监控已手动终止。")

from quantdash import QuantDash

qd = QuantDash(api_key="your_api_key_here")

# 批量获取 A 股（或者指定股票池）的最新实时价格
try:
    # 传入 universes 参数（支持 A 股市场 'CN_Stock'）
    quotes_df = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True)
    
    # 过滤出你关心的列进行展示
    if not quotes_df.empty:
        # 假设我们筛选出涨幅最大的前 5 只标的
        # 实际开发时可以利用这一步实现自选股监控或动态预警
        print("批量全市场快照获取成功，样本数据预览：")
        print(quotes_df[["symbol", "last_price", "change_percent", "volume"]].head())
        
except Exception as e:
    print(f"批量拉取行情失败: {e}")