shield_lock
金大哥
← 返回 Skill 列表
extension
分类: 数据与分析需要 API Key

BigQuant量化数据与策略回测

通过 BigQuant Python SDK 在本地 WorkBuddy 完成量化研究：云端金融数据读取、SQL 查询、因子分析与 BigTrader 本地回测。触发词：量化、量化交易、量化投资、量化选股、量化回测、宽客、因子选股、bigquant 数据查询、DAI 查询、bigquant SQL、搜索数据表、因子分析、A股数据、期货数据。

person作者: u_9e164d1ehubenterprise
download下载资源包

BigQuant 数据 + 回测技能（bigquant-data）

本技能让你在本地 WorkBuddy 环境中，仅凭安装 BigQuant SDK + 配置凭证，即可完成： 数据读取 → 数据查询 → 数据分析 → 策略回测 的完整量化研究工作流。

  • 数据侧：DAI（Data AI）模块，用 SQL 查询 BigQuant 云端 PB 级金融数据，通过 Arrow Flight 高速传输返回本地（pandas / polars / arrow）。
  • 分析侧：在 DAI 中完成计算后，把结果读取到本地做进一步的 pandas / polars 分析。
  • 回测侧：BigTrader 本地回测引擎，策略逻辑 100% 留在本地，云端只提供行情数据。

与 bigquant-sdk 技能的区别：本技能是聚焦且开箱即用的版本，自带可直接运行的 CLI 脚本， 适合在 WorkBuddy 里快速取数 / 分析 / 回测。两者可同时存在。

一、安装 SDK

要求 Python 3.11（Windows / macOS / Linux 均支持）。推荐使用托管 Python 3.11 + VS Code。

# 基础安装（数据查询 + 本地回测，固定版本）
pip install 'bigquant==0.1.11' 'bigquant-core==0.1.14' -i https://pypi.bigquant.com/simple/

# 本地回测（BigTrader 的 C++ 引擎核心包 bigtradercpp，必须显式安装）
pip install bigtradercpp -i https://pypi.bigquant.com/simple/

关键坑（必读）：上面已显式安装 bigtradercpp。若导入 from bigquant import bigtrader 仍报 ModuleNotFoundError: bigtradercpp（极少见，多见于环境被清理后），用官方包管理器重装即可：

bq pkg install bigtradercpp      # 官方方式，等价于 pip 但由 bq 管理
# 或兜底：直接 pip 安装
pip install bigtradercpp -i https://pypi.bigquant.com/simple/

用 bq pkg list 可查看受管理包的安装状态（含 bigtradercpp）。

安装后验证：

bq --version     # 输出版本即安装成功；若命令找不到，把 ~/.local/bin 加入 PATH

若 bq 命令找不到（macOS/Linux）：export PATH="$HOME/.local/bin:$PATH"

二、登录认证

需 BigQuant 凭证（AK/SK）。在 https://bigquant.com 登录 → 我的 > API Keys > 创建，获取 AK.SK 格式凭证。

方式一：API Key 认证（推荐）

bq auth --apikey <AK.SK>        # 明文传入
# 或交互式输入（不回显，更安全）
bq auth configure
# 验证登录状态
bq auth status

方式二：代码中初始化（已 bq auth 配置后，SDK 会自动从配置文件加载，无需显式调用）

import bigquant
bigquant.init_from_config()          # 默认读取 ~/.bigquant/config.json
# 或显式传 AK/SK
bigquant.init(ak="YOUR_AK", sk="YOUR_SK")
print(bigquant.whoami())             # 查看当前用户

配置文件位置：~/.bigquant/config.json（由 bq auth 自动创建维护）。 AK/SK 认证支持全部功能（含 Arrow Flight 高速传输）；Token 认证仅支持 HTTP，不支持 Arrow Flight。

三、数据来源（详见数据平台）

所有可用的数据表、字段、分区与更新说明，以 BigQuant 数据平台为准： 👉 https://bigquant.com/data/home

优先查 cn_stock_prefactors（预计算因子总表）：这是 BigQuant 集成大量常用衍生指标的「总表」，已预计算 估值（如 float_market_cap 流通市值、total_market_cap 总市值、pe_ttm、pb、ps_ttm、dv_ratio_ttm…）、 技术（如 return_1/return_5/return_20 区间收益、turnover_ratio 换手率、amplitude 振幅、avg_amount 均额…）等因子。 取数时先到这张表找目标字段；能直接取到就别去原始表（cn_stock_bar1d / cn_stock_valuation 等）重复计算。 只有当字段确实不在 cn_stock_prefactors 时，再去对应的原始表查询。

