BigQuant-SDK 使用文档

由small_q创建，最终由small_q更新于2025-12-11 08:13 被浏览 576 用户

本文件提供 BigQuant Python API 的使用说明，包括

Bigquant 用户管理
Dai 数据管理
Papertrading 模拟交易管理
Account 账户管理
Bigtrader 交易管理
Aistuido 管理
Fai 分布式执行管理等功能。

安装方式

安装 bigquant-sdk（Pyhton）

加群获取最新版本安装包

前置代码（Python）

from bigquant import dai, papertrading, account, bigtrader, aistuido, fai

1 Bigquant 用户管理

1.1 默认用户首次登陆

命令： bq --save-auth --aksk ak.sk\n功能：默认用户首次使用 AK/SK 登陆\n参数：

ak.sk: str， bigquant 平台注册的API Keys

API Keys 获取步骤：
- 登录 BigQuant 平台：https://bigquant.com
- 点击 ”用户名“ 进入”个人中心“ → ”API Keys“
- 点击 ”新增访问凭证“，保存新 AKSK

返回：“✓ Config saved to XXX”

1.2 多用户登陆

方法名：bigquant.from_config()\n功能：除默认用户，登陆其他用户\n参数：

path：str，其他用户配置文件的绝对路径

配置文件（json 格式）示例：
```
{
  "auth": {
    "ak": "XXX",
    "sk": "XXX"
  }
}
```

返回：Bigquant 对象

注意：其他用户调用模块功能时，需完整按照 用户.模块名.函数 的方式

1.3 获取用户信息

方法名：Bigquant.whoami()\n功能：获取用户信息

参数：无

返回：dict（包含用户名、用户ID等）

2 Dai 数据管理

2.1 自定义 UDF 函数

类名： dai.DaiUDF

功能：自定义 UDF 函数定义

参数：

name : str，必填，UDF 函数名称
function : Callable，必填，Python 函数对象
parameters : Optional[List]，非必填，参数类型列表，默认为 None
return_type : Optional[Any]，非必填，返回值类型，默认为 None
type : Optional[Any]，非必填，UDF 类型，默认为 None
null_handling : Optional[Any]，非必填，NULL 值处理方式，默认为 None
exception_handling : Optional[Any]，非必填，异常处理方式，默认为 None
side_effects : bool，非必填，是否有副作用，默认为 False

说明：用于在 dai.query() 中传递自定义函数，SDK 会自动序列化函数代码并在云端执行

2.2 远程查询数据\n方法名：dai.query()\n功能：远程查询\n参数：

sql: str，必填，mysql 查询语句
udf_list：List[DaiUDF]，非必填，UDF 函数列表，默认为 [ ]
full_db_scan： bool，非必填，是否允许全表查询，默认为 False
filters：Dict[str, List[Any]]，非必填，过滤条件 {"column": ["value1", "value2"]}，默认为 {}
bind_relations：Dict[str, Any]，非必填，绑定本地数据到 SQL 查询 {"name": DataFrame/Table}，默认为 None
params：Dict[str, Any]，非必填，查询参数 {"param_name": "value"}，默认为 None
compression：bool，非必填，是否启用字符串压缩，默认为 False
resource_spec_id：str，AIStudio 资源规格 ID（SDK 专属），默认为 D0(1C/6G) 免费
space_id：str，AIStudio 空间 ID（SDK 专属），默认为主空间

返回：QueryResult 对象

2.3 数据源管理

类名：dai.DataSource

功能：数据源管理

参数：

datasource_id : str，必填，数据源 ID

方法：见下

2.3.1 读取 BDB 数据

方法名： dai.DataSource.read_bdb()

功能：读取 BDB 数据

参数：

as_type : Type，非必填，返回类型，默认为 pa.Table，支持 pd.DataFrame, pa.Table
partition_filter : Optional[Dict[str, Union[tuple, set]]]，非必填，分区过滤条件，默认为 None
- tuple: 表示范围，如 ("2024-01-01", "2024-12-31")
- set: 表示特定值，如 {"000001.SZ", "600000.SH"}
columns : Optional[List[str]]，非必填，要读取的列名列表，默认为 None（读取所有列）

