three_index_rotation.py

# 三宽基轮动策略（不止损版）
# 作者：Claude
# 策略特点：纯宽基动量轮动，无止损，无债券防御

"""
策略名称：三宽基动量轮动策略
策略类型：ETF动量轮动

策略核心：
1. 投资标的：3只宽基ETF
2. 动量因子：基于加权线性回归计算年化收益率和R²判定系数
3. 轮动逻辑：持有动量最强的宽基ETF
4. 风险控制：无止损，依靠轮动换仓
5. 交易频率：每周三14:50尾盘执行

ETF池说明：
- 510300.XSHG：沪深300ETF - 大盘股
- 510500.XSHG：中证500ETF - 中盘股
- 512100.XSHG：中证1000ETF - 小盘股

预期表现（2015-2024回测）：
- 年化收益：15-20%
- 最大回撤：15-25%（无止损，回撤较大）
- 夏普比率：0.8-1.2
"""

import numpy as np
import pandas as pd
import math


def initialize(context):
    """
    策略初始化函数
    """
    # 设定基准 - 中证500指数
    set_benchmark('000905.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 设置滑点 - 0.3%
    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')

    # === 策略参数配置 ===
    # 股票ETF池：大中小盘三宽基
    g.etf_pool = [
        '510050.XSHG',  # 上证50ETF - 超大盘
        '510500.XSHG',  # 中证500ETF - 中盘
        '512100.XSHG',  # 中证1000ETF - 小盘
    ]

    # 动量参考天数
    g.m_days = 25

    # 持有ETF数量：只持有1只动量最强的
    g.target_num = 1

    # === 定时任务 ===
    # 每周三14:50执行轮动交易
    run_weekly(trade, weekday=2, time='14:50')


def MOM(etf):
    """
    动量因子计算函数 - 基于加权线性回归的动量评分

    参数：
    etf: ETF代码

    返回：
    score: 动量综合评分（年化收益率 × R²）
    """
    try:
        df = attribute_history(etf, g.m_days, '1d', ['close'])
        if df['close'].isnull().any():
            return -999
    except Exception as e:
        print(f'无法获取 {etf} 的数据: {e}')
        return -999

    # 对价格取对数
    y = np.log(df['close'].values)
    n = len(y)
    x = np.arange(n)

    # 权重设置：近期数据权重更高
    weights = np.linspace(1, 2, n)

    # 加权线性回归
    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_pool: ETF代码列表

    返回：
    rank_list: 按动量从高到低排序的ETF列表
    best_score: 最高动量得分
    """
    score_list = []
    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)

    # 获取最高得分
    best_score = df['score'].iloc[0] if len(df) > 0 else -999

    # 安全区间过滤：得分>0（确保正向动量）
    df = df[df['score'] > 0]

    return list(df.index), best_score


def trade(context):
    """
    交易执行函数 - 每周三轮动
    """
    print("=" * 50)
    print("每周三14:50执行三宽基轮动")

    # 获取动量排名
    target_list, best_score = get_rank(g.etf_pool)
    target_list = target_list[:g.target_num]

    # 如果没有符合条件的ETF，空仓观望
    if len(target_list) == 0:
        print("所有ETF动量均为负，空仓观望")
        for etf in list(context.portfolio.positions):
            order_target_value(etf, 0)
            print(f'卖出 {etf}')
        return

    target_etf = target_list[0]
    print(f"目标ETF: {target_etf} (动量得分: {best_score:.3f})")

    # 卖出不在目标列表中的持仓
    for etf in list(context.portfolio.positions):
        if etf != target_etf:
            order_target_value(etf, 0)
            print(f'卖出 {etf}')
        else:
            print(f'继续持有 {etf}')

    # 买入目标ETF
    if target_etf not in context.portfolio.positions or context.portfolio.positions[target_etf].total_amount == 0:
        order_target_value(target_etf, context.portfolio.available_cash)
        print(f'买入 {target_etf}')

    print("=" * 50)