移植了一个简单的策略到QMT

*******9183 · 发表于 2024-10-23 16:46:41

昨天移植了一个ETF动量策略，有粉丝表示持仓太多、计算量大、还要按月轮动，太复杂了，让我来移植一个简单易懂的。于是我就应粉丝要求，找一个持仓只有一只ETF、只有一个计算函数、每天计算一次决定调仓与否的策略。

这个策略比较简单，在回测的时候，QMT比较给力，所以很快就移植完毕、回测成功。

这个策略的回测收益平稳，回撤不大。对于回测结果，有几点要说明：

1、回测的成本设置为0。尽管etf的交易手续费几乎可以忽略不计，但是实际上，手续费和滑点还是影响收益的，大家在回测的时候可以根据实际情况设置一下。

2、动量天数设置为25天。这个25天，是回测了一些次数后，发现25天的效果最好。但历史最好不代表以后收益也是最好。所以，大家在回测时，可以多设置几次天数，看看大致是个什么收益水平，将来在仿真或模拟时，做到心里有数。

3、etf池选用了黄金、纳指等近几年表现好的品种。这个看起来有一些事后诸葛亮，但是，在实际的仿真或模拟时，我们选择品种，肯定也会选择有潜力的etf，如果从现在开始仿真或模拟，或许我们就不会加入黄金、纳指，而是加入超跌的A股宽基ETF和其他处于低部的ETF。因此，先用了近几年走势好的品种进行回测，不能说没有意义，而是我们应该做到选出长期会带来较好收益的ETF进行轮动。当然，保守起见，也可以选择一些这几年涨势不好的宽基或行业ETF来回测。

代码如下，这次是完整的，大家可以直接复制，进行回测。

#encoding:gbk


import numpy as np
import pandas as pd
import math
#初始化函数 
def init(C):
    C.acct = '********'
    C.acct_type = 'STOCK'


    C.etf_pool = [
        '518880.SH', #黄金ETF（大宗商品）
        '513100.SH', #纳指100（海外资产）
        '159915.SZ', #创业板100（成长股，科技股，中小盘）
        '510180.SH', #上证180（价值股，蓝筹股，中大盘）
    ]
    for i in C.etf_pool:
        download_history_data(i,'1d','','')
        download_history_data(i,'1m','','')
    C.m_days = 25 #动量参考天数


def handlebar(C):

    trade(C) #每天运行确保即时捕捉动量变化


def get_rank(C,etf_pool):
    score_list = []
    start_time = timetag_to_datetime(C.get_bar_timetag(C.barpos-C.m_days),'%Y%m%d')
    end_time = timetag_to_datetime(C.get_bar_timetag(C.barpos),'%Y%m%d')
    for etf in etf_pool:
        data = C.get_market_data_ex(fields=["close"],stock_code=[etf], period = "1d", start_time = start_time, end_time = end_time,count=C.m_days)
        df = data[etf]
        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)  
    return rank_list


# 交易
def trade(C):
    # 获取动量最高的一只ETF
    target_num = 1  
    target_list = get_rank(C,C.etf_pool)[:target_num]
    #获取持仓信息
    holdings = get_trade_detail_data(C.acct, C.acct_type, 'position')
    #获取股票的代码和持仓数量的字典
    holdings = {i.m_strInstrumentID + '.' + i.m_strExchangeID : i.m_nCanUseVolume for i in holdings}
    # 卖出  
    hold_list = holdings
    for etf in hold_list:
        if etf not in target_list:
            passorder(24, 1101, C.acct, etf, 10, 0, holdings.get(etf,0), '', 1 , '', C)
            print(timetag_to_datetime(C.get_bar_timetag(C.barpos),'%Y%m%d'), '卖出' + str(etf))
        else:
            print( timetag_to_datetime(C.get_bar_timetag(C.barpos),'%Y%m%d'),'继续持有' + str(etf))
    # 买入
    for i in get_trade_detail_data(C.acct,C.acct_type,"account"):
        cash = i.m_dAvailable
    #hold_list = list(context.portfolio.positions)
    if len(hold_list) < target_num:
        value = cash / (target_num - len(hold_list))
        for etf in target_list:
            if holdings.get(etf,0) == 0:
                passorder(23, 1102, C.acct, etf, 0, 0, value, '', 1 , '', C)
                print(timetag_to_datetime(C.get_bar_timetag(C.barpos),'%Y%m%d