BigQuant-SDK API 手册

由small_q创建，最终由small_q更新于2026-04-27 08:52 被浏览 707 用户

BigQuant Financial Quantitative Toolbox - 金融量化工具箱 Python SDK

目录

1. 顶层模块 (bigquant)
2. 数据模块 (dai)
3. 模拟交易模块 (papertrading)（暂未开放）
4. AIStudio 模块 (aistudio)（暂未开放）
5. 分布式计算模块 (fai)（暂未开放）
6. 错误处理

1. 顶层模块 (bigquant)

import bigquant

bigquant.init()

bigquant.init(
    *,
    ak: str | None = None,
    sk: str | None = None,
    token: str | None = None,
) -> None

显式初始化 SDK。

参数：

ak/sk 和 token 二选一，ak/sk 支持全部功能。

示例：

bigquant.init(ak="YOUR_AK", sk="YOUR_SK")

bigquant.init_from_config()

bigquant.init_from_config(path: str | None = None) -> None

从配置文件初始化 SDK。

参数：

异常：

• FileNotFoundError：配置文件不存在
• ValueError：配置文件中没有 ak/sk 或 token

bigquant.init_from_token()

bigquant.init_from_token(token: str) -> None

使用 Token 初始化 SDK（仅支持 HTTP 接口）。

bigquant.init_from_password()

bigquant.init_from_password(username: str, password: str) -> None

使用用户名密码登录初始化 SDK。

bigquant.whoami()

bigquant.whoami() -> dict | None

获取当前登录用户信息。

返回值：

{
    "id": 12345,
    "username": "your_name",
    "email": "you@example.com",
    # ... 其他字段
}

bigquant.close()

bigquant.close() -> None

关闭所有客户端连接（HTTP + Arrow Flight）。

2. 数据模块 (dai)

from bigquant import dai

dai.query()

dai.query(
    sql: str,
    *,
    full_db_scan: bool = False,
    filters: dict[str, list] | None = None,
) -> QueryResult

执行 SQL 查询，通过 Arrow Flight 直连云端数据（高性能）。

参数：

返回值： QueryResult

异常：

• AssertionError：Token 认证时使用 Arrow Flight 模式（需要 AK/SK）

示例：

# 基础查询（带 filters 避免全表扫描）
result = dai.query(
    "SELECT * FROM cn_stock_bar1d WHERE close > 10",
    filters={"date": ["2024-01-01", "2024-12-31"]},
)

QueryResult

dai.query() 的返回值，提供多种格式转换方法。

QueryResult.df()

result.df() -> pd.DataFrame

转换为 pandas DataFrame。重复列名自动添加 _N 后缀。

QueryResult.arrow()

result.arrow() -> pa.Table

返回 PyArrow Table。

QueryResult.pl()

result.pl() -> polars.DataFrame

转换为 Polars DataFrame。

QueryResult.fetchall()

result.fetchall() -> list[dict]

返回所有行的列表，每行为字典。

QueryResult.fetch_arrow_reader()

result.fetch_arrow_reader(batch_size: int = 1_000_000) -> pa.RecordBatchReader

获取流式读取器，适合大数据场景分批读取。

DataSource

数据源管理类，支持读写自定义数据。

from bigquant import dai

DataSource(datasource_id)

DataSource(datasource_id: str)

通过 ID 引用已有数据源。

DataSource.read_bdb()

ds.read_bdb(
    as_type: type = pa.Table,
    partition_filter: dict[str, tuple | set] | None = None,
    columns: list[str] | None = None,
) -> pd.DataFrame | pa.Table

读取数据。

参数：

示例：

ds = DataSource("my_signal_data")
df = ds.read_bdb(
    as_type=pd.DataFrame,
    partition_filter={
        "date": ("2024-01-01", "2024-12-31"),   # 范围
        "instrument": {"000001.SZ", "600519.SH"}, # 枚举
    },
    columns=["date", "instrument", "my_signal"],
)

DataSource.write_bdb()