返回：根据 as_type 参数返回 pd.DataFrame 或 pa.Table

2.3.2 写入 BDB 数据

方法名：dai.DataSource.write_bdb()

功能：写入 BDB 数据

参数：

• data : Union[pd.DataFrame, pa.Table]，必填，要写入的数据

• update_logs : Optional[Union[bool, Dict]]，非必填，是否记录更新日志，默认为 None

• update_msg : Optional[str]，非必填，更新备注信息，默认为 None

• id : Optional[str]，非必填，数据源 ID，默认为 None（创建临时数据源）

• partitioning : Optional[List[str]]，非必填，分区列，默认为 None

• indexes : Optional[List[str]]，非必填，索引列，默认为 None

• excludes : Optional[Set[str]]，非必填，排除字段，默认为 None

• unique_together : Optional[List[str]]，非必填，唯一约束，默认为 None

• on_duplicates : str，非必填，冲突处理策略，默认为 "last"，可选值：["last", "first", "error", "none"]

• sort_by : Optional[List[Tuple[str, str]]]，非必填，排序，格式为 [("field", "ascending/descending"), ...]，默认为 None

• preserve_pandas_index : bool，非必填，是否保留 pandas 索引，默认为 False

• docs : Optional[Dict[str, Any]]，非必填，文档，默认为 None

• timeout : int，非必填，写入锁的超时时间（秒），默认为 300

• extra : str，非必填，额外信息，默认为 ""

• base_ds : Optional[DataSource]，非必填，继承 extra 参数，默认为 None

• overwrite : bool，非必填，是否覆盖已有数据，默认为 False

• max_threads : Optional[int]，非必填，最大线程数，默认为 None

• preserve_order : bool，非必填，是否保持顺序，默认为 False

返回： DataSource 对象

2.4 查询结果

类名：dai.QueryResult

功能：查询结果包装类，提供多种数据格式转换方法

方法：见下

2.4.1 获取 Arrow Table

方法名：dai.QueryResult.arrow()

功能：返回 Arrow Table 格式的查询结果

参数：无

返回：pa.Table 对象

2.4.2 转换为 pandas DataFrame

方法名：dai.QueryResult.df()

功能：转换为 pandas DataFrame

参数：无

返回：pd.DataFrame 对象

2.4.3 转换为 Polars DataFrame

方法名：dai.QueryResult.pl()

功能：转换为 Polars DataFrame

参数：无

返回：polars.DataFrame 对象

2.4.4 获取所有行

方法名：dai.QueryResult.fetchall()

功能：获取所有行为列表

参数：无

返回：list，所有行的列表

2.4.5 获取流式读取器

方法名：dai.QueryResult.fetch_arrow_reader()

功能：获取 Arrow 流式读取器，用于大数据场景分批读取

参数：

batch_size : int，非必填，每批数据的行数，默认为 1000000

返回：pyarrow.RecordBatchReader 对象

3 strategy 模块：策略查询

3.1 获取策略列表\n函数：strategy.get_strategy(page=1, size=10)\n功能：分页获取策略列表\n参数：

page: int，非必填，页码
size: int，非必填，页容量\n返回值：dict（data 为策略列表，含分页信息）\n示例：

my_strategies = strategy.get_strategy()
print(my_strategies)

my_strategies_page2 = strategy.get_strategy(page=2, size=3)
print(my_strategies_page2)

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

strategy_id = input('请输入策略id:')

# 获取策略列表（默认首页）
res = strategy.get_strategy()
print(res)

# 获取第二页，每页3条
res = strategy.get_strategy(page=2, size=3)
print(res)

3.2 获取持仓详情\n函数：strategy.get_position(strategy_id=None, account_id=None, page=1, size=10, df=False)\n功能：根据策略 ID 或账户 ID 获取持仓详情\n参数：

strategy_id: str，与 account_id 二选一
account_id: str，与 strategy_id 二选一
page: int，非必填
size: int，非必填
df: bool，非必填，是否返回 DataFrame\n返回值：
df=False：dict
df=True：pandas.DataFrame\n示例：

positions_json = strategy.get_position(strategy_id='abcd1234', page=2, size=5)
print(positions_json)

