Python SDK 简介

quantdb-sdk 是 QuantDB 官方 Python SDK，当前版本 0.2.10，已发布至 PyPI，可直接 pip 安装。它封装全部 API 端点，提供同步和异步两种模式，支持 API Key 与用户名/密码两种鉴权方式。

安装

pip install quantdb-sdk

开发环境也可以从源码安装：

# 源码目录执行
pip install -e .

依赖会自动安装 requests、pandas、httpx、duckdb。开发测试依赖：

pip install -e ".[dev]"

初始化客户端

方式一：API Key（推荐）

import os
from quantdb_sdk import QuantDBClient

# 不要把 API Key 提交到源码；从部署环境读取
client = QuantDBClient(api_key=os.environ["QUANTDB_API_KEY"])

方式二：用户名密码

client = QuantDBClient(
    username="trader01",
    password="pass1234"
)

方法一览

账户信息

get_me — 当前用户信息

me = client.get_me()
print(me["username"], me["email"])

get_usage — 用量与订阅

返回已用流量、订阅限额、剩余可用量以及当前订阅状态。

usage = client.get_usage()
print(f"已用: {usage['used_gb']:.2f} GB")
print(f"限额: {usage['limit_gb']:.1f} GB")
print(f"剩余: {usage['remaining_gb']:.2f} GB")
print(f"订阅状态: {usage['subscription'].get('status')}")

API Key 管理

专业版月付和年付均支持 1 个有效 API Key。API Key 用于 SDK 与服务端鉴权。可通过 delete_api_key(key_id)（传入 list_api_keys() 返回的 ID）删除已有 Key 后重新创建。

# 列出已有 Key
keys = client.list_api_keys()
print(keys)

# 创建新 Key
new_key = client.create_api_key(description="量化服务器")
print(new_key["key"])

订阅与支付

SDK 内可直接查询套餐并创建支付宝订单。创建订单后会返回 pay_url，在浏览器中打开完成支付。

# 列出套餐
plans = client.list_plans()
for p in plans:
    print(p["id"], p["name"], p["amount_yuan"])

# 创建订单（示例：月付 ¥99）
order = client.create_order(plan_id="pro-monthly")
print(order["order_id"], order["pay_url"])

# 查询订单状态
status = client.get_order(order_id=order["order_id"])
print(status["status"])

V1 / V2 数据布局与流量策略

QuantDB 数据采用两种物理布局：V1（按股票的全历史文件 {Symbol}.parquet）和 V2（按交易日的全市场分区 dt=YYYYMMDD/data.parquet）。下载、Manifest、DataFrame、K 线和 Tick 接口均支持 layout="auto" | "v1" | "v2"。

V2 数据集（COS 纯 V2 按日分区）

以下数据集在 COS 上仅保留 V2 按日分区（dt=YYYYMMDD/data.parquet），不提供 V1 逐股票文件。V2 为按交易日切片，每个交易日更新：新增当日分区，历史修订通过 release patch 下发（sync_dataset() 自动应用）：

V1 数据集（逐股票文件，无 V2 分区）

以下数据集仅提供 V1 逐股票文件（{Symbol}.parquet），不支持 V2 按日分区。V1 为数据有变化才更新，sync_dataset() 自动使用 V1 Manifest 以 ETag + 文件大小做增量同步：

layout="auto" 是推荐默认值：优先 V2 按日分区，V2 覆盖不完整时回退 V1（仅对仍保留 V1 文件的数据集有效）。需要严格 V2 时用 layout="v2"（覆盖不完整会报错），需要强制逐股票文件时用 layout="v1"。

query_kline — K 线数据（下载）

下载 K 线 Parquet 切片后客户端解析，按下载流量计费。auto 始终优先 V2 按日分区：有日期范围时聚合多日分区并按标的、日期去重，覆盖不完整时回退 V1；无日期范围时走 V2 全量（manifest 列出的所有分区）。

def query_kline(
    symbol: str,
    adj_type: str = "unadjusted",
    start_date: str = None,
    end_date: str = None,
    fields: str = "open,high,low,close,volume,amount",
    limit: int = None,
    layout: str = "auto"
) -> pd.DataFrame

# 查询贵州茅台前复权日线
df = client.query_kline("600519.SH", adj_type="forward",
    start_date="2026-01-01", end_date="2026-07-22")
print(df.head())

# 只查收盘价和成交量
df = client.query_kline("000001.SZ", fields="close,volume", limit=10)

query_tick — Tick 数据（下载）

下载 Tick Parquet 切片后客户端解析，按下载流量计费。trade_date 按 8 位数字 YYYYMMDD 格式指定交易日，下载对应 {SYM}_{YYYYMMDD}.parquet。