DataSource.write_bdb(
    data: pd.DataFrame | pa.Table,
    *,
    id: str | None = None,
    partitioning: list[str] | None = None,
    indexes: list[str] | None = None,
    overwrite: bool = False,
    on_duplicates: str = "last",
    sort_by: list[tuple[str, str]] | None = None,
    docs: dict | None = None,
    timeout: int = 300,
    update_msg: str | None = None,
) -> DataSource

写入数据，返回 DataSource 对象。

参数：

异常：

• TypeError：data 类型不是 DataFrame 或 Table
• ValueError：data 为空

dai.search_datasources()

dai.search_datasources(
    keyword: str,
    category: str | None = None,
    page: int = 1,
    size: int = 20,
    fetch_all: bool = False,
) -> list[dict]

搜索平台数据表，支持中文/英文关键词。

参数：

dai.get_datasource_schema()

dai.get_datasource_schema(datasource_id: str) -> dict

获取数据表字段 schema。

返回值：

{
    "datasource_id": "cn_stock_bar1d",
    "cn_name": "股票日线行情",
    "description": "...",
    "fields": [
        {
            "name": "date",
            "type": "string",
            "description": "交易日期",
            "primary": True,
            "rank": 1,
        },
        # ...
    ]
}

dai.get_datasource_updates()

dai.get_datasource_updates(
    datasource_id: str,
    page: int = 1,
    size: int = 20,
) -> list[dict]

获取数据表更新记录。每条记录包含 created_at、data.rows、data.cols。

dai.list_datasources()

dai.list_datasources(
    constraints: dict | None = None,
    page: int = 1,
    size: int = 100,
    order_by: str = "-updated_at",
) -> list[dict]

获取当前用户创建的 DataSource 列表。

dai.list_user_datasources()

dai.list_user_datasources(
    constraints: dict | None = None,
    page: int = 1,
    size: int = 100,
    order_by: str = "-updated_at",
) -> list[dict]

获取当前用户订阅的 DataSource 列表。

dai.list_space_datasources()

dai.list_space_datasources(
    space_id: str | None = None,
    constraints: dict | None = None,
    page: int = 1,
    size: int = 100,
    order_by: str = "-updated_at",
) -> list[dict]

获取指定空间已上架的 DataSource 列表。

3. 模拟交易模块 (papertrading)

暂未开放：PaperTrading SDK 接口正在开发中，以下 API 文档仅供参考，暂不可用。

from bigquant import papertrading

papertrading.list()

papertrading.list(
    constraints: dict | None = None,
    order_by: list[str] | None = None,
    include_fields: list[str] | None = None,
    page: int = 1,
    size: int = 1000,
) -> dict

获取策略列表。

返回值：

{
    "items": [...],   # 策略列表
    "count": 10,      # 总数量
    "page": 1,
    "size": 20,
    "total": 1,       # 总页数
}

papertrading.get()

papertrading.get(strategy_id: str) -> Strategy

获取策略对象。

异常：

• ValueError：策略不存在

papertrading.delete()

papertrading.delete(strategy_id: str) -> dict

删除策略（含取消订阅数据、删除任务）。已分享的策略需先调用 unshare()。

papertrading.unshare()

papertrading.unshare(strategy_id: str) -> dict

取消策略分享。

Strategy 对象

通过 papertrading.get(strategy_id) 获取。

strategy.strategy_info

策略基本信息字典，包含 id、strategy_name、status、frequency、benchmark_instrument 等字段。

strategy.get_positions()

strategy.get_positions(
    constraints: dict | None = None,
    order_by: list[str] | None = None,
    page: int = 1,
    size: int = 100,
) -> dict

获取持仓信息。返回 {"items": [...], "count": N, ...}。

每条持仓包含：instrument、name、current_qty、cost_price、last_price、market_value、position_pnl、profit_ratio、posi_direction。

strategy.get_orders()

strategy.get_orders(
    constraints: dict | None = None,
    order_by: list[str] | None = None,
    page: int = 1,
    size: int = 100,
) -> dict

获取已成交订单。每条订单包含：instrument、direction（1=买/2=卖）、order_qty、order_price、average_price、filled_qty、status_msg。

