《QMT量化实战系列》复现聚宽年化30%+的ETF轮动策略

今天，公众号将为全网读者带来QMT量化实战系列！

公众号将为大家多维度、多策略、多场景来讲述QMT量化平台的实践应用。同时，我们对每段代码都做了解读说明，愿你在Quant的道路上学有所获！

此系列将由浅入深，每月1~2期，大家敬请期待！

今日复现策略：《聚宽年化30%+的ETF轮动策略》

策略逻辑解读

🧠 策略核心

从4只ETF中选择动量最强的1只持有，其余清仓。

⚙️ 动量评分

• 取每只ETF近25日收盘价
• 回归计算上涨趋势（斜率）
• 年化收益 × 拟合优度 R2R^2R2 = 动量得分
• 分数越高代表走势越强

🔁 交易逻辑

卖出：不在得分最高的ETF清仓

买入：若目标ETF未持有，用可用资金买入

⏱️ 执行频率

每日调仓一次

聚宽策略定义

import numpy as np
import pandas as pd
import math  # 需要导入 math 模块，因为你在代码中使用了 math.pow 和 math.exp


# 初始化函数
def initialize(context):
    # 设定基准
    set_benchmark('000300.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    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')  # 每天运行确保即时捕捉动量变化


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) 
    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))
    # 买入
    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))

QMT复现过程

step 1：函数对比

聚宽	QMT	说明
attribute_history	get_market_data_ex	获取历史数据
`...portfolio.positions `	get_trade_detail_data	获取当前持仓
...olio.available_cash	get_trade_detail_data	获取当前资金
order_target_value	order_target_value	下单(QMT此函数非实盘)
record	draw_text	绘图函数

step 2：get_rank函数

def get_rank(ContextInfo):
"""
对 ETF 进行动量评分，返回得分从高到低的ETF列表
动量评分 = 年化收益 * 拟合优度(R2)
"""
bar_date = timetag_to_datetime(ContextInfo.get_bar_timetag(ContextInfo.barpos - 1), '%Y%m%d%H%M%S')
score_list = []
etf_pool = ContextInfo.etf_pool 
df_list = df = ContextInfo.get_market_data_ex(['close'],etf_pool, end_time=bar_date, period='1d', count=ContextInfo.m_days, subscribe=False)
# 获取过去N日的收盘价数据 
for etf in etf_pool:
        df = df_list[etf] 
if df is None or df.empty or df['close'].isnull().all() or len(df) < 5:
print(f'{etf}行情数据不足 跳过')
continue
# 计算对数收益回归斜率（近似日收益率）
        y = df['log'] = np.log(df.close)
        x = df['num'] = np.arange(df.log.size)


        slope, intercept = np.polyfit(x, y, 1)


# 年化收益率 = e^(slope * 250) - 1
        annualized_returns = math.pow(math.exp(slope), 250) - 1


# 拟合优度R2
        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_score = pd.DataFrame(index=etf_pool, data={'score': score_list})
    df_score = df_score.sort_values(by='score', ascending=False)
return list(df_score.index)

复现结果验证

交易明细验证

需要进量化交流群的可以加微信