JoinQuant平台如何解决create_backtest并发超过10个的排队问题？

在 JoinQuant（聚宽）平台的研究模块中，使用 create_backtest 批量创建回测是进行策略参数调优（Grid Search）的常见方法。然而，平台的服务器资源是有限的，通常会对单个用户的并发回测数量进行限制。

超过并发限制会怎样？

如果在研究环境中通过循环瞬间提交大量（如超过10个）的回测请求，通常会遇到以下情况：

1. 请求被拒绝或报错：API 可能会抛出异常，提示达到并发上限或请求过于频繁。
2. 资源排队超时：部分回测可能会一直处于 none（未开始）状态，甚至因为长时间获取不到计算资源而变为 failed。

如何实现排队创建回测？

JoinQuant 官方 API 并没有直接提供“自动排队等待”的参数，但我们可以利用 get_backtest(backtest_id).get_status() 方法，自己在 Python 代码中实现一个任务队列控制逻辑。

核心思路是：维护一个“正在运行”的回测 ID 列表，每次提交新回测前，检查当前正在运行的回测数量。如果达到了设定的并发上限（例如 5 个或 10 个），则让程序 time.sleep() 暂停等待，直到有回测完成（状态变为 done、failed 等）腾出空位后，再提交下一个回测。

Python 排队执行源码示例

以下是一个完整的批量参数调优并控制并发排队的示例代码，您可以直接在研究环境中使用：

import time
from jqdata import *

def run_queued_backtests(param_list, max_concurrent=5):
    """
    控制并发数量的批量回测函数
    :param param_list: 包含回测参数字典的列表
    :param max_concurrent: 最大允许的并发回测数量
    """
    active_backtests = []     # 记录正在运行或排队中的回测ID
    completed_backtests = []  # 记录已结束的回测ID

    for i, params in enumerate(param_list):
        # 1. 检查当前活跃的回测数量，如果达到上限则等待
        while True:
            still_active = []
            for bt_id in active_backtests:
                # 获取回测状态
                status = get_backtest(bt_id).get_status()
                # 'none': 未开始, 'running': 正在进行
                if status in ['none', 'running']:
                    still_active.append(bt_id)
                else:
                    completed_backtests.append(bt_id)
            
            active_backtests = still_active
            
            # 如果当前活跃任务数小于最大并发数，跳出等待循环，提交新任务
            if len(active_backtests) < max_concurrent:
                break
            
            # 达到并发上限，休眠一段时间后再次轮询检查
            print(f"当前并发数已达 {max_concurrent}，等待 30 秒后重试...")
            time.sleep(30) 
        
        # 2. 提交新的回测任务
        try:
            print(f"正在提交第 {i+1}/{len(param_list)} 个回测...")
            bt_id = create_backtest(**params)
            print(f"回测创建成功，ID: {bt_id}")
            active_backtests.append(bt_id)
        except Exception as e:
            print(f"创建回测失败: {e}")
            # 可以在这里加入失败重试逻辑
            
    # 3. 所有任务提交完毕后，等待最后一批任务执行完成（可选）
    print("所有回测任务已提交，等待最后一批执行完毕...")
    while len(active_backtests) > 0:
        still_active = []
        for bt_id in active_backtests:
            if get_backtest(bt_id).get_status() in ['none', 'running']:
                still_active.append(bt_id)
            else:
                completed_backtests.append(bt_id)
        active_backtests = still_active
        if len(active_backtests) > 0:
            time.sleep(30)
            
    print("批量回测全部完成！")
    return completed_backtests

# ================= 使用示例 =================
# 假设我们要测试 20 组不同的参数
params_to_test = []
for x in range(20):
    # 构造 create_backtest 需要的参数字典
    # 注意：extras 参数用于将外部变量传递给策略中的 g 对象
    bt_params = {
        "algorithm_id": "你的策略ID", 
        "start_date": "2023-01-01",
        "end_date": "2023-06-01",
        "frequency": "day",
        "initial_cash": 100000,
        "name": f"参数调优_测试组_{x}",
        "extras": {'my_param': x}  # 策略里可以通过 g.my_param 获取
    }
    params_to_test.append(bt_params)

# 执行批量回测，限制同时最多运行 5 个
# all_bt_ids = run_queued_backtests(params_to_test, max_concurrent=5)

注意事项

1. 积分消耗：如果您的免费回测时长已用完，create_backtest 可能会失败。如果愿意消耗积分，请在参数中加上 use_credit=True。
2. 轮询频率：建议 time.sleep() 的时间设置在 10~30 秒左右，不要设置得太短（如 1 秒），以免频繁调用 get_backtest 触发 API 频率限制。
3. 结果收集：回测完成后，您可以再次遍历 completed_backtests 列表，使用 get_backtest(id).get_risk() 等方法批量提取夏普比率、最大回撤等指标，从而实现自动化的参数寻优分析。