聚宽平台如何构建动态ETF池并进行轮动？（附 Python 源码）

在聚宽（JoinQuant）量化交易平台上，构建动态ETF池并进行轮动是实现多资产配置和动量交易的经典策略。通过动态筛选高流动性、具备动量效应的ETF，可以有效规避单一资产的下行风险。

一、 核心设计思路

1. 动态筛选（流动性过滤）：利用 get_all_securities 获取所有场内ETF，并通过 get_bars 或 attribute_history 获取过去一段时间的日均成交额，剔除流动性差的“僵尸ETF”。
2. 分类筛选：通过聚宽的 get_security_info 识别标的类型，确保筛选出的是场内ETF（etf）。
3. 动量信号计算：计算候选ETF池中各标的在过去N个交易日（如20日）的涨幅（动量），进行降序排列。
4. 定期轮动与调仓：使用 run_monthly 或 run_weekly 定时运行调仓函数，卖出不在新推荐池中的ETF，买入动量最强的Top K只ETF，并使用 order_target_value 调整至等权重。

二、 聚宽策略源码实现

以下是完整的聚宽Python策略代码，展示了如何动态维护ETF池并按月进行动量轮动：

import pandas as pd
import numpy as np
from jqdata import *

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 设定交易税费：买入万分之三，卖出万分之三，无印花税（场内基金不收印花税）
    set_order_cost(OrderCost(close_tax=0, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='fund')
    
    # 策略参数设置
    g.momentum_window = 20  # 动量计算周期（天）
    g.top_k = 3             # 最终持仓的ETF数量
    g.min_daily_money = 10000000  # 过滤门槛：近5日日均成交额需大于1000万
    
    # 每月第一个交易日开盘运行调仓
    run_monthly(handle_rotation, monthday=1, time='9:30', reference_security='000300.XSHG')

def get_dynamic_etf_pool(context):
    """动态构建高流动性ETF池"""
    # 1. 获取所有上市的ETF基金
    all_funds = get_all_securities(['etf'], date=context.current_dt.date())
    etf_list = list(all_funds.index)
    
    active_etfs = []
    # 2. 过滤流动性（成交额）
    for etf in etf_list:
        # 获取近5天的成交额数据
        hist = attribute_history(etf, 5, '1d', ['money', 'paused'])
        if hist is not None and not hist.empty:
            # 排除停牌且日均成交额达标的ETF
            if hist['paused'].iloc[-1] == 0 and hist['money'].mean() >= g.min_daily_money:
                active_etfs.append(etf)
                
    return active_etfs

def handle_rotation(context):
    """轮动调仓逻辑"""
    # 1. 获取当前动态ETF池
    etf_pool = get_dynamic_etf_pool(context)
    if not etf_pool:
        log.warn("当前可交易ETF池为空！")
        return
        
    # 2. 计算动量信号（过去N天的收益率）
    momentum_scores = {}
    for etf in etf_pool:
        close_data = attribute_history(etf, g.momentum_window + 1, '1d', ['close'])
        if len(close_data) >= g.momentum_window:
            # 计算涨幅
            return_rate = (close_data['close'].iloc[-1] - close_data['close'].iloc[0]) / close_data['close'].iloc[0]
            momentum_scores[etf] = return_rate
            
    # 3. 按动量得分降序排列，选出Top K
    sorted_etfs = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    target_etfs = [item[0] for item in sorted_etfs[:g.top_k] if item[1] > 0] # 仅选择动量为正的ETF
    
    log.info(f"今日调仓，目标持仓ETF: {target_etfs}")
    
    # 4. 执行调仓：先卖出不在目标池中的持仓
    current_positions = list(context.portfolio.positions.keys())
    for etf in current_positions:
        if etf not in target_etfs:
            order_target(etf, 0)
            log.info(f"卖出不符合轮动条件的ETF: {etf}")
            
    # 5. 买入或调整目标ETF至等权重
    if len(target_etfs) > 0:
        target_value = context.portfolio.total_value / len(target_etfs)
        for etf in target_etfs:
            order_target_value(etf, target_value)
            log.info(f"调整ETF {etf} 仓位至目标价值: {target_value:.2f}")

三、 关键API与注意事项

• get_all_securities(['etf'], date)：动态获取指定日期在市的全部ETF，避免引入未来函数（如买入已退市或未上市的ETF）。
• attribute_history：在调仓日获取历史成交额与收盘价。注意，在日频策略中，该函数返回的数据不包含调仓日当天，确保了回测的真实性。
• 真实价格模式：在 initialize 中必须设置 set_option('use_real_price', True)。因为ETF存在分红或拆分，真实价格模式能准确处理账户中的现金和份额变动。