positions_df = strategy.get_position(strategy_id='abcd1234', df=True)
print(positions_df)

返回字段：

account_type: 账户类型
account_id: 账户ID
trading_day: 交易日
exchange: 交易所
instrument: 代码
name: 代码名称
posi_direction: 持仓方向(1=多头, 2=空头)
current_qty: 持仓数量
available_qty: 可用数量
today_qty: 今日持仓数量
today_available_qty: 今日可用数量
cost_price: 持仓成本价格
last_price: 持仓最新价
market_value: 持仓市值
margin: 保证金占用
position_pnl: 盈亏金额
hedge_flag: 投保标记(1=投机)
sum_buy_value: 累计买入金额
sum_sell_value: 累计卖出金额
commission: 累计交易费用
dividend_qty: 当日分红数量
dividend_cash: 当日分红金额
open_date: 开仓日期
open_price: 开仓价格
settlement_price: 结算价
profit_ratio: 盈亏比率
initial_qty: 日初持仓数量
hold_days: 持有天数

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

strategy_id = input('请输入策略id:')

# 使用策略ID返回JSON格式
res = strategy.get_position(strategy_id=strategy_id)
print(res)

# 返回DataFrame格式
positions_df = strategy.get_position(strategy_id=strategy_id, df=True)
print(positions_df)

3.3 获取交易详情\n函数：strategy.get_order(strategy_id=None, account_id=None, daily_id=None, trading_day=None, page=1, size=10, df=False)\n功能：获取交易记录，可按策略ID、账户ID或实时策略ID查询

参数：

strategy_id: str，与 account_id、daily_id 三者择一
account_id: str，与 strategy_id、daily_id 三者择一
daily_id: str，实时策略ID，与其他两者择一
trading_day: str，非必填，格式支持 yyyyMMdd 或 yyyy-MM-dd（实时策略）
page: int，非必填
size: int，非必填
df: bool，非必填，是否返回 DataFrame

返回值：dict 或 DataFrame

示例：

orders_json = strategy.get_order(strategy_id=strategy_id)
print("策略ID查询结果（JSON格式）:")
print(orders_json)

orders_page = strategy.get_order(strategy_id=strategy_id, page=1, size=5)
print("\n策略ID分页查询结果:")
print(orders_page)

orders_df = strategy.get_order(strategy_id=strategy_id, df=True)
print("\n策略ID查询结果（DataFrame格式）:")
print(orders_df)

rt_orders_today = strategy.get_order(daily_id=daily_id)
print("\n实时策略今日交易查询结果:")
print(rt_orders_today)

rt_orders_on_day = strategy.get_order(daily_id=daily_id, trading_day='20250814')
print("\n实时策略指定日期交易查询结果:")
print(rt_orders_on_day)

realtime_orders_df = strategy.get_order(daily_id=daily_id, df=True)
print("\n实时策略查询结果（DataFrame格式）:")
print(realtime_orders_df)

\n返回字段：

strategy_id: 策略ID
account_type: 账户类型
account_id: 账户ID
trading_day: 交易日
exchange: 交易所
instrument: 代码
name: 代码名称
direction: 买卖方向(1=买入, 2=卖出)
offset_flag: 开平标志(0=开仓, 1=平仓, 2=平今)
order_qty: 委托数量
order_price: 委托价格
average_price: 成交均价
filled_qty: 成交数量
filled_dt: 最新成交时间
cancel_dt: 撤单时间
order_status: 委托状态（0=未成交, 1=部分成交, 2=全部成交, ...）
status_msg: 状态信息
order_type: 委托类型(0=限价, U=市价)
order_property: 委托属性
hedge_flag: 投保标记(1=投机)
order_dt: 委托时间
order_key: 本地订单标识
order_sysid: 交易所报单编号
entrust_no: 柜台委托编号
algo_order_id: 算法单号
commission: 交易费用
realized_pnl: 平仓盈亏

运行示例(可直接复制运行)：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

strategy_id = input('请输入普通策略id:')
daily_id = input('请输入实时策略id:')

res = strategy.get_order(strategy_id=strategy_id)
print(res)

