RSI超买/超卖交易策略
一、交易策略解释
核心思想
RSI超买/超卖反转策略的核心思想是基于相对强弱指标(Relative Strength Index, RSI)在达到极端水平时，市场可能出现价格反转的原理。该策略认为：
当RSI值超过70进入超买区域后，再回落至70以下时，表明市场可能已达到短期高点，价格有下跌趋势的可能，产生做空信号。
当RSI值低于30进入超卖区域后，再回升至30以上时，表明市场可能已达到短期低点，价格有上涨趋势的可能，产生做多信号。
这种策略本质上是一种均值回归(Mean Reversion)思想，认为价格在短期内大幅偏离其均值后，有回归至均值的趋势。RSI作为一个动量指标，帮助交易者识别价格动量的极端值及其随后的反转。
理论基础
RSI指标由J. Welles Wilder Jr.于1978年在其著作《新型技术交易系统》中首次提出。该指标的理论基础建立在以下几个方面：
-
价格动量与反转理论：市场过度扩张理论(Market Overextension Theory)指出，价格在短期内不可能无限制地向一个方向发展，当达到极端水平时，会因供需平衡的调整而反转。
-
实证研究支持：多项学术研究证实了RSI在预测短期市场反转方面的有效性。例如，Andrew Lo和Jasmina Hasanhodzic在《市场的适应性本质》中发现，RSI在识别超买和超卖条件方面具有统计显著性。
-
技术分析理论支持：价格与震荡指标的背离(Divergence)理论认为，当价格趋势与技术指标趋势不一致时，往往预示着价格趋势可能即将逆转。
策略适用场景
-
区间震荡市场：在缺乏明显趋势的横盘市场中，价格往往在一定区间内震荡，RSI指标能有效捕捉区间的上下边界，提供较为准确的反转信号。
-
避免强趋势市场：在强烈的单向趋势市场中，RSI可能长时间保持在超买或超卖区域，反转信号可能导致过早入场，因此该策略不适合强趋势市场。
-
与其他指标结合使用：结合其他技术指标如移动平均线、MACD或布林带等，可以提高信号的可靠性。例如，只在价格位于移动平均线下方时考虑RSI超卖信号。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
RSI指标的标准公式为：
RSI = 100 - (100 / (1 + RS))
RS = 平均上涨点数 / 平均下跌点数
本策略中，使用天勤SDK的内置RSI函数计算相对强弱指标，并通过缩短RSI周期(6日)，使指标对价格变化更加敏感，并调整传统的超买超卖阈值(65/35)以获取更多交易信号。
交易逻辑
入场条件：
- 做多信号：RSI从低于35回升至35以上，且之前的RSI值已经进入过超卖区域（RSI < 35）
- 做空信号：RSI从高于65回落至65以下，且之前的RSI值已经进入过超买区域（RSI > 65）
出场条件：
- 止盈：当RSI达到反向的极端区域
- 多头持仓时RSI升至65以上
- 空头持仓时RSI降至35以下
- 止损：使用固定止损，设置为入场点的1%
- 时间止损：如果持仓超过10个交易日没有触发止盈或止损，则强制平仓
风险管理：
- 资金风险控制：每次交易风险敞口不超过账户总资金的2%，通过止损比例和头寸大小配合实现
- 指标确认机制：通过跟踪RSI是否真正进入过超买/超卖区域，避免边界波动产生的错误信号
回测
回测初始设置
- 测试周期： 2020 年 4 月 20 日 - 2020 年 11 月 20 日
- 交易品种： DCE.a2101
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from tqsdk.ta import RSI
from tqsdk.tafunc import time_to_str
# ===== 全局参数设置 =====
SYMBOL = "DCE.a2101"
POSITION_SIZE = 500 # 每次交易手数
START_DATE = date(2020, 4, 20) # 回测开始日期
END_DATE = date(2020, 11, 20) # 回测结束日期
# RSI参数
RSI_PERIOD = 6 # RSI计算周期，
OVERBOUGHT_THRESHOLD = 65 # 超买阈值
OVERSOLD_THRESHOLD = 35 # 超卖阈值
# 风控参数
STOP_LOSS_PERCENT = 0.01 # 止损百分比
TIME_STOP_DAYS = 10 # 缩短时间止损天数
# ===== 全局变量 =====
current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓
entry_price = 0 # 开仓价格
stop_loss_price = 0 # 止损价格
entry_date = None # 开仓日期
was_overbought = False # 是否曾进入超买区域
was_oversold = False # 是否曾进入超卖区域
# ===== 策略开始 =====
print("开始运行RSI超买/超卖反转策略...")
# 创建API实例
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账号", "快期密码"))
# 订阅合约的K线数据
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24) # 日线数据
# 创建目标持仓任务
target_pos = TargetPosTask(api, SYMBOL)
try:
while True:
# 等待更新
api.wait_update()
# 如果K线有更新
if api.is_changing(klines.iloc[-1], "datetime"):
# 确保有足够的数据计算指标
if len(klines) < RSI_PERIOD + 10:
continue
# 使用天勤的RSI函数计算
rsi = RSI(klines, RSI_PERIOD)
current_rsi = float(rsi.iloc[-1].iloc[0])
previous_rsi = float(rsi.iloc[-2].iloc[0])
# 更新超买/超卖状态
if previous_rsi > OVERBOUGHT_THRESHOLD:
was_overbought = True
if previous_rsi < OVERSOLD_THRESHOLD:
was_oversold = True
# 获取最新数据
current_price = float(klines.close.iloc[-1])
current_timestamp = klines.datetime.iloc[-1]
current_datetime = time_to_str(current_timestamp) # 使用time_to_str转换时间
# 打印当前状态
print(f"日期: {current_datetime}, 价格: {current_price:.2f}, RSI: {current_rsi:.2f}")
# ===== 交易逻辑 =====
# 空仓状态 - 寻找开仓机会
if current_direction == 0:
# 多头开仓条件：RSI从超卖区域回升
if was_oversold and previous_rsi < OVERSOLD_THRESHOLD and current_rsi > OVERSOLD_THRESHOLD:
current_direction = 1
target_pos.set_target_volume(POSITION_SIZE)
entry_price = current_price
stop_loss_price = entry_price * (1 - STOP_LOSS_PERCENT)
entry_date = current_timestamp # 存储时间戳
print(f"多头开仓: 价格={entry_price:.2f}, 止损={stop_loss_price:.2f}")
was_oversold = False # 重置超卖状态
# 空头开仓条件：RSI从超买区域回落
elif was_overbought and previous_rsi > OVERBOUGHT_THRESHOLD and current_rsi < OVERBOUGHT_THRESHOLD:
current_direction = -1
target_pos.set_target_volume(-POSITION_SIZE)
entry_price = current_price
stop_loss_price = entry_price * (1 + STOP_LOSS_PERCENT)
entry_date = current_timestamp # 存储时间戳
print(f"空头开仓: 价格={entry_price:.2f}, 止损={stop_loss_price:.2f}")
was_overbought = False # 重置超买状态
# 多头持仓 - 检查平仓条件
elif current_direction == 1:
# 止损条件
if current_price <= stop_loss_price:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止损平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止盈条件：RSI进入超买区域
elif current_rsi > OVERBOUGHT_THRESHOLD:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 时间止损
elif (current_timestamp - entry_date) / (60 * 60 * 24) >= TIME_STOP_DAYS: # 计算天数差
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头时间止损: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 空头持仓 - 检查平仓条件
elif current_direction == -1:
# 止损条件
if current_price >= stop_loss_price:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止损平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止盈条件：RSI进入超卖区域
elif current_rsi < OVERSOLD_THRESHOLD:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 时间止损
elif (current_timestamp - entry_date) / (60 * 60 * 24) >= TIME_STOP_DAYS: # 计算天数差
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头时间止损: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
except BacktestFinished as e:
print("回测结束")
api.close()