df = client.query_tick(
    symbol="000001.SZ",
    trade_date="20260720",
    start_ts="09:30:00",
    end_ts="11:30:00",
    limit=5000
)

query_stock_list — 股票列表

df = client.query_stock_list(keyword="茅台")
print(df)

# 查全部
df = client.query_stock_list(limit=10000)

query_calendar — 交易日历

df = client.query_calendar(
    start_date="2026-01-01",
    end_date="2026-12-31"
)

preview_as_df — 数据预览

预览 Parquet 尾部数据，不产生流量费用。

df = client.preview_as_df(
    category_id="1",
    sub_category="daily_forward",
    symbol="600519.SH",
    limit=30
)

download_file — 下载文件

从 COS 流式下载原始 Parquet 切片到本地，返回文件保存路径。

# 下载到默认目录 (D:\QuantDB\downloads 或 ~/QuantDB/downloads)
path = client.download_file("1", "daily_forward", symbol="600519.SH")
print(f"保存到: {path}")

# 指定保存目录
path = client.download_file("1", "daily_forward",
    symbol="600519.SH", save_dir="./data")

load_as_df — 直读 DataFrame

直接将远端 Parquet 加载到内存 DataFrame，不落盘。适合中小数据集。

df = client.load_as_df("1", "daily_forward", symbol="600519.SH")
print(f"行数: {len(df)}, 列: {list(df.columns)}")

sync_dataset — 本地增量同步

以服务端 release cursor 增量下载 V2 daily 与 patch 文件；下载使用临时 .part 文件，SHA-256 校验和原子替换均成功后才推进 cursor。财务、ETF/可转债等尚无 V2 release 的数据集自动使用 V1 Manifest，并以 ETag + 文件大小校验增量同步。

# 首次同步会建立 quantdb_sync.sqlite 本地状态库
result = client.sync_dataset("daily_forward", save_dir="D:/quantdb-data")
print(result["layout"], len(result["downloaded"]))

# 财务 V1 同步：同样保存 SQLite 状态并支持断点重试
financial = client.sync_dataset("balance", save_dir="D:/quantdb-data")

# 异步客户端具有对等能力
result = await async_client.a_sync_dataset("daily_forward", "D:/quantdb-data")

query_local — 本地 SQL 查询

基于 DuckDB 对本地已下载的 Parquet 文件执行 SQL 查询。文件不存在时自动下载（首次调用会消耗下载流量），已下载后再次查询零流量费用。

# 仅传 WHERE 条件；SDK 固定查询已下载文件
df = client.query_local("close > 1500 AND volume > 1000000",
    "1", "daily_forward", symbol="600519.SH")
    

异步客户端

基于 httpx 的异步版本，适用于 asyncio 量化框架。方法名加 a_ 前缀。

from quantdb_sdk import AsyncQuantDBClient
import asyncio

async def main():
    client = AsyncQuantDBClient(api_key="qdb_abc...")

    # 并发查询多只股票
    tasks = [
        client.a_query_kline("600519.SH", adj_type="forward"),
        client.a_query_kline("000858.SZ", adj_type="forward"),
        client.a_query_stock_list(keyword="银行"),
    ]
    results = await asyncio.gather(*tasks)
    for df in results:
        print(df.head())

asyncio.run(main())

批量异步请求建议用 asyncio.Semaphore(8) 限制并发，避免触发 429 限流或占满本地网络与内存。

异常处理与重试

SDK 将 HTTP 错误映射为明确异常：AuthError（401/403）、InsufficientTrafficError（402）、ValidationError（422）、RateLimitError（429）和 ServerError（5xx）。认证、订阅和参数错误应直接提示用户；429/5xx 可使用指数退避后重试。

from quantdb_sdk import RateLimitError, ServerError

try:
    df = client.query_kline("600519.SH")
except (RateLimitError, ServerError):
    # 生产脚本应记录失败，并按指数退避重试
    raise

完整示例

from quantdb_sdk import QuantDBClient

# 1. 初始化
client = QuantDBClient(api_key="qdb_abc123...")

# 2. 查用量与订阅
usage = client.get_usage()
print(f"已用: {usage['used_gb']:.2f} GB")
print(f"免费额度: {usage['limit_gb']:.1f} GB")
print(f"信用额度: {usage['credit_gb']:.1f} GB")
print(f"剩余: {usage['remaining_gb']:.2f} GB")

# 3. 下载 K 线 Parquet（消耗流量）
df_kline = client.query_kline("600519.SH", adj_type="forward",
    start_date="2026-01-01", end_date="2026-07-22")

