用 Python 调用 WorldQuant Brain API 做 Alpha 回测

为什么要使用 API？

虽然 WorldQuant Brain 提供了网页端 Alpha 编写与回测功能，但网页端更适合单个 Alpha 的开发与调试，不支持循环、批量参数搜索、多进程回测等程序化研究流程。

对于以下场景：

批量生成 Alpha
自动化因子挖掘
参数扫描（Decay、Neutralization 等）
大规模回测实验
AI 自动生成 Alpha

使用 API 往往会更加高效。

本文结合 Python 示例代码，简要介绍 WorldQuant Brain API 的基本使用方法。

认证（Authentication）

认证接口：

1	https://api.worldquantbrain.com/authentication

Brain API 使用 HTTP Basic Authentication 进行身份认证，可以封装一个简单的登录函数：

def init_session(credentials_file='./brain_credentials.txt',
                 auth_token_ttl=DEFAULT_AUTH_TOKEN_TTL):
    """
    初始化 WorldQuant Brain API 会话连接
    """
    with open(expanduser(credentials_file)) as f:
        username, password = json.load(f)

    sess = requests.Session()
    sess.auth = HTTPBasicAuth(username, password)

    resp = sess.post(
        "https://api.worldquantbrain.com/authentication"
    )

    if resp.status_code not in (200, 201):
        raise RuntimeError(
            f"Authentication failed: "
            f"{resp.status_code} {resp.text}"
        )

    sess._auth_credentials_file = credentials_file
    sess._auth_time = time()
    sess._auth_ttl = auth_token_ttl

    return sess

其中 brain_credentials.txt 存储账号和密码：

[
    "your_email@example.com",
    "your_password"
]

认证成功后，服务器会返回一个带有效期的 Token。

当前返回的 Token 默认有效期约为 4 小时（14400 秒），因此长时间运行的程序通常需要定期重新认证。

提交 Alpha 回测

Brain 的网页端点击 Simulate 按钮，本质上也是向 Simulation 接口提交请求。

提交回测：

sim_resp = sess.post(
    "https://api.worldquantbrain.com/simulations",
    json=simulation_data
)

其中 simulation_data 包含：

回测设置（Settings）
Alpha 表达式（Formula）

例如：

def get_default_simulation(alpha_expr):
    return {
        "type": "REGULAR",
        "settings": {
            "instrumentType": "EQUITY",
            "region": "USA",
            "universe": "TOP3000",
            "delay": 1,
            "decay": 0,
            "neutralization": "INDUSTRY",
            "truncation": 0.08,
            "pasteurization": "ON",
            "unitHandling": "VERIFY",
            "nanHandling": "OFF",
            "language": "FASTEXPR",
            "visualization": False
        },
        "regular": alpha_expr
    }

1
2
3

simulation_data = get_default_simulation(
    "rank(-returns)"
)

提交后即可开始回测。

查询回测进度

提交 Simulation 后，服务器不会立即返回结果。

成功提交时，响应头中会包含一个：

Location

字段：

1	sim_progress_url = sim_resp.headers["Location"]

该地址用于查询当前回测任务状态。

随后需要轮询该地址：

while True:

    sim_progress_resp = sess.get(
        sim_progress_url
    )

    retry_after = float(
        sim_progress_resp.headers.get(
            "Retry-After", 0
        )
    )

    if retry_after == 0:
        break

    sleep(retry_after)

当任务完成后即可获取 Alpha ID：

alpha_id = (
    sim_progress_resp.json()
    .get("alpha")
)

Alpha ID 是后续获取结果的重要标识。

如果 Alpha ID 为空，通常表示任务未成功完成，需要结合返回状态码和错误信息进一步分析原因。

获取 Alpha 详细信息

获得 Alpha ID 后，可以进一步获取回测结果：

1
2
3

alpha_details = sess.get(
    f"https://api.worldquantbrain.com/alphas/{alpha_id}"
)

返回内容包括：

Sharpe
Fitness
Turnover
Returns
Drawdown
Checks

等统计指标。

批量获取数据字段（Data Fields）

除了提交回测之外，API 还允许获取平台支持的数据字段。

接口：

1	/data-fields

示例：

df = get_datafields(
    sess,
    search_scope,
    dataset_id="fundamental6"
)

其中：

search_scope

用于指定搜索范围：

{
    "instrumentType": "EQUITY",
    "region": "USA",
    "delay": 1,
    "universe": "TOP3000"
}

对应网页端中的：

Instrument Type
Region
Delay
Universe

dataset_id

用于指定数据集，例如：

fundamental6
analyst4
model16
news12

函数会自动分页获取全部字段并返回 DataFrame。

API 在 Alpha 挖掘中的应用

批量获取 Data Field 后，可以构建参数化 Alpha 模板：

1	group_rank(FIELD, industry)

然后自动替换：

FIELD

为不同数据字段：

earnings_yield
oper_margin
sales_growth
book_value
...

形成大量候选 Alpha。

例如：

group_rank(earnings_yield, industry)

group_rank(oper_margin, industry)

group_rank(sales_growth, industry)

再结合 API 自动提交回测，即可快速筛选具有潜力的 Alpha。

这种方式也是许多自动化 Alpha Mining 工具的重要基础。

API 的主要优势

相比网页端手动操作，API 具有以下优势：

批量回测

自动生成并提交大量 Alpha。

并行计算

普通账户通常允许同时运行多个 Simulation 任务，可以利用多进程提高研究效率。

参数扫描

自动测试不同：

Decay
Neutralization
Truncation
Universe

设置。

自动化 Alpha 挖掘

结合：

Python
LLM
因子模板

构建完整的 Alpha Mining Pipeline。