使用建议：

  • 先到数据平台浏览/搜索你想用的数据，确认字段在 cn_stock_prefactors 还是其他表，拿到准确的表名与字段名。
  • 拿到表名后，可用 SDK 在本地查看 schema / 更新记录（见第五节），或直接写 SQL 查询。
  • 数据平台未涵盖的私有数据，可用 dai.DataSource 读写自定义数据（见第五节）。

本技能不内置「数据表清单」——数据字典随平台更新，请以数据平台实时内容为准。 字段是否已在 cn_stock_prefactors 预计算，请以数据平台该表 schema 为准（也可用 scripts/search_table.py cn_stock_prefactors 查看）。

四、核心取数原则：在 DAI 中计算，再读取

这是本技能最重要的使用约定。 取数时，尽量把筛选、字段计算、聚合、窗口函数、JOIN 全部写进 SQL，交给 DAI 在云端分布式计算；只在最后一步用 .df() 把计算结果读取到本地。

✅ 推荐（计算在云端，只拉结果）：

df = dai.query("""
    SELECT date, instrument,
           close / LAG(close, 5) OVER (PARTITION BY instrument ORDER BY date) AS return_5,
           CORR(volume, close) OVER (PARTITION BY instrument) AS vol_close_corr
    FROM cn_stock_bar1d
    WHERE date BETWEEN '2024-01-01' AND '2024-12-31'
""", filters={"date": ["2024-01-01", "2024-12-31"]}).df()

❌ 避免（先拉全表到本地再算，慢且费流量）：

# 不要这样：拉整张表到本地再用 pandas 过滤/计算
df = dai.query("SELECT * FROM cn_stock_bar1d").df()
df = df[df["date"] >= "2024-01-01"]

要点：

  1. 字段优先从 cn_stock_prefactors 取：估值 / 技术 / 衍生因子先查这张预计算总表；没有再查原始表（如 cn_stock_bar1d、cn_stock_valuation、cn_stock_fundamentals）。优先用预计算因子，避免重复计算、更快更省。
  2. 有日期/代码范围时，务必同时传 filters 参数做分区过滤，避免全表扫描（否则报错或极慢）。
  3. 分区过滤字段通常是 date、instrument，需与 WHERE 条件一致。
  4. 计算量大的衍生指标、因子、聚合，优先在 SQL 里完成。
  5. 本地只拿到「已经算好的结果」做轻量分析或喂给回测。

五、数据查询（DAI）

DAI 是核心数据模块，通过 SQL 查询云端数据，支持 Arrow Flight 高性能传输。本地导入： from bigquant import dai（网页端用 import dai）。

基础查询（带 filters）

from bigquant import dai

df = dai.query("""
    SELECT date, instrument, open, high, low, close, volume
    FROM cn_stock_bar1d
    WHERE date >= '2024-01-01' AND date <= '2024-03-31'
""", filters={"date": ["2024-01-01", "2024-03-31"]}).df()

print(df.head())

重要：查询有日期范围时，务必同时传 filters，避免全表扫描。

结果格式转换

result = dai.query("SELECT * FROM cn_stock_bar1d LIMIT 100", full_db_scan=True)

df      = result.df()              # pandas DataFrame
table   = result.arrow()           # PyArrow Table
pl_df   = result.pl()              # Polars DataFrame
rows    = result.fetchall()        # Python 列表（每行为 dict）
# 大数据流式读取（节省内存）
reader  = result.fetch_arrow_reader(batch_size=10000)
for batch in reader:
    print(batch.num_rows)

过滤条件（filters）

# 日期范围过滤
df = dai.query("SELECT * FROM cn_stock_bar1d WHERE close > 10",
               filters={"date": ["2024-01-01", "2024-12-31"]}).df()

# 多字段过滤
df = dai.query("SELECT * FROM cn_stock_bar1d",
               filters={"date": ["2024-01-01", "2024-12-31"],
                         "instrument": ["000001.SZ", "600519.SH"]}).df()

DataSource 读写（自定义数据）

