摘要：策略步骤：每天计算四个ETF的动量得分，选最高分的持有，切换持仓。动量得分结合了收益和趋势的确定性，避免选择波动大但趋势不明显的标的。
策略步骤：每天计算四个ETF的动量得分，选最高分的持有，切换持仓。动量得分结合了收益和趋势的确定性，避免选择波动大但趋势不明显的标的。
该策略是一个基于动量效应的ETF轮动策略
其核心原理：
1. 投资标的配置
选择四类差异化资产ETF构成投资池：
- 黄金ETF（大宗商品抗通胀）
- 上证180ETF（A股价值股）
通过分散配置不同资产类别，实现风险对冲。
2. 动量因子计算
采用改良版动量评估方法，每日计算两个关键指标：
- 年化收益趋势：通过对数价格25日线性回归的斜率计算，反映价格变化速度
公式：annualized_returns = e^(回归斜率)^250 - 1
- 趋势确定性：使用R平方判定系数，衡量价格趋势的稳定性（0-1之间）
最终动量得分 = 年化收益 × R平方
这种方法既考虑收益幅度，又过滤了波动大的伪趋势。
3. 交易逻辑
- 每日开盘时计算各ETF动量得分
- 全仓持有得分最高的ETF
- 当排名变化时立即切换持仓
- 每次仅持有1个ETF，实现极致轮动
4. 风险控制
- 设置防未来函数校验
- 采用真实价格回测
- 设定较低的交易成本（万2佣金）
- 通过多资产配置分散风险
策略优势：通过量化模型捕捉强势资产，利用不同资产间的低相关性平滑收益曲线。实证研究表明，动量效应在3-6个月周期内有效性较高，25日参数正是捕捉这一周期特征。
测试结果:
年化收益34%，11年收益2300%，夏普比例1.2，胜率55%，盈亏比2.66，回撤率30%
聚宽策略源码
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 #动量参考天数
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))
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))
# 买入
if len(hold_list)
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))
来源：简易交易