发布于vn.py社区公众号【vnpy-community】
原文作者： 丛子龙 | 发布时间：2023-10-13
RSI相对强度指数是技术分析中常用的指标之一，由J. Welles Wilder Jr.在其1978年的著作《技术交易系统的新概念》中开发并引入技术分析中使用。RSI衡量价格变动的速度和幅度，该指标绘制在0到100的范围内。
许多技术分析类的书籍中常常见到将RSI用于均值回归交易，为人熟知的用法之一是在RSI超过70时卖出资产，当RSI跌破30时买入资产。然而，RSI也可以用作动量趋势指标，比如在上升趋势中，RSI通常在40到80之间波动，在下降趋势中在20到60之间波动。
本篇文章将会介绍三套围绕RSI构建的多头短线择时策略（策略思路来源于【QuantifiedStrategies】网站），分别是：
1. RSI经典策略
2. RSI区域动量策略
3. RSI-IBS策略
在文章的结尾，我们将会组合上述三个策略中的核心信号，构建一个新的RSI多信号集成策略，并在SPY（标普500ETF基金）和IF（沪深300股指期货）上进行回测。
RSI经典策略
回测绩效
由于【QuantifiedStrategies】网站文章中主要使用了SPY的日线数据进行回测，以下回测结果也基于同样数据：
基本信息
核心逻辑
策略的核心思想是在市场情绪低迷时，买入以寻求市场反弹，然后在价格触及昨日高点时卖出：
- 当市场上的RSI指标低于设定阈值时，即被认为处于超卖状态；
- 计算每次交易的头寸大小；
- 如果当前无持仓且市场满足买入条件，则执行超价买入操作；
- 如果当前持有多头仓位，则挂出卖单，价格为昨日高点，这里不使用超价卖出。
初始化策略
class RsiStrategy(CtaTemplate):
"""经典RSI策略"""
# 计算RSI指标的窗口
rsi_window: int = 2
# RSI低阈值
rsi_lower: int = 15
# 持仓周期限制
max_holding: int = 9
# 风险资金
risk_capital: int = 1_000_000
parameters = [
"rsi_window",
"rsi_lower",
"max_holding",
"risk_capital",
]
variables = []
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
# 加载足够的历史数据来计算指标
self.load_bar(60)
指标计算&交易执行
def on_bar(self, bar: BarData):
"""
K线数据推送
"""
# 撤销之前发出的委托
self.cancel_all()
am = self.am
am.update_bar(bar)
if not am.inited:
return
# 计算RSI指标
rsi_value = am.rsi(self.rsi_window)
# 保存昨日高点
prev_high = am.high[-1]
# 判断是否要进行交易
long_signal = rsi_value <= self.rsi_lower
# 计算每次交易的头寸
self.trading_size = int(self.risk_capital / am.close[-1] / 100) * 100
# 如果无持仓，且满足条件，则直接开仓
if self.pos == 0 and long_signal:
self.buy(bar.close_price * 1.05, self.trading_size)
# 如果持仓，且满足条件，则直接平仓
if self.pos > 0:
self.sell(prev_high, self.pos)
# 推送UI更新
self.put_event()
RSI区域动量策略
回测绩效
本策略的独特之处在于其较低的回撤率（-7.99%），如此低的回撤率也造就了漂亮的收益回撤比：17.62。
本策略一旦开仓就通常保持在上升轨道上，这意味着它在大部分时间内不会暴露于市场风险之下。这种特性使得这个策略在不稳定市场中具有较强的抗跌能力，有助于保护利润。
由于在大部分时间内没有仓位，该策略可以与其他策略相互配合提供额外收益。低回撤和稳定的增长趋势为其提供了与其他更高风险策略相互协同的机会，以实现更好的综合投资表现。
基本信息
核心逻辑
该策略包括两个指标：
RSI多头范围：RSI过去N天内在40到100之间波动。
RSI多头动量：RSI的极值高点在N天内大于70。
本次回测将使用100天的回望窗口和14天的RSI，这意味着为了触发RSI多头范围的信号，RSI必须在过去的100天内在40到100之间波动。
有了这个理念，交易逻辑非常简单：
当RSI多头范围和多头动量条件都成立时，开仓。
当RSI多头范围和多头动量条件都不再成立时，平仓。
初始化策略
class RsiRangeMomStrategy(CtaTemplate):
"""RSI区域动量策略"""
# 计算RSI指标的窗口
rsi_window: int = 14
# RSI低阈值
rsi_lower: int = 40
# RSI高阈值
rsi_upper: int = 100
# RSI极值阈值
rsi_highest: int = 70
# 风险资金
risk_capital: int = 1_000_000
parameters = [
"rsi_window",
"rsi_lower",
"rsi_upper",
"rsi_highest",
"risk_capital",
]
variables = []
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
# 加载足够的历史数据来计算指标
self.load_bar(150)
信号指标计算
def on_bar(self, bar: BarData):
# 撤销所有挂单
self.cancel_all()
am = self.am
am.update_bar(bar)
if not am.inited:
return
# 计算rsi的值，并返回一个【rsi_window】长度的数组
rsi_array: np.ndarray = am.rsi(self.rsi_window, array=True)
# 计算该rsi数组的平均值，使用【np.nanmean】的原因是因为返回的数组中包含NaN值
mean_value: float = np.nanmean(rsi_array)
# 将NaN值使用【mean_value】填充
rsi_array: np.ndarray = np.nan_to_num(rsi_array, nan=mean_value)
目标交易执行
# 使用历史rsi值计算趋势是否处于牛市区间
# 判断标准为本段历史中所有rsi的值是否都在【rsi_lower】与【rsi_upper】之间
bull_range_signal: bool = (np.all(rsi_array > self.rsi_lower) and
np.all(rsi_array < self.rsi_upper))
# 判断本段历史中是否存在较强的rsi值，只要有一个超过设定的【rsi_highest】即成立
bull_mom_signal: bool = np.any(rsi_array > self.rsi_highest)
# 计算开仓数量
trading_size: int = (int(self.risk_capital / bar.close_price / 100)
* 100)
# 判断开仓信号
if self.pos == 0:
# 如果牛市区间以及极值信号都出现，满仓入场
if bull_range_signal and bull_mom_signal:
self.buy(bar.close_price*1.05, trading_size)
# 判断平仓信号
if self.pos > 0:
# 如果信号不再成立，清仓
if not (bull_range_signal or bull_mom_signal):
self.sell(bar.close_price*0.95, self.pos)
# 推送UI更新
self.put_event()
RSI-IBS 策略
回测绩效
基本信息
核心逻辑
策略中用到的IBS（内部K线强度），指标计算公式如下：
(Close - Low) / (High - Low)
IBS指标的波动范围从0到1，测量收盘价相对于日内高低点的位置，较低的值被认为是看涨的，而较高的值则是短期看跌的，IBS的基本假设是市场具有均值回归的特性。
策略交易逻辑为：
- 计算RSI和IBS指标；
- 根据计算的RSI和IBS指标，判断是否满足开仓条件；
- 如果没有持仓且满足开仓条件，执行买入。如果已有多头仓位且当日收盘价高于昨日收盘价，则执行卖出。
初始化策略
class RsiIbsStrategyOG(CtaTemplate):
"""RSI-IBS策略"""
# rsi指标窗口
rsi_window: int = 21
# 入场rsi指标阈值
rsi_entry: int = 45
# 入场ibs指标阈值
ibs_entry: float = 0.25
# 风险资金
risk_capital: int = 1_000_000
parameters = [
"rsi_window",
"rsi_entry",
"ibs_entry",
]
variables = []
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
self.prev_close = 0
self.load_bar(60)
指标交易计算&交易执行
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
# 撤销之前发出的委托
self.cancel_all()
# 对am更新bar数据
am = self.am
am.update_bar(bar)
if not am.inited:
return
# 计算开仓手数
trading_size = int(self.risk_capital / bar.close_price / 100) * 100
# 计算rsi指标数值
rsi_value = am.rsi(self.rsi_window)
# 计算ibs指标数值
ibs_value = ((bar.close_price - bar.low_price) /
(bar.high_price - bar.low_price))
# 分别计算开仓信号
cond_1 = rsi_value < self.rsi_entry
cond_2 = ibs_value < self.ibs_entry
# 汇总合成信号
long_signal = cond_1 and cond_2
# 执行开仓操作
if self.pos == 0 and long_signal:
self.buy(bar.close_price * 1.05, trading_size)
# 计算平仓信号，并执行平仓操作
if self.pos > 0:
if bar.close_price > self.prev_close:
self.sell(bar.close_price * 0.95, self.pos)
# 记录当今bar的收盘价
self.prev_close = bar.close_price
# 推送UI更新
self.put_event()
RSI多信号集成策略
前文中的三套策略虽然都围绕RSI指标开发，但由于核心思路的区别，其回测绩效曲线还是体现出了较低的相关性。那么下一步的研究方向，就是把三套策略中的交易信号提取出来后，集成组合成为一个新的策略，看看能否达到更优秀的整体绩效。
为了实现信号的集成组合，需要对之前的策略代码进行调整，拆分成为策略信号和交易执行两块部分，具体逻辑流程看了下图应该会有一个更加清晰直观的理解：
当任一策略信号给出True值即入场做多，当所有策略信号都返回False值或达到止损目标时平仓离场。
回测绩效
代码实现
信号生成部分被封装在独立的信号类中，分别是：
1. RsiSignal
2. RsiRangeMomSignal
3. RsiIbsSignal
实现这些信号的方式并没有什么特别之处，单单是将前述三个策略的信号生成部分截取出来封装成一个可以返回布尔值（信号）的函数即可。
RsiSignal
class RsiSignal:
def __init__(
self,
rsi_window: int = 2,
rsi_lower: int = 15
):
# 计算RSI指标的窗口
self.rsi_window: int = rsi_window
# RSI低阈值
self.rsi_lower: int = rsi_lower
def calculate_signal(self, am: ArrayManager) -> bool:
# 计算RSI指标
rsi_value = am.rsi(self.rsi_window)
# 判断是否要进行交易
return rsi_value <= self.rsi_lower
RsiRangeMomSignal
class RsiRangeMomSignal:
def __init__(
self,
rsi_window: int = 14,
rsi_lower: int = 40,
rsi_upper: int = 100,
rsi_highest: int = 70
) -> None:
# 计算RSI指标的窗口
self.rsi_window: int = rsi_window
# RSI低阈值
self.rsi_lower: int = rsi_lower
# RSI高阈值
self.rsi_upper: int = rsi_upper
# RSI极值阈值
self.rsi_highest: int = rsi_highest
def calculate_signal(self, am: ArrayManager) -> bool:
# 计算rsi的值，并返回一个【rsi_window】长度的数组
rsi_array: np.ndarray = am.rsi(self.rsi_window, array=True)
# 计算该rsi数组的平均值，使用【np.nanmean】的原因是因为返回的数组中包含NaN值
mean_value: float = np.nanmean(rsi_array)
# 将NaN值使用【mean_value】填充
rsi_array: np.ndarray = np.nan_to_num(rsi_array, nan=mean_value)
# 使用历史rsi值计算趋势是否处于牛市区间
# 判断标准为本段历史中所有rsi的值是否都在【rsi_lower】与【rsi_upper】之间
bull_range_signal: bool = (np.all(rsi_array > self.rsi_lower) and
np.all(rsi_array < self.rsi_upper))
# 判断本段历史中是否存在较强的rsi值，只要有一个超过设定的【rsi_highest】即成立
bull_mom_signal: bool = np.any(rsi_array > self.rsi_highest)
return bull_range_signal and bull_mom_signal
RsiIbsSignal
class RsiIbsSignal:
def __init__(
self,
rsi_window: int = 21,
rsi_entry: int = 45,
ibs_entry: float = 0.25
):
# rsi指标窗口
self.rsi_window: int = rsi_window
# 入场rsi指标阈值
self.rsi_entry: int = rsi_entry
# 入场ibs指标阈值
self.ibs_entry: float = ibs_entry
def calculate_signal(self, am: ArrayManager) -> bool:
# 计算rsi指标数值
rsi_value = am.rsi(self.rsi_window)
# 计算ibs指标数值
ibs_value = ((am.close[-1] - am.low[-1]) /
(am.high[-1] - am.low[-1]))
# 计算开仓信号
cond_1 = rsi_value < self.rsi_entry
cond_2 = ibs_value < self.ibs_entry
return cond_1 and cond_2
前文已经详细讲解过各个信号的生成逻辑，这里便不再赘述。
信号合成&交易执行
在主策略的【on_init】函数下，将上述三个信号类实例化为成员对象，并分别传入量价数据缓存容器（通过ArrayManager类的封装）：
class RsiEnsembleStrategy(CtaTemplate):
""""""
author = "Tony"
rrms_rsi_window: int = 14
rrms_rsi_lower: int = 40
rrms_rsi_upper: int = 100
rrms_rsi_highest: int = 70
ris_rsi_window: int = 21
ris_rsi_entry: int = 45
ris_ibs_entry: float = 0.25
rs_rsi_window: int = 2
rs_rsi_lower: int = 15
# 风险资金
risk_capital: int = 1_000_000
parameters = [
"rrms_rsi_window",
"rrms_rsi_lower",
"rrms_rsi_upper",
"rrms_rsi_highest",
"ris_rsi_window",
"ris_rsi_entry",
"ris_ibs_entry",
"rs_rsi_window",
"rs_rsi_lower",
]
variables = []
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
# 初始化信号生成器实例
self.rrms = RsiRangeMomSignal(
self.rrms_rsi_window,
self.rrms_rsi_lower,
self.rrms_rsi_upper,
self.rrms_rsi_highest
)
self.ris = RsiIbsSignal(
self.ris_rsi_window,
self.ris_rsi_entry,
self.ris_ibs_entry
)
self.rs = RsiSignal(
self.rs_rsi_window,
self.rs_rsi_lower,
)
self.load_bar(150)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
# 撤销之前发出的委托
self.cancel_all()
# 对am更新bar数据
am = self.am
am.update_bar(bar)
if not am.inited:
return
# 计算开仓手数
trading_size = int(self.risk_capital / bar.close_price / 100) * 100
# 传入am，计算三个信号的值
rrms_signal = self.rrms.calculate_signal(am)
ris_signal = self.ris.calculate_signal(am)
rs_signal = self.rs.calculate_signal(am)
# 如果任一信号成立则进行开仓
if self.pos == 0:
if rrms_signal or ris_signal or rs_signal:
self.buy(bar.close_price*1.05, trading_size)
# 如果三个信号都不成立则进行平仓
if self.pos != 0:
if not rrms_signal and not ris_signal and not rs_signal:
self.sell(bar.close_price*0.95, abs(self.pos))
# 移动止损逻辑
elif self.pos > 0:
self.sell(bar.close_price*0.95, abs(self.pos), stop=True)
# 推送UI更新
self.put_event()
回测数据和参数
该策略历史回测需要用到的SPY历史数据，可以下载zip文件后解压，找到其中load_bar_data.py脚本文件，然后用Python运行即可自动导入数据库。
具体的回测参数配置如下：
股指IF回测绩效
本文选择的回测数据时间段中SPY整体处于长周期大牛市，因此多头逻辑的交易策略可能天然具有明显优势（毕竟简单买入做多就能赚钱），所以这里选择使用IF股指期货来作为策略有效性的交叉验证：
可以看出，RSI多信号集成策略在IF上的绩效也是相当可观的，虽然自2021年起策略的有效性变差了许多，但是仍然将回撤保持在了一个可控的范围，这也体现了CTA策略中多信号组合的优势。
关于向量化计算
本文代码中多次运用了numpy库中提供的向量化计算函数，例如【np.nan】、【np.mean】、【np.any】、【np.all】等。向量化计算是一种使用数组或矢量操作来处理数据的方法，它具有性能提升、代码简洁、可读性高、跨平台性，适用于大规模数据等优势。
完整策略代码和回测数据文件，可以通过【VeighNa进阶用户交流群】获取：
免责声明
文章中的信息或观点仅供参考，作者不对其准确性或完整性做出任何保证。读者应以其独立判断做出投资决策，作者不对因使用本报告的内容而引致的损失承担任何责任。