orders_df = strategy.get_order(strategy_id=strategy_id, df=True)
print(orders_df)

res_today = strategy.get_order(daily_id=daily_id)
print("实时策略今日交易查询结果:")
print(res_today)

res_specific = strategy.get_order(daily_id=daily_id, trading_day='20250814')
print("\n实时策略指定日期交易查询结果:")
print(res_specific)

realtime_orders_df = strategy.get_order(daily_id=daily_id, df=True)
print("\n实时策略查询结果（DataFrame格式）:")
print(realtime_orders_df)

3.4 获取计划交易详情\n函数：strategy.get_planned_order(strategy_id=None, account_id=None, trading_day=Today, page=1, size=10, df=False)\n功能：根据策略或账户ID获取指定交易日的计划交易

参数：

strategy_id: str，与 account_id 二选一
account_id: str，与 strategy_id 二选一
trading_day: str，非必填，格式 yyyy-MM-dd，默认今天
page: int，非必填
size: int，非必填
df: bool，非必填，是否返回 DataFrame

返回值：

df=False：dict（包含计划交易列表及分页信息）
df=True：DataFrame

示例：

planned_orders_today = strategy.get_planned_order(strategy_id='abcd1234')
print("今日计划交易查询结果:")
print(planned_orders_today)

planned_orders_specific_day = strategy.get_planned_order(strategy_id='abcd1234', trading_day='2025-07-11', page=2, size=4)
print("\n指定日期计划交易查询结果:")
print(planned_orders_specific_day)

planned_orders_df = strategy.get_planned_order(strategy_id='abcd1234', df=True)
print("\n计划交易查询结果（DataFrame格式）:")
print(planned_orders_df)

\n返回字段：

planned_order_id: 计划订单ID
strategy_id: 策略ID
account_type: 账户类型
account_id: 账户ID
trading_day: 交易日
order_dt: 待下单时间
exchange: 交易所
instrument: 代码
name: 代码名称
direction: 买卖方向(1=买入, 2=卖出)
offset_flag: 开平标志(0=开仓, 1=平仓, 2=平今)
original_order_qty: 原始委托数量
order_qty: 委托数量
order_price: 委托价格
order_type: 委托类型(0=限价, U=市价)
order_status: 委托状态（0=未成交, 1=部分成交, 2=全部成交, ...）
status_msg: 状态描述
order_params: 下单条件参数
order_placed_dt: 实际发单时间
order_key: 本地订单标识
entrust_no: 委托号
algo_order_id: 算法单号
stop_loss_price: 止损价格
stop_profit_price: 止盈价格
created_at: 创建时间

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

strategy_id = input('请输入策略id:')


res_today = strategy.get_planned_order(strategy_id=strategy_id)
res_specific = strategy.get_planned_order(strategy_id=strategy_id, trading_day='2025-07-11', page=1, size=4)
planned_orders_df = strategy.get_planned_order(strategy_id=strategy_id, df=True)

print(res_today, res_specific, planned_orders_df)

3.5 获取策略绩效数据\n函数：strategy.get_performance(strategy_id)\n功能：获取策略绩效指标

参数：

strategy_id: str，必填

返回值：dict

示例：

perf = strategy.get_performance(strategy_id='abcd1234')
print(perf)

返回字段：

strategy_id: 策略ID
account_type: 账户类型
account_id: 账户ID
trading_day: 交易日
trading_day_index: 累计交易日
performance_type: 绩效类型
today_return: 当日收益率
cumulative_return: 累计收益率
annualized_return: 年化收益率
max_drawdown: 最大回撤
total_market_value: 持仓市值
total_margin: 保证金占用
balance: 当前现金
portfolio_value: 总资产
max_portfolio_value: 最大资产
benchmark: 基准数据
risk_indicator: 风险指标
today_sum_buy_value: 今日总买入金额
today_sum_sell_value: 今日总卖出金额
today_commission: 今日交易费用
total_buy_value: 累计买入金额
total_sell_value: 累计卖出金额
total_commission: 累计交易费用
grade: 策略分数
ten_day_return: 近十日收益率
week_return: 近一周收益率
month_return: 近一月收益率
three_month_return: 近三月收益率
six_month_return: 近六月收益率
year_return: 近一年收益率
total_close_count: 累计平仓次数
total_profit_count: 累计盈利次数
total_profit_value: 累计盈利金额
total_loss_value: 累计亏损金额
pnl_ratio: 盈亏比率
win_percent: 胜率
premium: 权利金收支
today_realized_pnl: 当日平仓盈亏

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

