Python SDK 简介

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

安装

pip install quantdb-sdk

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

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

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

pip install -e ".[dev]"

初始化客户端

方式一：API Key（推荐）

from quantdb_sdk import QuantDBClient

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

方式二：用户名密码

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

自定义服务地址

client = QuantDBClient(
    api_host="http://your-server:5000",
    api_key="qdb_abc..."
)

方法一览

账户信息

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')}")

register / forgot_password / reset_password

# 注册
client.register(username="trader01", email="trader@example.com", password="your-pass")

# 忘记密码
client.forgot_password(email="trader@example.com")

# 使用邮件中的 token 重置密码
client.reset_password(token="reset-token-xxx", new_password="new-pass")

API Key 管理

专业版月付和年付均支持 1 个有效 API Key。API Key 用于 SDK 与服务端鉴权。可通过 delete_api_key(key_token) 删除已有 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"])

query_kline — K 线数据（下载）

下载 K 线 Parquet 切片后客户端解析，按下载流量计费。SDK 内部调用 load_as_df 下载整个 symbol 的日线 parquet，再按日期、字段过滤。

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 = 10000
) -> 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 切片后客户端解析，按下载流量计费。按交易日下载 {SYM}_{YYYYMMDD}.parquet。

df = client.query_tick(
    symbol="000001.SZ",
    trade_date="2026-07-20",
    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)}")

query_local — 本地 SQL 查询

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

# 方式一：传 WHERE 条件（自动补全 SELECT * FROM）
df = client.query_local("close > 1500 AND volume > 1000000",
    "1", "daily_forward", symbol="600519.SH")

# 方式二：传完整 SQL
df = client.query_local("SELECT AVG(close) as avg_close FROM data WHERE trade_date >= '2026-01-01'",
    "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())

完整示例

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)

典型使用场景

场景一：均线策略回测

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="2026-07-20")

# 计算买卖价差
df["spread"] = df["ask_price"].apply(lambda x: x[0]) - df["bid_price"].apply(lambda x: x[0])

# 计算大单成交
df["amount_diff"] = df["amount"].diff()
df["big_order"] = df["amount_diff"] > 1000000

# 统计大单占比
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)}")