strategy.get_planned_orders()

strategy.get_planned_orders(
    constraints: dict | None = None,
    order_by: list[str] | None = None,
    page: int = 1,
    size: int = 100,
) -> dict

获取计划订单（待交易信号）。

strategy.get_performances()

strategy.get_performances(fields: list[str] | None = None) -> list

获取绩效数据列表，每条为按字段顺序排列的列表：

4. AIStudio 模块 (aistudio)

暂未开放：AIStudio SDK 接口正在开发中，以下 API 文档仅供参考，暂不可用。

from bigquant import aistudio

aistudio.list()

aistudio.list(space_id: str = DEFAULT_SPACE_ID) -> list[dict]

获取可用资源规格列表。

返回值每条包含：

aistudio.start()

aistudio.start(
    resource_spec_id: str = DEFAULT_RESOURCE_SPEC_ID,
    space_id: str = DEFAULT_SPACE_ID,
) -> AIStudio

启动 AIStudio 实例。默认使用免费规格 D0（1C/6G）。

注意： 需要 AK/SK 认证，Token 认证不支持。

异常：

• RuntimeError：60 秒内无法启动

AIStudio 对象

通过 aistudio.start() 获取。

studio.run()

studio.run(
    script_or_code: str,
    *,
    is_code: bool = False,
    timeout: int = 300,
) -> str

在云端执行脚本或代码。

参数：

支持 .py 和 .ipynb 文件。

异常：

• ValueError：文件路径不是绝对路径
• FileNotFoundError：文件不存在
• RuntimeError：执行失败或超时

studio.close()

studio.close() -> None

关闭 AIStudio 实例，释放云端资源。

5. 分布式计算模块 (fai)

暂未开放：FAI 分布式计算 SDK 接口正在开发中，以下 API 文档仅供参考，暂不可用。

from bigquant import fai

需要额外安装：pip install 'bigquant[fai]' -i https://pypi.bigquant.com/simple/

fai.create_cluster()

fai.create_cluster(
    cluster_name: str,
    num_workers: int,
    worker_cpus: int,
    worker_memory: str | int,
    worker_gpus: int = 0,
    image: str | None = "",
    max_no_fai_run_time: int = 300,
) -> Cluster

创建分布式计算集群。

参数：

fai.get_cluster()

fai.get_cluster(cluster_id: str) -> Cluster

获取已有集群对象。

fai.list_clusters()

fai.list_clusters(all_space: bool = False) -> list[dict]

获取集群列表。

fai.remote

@fai.remote
def my_func(x):
    return x * 2

# 带参数
@fai.remote(num_cpus=2, memory="4G")
def heavy_func(data):
    ...

将函数标记为远程执行。支持 Ray 的所有 remote 选项（num_cpus、num_gpus、memory、max_retries 等）。

fai.get()

fai.get(object_refs) -> Any

获取远程函数执行结果（阻塞等待）。

fai.wait()

fai.wait(
    object_refs: list,
    num_returns: int = 1,
    timeout: float | None = None,
) -> tuple[list, list]

等待部分远程任务完成，返回 (ready, not_ready) 元组。

fai.put()

fai.put(value) -> ObjectRef

将对象存入分布式对象存储，返回引用。

fai.shutdown()

fai.shutdown() -> None

关闭 Ray 连接。

Cluster 对象

通过 fai.create_cluster() 或 fai.get_cluster() 获取。

cluster.init()

cluster.init() -> None

连接到集群（自动启动并等待就绪）。必须在使用 @fai.remote 前调用。

cluster.start_cluster() / stop_cluster()

cluster.start_cluster() -> dict
cluster.stop_cluster() -> dict

启动/停止集群。

cluster.delete_cluster()

cluster.delete_cluster() -> dict

删除集群，释放所有资源。

cluster.wait_cluster()

cluster.wait_cluster(
    status: str,
    min_available_workers: int = 1,
    timeout: int = 120,
) -> None

等待集群达到指定状态（"Running" 或 "Stopped"）。

6. 错误处理

常见 Python 异常：

\