2507 核心资产动量轮动完善 bug优化参数，10 年26 倍

全球核心资产动量轮动策略

核心思想

策略特点

1. 全球资产配置：覆盖 A 股、港股、美股和黄金等多元资产类别
2. 动量因子驱动：基于 25 日动量指标进行资产排序和选择
3. 双重择时系统：结合 RSRS 指标和均线系统进行买卖决策
4. 严格风险控制：设置 80% 的极端止损线
5. 低交易频率：每日 11:20 进行调仓，降低交易成本

交易规则

1. 资产池构成：
   • 创业板 ETF (159915.XSHE)
   • 纳斯达克 ETF (513100.XSHG)
   • 恒生科技 ETF (159740.XSHE)
   • 黄金 ETF (518880.XSHG)
   • 沪深 300ETF (510300.XSHG)
2. 动量计算规则：
   • 计算 25 日对数收益率线性回归斜率
   • 结合 R² 指标调整动量得分
   • 每日选择得分最高的 1 只资产持有
3. 择时信号规则：
   • RSRS 指标：基于沪深 300 指数高低价回归的标准化得分
   • 均线系统：20 日均线与 23 日均线比较
   • 买入信号：RSRS>0.7 且 20MA>23MA
   • 卖出信号：RSRS<-0.7 且 20MA<23MA
4. 风险控制规则：
   • 单资产持仓亏损达到 80% 时强制止损
   • 每日开盘检查止损条件
   • 过滤 ST 股、退市股、停牌股和涨跌停股

优化改进点

1. 修复了原策略动量排序中的 NaN 值处理问题
2. 将字典数据结构改为 DataFrame 格式提高稳定性
3. 基于相关性和波动性优化资产选择
4. 调整动量参数至 25 日表现最佳
5. 完善止损机制和交易执行逻辑

绩效表现

• 回测周期：2015.1.1-2025.2.22
• 累计收益：26 倍
• 年化收益：40%
• 最大回撤：24.72%
• 主要回撤原因：单资产持仓遭遇突发暴跌

实时

#导入函数库
from jqdata import *
import numpy as np

#初始化函数 
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    # set_option("avoid_future_data", True)
    # 将滑点设置为0
    set_slippage(FixedSlippage(0.001))
    # 设置交易成本万分之三
    set_order_cost(OrderCost(open_tax=0, close_tax=0, open_commission=0.0003, close_commission=0.0003, close_today_commission=0, min_commission=5),
                   type='fund')
    # 过滤order中低于error级别的日志
    log.set_level('order', 'error')
    # 初始化各类全局变量
    #股票池
    g.stock_pool = [

        '159915.XSHE', #创业板
        '513100.XSHG', #纳指
        '159740.XSHE', #恒生科技
        '518880.XSHG',#黄金ETF
        '510300.XSHG', # 沪深300ETF
        # '510880.XSHG', #红利etf        
        # '512150.XSHG', #A50
        # '510500.XSHG', # 中证500ETF
        # '511880.XSHG',#银华日利
        # '511090.XSHG', #30年国债etf
        
    ]
    #动量轮动参数
    g.stock_num = 1 #买入评分最高的前stock_num只股票
    g.momentum_day = 25 #最新动量参考最近momentum_day的
    #rsrs择时参数
    g.ref_stock = '000300.XSHG' #用ref_stock做择时计算的基础数据
    g.N = 18 # 计算最新斜率slope，拟合度r2参考最近N天
    g.M = 600 # 计算最新标准分zscore，rsrs_score参考最近M天
    g.score_threshold = 0.7 # rsrs标准分指标阈值
    #ma择时参数
    g.mean_day = 20 #计算结束ma收盘价，参考最近mean_day
    g.mean_diff_day = 3 #计算初始ma收盘价，参考(mean_day + mean_diff_day)天前，窗口为mean_diff_day的一段时间
    g.slope_series = initial_slope_series()[:-1] # 除去回测第一天的slope，避免运行时重复加入
    # 设置交易时间，每天运行
    run_daily(my_trade, time='11:20', reference_security='000300.XSHG')
    run_daily(check_lose, time='open', reference_security='000300.XSHG')
    run_daily(print_trade_info, time='15:30', reference_security='000300.XSHG')
    # run_daily(check_break_ma5, '14:25')
    
#1-1 选股模块-动量因子轮动 
#基于股票年化收益和判定系数打分,并按照分数从大到小排名
# def get_rank(stock_pool):
#     score_list = []
#     for stock in g.stock_pool:
#         data = attribute_history(stock, g.momentum_day, '1d', ['close'])
#         y = data['log'] = np.log(data.close)
#         x = data['num'] = np.arange(data.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)
#     stock_dict=dict(zip(g.stock_pool, score_list))
#     sort_list=sorted(stock_dict.items(), key=lambda item:item[1], reverse=True) #True为降序   该排序如何初选缺失值排序会出现错误
#     print(sort_list)
#     code_list=[]
#     for i in range((len(g.stock_pool))):
#         code_list.append(sort_list[i][0])
#     rank_stock = code_list[0:g.stock_num]
#     print(code_list[0:5])
#     return rank_stock

关键函数解锁后查看：

#3-1 过滤模块-过滤停牌股票
#输入选股列表，返回剔除停牌股票后的列表
def filter_paused_stock(stock_list):
	current_data = get_current_data()
	return [stock for stock in stock_list if not current_data[stock].paused]

#3-2 过滤模块-过滤ST及其他具有退市标签的股票
#输入选股列表，返回剔除ST及其他具有退市标签股票后的列表
def filter_st_stock(stock_list):
	current_data = get_current_data()
	return [stock for stock in stock_list
			if not current_data[stock].is_st
			and 'ST' not in current_data[stock].name
			and '*' not in current_data[stock].name
			and '退' not in current_data[stock].name]