strategy_id = input('请输入策略id:')

res = strategy.get_performance(strategy_id=strategy_id)
print(res)

3.6 获取实时成交详情（实时策略）

函数：strategy.get_success(daily_id, trading_day=None, df=False)\n功能：根据实时策略ID查询成交明细，可指定交易日

参数：

daily_id: str，必填
trading_day: str，非必填，格式支持 yyyyMMdd 或 yyyy-MM-dd
df: bool，非必填，是否返回 DataFrame

返回值：dict 或 DataFrame

示例：

rt_trades_today = strategy.get_success(daily_id=daily_id)
print("实时策略今日成交查询结果:")
print(rt_trades_today)

rt_trades_on_day = strategy.get_success(daily_id=daily_id, trading_day='2025-08-14')
print("\n实时策略指定日期成交查询结果:")
print(rt_trades_on_day)

rt_trades_df = strategy.get_success(daily_id=daily_id, df=True)
print("\n实时策略成交查询结果（DataFrame格式）:")
print(rt_trades_df)

返回字段：

strategy_id：策略ID
account_type：账户类型
account_id：账户ID
trading_day：交易日
exchange：交易所
instrument：代码
name：代码名称
direction：买卖方向（1表示买入，2表示卖出）
offset_flag：开平标志（0表示开仓，1表示平仓，2表示平今）
filled_qty：成交数量
filled_price：成交价格
filled_money：成交金额
trade_type：成交类型
hedge_flag：投保标记
trade_dt：成交时间
trade_id：成交编号
order_key：本地订单标识
order_sysid：交易所报单编号
entrust_no：柜台委托编号
algo_order_id：算法单（母单）号
commission：交易费用
realized_pnl：平仓盈亏

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)


daily_id = input('请输入实时策略id:')

res_today = strategy.get_success(daily_id=daily_id)
res_specific = strategy.get_success(daily_id=daily_id, trading_day='2025-08-14')
res_df = strategy.get_success(daily_id=daily_id, df=True)

print(res_today, res_specific, res_df)

3.7 查看实时策略运行日志（实时策略）

函数：strategy.get_log(daily_id, df=False)\n功能：根据实时策略ID查询成交明细，可指定交易日

参数：

daily_id: str，必填
df: bool，非必填，是否返回 DataFrame

返回值：dict

示例

    # 使用实时策略ID查询日志流，用Ctrl+C退出后，返回DataFrame格式的历史日志
    res = strategy.get_log(daily_id=daily_id)
    print(res)

    # 使用实时策略ID查询日志流，用Ctrl+C退出后，返回DataFrame格式的历史日志
    logs_df = strategy.get_log(daily_id=daily_id, df=True)
    print(logs_df)

运行示例（可直接复制运行）：

from bigquant.api import user, strategy, run

# 登录（任选其一）

# 方式一：使用 AK/SK
keypair = input('请输入你的公钥.私钥:')
user.login(keypair=keypair)

# 方式二：使用用户名密码
# username = input("请输入用户名: ")
# password = input("请输入密码: ")
# user.login(username=username, password=password)

daily_id = input('请输入实时策略id:')

# 查看实时策略的日志(流式输出，按Ctrl+C结束)
res = strategy.get_log(daily_id=daily_id)
print(res)

# 查看实时策略的日志(流式输出，按Ctrl+C结束)，并在结束后自动按dataFrame格式打印历史日志
# logs_df = strategy.get_log(daily_id=daily_id, df=True)
# print(logs_df)

dai 模块可以后续开放吗？
DAI后续会开放
可以，虽然我觉得用线上也不错。
小q能看下我在平台建议里关于集合竞价因子在早上更新的建议吗。
SDK的权限需要申请吗？对哪些用户开放？