# 4. 查股票列表
df_stocks = client.query_stock_list(keyword="贵州")

# 5. 下载 Parquet 到本地（订阅含 30 GB/月，超出 ¥1/GB）
path = client.download_file("1", "daily_forward", symbol="600519.SH")

# 6. 本地分析
df_local = client.query_local(
    "close > 1500 AND trade_date >= '2026-06-01'",
    "1", "daily_forward", symbol="600519.SH"
)
print(df_local)

DuckDBWarehouse 本地数据仓库

SDK 内置 DuckDBWarehouse，支持将多个已下载的 Parquet 文件注册为 DuckDB View 视图，在本地毫秒级执行零流量 SQL JOIN 关联查询。

from quantdb_sdk import QuantDBClient, DuckDBWarehouse

client = QuantDBClient(api_key="qdb_abc123...")
wh = client.get_local_warehouse()

# 挂载 K 线与财务报表
wh.mount_dataset("1", "daily_unadjusted", symbol="600519.SH", view_name="kline")
wh.mount_dataset("3", "income", symbol="600519.SH", view_name="income")

# 零流量本地 JOIN 查询
df_join = wh.query("""
    SELECT k.trade_date, k.close, i.net_profit_excl_min_int_inc 
    FROM kline k 
    LEFT JOIN income i ON k.trade_date = i.m_anntime
""")
print(df_join.head())

365 维 ML 模型训练集拉取示例

# 直接将全量 365 维 AI/ML 因子训练宽表装载至内存 (带 ETag 进程防重缓存)
	df_factors = client.load_as_df(category_id="6", sub_category="l1_l2_factors", symbol="train_factors_2026q1")
	print(df_factors.shape)  # (175217, 367)

# L1 日频因子支持 V2 按日范围查询（高效，只下载所需日期分区）
df_l1 = client.load_as_df(category_id="6", sub_category="l1_factors",
    start_date="2026-01-01", end_date="2026-03-31", layout="v2")

# L2 高频因子（34 个交易日，2026-01-05 ~ 2026-02-27）
df_l2 = client.load_as_df(category_id="6", sub_category="l2_factors",
    start_date="2026-01-05", end_date="2026-02-27", layout="v2")

典型使用场景

场景一：均线策略回测

import pandas as pd

# 获取前复权日线
df = client.query_kline("600519.SH", adj_type="forward",
    start_date="2025-01-01", end_date="2026-07-22")

# 计算均线
df["ma5"] = df["close"].rolling(5).mean()
df["ma20"] = df["close"].rolling(20).mean()

# 生成交易信号
df["signal"] = 0
df.loc[df["ma5"] > df["ma20"], "signal"] = 1  # 买入
df.loc[df["ma5"] < df["ma20"], "signal"] = -1  # 卖出

# 计算策略收益
df["return"] = df["close"].pct_change()
df["strategy_return"] = df["signal"].shift(1) * df["return"]

cumulative = (1 + df["strategy_return"]).cumprod()
print(f"策略累计收益: {cumulative.iloc[-1]:.2%}")

场景二：多因子选股

# 读取估值数据
df = client.load_as_df("5", "valuation", symbol="600519.SH")

# 筛选低估值 + 高股息
filtered = df[(df["pe_ttm"] < 30) &
              (df["pb"] < 5) &
              (df["dividend_rate"] > 0.02)]

# 按 PE 排序
selected = filtered.sort_values("pe_ttm").head(10)
print(selected[["time", "pe_ttm", "pb", "dividend_rate"]])

场景三：Tick 数据高频分析

# 查询指定交易日 Tick 分笔数据
df = client.query_tick("000001.SZ", trade_date="20260720")

# 计算单笔成交金额变化（单位：万元）
df["amount_diff"] = df["amount"].diff()

# 筛选大单成交（例如单笔成交 > 10 万元）
df["big_order"] = df["amount_diff"] > 10.0

# 统计大单成交笔数占比
big_order_ratio = df["big_order"].mean()
print(f"大单成交笔数占比: {big_order_ratio:.2%}")

场景四：财务数据筛选

# 读取资产负债表
df = client.load_as_df("3", "balance", symbol="600519.SH")

# 计算关键指标
df["debt_ratio"] = df["tot_liab"] / df["tot_assets"]
df["current_ratio"] = df["total_current_assets"] / df["total_current_liability"]

# 筛选健康企业
healthy = df[(df["debt_ratio"] < 0.5) & (df["current_ratio"] > 1.0)]
print(f"健康记录数: {len(healthy)}")