#3-3 过滤模块-过滤涨停的股票
#输入选股列表，返回剔除未持有且已涨停股票后的列表
def filter_limitup_stock(context, stock_list):
	last_prices = history(1, unit='1m', field='close', security_list=stock_list)
	current_data = get_current_data()
	# 已存在于持仓的股票即使涨停也不过滤，避免此股票再次可买，但因被过滤而导致选择别的股票
	return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
			or last_prices[stock][-1] < current_data[stock].high_limit]

#3-4 过滤模块-过滤跌停的股票
#输入股票列表，返回剔除已跌停股票后的列表
def filter_limitdown_stock(context, stock_list):
	last_prices = history(1, unit='1m', field='close', security_list=stock_list)
	current_data = get_current_data()
	return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
			or last_prices[stock][-1] > current_data[stock].low_limit]

 

#4-1 交易模块-自定义下单
#报单成功返回报单(不代表一定会成交),否则返回None,应用于
def order_target_value_(security, value):
	if value == 0:
		log.debug("Selling out %s" % (security))
	else:
		log.debug("Order %s to value %f" % (security, value))
	# 如果股票停牌，创建报单会失败，order_target_value 返回None
	# 如果股票涨跌停，创建报单会成功，order_target_value 返回Order，但是报单会取消
	# 部成部撤的报单，聚宽状态是已撤，此时成交量>0，可通过成交量判断是否有成交
	return order_target_value(security, value)

#4-2 交易模块-开仓
#买入指定价值的证券,报单成功并成交(包括全部成交或部分成交,此时成交量大于0)返回True,报单失败或者报单成功但被取消(此时成交量等于0),返回False
def open_position(security, value):
	order = order_target_value_(security, value)
	if order != None and order.filled > 0:
		return True
	return False

#4-3 交易模块-平仓
#卖出指定持仓,报单成功并全部成交返回True，报单失败或者报单成功但被取消(此时成交量等于0),或者报单非全部成交,返回False
def close_position(position):
	security = position.security
	order = order_target_value_(security, 0)  # 可能会因停牌失败
	if order != None:
		if order.status == OrderStatus.held and order.filled == order.amount:
			return True
	return False

#4-4 交易模块-调仓
#当择时信号为买入时开始调仓，输入过滤模块处理后的股票列表，执行交易模块中的开平仓操作
def adjust_position(context, buy_stocks):
	for stock in context.portfolio.positions:
		if stock not in buy_stocks:
			log.info("[%s]已不在应买入列表中" % (stock))
			position = context.portfolio.positions[stock]
			close_position(position)
		else:
			log.info("[%s]已经持有无需重复买入" % (stock))
	# 根据股票数量分仓
	# 此处只根据可用金额平均分配购买，不能保证每个仓位平均分配
	position_count = len(context.portfolio.positions)
	if g.stock_num > position_count:
		value = context.portfolio.cash / (g.stock_num - position_count)
		for stock in buy_stocks:
			if context.portfolio.positions[stock].total_amount == 0:
				if open_position(stock, value):
					if len(context.portfolio.positions) == g.stock_num:
						break

#4-5 交易模块-择时交易
#结合择时模块综合信号进行交易
def my_trade(context):
    #获取选股列表并过滤掉:st,st*,退市,涨停,跌停,停牌
    check_out_list = get_rank(g.stock_pool)
    check_out_list = filter_st_stock(check_out_list)
    check_out_list = filter_limitup_stock(context, check_out_list)
    check_out_list = filter_limitdown_stock(context, check_out_list)
    check_out_list = filter_paused_stock(check_out_list)
    print('今日自选股:{}'.format(check_out_list))
    #获取综合择时信号
    timing_signal = get_timing_signal(g.ref_stock)
    print('今日择时信号:{}'.format(timing_signal))
    #开始交易
    if timing_signal == 'SELL':
        for stock in context.portfolio.positions:
            position = context.portfolio.positions[stock]
            close_position(position)
    elif timing_signal == 'BUY' or timing_signal == 'KEEP':
        if len(check_out_list) == 0:
            for stock in context.portfolio.positions:
                position = context.portfolio.positions[stock]
                close_position(position)
        else:    
            adjust_position(context, check_out_list)
    else:
        pass

#4-6 交易模块-止损
#检查持仓并进行必要的止损操作
def check_lose(context):
    for position in list(context.portfolio.positions.values()):
        securities=position.security
        cost=position.avg_cost
        price=position.price
        ret=100*(price/cost-1)
        value=position.value
        amount=position.total_amount
        #这里设定80%止损几乎等同不止损，因为止损在指数etf策略中影响不大
        if ret <=-80:
            order_target_value(position.security, 0)
            print("！！！！！！触发止损信号: 标的={},标的价值={},浮动盈亏={}% ！！！！！！"
                .format(securities,format(value,'.2f'),format(ret,'.2f')))

#5-1 复盘模块-打印
#打印每日持仓信息
def print_trade_info(context):
    #打印当天成交记录
    trades = get_trades()
    for _trade in trades.values():
        print('成交记录：'+str(_trade))
    #打印账户信息
    for position in list(context.portfolio.positions.values()):
        securities=position.security
        cost=position.avg_cost
        price=position.price
        ret=100*(price/cost-1)
        value=position.value
        amount=position.total_amount    
        print('代码:{}'.format(securities))
        print('成本价:{}'.format(format(cost,'.2f')))
        print('现价:{}'.format(price))
        print('收益率:{}%'.format(format(ret,'.2f')))
        print('持仓(股):{}'.format(amount))
        print('市值:{}'.format(format(value,'.2f')))
    print('一天结束')
    print('———————————————————————————————————————分割线————————————————————————————————————————')

2025-04-05