import pandas as pd
from bigquant import dai

# 写入（需开通写入权限）
ds = dai.DataSource.write_bdb(
    data=my_df, id="my_signal_data", partitioning=["date"], overwrite=True,
)
# 读取
ds = dai.DataSource("my_signal_data")
df = ds.read_bdb(as_type=pd.DataFrame,
                 partition_filter={"date": ("2024-01-01", "2024-01-31")},
                 columns=["date", "instrument", "my_signal"])

搜索与浏览数据表

from bigquant import dai

results = dai.search_datasources("股票日线")          # 搜索（中/英/表名/描述）
for item in results:
    print(item.get("id"), item.get("docs", {}).get("cn_name"))

schema = dai.get_datasource_schema("cn_stock_bar1d")   # 查看 schema
for f in schema["fields"]:
    print(f["name"], f["type"], f["description"])

updates = dai.get_datasource_updates("cn_stock_bar1d") # 更新记录
my_ds = dai.list_datasources()                         # 我创建的 DataSource

也可用内置脚本一键搜索：scripts/search_table.py "股票日线"。

六、数据分析

遵循「第四节：在 DAI 中计算，再读取」——把重计算放 SQL，本地只做轻量分析。

from bigquant import dai
import pandas as pd

# 1) DAI 中算好因子/指标，读取结果
df = dai.query("""
    SELECT date, instrument, close,
           close / LAG(close, 1) OVER (PARTITION BY instrument ORDER BY date) - 1 AS ret
    FROM cn_stock_bar1d
    WHERE date BETWEEN '2024-01-01' AND '2024-12-31'
""", filters={"date": ["2024-01-01", "2024-12-31"]}).df()

# 2) 本地做分析（pandas / polars 均可）
print(df.groupby("instrument")["ret"].describe())
print("全市场日收益相关系数:\n", df.pivot(index="date", columns="instrument", values="ret").corr())

内置脚本 scripts/analyze.py 封装了上面的流程（传 SQL + filters，自动 .df() 后输出统计/CSV，可选绘图）：

python scripts/analyze.py \
  --sql "SELECT date, instrument, close FROM cn_stock_bar1d WHERE date>='2024-01-01'" \
  --filters date:2024-01-01:2024-12-31 \
  --output analyzed.csv

七、本地回测（BigTrader）

BigTrader 是本地回测引擎，策略逻辑完全在本地运行（隐私好、响应快）。本地导入： from bigquant import bigtrader。

基础回测

from bigquant import bigtrader

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0003, min_cost=5.0, tax_ratio=0.0005))
    context.security = context.instruments[0]

def handle_data(context, data):
    if context.security not in context.portfolio.positions:
        context.order_target_percent(context.security, 1.0)

performance = bigtrader.run(
    market=bigtrader.Market.CN_STOCK,     # 或字符串 "cn_stock"
    frequency=bigtrader.Frequency.DAILY,  # 或 "1d"
    start_date="2024-01-01",
    end_date="2024-01-31",
    instruments=["000001.SZ"],
    capital_base=1_000_000,
    benchmark="000300.SH",
    initialize=initialize,
    handle_data=handle_data,
)

print(performance.summary)               # 绩效摘要（dict）
print(performance.raw_perf)              # 每日绩效 DataFrame

查看回测结果

s = performance.summary
print("夏普比率:", s["sharp_ratio"])
print("最大回撤:", s["max_drawdown"])
print("年化收益:", s["annual_return_ratio"])
print("累计收益:", s["return_ratio"])
print("胜率:", s["win_ratio"])

多资产 / 多标的策略

def initialize(context):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0003))
    context.stocks = ["000001.SZ", "000002.SZ", "600519.SH"]

def handle_data(context, data):
    for s in context.stocks:
        context.order_target_percent(s, 1.0 / len(context.stocks))

performance = bigtrader.run(
    market="cn_stock", frequency="1d",
    start_date="2024-01-01", end_date="2024-01-31",
    instruments=context_stocks, initialize=initialize, handle_data=handle_data,
)

示例：小市值策略（scripts/small_cap.py）

内置 scripts/small_cap.py 是一个可直接运行的小市值策略模板：在 DAI 中按 total_market_cap 升序选最小的 N 只 A 股，每 N 个交易日等权调仓。

