如何用迅投QMT实现ETF动量轮动策略

#encoding:gbk
'''
迅投QMT ETF动量轮动策略
作者：子晓聊技术--AI辅助编程
时间：2025-10-24
策略原理：每日计算ETF池中各标的的动量得分，全仓持有得分最高的ETF
动量得分 = 年化收益趋势 × 趋势确定性(R平方)
参考：聚宽年化30%+的ETF轮动策略逻辑
'''
import pandas as pd
import numpy as np

def init(ContextInfo):
    # 初始化配置
    ContextInfo.account = ''  # 账户号，实盘时填写
    ContextInfo.account_type = 'STOCK'  # 账户类型
    ContextInfo.capital = 100000  # 初始资金
    ContextInfo.start='20250101 00:00:00'
    #结束时间
    ContextInfo.end='20251023 00:00:00'
    # ETF投资池配置
    ContextInfo.etf_pool = [
        '518880.SH',  # 黄金ETF(大宗商品抗通胀)
        '513100.SH',  # 纳指ETF(海外科技股代表)
        '159915.SZ',  # 创业板ETF(A股成长股)
        '510180.SH'   # 上证180ETF(A股价值股)
    ]
    ContextInfo.buy_ratio=1
    ContextInfo.sell_ratio=0

    # 动量计算参数
    ContextInfo.m_days = 25  # 动量计算周期(25日)

    # 订阅行情数据
    ContextInfo.set_universe(ContextInfo.etf_pool)
    print('ETF动量轮动策略初始化完成')
    print('ETF池:', ContextInfo.etf_pool)
    print('动量计算周期:', ContextInfo.m_days, '天')

def handlebar(ContextInfo):
    '''每根K线触发的主逻辑'''
    d=ContextInfo.barpos

    #获取当前K线日期
    #精确到分钟小周期,不然有未来函数
    today_1=timetag_to_datetime(ContextInfo.get_bar_timetag(d),'%Y%m%d%H%M%S')
    print(today_1,"当前时间*****")
    # 获取当前持仓
    hold_stock = get_position(ContextInfo, ContextInfo.account, ContextInfo.account_type)
    if hold_stock.shape[0] > 0:
        current_hold = hold_stock['证券代码'].tolist()[0]  # 全仓只持有一个ETF
    else:
        current_hold = None

    # 计算各ETF动量得分并排序
    rank_list = get_rank(ContextInfo, today_1)


    if not rank_list:  # 检查排名列表是否为空
        print("动量得分计算异常，跳过本次调仓")
        return

    # 获取动量得分最高的ETF
    target_etf = rank_list[0]
    print(f"动量排名结果: {rank_list}")
    print(f"目标ETF: {target_etf}, 当前持仓: {current_hold}")

    # 交易逻辑：如果目标ETF与当前持仓不同，则调仓
    if current_hold != target_etf:
        # 先清空现有持仓
        if current_hold:
            order_target_percent(current_hold, 0, ContextInfo, ContextInfo.account)
            print(f'卖出 {current_hold}')

        # 买入目标ETF
        order_target_percent(target_etf, 1.0, ContextInfo, ContextInfo.account)  # 全仓买入
        print(f'买入 {target_etf}')
    else:
        print(f'继续持有 {current_hold}')

def get_rank(ContextInfo, today_1):
    '''
    计算ETF动量得分并排序
    动量得分 = 年化收益趋势 × 趋势确定性(R平方)
    返回：按得分从高到低排序的ETF列表
    '''
    import math
    score_list = []

    for etf in ContextInfo.etf_pool:
        try:
            start = str(ContextInfo.start)[:8]
            #print(start)
            # 获取历史数据
            hist_data = ContextInfo.get_market_data_ex(
                fields=['close'], 
                stock_code=[etf], 
                period='1d', 
                start_time= start, 
                end_time=today_1, 
                fill_data=True, 
                subscribe=False
            )
            #print(hist_data)
            if etf not in hist_data or hist_data[etf] is None:
                print(f"{etf} 数据获取失败")
                continue

            df = hist_data[etf]
            #print(df)        
            if df is None or len(df) < ContextInfo.m_days:
                print(f"{etf} 数据量不足")
                continue

            # 使用收盘价计算动量
            closes = df['close'].values
            if len(closes) < ContextInfo.m_days:
                continue

            # 取最近m_days个数据
            closes = closes[-ContextInfo.m_days:]

            # 计算对数价格序列
            log_prices = np.log(closes)
            x = np.arange(len(log_prices))

            # 线性回归计算斜率(收益趋势)
            slope, intercept = np.polyfit(x, log_prices, 1)

            # 计算年化收益率
            annualized_returns = math.pow(math.exp(slope), 250) - 1

            # 计算R平方(趋势确定性)
            y_pred = slope * x + intercept
            y_mean = np.mean(log_prices)
            ss_tot = np.sum((log_prices - y_mean)**2)
            ss_res = np.sum((log_prices - y_pred)**2)

            if ss_tot == 0:
                r_squared = 0
            else:
                r_squared = 1 - (ss_res / ss_tot)

            # 动量得分 = 年化收益 × R平方
            score = annualized_returns * r_squared
            score_list.append((etf, score))

            print(f"{etf}: 年化收益{annualized_returns:.3f}, R平方{r_squared:.3f}, 得分{score:.3f}")

        except Exception as e:
            print(f"计算{etf}动量得分出错: {str(e)}")
            continue

    if not score_list:
        return []

    # 按得分从高到低排序
    score_list.sort(key=lambda x: x[1], reverse=True)
    return [item[0] for item in score_list]

def get_position(ContextInfo, accountid, datatype):
    '''获取当前持仓信息'''
    positions = get_trade_detail_data(accountid, datatype, 'position')
    data = pd.DataFrame()

    if len(positions) > 0:
        df = pd.DataFrame()
        for dt in positions:
            df_temp = pd.DataFrame({
                '证券代码': [dt.m_strInstrumentID + '.' + dt.m_strExchangeID],
                '证券名称': [dt.m_strInstrumentName],
                '持仓量': [dt.m_nVolume],
                '可用数量': [dt.m_nCanUseVolume],
                '成本价': [dt.m_dOpenPrice],
                '市值': [dt.m_dInstrumentValue]
            })
            data = pd.concat([data, df_temp], ignore_index=True)
    return data