2422 ETF 动量轮动止损策略 带上止损的核心资产轮动才更安心，低回撤，高收益率 47966

ETF 动量轮动止损策略

核心逻辑

1. 动量因子计算：

• 计算窗口：25 日历史数据
• 双因子评分：
  • 年化收益因子：对数收盘价线性回归斜率年化值
  • 趋势质量因子：判定系数 R²（衡量趋势稳定性）
• 最终得分：年化收益 × R²，筛选出兼具高收益与稳定趋势的 ETF

2. 持仓管理规则：

• 买入条件：
  • 动量得分最高且满足均值突破（当前价 > 4 日均线）
  • 空仓时恢复买入需满足前一日动量方向
• 卖出条件：
  • 调仓信号：当日最优 ETF 变化
  • 止损条件：单日跌幅超 4% 或跌破 10 日最高价
• 特殊处理：记录前一日持仓，避免频繁调仓

3. 风险控制体系：

• 动态止损：持仓 ETF 触发「当前价 < 10 日最高价 ×0.96」时强制平仓
• 趋势延续验证：空仓后需确认标的仍处 4 日均线上方才恢复建仓
• 仓位管理：全仓单品种持有，降低摩擦成本

核心指标：

• 动量得分 = 年化收益斜率 × 趋势判定系数
• 趋势健康度 = 当前价 / 10 日最高价
• 短期动能 = 当前价 / 4 日均线

 

策略特点：

1. 多维度趋势验证：同时考量收益率、趋势稳定性、短期动能
2. 差异化风控：持仓品种适用严格止损，候选品种需突破确认
3. 资产配置特性：覆盖黄金（抗通胀）、纳指（科技成长）、创业板（A 股成长）、180 指数（A 股价值）四大类资产
4. 交易成本优化：使用 ETF 作为标的，佣金设置匹配场内基金标准
5. 防止过度交易：通过g.etf_pre变量避免次日立即反向操作

策略代码

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.000))
    # 设置交易成本
    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（海外资产）
        '159915.XSHE', #创业板100（成长股，科技股，中小盘）
        '510180.XSHG', #上证180（价值股，蓝筹股，中大盘）
    ]
    g.m_days = 25 #动量参考天数
    g.etf_info = initial_etf_info(g.etf_pool, 100)
    g.etf_pre = None
    run_daily(update_etf_table, '7:00') #每天运行确保即时捕捉动量变化
    run_daily(trade, '9:30') #每天运行确保即时捕捉动量变化

# 初始化股票信息
def initial_etf_info(etf_pool, ndays=100):
    etf_info = {}
    for etf in etf_pool:
        etf_df = attribute_history(etf, ndays,'1d',['close'],skip_paused=True, df=True, fq='pre')
        etf_info[etf] = list(etf_df.close)
    return(etf_info)

def update_etf_table(context):
    for etf in g.etf_pool:
        etf_df = attribute_history(etf, 1,'1d',['close'],skip_paused=True, df=True, fq='pre')
        g.etf_info[etf].append(etf_df.close[-1])
        g.etf_info[etf].pop(0)

关键函数解锁后查看：

# 交易
def trade(context):
    # 获取动量最高的一只ETF
    sel_etf = get_rank(g.etf_pool)[0]
    hold_etf = None
    # 卖出    
    hold_list = list(context.portfolio.positions)
    if(len(hold_list)>0): hold_etf = hold_list[0]
    sel_worth = evaluate_etf_worth(sel_etf, 0)
    hold_worth = evaluate_etf_worth(hold_etf, 1)
    print(f"持有{hold_etf}={hold_worth}, 选择{sel_etf}={sel_worth}")
    if(hold_etf != None):
        g.etf_pre = hold_etf
        if sel_etf not in hold_list:
            order_target_value(hold_etf, 0)
            print('调仓卖出' + str(hold_etf))
        elif hold_worth==0:
            order_target_value(hold_etf, 0)
            print('止损卖出' + str(hold_etf))
        else:
            print('继续持有' + str(hold_etf))
    # 买入
    if (sel_etf != g.etf_pre):
        order_target_value(sel_etf, context.portfolio.available_cash)
        print('调仓买入' + str(sel_etf))
    elif (len(hold_list)==0)and(sel_etf == g.etf_pre)and(sel_worth==1):
        order_target_value(sel_etf, context.portfolio.available_cash)
        print('恢复买入' + str(sel_etf))

2025-02-25