ETF_simple_rotation.py

# 克隆自聚宽文章：https://www.joinquant.com/post/61829
# 标题：ETF核心资产轮 动，非每 日 轮动 ，滑点0.003
# 作者：只做ETF

# ETF策略之核心资产轮动 - 详细解说
"""
策略名称：ETF核心资产动量轮动策略
策略作者：JoinQuant社区
策略类型：ETF动量轮动
策略特点：完全可实盘，风险可控，逻辑清晰

策略核心原理：
1. 投资标的：34只多资产类别ETF（宽基、行业、商品、债券、海外），实现资产配置分散化
2. 动量因子：基于加权线性回归计算年化收益率和R²判定系数
3. 轮动逻辑：每周三选择动量最强的3只ETF持有，实现强者恒强
4. 风险控制：动量得分安全区间过滤，避免追高杀跌
5. 交易频率：每周三14:50尾盘执行一次，大幅降低交易成本

实盘适用性评估：
✅ 完全可实盘 - 策略逻辑清晰，风险可控
✅ 交易成本合理 - 佣金设置符合实际
✅ 防未来函数 - 避免数据泄露
✅ 滑点控制 - 设置合理滑点
✅ 风险分散 - 多资产类别配置

策略优势：
1. 简单有效：动量因子在A股市场长期有效
2. 风险分散：投资不同资产类别，降低单一市场风险
3. 自动化交易：每周三自动执行，大幅减少过度交易
4. 成本可控：ETF交易成本低，流动性好
5. 频率优化：每周交易一次，极大降低交易成本，提高策略稳定性

注意事项：
1. 动量策略在趋势反转时可能产生较大回撤
2. 需要定期监控策略表现，适时调整参数
3. 建议配合其他策略使用，实现策略组合
"""

import numpy as np
import pandas as pd
import math
# from redistrade import *  # 实盘交易模块，回测环境不可用，已注释


#初始化函数 - 策略配置和参数设置
def initialize(context):
    """
    策略初始化函数，设置交易环境、参数和定时任务
    """
    # 设定基准 - 以沪深300指数作为业绩比较基准
    set_benchmark('000905.XSHG')
    # 用真实价格交易 - 避免使用复权价格，更贴近实盘
    set_option('use_real_price', True)
    # 打开防未来函数 - 防止使用未来数据，确保策略可实盘
    set_option("avoid_future_data", True)
    # 设置滑点 - 固定滑点0.1%，模拟实际交易中的价格冲击
    set_slippage(FixedSlippage(0.003))
    # 设置交易成本 - ETF交易成本较低，符合实际交易情况
    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 = [
        # === 宽基指数类 ===
        '510300.XSHG',  # 沪深300ETF - 大盘价值股
        '510500.XSHG',  # 中证500ETF - 中盘股
        '512100.XSHG',  # 中证1000ETF - 小盘股
        '562660.XSHG',  # 中证2000ETF - 微盘股
        '159915.XSHE',  # 创业板ETF - 成长股
        '159949.XSHE',  # 创业板50ETF - 创业板龙头
        '159967.XSHE',  # 创成长ETF - 创业板成长
        '588000.XSHG',  # 科创50ETF - 科创板龙头
        '588220.XSHG',  # 科创100ETF - 科创板中小盘
        '512890.XSHG',  # 红利低波ETF - 红利策略
        # === 行业主题类（动量有效）===
        '512660.XSHG',  # 军工ETF - 国防军工
        '512480.XSHG',  # 半导体ETF - 科技芯片
        '512760.XSHG',  # 半导体设备ETF - 半导体设备
        '516160.XSHG',  # 新能源ETF - 南方中证新能源
        '515030.XSHG',  # 新能源车ETF - 电动车产业链
        '515790.XSHG',  # 光伏ETF - 光伏产业
        '512690.XSHG',  # 酒ETF - 白酒消费
        '159928.XSHE',  # 消费ETF - 主要消费
        '512170.XSHG',  # 医疗ETF - 医疗健康
        '560080.XSHG',  # 中药ETF - 中药板块
        '562500.XSHG',  # 机器人ETF - 机器人产业
        '515070.XSHG',  # 人工智能ETF - AI产业
        '515880.XSHG',  # 通信ETF - 通信行业
        '512400.XSHG',  # 有色金属ETF - 有色金属
    ]
    # 已删除动量无效的ETF：
    # - 债券类(511010, 511090): 波动率太低，动量信号无效
    # - 商品类(518880黄金, 159985豆粕): 与股市低相关，动量逻辑不同
    # - 海外类(513100, 513500, 159920, 513180): T+1价格滞后，动量失真
    # - 窄基周期类(515220煤炭, 512800银行, 512880证券, 561260石油, 159981能源化工): 动量持续性差
    g.m_days = 25 # 动量参考天数 - 约1个月交易日的动量周期
    g.strategy = 'smallETF'  # 策略标识名
    
    # 定时任务设置
    # 每周三尾盘14:50执行交易，当日走势基本确定，避免盘中波动
    run_weekly(trade, weekday=3, time='14:50')  # 每周三14:50执行交易
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')

    
def MOM(etf):
    """
    动量因子计算函数 - 基于加权线性回归的动量评分
    核心思想：近期价格变化比远期价格变化更重要

    计算步骤：
    1. 获取历史价格数据
    2. 对价格取对数，转换为线性关系
    3. 使用加权线性回归拟合价格趋势
    4. 计算年化收益率和R²判定系数
    5. 综合评分 = 年化收益率 × R²

    参数：
    etf: ETF代码

    返回：
    score: 动量综合评分
    """
    # 获取25个交易日的收盘价数据
    try:
        df = attribute_history(etf, g.m_days, '1d', ['close'])
    except Exception as e:
        # ETF不存在或数据不足时，返回极低的动量得分
        print(f'无法获取 {etf} 的数据: {e}')
        return -999
    # 对价格取对数，使价格变化更符合线性关系
    y = np.log(df['close'].values)
    n = len(y)  
    x = np.arange(n)  # 时间序列 [0, 1, 2, ..., n-1]
    
    # 权重设置：近期数据权重更高（1-2线性递增）
    weights = np.linspace(1, 2, n)  # 线性增加权重，最近数据权重为2
    
    # 加权线性回归拟合价格趋势
    slope, intercept = np.polyfit(x, y, 1, w=weights)
    
    # 计算年化收益率：将日收益率转换为年化收益率
    annualized_returns = math.pow(math.exp(slope), 250) - 1
    
    # 计算R²判定系数，衡量趋势的稳定性
    residuals = y - (slope * x + intercept)  # 残差
    weighted_residuals = weights * residuals**2  # 加权残差平方和
    r_squared = 1 - (np.sum(weighted_residuals) / np.sum(weights * (y - np.mean(y))**2))
    
    # 综合评分：收益率 × 趋势稳定性
    score = annualized_returns * r_squared
    return score

