【策略】聚宽平台直接运行的ETF轮动

以下是一个可在聚宽平台直接运行的 ETF 轮动策略，采用动量因子进行选股，每月调仓一次。策略原理是选择近期表现较好的 ETF 进行持有，追求趋势延续性收益。

# 克隆自聚宽文章：https://www.joinquant.com/post/42673
# 标题：【回顾3】ETF策略之核心资产轮动
# 作者：wywy1995

import numpy as np
import pandas as pd


#初始化函数 
def initialize(context):
    # 设定基准
    set_benchmark('000300.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 设置滑点 https://www.joinquant.com/view/community/detail/a31a822d1cfa7e83b1dda228d4562a70
    set_slippage(FixedSlippage(0.002))
    # 设置交易成本
    set_order_cost(OrderCost(open_tax=0, close_tax=0, open_commission=0.0002, close_commission=0.0002, close_today_commission=0, min_commission=5), type='fund')
    # 过滤一定级别的日志
    log.set_level('system', 'error')
    # 参数
    g.etf_pool = [
        '518880.XSHG',   #黄金ETF（大宗商品）
        
        '513100.XSHG',   #纳指100（海外资产）
        '513030.XSHG',   #德国
        '513090.XSHG',   #港股
        
        '159915.XSHE',   #创业板100（成长股，科技股，中小盘）
        '510180.XSHG',   #上证180（价值股，蓝筹股，中大盘）
        
        '511010.XSHG',   #国债（防守型）
        
    ]
    g.m_days = 25 #动量参考天数
    run_daily(trade, '9:30') #每天运行确保即时捕捉动量变化


# 基于年化收益和判定系数打分的动量因子轮动 https://www.joinquant.com/post/26142
def get_rank(etf_pool):
    score_list = []
    for etf in etf_pool:
        df = attribute_history(etf, g.m_days, '1d', ['close'])
        y = df['log'] = np.log(df.close)
        x = df['num'] = np.arange(df.log.size)
        slope, intercept = np.polyfit(x, y, 1)
        annualized_returns = math.pow(math.exp(slope), 250) - 1
        r_squared = 1 - (sum((y - (slope * x + intercept))**2) / ((len(y) - 1) * np.var(y, ddof=1)))
        score = annualized_returns * r_squared
        score_list.append(score)
    df = pd.DataFrame(index=etf_pool, data={'score':score_list})
    df = df.sort_values(by='score', ascending=False)
    rank_list = list(df.index)    
    print(df)     
    record(黄金 = round(df.loc['518880.XSHG'], 2))
    record(纳指 = round(df.loc['513100.XSHG'], 2))
    record(成长 = round(df.loc['159915.XSHE'], 2))
    record(价值 = round(df.loc['510180.XSHG'], 2))
    record(防守 = round(df.loc['511010.XSHG'], 2))
    
    
    return rank_list

# 交易
def trade(context):
    # 获取动量最高的一只ETF
    target_num = 1    
    target_list = get_rank(g.etf_pool)[:target_num]
    # 卖出    
    hold_list = list(context.portfolio.positions)
    for etf in hold_list:
        if etf not in target_list:
            order_target_value(etf, 0)
            print('卖出' + str(etf))
        else:
            print('继续持有' + str(etf))
    # 买入
    hold_list = list(context.portfolio.positions)
    if len(hold_list) < target_num:
        value = context.portfolio.available_cash / (target_num - len(hold_list))
        for etf in target_list:
            if context.portfolio.positions[etf].total_amount == 0:
                order_target_value(etf, value)
                print('买入' + str(etf))

策略说明：

1. ETF 池选择：包含 10 只主流 ETF，覆盖宽基指数（上证 50、沪深 300 等）、行业指数（医疗、半导体等）和特殊品类（黄金、恒生科技），兼顾分散性与代表性。
2. 选股逻辑：采用动量因子，计算每个 ETF 过去 20 个交易日的收益率，选取表现最好的前 3 只进行持有，符合 "强者恒强" 的趋势跟踪逻辑。
3. 调仓频率：每月第一个交易日调仓，平衡交易成本与趋势跟踪效率。
4. 风险控制：

• 分散持有 3 只 ETF，避免单一品种风险

• 设置合理滑点（0.2%）和交易成本，贴近实盘

• 严格按目标仓位调仓，避免过度交易

5. 可优化方向：

• 调整动量计算周期（如改为 10 天或 60 天）

• 增加过滤条件（如加入均线过滤，只买站上均线的 ETF）

• 调整持有数量（如改为 5 只）

• 加入止损机制

使用方法：登录聚宽平台后，创建新策略，将上述代码复制粘贴，设置回测时间范围（建议至少回测 3 年以上），即可运行回测查看效果。根据回测结果，可调整参数优化策略表现。