JoinQuant 聚宽如何实现策略参数自动优化与批量回测？

在 JoinQuant（聚宽）平台上，虽然没有直接的“一键参数网格搜索”按钮，但你可以通过**研究环境（Research）**结合平台提供的 create_backtest API，非常灵活地实现自动设置参数、批量回测并寻找最优收益率的功能。

核心实现思路

利用 create_backtest 函数中的 extras 参数。extras 允许你传入一个字典，这个字典中的键值对会在回测初始化后自动赋值给策略的全局变量 g。因此，你可以：

1. 将策略代码写成字符串。
2. 在研究环境中利用 Python 的循环（如 itertools.product）生成所有可能的参数组合。
3. 遍历这些组合，通过 create_backtest 批量提交回测任务。
4. 使用 get_backtest 获取每个回测的结果（如累计收益率、夏普比率等），并进行排序，找出最优参数。

具体实现步骤与代码示例

以下是一个在研究环境中运行的完整示例，假设我们要优化一个双均线策略的快线（fast_length）和慢线（slow_length）参数：

1. 定义策略代码

将你的策略代码定义为一个多行字符串。注意在 initialize 中给参数设置默认值，这些值会被 extras 覆盖。

# 策略代码字符串
strategy_code = """
import jqdata

def initialize(context):
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    
    # 设置默认参数，这些会被 extras 覆盖
    if not hasattr(g, 'fast_length'):
        g.fast_length = 5
    if not hasattr(g, 'slow_length'):
        g.slow_length = 20
        
    g.security = '000300.XSHG'
    run_daily(market_open, time='every_bar')

def market_open(context):
    security = g.security
    fast_l = g.fast_length
    slow_l = g.slow_length
    
    # 获取历史价格
    close_data = attribute_history(security, slow_l + 1, '1d', ['close'])
    if len(close_data) < slow_l:
        return
        
    fast_ma = close_data['close'][-fast_l:].mean()
    slow_ma = close_data['close'][-slow_l:].mean()
    
    cash = context.portfolio.available_cash
    
    # 金叉买入，死叉卖出
    if fast_ma > slow_ma and context.portfolio.positions[security].closeable_amount == 0:
        order_value(security, cash)
    elif fast_ma < slow_ma and context.portfolio.positions[security].closeable_amount > 0:
        order_target(security, 0)
"""

2. 编写批量回测与寻优脚本

在研究环境中运行以下代码，生成参数网格并提交回测：

import itertools
import time
import pandas as pd
from jqdata import *

# 1. 定义参数网格
fast_lengths = [5, 10, 15]
slow_lengths = [20, 30, 60]

# 生成所有参数组合
param_grid = list(itertools.product(fast_lengths, slow_lengths))
# 过滤掉不合理的参数（如快线大于等于慢线）
valid_params = [(f, s) for f, s in param_grid if f < s]

print(f"共生成 {len(valid_params)} 组有效参数，开始批量回测...")

backtest_ids = []
results_list = []

# 2. 提交回测任务
for fast, slow in valid_params:
    # 构造 extras 参数，键名需与策略中 g.xxx 对应
    extra_vars = {'fast_length': fast, 'slow_length': slow}
    
    params = {
        "algorithm_id": "your_algorithm_id_here", # 随便填一个占位符，因为我们用 code 覆盖
        "start_date": "2020-01-01",
        "end_date": "2021-01-01",
        "frequency": "day",
        "initial_cash": 100000,
        "extras": extra_vars,
        "name": f"MA_Optimize_{fast}_{slow}"
    }
    
    # 提交回测
    bt_id = create_backtest(code=strategy_code, **params)
    backtest_ids.append({'bt_id': bt_id, 'fast': fast, 'slow': slow})
    print(f"已提交回测: 快线={fast}, 慢线={slow}, ID={bt_id}")
    time.sleep(1) # 避免请求过快

# 3. 等待回测完成并获取结果
print("等待回测完成...")
for item in backtest_ids:
    bt_id = item['bt_id']
    while True:
        gt = get_backtest(bt_id)
        status = gt.get_status()
        if status == 'done':
            # 获取风险指标（包含收益率等）
            risk = gt.get_risk()
            if risk:
                item['returns'] = risk.get('returns', 0)
                item['sharpe'] = risk.get('sharpe', 0)
                item['max_drawdown'] = risk.get('max_drawdown', 0)
            else:
                item['returns'] = None
            break
        elif status in ['failed', 'canceled', 'deleted']:
            print(f"回测 {bt_id} 失败或被取消")
            item['returns'] = None
            break
        time.sleep(5) # 每5秒检查一次状态

# 4. 分析并输出最优结果
results_df = pd.DataFrame(backtest_ids)
# 按收益率降序排列
results_df = results_df.sort_values(by='returns', ascending=False)

print("\n参数优化结果排名：")
print(results_df[['fast', 'slow', 'returns', 'sharpe', 'max_drawdown']])

best_param = results_df.iloc[0]
print(f"\n最优参数组合：快线={best_param['fast']}, 慢线={best_param['slow']}，最高收益率={best_param['returns']:.2%}")

注意事项

1. 积分消耗：批量回测可能会消耗较多平台资源。如果每个自然日内编译运行、回测超过免费时间，继续运行每30分钟需消耗2积分。可以在 create_backtest 中设置 use_credit=True 允许消耗积分。
2. 过拟合风险：参数优化（网格搜索）极易导致过拟合。建议将数据集分为样本内（In-Sample，用于寻找最优参数）和样本外（Out-of-Sample，用于验证该参数是否真的有效），不要盲目相信回测最高收益的参数组合。