QuantDB Python SDK

QuantDB 量化数据平台官方 Python SDK，提供同步与异步两种客户端，封装了全部 API 端点。

安装

pip install quantdb-sdk

开发依赖：

pip install -e ".[dev]"

快速开始

API Key 鉴权（推荐）

from quantdb_sdk import QuantDBClient

client = QuantDBClient(api_key="qdb_xxx...")

# 查询用量
usage = client.get_usage()
print(f"已用: {usage['used_gb']:.2f} GB, 剩余: {usage['remaining_gb']:.2f} GB")

# K 线查询（免费）
df = client.query_kline(
    "600519.SH",
    adj_type="forward",
    start_date="2026-01-01",
    end_date="2026-07-22",
)
print(df.head())

用户名密码鉴权

client = QuantDBClient(username="admin", password="admin123")

核心功能

• 数据查询：K 线、Tick 通过下载 Parquet 切片后客户端解析（消耗流量）；股票列表、交易日历、元数据走网关 JSON（不计流量）。
• 数据下载：Parquet 文件下载或直读 DataFrame，计入订阅流量。
• 本地分析：基于 DuckDB 对本地 Parquet 执行 SQL。
• 账户管理：查询用户信息、用量、API Key、订阅与订单。
• 异步客户端：基于 httpx，适用于 asyncio 量化框架。

V1 / V2 数据布局

COS 同时保留 V1（按股票历史文件）和 V2（按交易日全市场分区）。所有下载相关接口均可传入 layout="auto" | "v1" | "v2"。默认 auto 的规则是：给出 K 线日期范围时优先 V2；若任一 交易日没有 V2 分区，则整次请求回退 V1，绝不混合两种口径；未给日期范围时读取 V1 全历史文件。

# 按日期范围优先 V2；覆盖不完整时自动回退 V1
df = client.query_kline("600519.SH", start_date="2026-07-01", end_date="2026-07-24")

# 强制指定物理布局；layout="v2" 缺日时会明确报错
latest = client.download_file("1", "daily_forward", trade_date="2026-07-24", layout="v2")
history = client.download_file("1", "daily_forward", symbol="600519.SH", layout="v1")

# 以发布清单为 cursor 做原子化增量同步（含 V2 patch）
result = client.sync_dataset("daily_forward", save_dir="D:/quantdb-data")

# 财务、ETF/可转债等尚无 V2 release 的数据集自动按 V1 Manifest 增量同步，
# 使用 ETag + size 校验；也可从命令行执行：
# quantdb sync qdb_xxx balance --save-dir D:/quantdb-data
financial = client.sync_dataset("balance", save_dir="D:/quantdb-data")

加速批量下载：建议从 8 个工作线程开始

sync_dataset() 单次调用会按发布顺序串行下载，以保证本地 SQLite 同步状态和 release cursor 一致。 当需要下载多个独立标的或文件时，可由调用方并行调度；建议先使用 8 个工作线程，再结合网络带宽、磁盘写入能力和账户流量配额调整。

from concurrent.futures import ThreadPoolExecutor
from quantdb_sdk import QuantDBClient

API_KEY = "qdb_xxx..."
symbols = ["600519.SH", "000001.SZ", "600036.SH"]

def download_one(symbol: str) -> str:
    # 每个 worker 使用独立客户端，避免跨线程共享 HTTP Session。
    with_client = QuantDBClient(api_key=API_KEY)
    return with_client.download_file(
        "1", "daily_forward", symbol=symbol, save_dir="D:/quantdb-data"
    )

with ThreadPoolExecutor(max_workers=8) as pool:
    files = list(pool.map(download_one, symbols))

不要对相同 save_dir 并行调用多个 sync_dataset()：它们会共同写入 quantdb_sync.sqlite，可能产生锁竞争。批量同步本身仍建议一次一个数据集执行。

流量说明

免费注册用户获赠 100 MB 一次性体验流量；订阅用户每月含 30 GB 下载流量，超出部分按 ¥1/GB 从账户余额扣减。余额不足时下载会被拦截。

文档

完整文档请访问：https://www.quantmindai.cn/docs/sdk.html