# 标题：多因子宽基ETF择时轮动改进版-高收益大资金低回撤
# 作者：养家大哥
# 标题：ETF动量轮动RSRS择时-V15.0，2023/3/23
# 作者：养家大哥
# 标题：动量ETF轮动RSRS择时-v16
# 作者：杨德勇
# v2 养家大哥的思路：
# 趋势因子的特点是无法及时判断趋势的变向，往往趋势变向一段时间后才能跟上，
# 巨大回撤往往就发生在这种时候。因此基于动量因子的一阶导数，衡量趋势的潜在变化速度，
# 若变化速度过快则空仓，反之则按原计划操作。
# 可以进一步发散，衡量动量因子的二阶导、三阶导等等，暂时只测试过一阶导，就是目前这个升级2版本。
from jqdata import * # 从jqdata模块导入所有内容
import numpy as np # 导numpy库并设置别名np
from jqlib.technical_analysis import *
#初始化函数
def initialize(context):
set_benchmark('399006.XSHE') # 设定基准
set_option('use_real_price', True) # 用真实价格交易
set_option("avoid_future_data", True) # 打开防未来函数
set_slippage(FixedSlippage(0.001)) # 将滑点设置为0.001
# 设置交易成本，包括印花税、佣金等，这里设置为基金的交易成本
set_order_cost(OrderCost(open_tax=0, close_tax=0.000, open_commission=0.0001, close_commission=0.0001, close_today_commission=0, min_commission=0),
type='fund')
log.set_level('order', 'error') # 设置日志级别，只记录订单错误信息
g.stock_pool = [ # 备选池：用流动性和市值更大的50ETF分别代替宽指ETF，500与300ETF保留一个
# ======== 大盘 ===================
'510300.XSHG', # 沪深300ETF
'510050.XSHG', # 上证50ETF
# '510180.XSHG', # 上证180 （用于替换上证50或沪深300，其与创业板有重合）
'159949.XSHE', # 创业板500
# '159915.XSHE', # 创业指数，替代创业500
# '510500.XSHG', # 500ETF
# '159915.XSHE', # 创业板 ETF
'159928.XSHE', # 中证消费ETF
# '512120.XSHG', # 医药50ETF
# '510880.XSHG', # 红利ETF
# '512100.XSHG', # 中证1000
# '159845.XSHE', # 中证1000
]
# 设置策略参数，如买入评分最高的股票数量、动量参考天数等
g.stock_num = 1 # 买入评分最高的前stock_num只股票
g.momentum_day = 20 # 最新动量参考最近momentum_day天
g.ref_stock = '000300.XSHG' # 用ref_stock做择时计算的基础数据
g.N = 18 # 计算最新斜率slope，拟合度r2参考最近N天
g.M = 600 # 计算最新标准分zscore，rsrs_score参考最近M天(600)
g.K = 8 # 计算 zscore 斜率的窗口大小
g.biasN = 90 # 乖离动量的时间天数
g.lossN = 20 # 止损MA20
g.lossFactor = 1.005 # 下跌止损的比例，相对前一天的收盘价
g.SwitchFactor = 1.04 # 换仓位的比例，待换股相对当前持股的分数
g.Motion_1diff = 19 # 股票前一天动量变化速度门限
g.raiser_thr = 4.8 # 股票前一天上涨的比例门限
g.hold_stock = 'null' # 持有股票
g.score_thr = -0.68 # rsrs标准分指标阈值
g.score_fall_thr = -0.43 # 当股票下跌趋势时候， 卖出阀值rsrs
g.idex_slope_raise_thr = 12 # 判断大盘指数强势的斜率门限
g.slope_series,g.rsrs_score_history= initial_slope_series() # 除去回测第一天的slope，避免运行时重复加入；初始化斜率和rsrs得分历史数据
g.stock_motion = initial_stock_motion(g.stock_pool) # 除去回测第一天的动量；初始化股票动量数据
# 设置交易时间，每天运行
run_daily(my_trade_prepare, time='7:00', reference_security='000300.XSHG')
run_daily(my_trade, time='9:30', reference_security='000300.XSHG')
run_daily(my_sell2buy, time='9:35', reference_security='000300.XSHG')
run_daily(check_lose, time='open', reference_security='000300.XSHG')
# run_daily(print_trade_info, time='15:10', reference_security='000300.XSHG')
run_daily(pre_hold_check, time='11:25')
run_daily(hold_check, time='11:27')
# 初始化斜率和zscore数据,除去回测第一天的slope,zscores
def initial_slope_series():
length = g.N+g.M+g.K # 计算所需的天数，包括用于计算斜率的天数（g.N），用于计算zscore的天数（g.M），以及用于计算斜率变化的天数（g.K）
# 获取基准股票的历史数据，包括最高价（high）、最低价（low）和收盘价（close）
data = attribute_history(g.ref_stock, length, '1d', ['high', 'low', 'close'])
# 使用列表推导式和get_ols函数计算每段时间的斜率和拟合度
multe_data = [get_ols(data.low[i:i+g.N], data.high[i:i+g.N]) for i in range(length-g.N)]
slopes = [i[1] for i in multe_data] # 从回归分析结果中提取斜率
r2s = [i[2] for i in multe_data] # 从回归分析结果中提取拟合度
# 计算zscore，这是股票价格动量的一个度量，用于评估股票价格趋势的强度
zscores =[(get_zscore(slopes[i+1:i+1+g.M])*r2s[i+g.M]) for i in range(g.K)]
return (slopes,zscores) # 返回计算得到的斜率列表和zscore列表
# 获取初始化动量因子，除去回测第一天
def initial_stock_motion(stock_pool):
stock_motion = {} # 创建一个字典，用于存储每只股票的动量队列
for stock in stock_pool: # 遍历股票池中的每只股票
motion_que = [] # 初始化一个列表，用于存储当前股票的动量值
# 获取该股票的历史收盘价数据，这里的数据用于计算乖离率和动量
data = attribute_history(stock, g.biasN + g.momentum_day + 1, '1d', ['close'])
data = data[:-1] # 移除最后一天的数据，因为计算动量时需要使用到前一天的数据
bias = (data.close/data.close.rolling(g.biasN).mean())[-g.momentum_day:] # 计算乖离率，即当前价格相对于过去 g.biasN 天平均价格的偏离程度
# 使用 numpy 的 polyfit 函数对过去 g.momentum_day 天的乖离率数据进行一次多项式拟合
# 这里计算的是一次多项式（线性）拟合的斜率，即动量因子，拟合结果的斜率乘以 10000 得到最终的动量分数
score = np.polyfit(np.arange(g.momentum_day),bias/bias[0],1)[0].real*10000
motion_que.append(score) # 将计算得到的动量分数添加到当前股票的动量队列中
stock_motion[stock] = motion_que # 将当前股票的动量队列存储到 stock_motion 字典中，键为股票代码，值为动量队列
return(stock_motion) # 返回包含所有股票动量队列的字典
# 持仓检查，盘中动态止损：早盘结束后，若60分钟周期跌破MA20均线，并且当前价格相对昨天没有上涨，则卖出
def pre_hold_check(context):
if context.portfolio.positions: # 检查当前投资组合是否有持仓
for stk in context.portfolio.positions: # 遍历当前投资组合中的所有持仓股票
dt = attribute_history(stk,g.lossN+2,'60m',['close']) # 获取指定股票的历史数据，包括过去 g.lossN+2 个60分钟周期的收盘价
# 计算每个周期的收盘价与过去 g.lossN 个周期收盘价的移动平均值的比例
# 这可以视为一种动态的止损指标，如果比例小于1，意味着价格跌破了移动平均线
dt['man'] = dt.close/dt.close.rolling(g.lossN).mean()
# 检查最新的周期（即最后一个周期）的收盘价是否低于过去 g.lossN 个周期的移动平均值，如果是，则可能需要进行止损操作
if(dt.man[-1] < 1.0):
stk_dict = context.portfolio.positions[stk] # 获取当前股票的持仓信息
log.info("盘中可能止损，卖出：{}".format(stk)) # 记录日志信息，表示可能需要止损并卖出该股票
send_message("盘中可能止损，卖出：{}".format(stk)) # 发送消息，提示可能需要止损并卖出该股票
# 并且当前价格相对昨天没有上涨，则卖出
def hold_check(context):
current_data = get_current_data() # 获取当前市场数据
if context.portfolio.positions: # 检查当前投资组合是否有持仓
for stk in context.portfolio.positions: # 遍历当前投资组合中的所有持仓股票
yesterday_di = attribute_history(stk,1,'1d',['close']) # 获取昨天的收盘价数据
dt = attribute_history(stk,g.lossN+2,'60m',['close']) # 获取过去 g.lossN+2 个60分钟周期的收盘价数据
dt['man'] = dt.close/dt.close.rolling(g.lossN).mean() # 计算每个周期的收盘价与过去 g.lossN 个周期收盘价的移动平均值的比例
#log.info("man=%0f, last_price=%0f, yester=%0f"%(dt.man[-1], current_data[stk].last_price*1.006, yesterday_di['close'][-1]))
# 检查最新的周期（即最后一个周期）的收盘价是否低于过去 g.lossN 个周期的移动平均值，并且当前价格相比昨天的收盘价没有上涨
if((dt.man[-1] < 1.0) and (current_data[stk].last_price*g.lossFactor <= yesterday_di['close'][-1])):
#if (dt.man[-1] < 1.0):
stk_dict = context.portfolio.positions[stk] # 获取当前股票的持仓信息
log.info('准备平仓，总仓位:{}, 可卖出：{}, '.format(stk_dict.total_amount,stk_dict.closeable_amount)) # 记录日志信息，准备平仓
send_message("盘中止损，卖出：{}".format(stk)) # 发送消息，提示准备平仓
if(stk_dict.closeable_amount): # 如果有可卖出的持仓
order_target_value(stk,0) # 下单将持仓平仓至0
log.info('盘中止损',stk) # 记录日志信息，确认已执行止损
else:
log.info('无法止损',stk) # 记录日志信息，表示无法止损
# 计算股票池中各股票的动量因子并进行排名：由收益率动量改为相对MA90均线的乖离动量
def get_rank(context,stock_pool):
rank = [] # 初始化一个空列表，用于存储股票的排名信息
for stock in stock_pool: # 遍历股票池中的每只股票
data = attribute_history(stock, g.biasN + g.momentum_day, '1d', ['close']) # 获取每只股票 g.biasN + g.momentum_day 天的收盘价数据
bias = (data.close/data.close.rolling(g.biasN).mean())[-g.momentum_day:] # 计算乖离因子，即当前价格与过去 g.biasN 天收盘价的移动平均的比值
score = np.polyfit(np.arange(g.momentum_day),bias/bias[0],1)[0].real*10000 # 对过去 g.momentum_day 天的乖离数据进行一次多项式拟合，计算拟合的斜率，乘以 10000 得到最终的动量分数
adr = 100*(data.close[-1] - data.close[-2])/data.close[-2] # 计算股票的涨跌幅度，即最新价格相比前一天的百分比变化
if(stock == g.hold_stock): raise_x = g.SwitchFactor # 如果当前股票是持有中的股票，则调整其分数
else: raise_x = 1
# data = attribute_history(stock, g.momentum_day, '1d', ['close'])
# score = np.polyfit(np.arange(g.momentum_day),data.close/data.close[0],1)[0].real # 乖离动量拟合
#log.info("计算data.close[-1]=%f, data.close[-2]=%f,adr=%f"%(data.close[-1], data.close[-2], adr))
rank.append([stock, score*raise_x, adr]) # 将股票的代码、调整后的动量分数和涨跌幅度加入到排名列表中
g.stock_motion[stock].append(score) # 更新该股票的动量列表，将最新的动量分数添加到列表末尾
if(len(g.stock_motion[stock])>5):g.stock_motion[stock].pop(0) # 如果动量列表的长度超过 5，则移除最旧的动量值
#log.info('rsrs_score:')
str = '' # 构建一个字符串，用于记录和显示所有股票的排名信息
for item in rank:
str += "%s:%.2f:%.2f; "%(item[0], item[1], item[2])
log.info(str) # 记录所有股票的排名信息到日志中
rank = [ i for i in rank if math.isnan(i[1])==False ] # 移除排名列表中动量分数为 NaN 的股票
rank.sort(key=lambda x: x[1],reverse=True) # 对排名列表进行排序，按照动量分数从高到低排序
return rank[0] # 返回排名第一的股票信息
# 线性回归：复现statsmodels的get_OLS函数
# 用于执行线性回归并计算斜率、截距和 R-squared（决定系数）
def get_ols(x, y):
slope, intercept = np.polyfit(x, y, 1) # 使用 numpy 的 polyfit 函数对 x 和 y 进行一次多项式拟合，得到斜率和截距
# 计算 R-squared，即决定系数，表示模型对数据的拟合程度，计算公式为 1 - (残差平方和 / 总平方和)
r2 = 1 - (sum((y - (slope * x + intercept))**2) / ((len(y) - 1) * np.var(y, ddof=1)))
return (intercept, slope, r2) # 返回线性回归的截距、斜率和 R-squared
# 计算因子的标准化分数（z-score）
def get_zscore(slope_series):
# 计算斜率序列的平均值和标准差
mean = np.mean(slope_series)
std = np.std(slope_series)
return (slope_series[-1] - mean) / std # 计算最新斜率值的 z-score，即最新斜率值与平均值的差除以标准差
# 计算 z-score 斜率
def get_zscore_slope(z_scores):
y = z_scores # 将 z-score 序列作为 y 值
x = np.arange(len(z_scores)) # 创建一个与 z-score 序列等长的 x 值序列，用于线性回归
slope, intercept = np.polyfit(x, y, 1) # 对 x 和 y 进行线性回归，得到斜率和截距
return slope # 返回 z-score 斜率
# 只看RSRS因子值作为买入、持有和清仓依据，前版本还加入了移动均线的上行作为条件
def get_timing_signal(context,stock):
data = attribute_history(g.ref_stock, g.N, '1d', ['high', 'low', 'close']) # 获取基准股票的历史数据，包括最高价、最低价和收盘价，用于计算 RSRS 因子
intercept, slope, r2 = get_ols(data.low, data.high) # 使用 get_ols 函数进行线性回归，计算斜率、截距和 R-squared
g.slope_series.append(slope) # 将计算得到的斜率添加到斜率序列中
rsrs_score = get_zscore(g.slope_series[-g.M:]) * r2 # 计算 RSRS 因子的标准化分数（z-score），并乘以 R-squared 得到最终得分
g.rsrs_score_history.append(rsrs_score) # 将 RSRS 得分添加到得分历史序列中
rsrs_slope = get_zscore_slope(g.rsrs_score_history[-g.K:]) # 计算 RSRS 得分序列的斜率变化率
idex_slope = np.polyfit(np.arange(8), data.close[-8:],1)[0].real # 计算大盘指数收盘价的趋势斜率
# 移除斜率序列和得分历史序列中的第一个元素，保持序列长度一致
g.slope_series.pop(0)
g.rsrs_score_history.pop(0)
#record(rsrs_score=rsrs_score,rsrs_slope=rsrs_slope)
log.info('rsrs_slope {:.3f}'.format(rsrs_slope)+' rsrs_score {:.3f} '.format(rsrs_score) # 记录日志信息，包括 RSRS 斜率、得分和大盘指数斜率
+' idex_slope {:.3f} '.format(idex_slope))
# 使用摆动指数（Williams %R）判断市场趋势变化，优先级高于 RSRS 因子
WR2,WR1 = WR([g.ref_stock], check_date =context.previous_date, N = 21, N1 = 14, unit='1d', include_now=True)
# 如果摆动指数显示市场可能过度买入，返回买入信号；个人觉得这里应该返回卖出才对。
if WR1[g.ref_stock]>=97 and WR2[g.ref_stock] >=97: return "BUY"
# 如果 RSRS 斜率小于 0 且得分大于 0，表示上升趋势可能结束，返回卖出信号
if(rsrs_slope< 0 and rsrs_score >0):
return "SELL"
# 如果大盘指数斜率小于 0，RSRS 斜率大于 0 且得分低于下跌阈值，表示市场可能下跌，返回卖出信号
if(idex_slope<0) and (rsrs_slope>0) and (rsrs_score < g.score_fall_thr): return "SELL"
# 如果大盘指数斜率大于设定的上升阈值，且 RSRS 斜率大于 0，表示市场处于上升趋势，返回买入信号
if(idex_slope>g.idex_slope_raise_thr) and (rsrs_slope>0): return "BUY"
# 如果 RSRS 得分高于设定的买入阈值，返回买入信号
if (rsrs_score> g.score_thr) : return "BUY"
# 如果以上条件都不满足，返回卖出信号
else: return "SELL"
#4-2 交易模块-开仓
#买入指定价值的证券,如果报单成功并且有成交（无论是全部成交还是部分成交），如果报单失败或者报单成功但最终被取消（此时成交量为 0），则返回 False
def open_position(security, value):
order = order_target_value(security, value) # 下单买入指定价值的证券
if order != None and order.filled > 0: # 检查订单是否创建成功，并且是否至少有部分成交
return True # 返回 True
return False # 返回 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: # 检查订单是否全部成交并且订单状态为已成交（held）
return True # 返回 True
return False # 返回 False
# 调整投资组合中的仓位
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) # 尝试平仓该股票
g.hold_stock = 'null' # 将持有股票的状态设置为 'null'
return # 调整仓位后，结束函数执行
else:
pass # 如果股票在买入列表中，不执行任何操作
# 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: # 如果成功买入，并且持有的股票数量达到了目标数量，则记录该股票为当前持有股票
g.hold_stock = stock
break # 达到目标持仓数量后，结束循环
# 根据给定的股票列表买入股票
def buy_stocks(context, buy_stocks):
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: # 如果成功买入，并且持有的股票数量达到了目标数量，则记录该股票为当前持有股票
g.hold_stock = stock
break # 达到目标持仓数量后，结束循环
关键函数解锁后查看：
# 打印当天的成交记录
def print_trade_info(context):
#打印当天成交记录
trades = get_trades() # 获取当天的所有成交记录
for _trade in trades.values(): print('成交记录：'+str(_trade)) # 遍历成交记录并打印
#打印账户信息
print('———————————————————————————————————————分割线1————————————————————————————————————————')
2025-02-21 