python scripts/small_cap.py -s 2024-01-01 -e 2026-01-21 -n 20 -r 5 -c 1000000

要点：

  • 选股与过滤全部在 DAI（cn_stock_prefactors）中完成：用该表原生字段 st_status=0（排除 ST）、suspended=0（排除停牌）、list_sector IN (1,2)（仅主板）、list_days>365（排除次新股）、float_market_cap>5e8 等做股票池过滤；按 total_market_cap 升序取最小 N 只（小市值）；再用 bigtrader.TradingDaysRebalance 做调仓、内置 HandleDataLib.handle_data_weight_based 等权下单。
  • 数据筛选口径严格对齐官方模板（https://bigquant.com/wiki/doc/vHlB8S23iJ）：ST / 停牌 / 板块 / 上市天数均用 cn_stock_prefactors 原生字段，不再 JOIN 其他表 —— 符合「优先用 prefactors」原则。
  • 改动入口：--stock-num（每期只数）、--rebalance-days（调仓间隔）、--start-date/--end-date/--capital-base；因子与过滤条件在脚本 SQL 内改。

通用参数化模板 scripts/backtest.py 仍保留（支持 -i/-s/-e/-c/-m/-f/-b），适合从零写策略。 完整 API（回调、context/data 接口、枚举、撮合规则）见 references/bigtrader_api.md。

八、组合工作流（取数 → 分析 → 回测）

典型闭环：用 DAI 选出股票池 / 算好因子 → 本地分析验证 → 把标的/信号喂给 BigTrader 回测。

from bigquant import dai, bigtrader

# 1) DAI：取动量因子 Top-N 股票（优先用 cn_stock_prefactors 的预计算 return_20）
universe = dai.query("""
    SELECT instrument, return_20 AS mom_20
    FROM cn_stock_prefactors
    WHERE date = '2024-12-31'
    QUALIFY ROW_NUMBER() OVER (ORDER BY mom_20 DESC) <= 10
""", filters={"date": ["2024-12-31", "2024-12-31"]}).df()
top_stocks = universe["instrument"].tolist()

# 2) 本地轻量分析（可选）
print("入选股票:", top_stocks)

# 3) BigTrader：用选出的股票等权回测
def initialize(context):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0003))
    context.stocks = context.instruments

def handle_data(context, data):
    for s in context.stocks:
        context.order_target_percent(s, 1.0 / len(context.stocks))

performance = bigtrader.run(
    market="cn_stock", frequency="1d",
    start_date="2025-01-01", end_date="2025-06-30",
    instruments=top_stocks, capital_base=1_000_000,
    benchmark="000300.SH", initialize=initialize, handle_data=handle_data,
)

九、常见问题

Q：bq 命令找不到？ 把 ~/.local/bin 加入 PATH（macOS/Linux）：export PATH="$HOME/.local/bin:$PATH"。

Q：查询报错 full_db_scan=False but no filters provided？ 提供 filters 参数，或设置 full_db_scan=True：

df = dai.query(sql, filters={"date": ["2024-01-01", "2024-12-31"]}).df()  # 推荐
df = dai.query(sql, full_db_scan=True).df()                              # 允许全表扫描（慢）

Q：导入 bigtrader 报 ModuleNotFoundError: bigtradercpp？ 见第一节「关键坑」：bq pkg install bigtradercpp（或 pip install bigtradercpp -i https://pypi.bigquant.com/simple/）。

Q：本地 SDK 与网页端 AIStudio 区别？ 几乎一致。主要区别：网页端 import dai，本地 from bigquant import dai；网页端无需初始化，本地需 bq auth 配置凭证；本地 dai.query 默认走 Arrow Flight 直连（更快）。

Q：AK/SK 与 Token 认证区别？ AK/SK 支持全部功能（含 Arrow Flight）；Token 仅支持 HTTP 接口，不支持 Arrow Flight 直连。

十、相关资源

  • 数据平台（数据来源总入口）：https://bigquant.com/data/home
  • DAI 文档：https://bigquant.com/wiki/doc/PLSbc1SbZX
  • SDK 使用文档：https://bigquant.com/wiki/doc/vac4qwmQr4
  • BigTrader 引擎完整 API：references/bigtrader_api.md（本技能内置）
  • AK/SK 获取：https://bigquant.com 登录 → 我的 > API Keys