def get_rank(etf_pool):
    """
    ETF动量排名函数 - 基于动量评分对ETF进行排序和筛选
    核心功能：计算所有ETF的动量评分，按得分排序，并应用安全区间过滤
    
    风险控制逻辑：
    - 得分>0：确保有正向动量
    - 得分<=5：避免动量过高（可能意味着泡沫或过度投机）
    - 如果所有ETF得分都不在安全区间，则空仓避险
    
    参数：
    etf_pool: ETF代码列表
    
    返回：
    rank_list: 符合条件的ETF排名列表（按动量从高到低）
    """
    score_list = []
    # 计算每个ETF的动量评分
    for etf in etf_pool:
        score = MOM(etf)
        score_list.append(score)
    
    # 创建DataFrame并排序
    df = pd.DataFrame(index=etf_pool, data={'score':score_list})
    df = df.sort_values(by='score', ascending=False)  # 按得分降序排列
    
    # 安全区间过滤：得分在(0, 5]范围内
    # 得分>0：确保正向动量，避免负向趋势
    # 得分<=5：避免动量过高，防止追高风险
    df = df[(df['score'] > 0) & (df['score'] <= 5)]
    
    rank_list = list(df.index)
    
    # 风险控制：如果所有ETF都不符合条件，则空仓避险
    if len(rank_list) == 0:
        rank_list = []  # 空仓，等待更好的入场时机
    
    return rank_list

def trade(context):
    """
    交易执行函数 - 每周三轮动交易逻辑
    核心策略：卖出非目标ETF，买入动量最强的ETF
    
    交易流程：
    1. 获取动量排名最高的ETF
    2. 卖出当前持仓中不在目标列表的ETF
    3. 买入目标ETF（如果当前未持有）
    
    风险控制：
    - 持有3只ETF，分散投资于动量最强的标的
    - 如果所有ETF都不符合条件，则空仓避险
    - 每周三交易一次，大幅降低交易频率
    """
    print("每周三14:50尾盘执行交易")

    # 策略配置：持有3只动量最强的ETF，分散风险
    target_num = 3 
    
    # 获取动量排名最高的ETF（最多取target_num只）
    target_list = get_rank(g.etf_pool)[:target_num]
    
    # 卖出逻辑：卖出不在目标列表中的持仓ETF
    hold_list = list(context.portfolio.positions)
    for etf in hold_list:
        if etf not in target_list:
            # 清仓卖出非目标ETF
            order_target_value(etf, 0)
            print('卖出' + str(etf))
        else:
            print('继续持有' + str(etf))
    
    # 买入逻辑：买入目标ETF（如果当前未持有）
    hold_list = list(context.portfolio.positions)  # 重新获取持仓列表（卖出后可能变化）
    if len(hold_list) < target_num:
        # 计算每只目标ETF的买入金额
        value = context.portfolio.available_cash / (target_num - len(hold_list))
        
        for etf in target_list:
            # 如果当前未持有该ETF，则买入
            if context.portfolio.positions[etf].total_amount == 0:
                order_target_value(etf, value)
                print('买入' + str(etf))

def after_market_close(context):
    """
    收盘后处理函数 - 每日收盘后执行清理工作
    """
    pass  # 实盘时用于关闭RedisTrade连接，回测环境无需处理