问题描述
请写一个 RSI 背离检测的策略代码
解决方案
这是一个基于 QMT (Quantitative Trading) 平台的 RSI（相对强弱指标）背离策略 代码。
策略逻辑说明
RSI 背离通常被视为趋势反转的信号。本策略包含两种背离检测：
-
底背离（看涨/买入信号）：
- 价格创新低（当前低点 < 上一个低点）。
- RSI 指标未创新低（当前 RSI 低点 > 上一个 RSI 低点）。
- 过滤条件： RSI 处于超卖区域（例如 < 30）时信号更有效。
-
顶背离（看跌/卖出信号）：
- 价格创新高（当前高点 > 上一个高点）。
- RSI 指标未创新高（当前 RSI 高点 < 上一个 RSI 高点）。
- 过滤条件： RSI 处于超买区域（例如 > 70）时信号更有效。
QMT 策略代码
# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import talib
def init(ContextInfo):
"""
初始化函数
"""
# 设置资金账号 (请替换为您自己的资金账号)
ContextInfo.accID = 'YOUR_ACCOUNT_ID'
ContextInfo.set_account(ContextInfo.accID)
# 策略参数设置
ContextInfo.rsi_period = 14 # RSI 计算周期
ContextInfo.rsi_overbought = 70 # 超买阈值
ContextInfo.rsi_oversold = 30 # 超卖阈值
ContextInfo.lookback_window = 5 # 寻找局部极值的窗口大小（左右各n根k线）
ContextInfo.history_len = 100 # 获取历史数据的长度
# 设置股票池 (示例：浦发银行)
ContextInfo.target_stock = '600000.SH'
ContextInfo.set_universe([ContextInfo.target_stock])
# 设置交易费率等 (回测用)
ContextInfo.set_commission(0, [0.0003, 0.0013, 0.0003, 0.0003, 0.0003, 5])
def get_peaks_and_troughs(data_series, window=3):
"""
辅助函数：寻找序列中的局部高点(峰)和局部低点(谷)的索引
返回: peaks (list of indices), troughs (list of indices)
"""
peaks = []
troughs = []
# 遍历数据，除去头尾，寻找局部极值
for i in range(window, len(data_series) - window):
is_peak = True
is_trough = True
# 检查是否为局部最大值
for j in range(1, window + 1):
if data_series[i] <= data_series[i-j] or data_series[i] <= data_series[i+j]:
is_peak = False
break
# 检查是否为局部最小值
for j in range(1, window + 1):
if data_series[i] >= data_series[i-j] or data_series[i] >= data_series[i+j]:
is_trough = False
break
if is_peak:
peaks.append(i)
if is_trough:
troughs.append(i)
return peaks, troughs
def handlebar(ContextInfo):
"""
K线周期运行函数
"""
# 获取当前正在处理的股票代码
stock_code = ContextInfo.target_stock
# 获取历史行情数据 (High, Low, Close)
# 注意：get_market_data_ex 返回的是字典 {code: dataframe}
market_data = ContextInfo.get_market_data_ex(
['high', 'low', 'close'],
[stock_code],
period=ContextInfo.period,
count=ContextInfo.history_len,
dividend_type='front' # 前复权
)
if stock_code not in market_data:
return
df = market_data[stock_code]
# 确保数据量足够计算 RSI
if len(df) < ContextInfo.rsi_period + ContextInfo.lookback_window * 2:
return
# 提取价格序列
close_prices = df['close'].values
low_prices = df['low'].values
high_prices = df['high'].values
# 计算 RSI 指标
# talib.RSI 需要 numpy array 类型
rsi_values = talib.RSI(close_prices, timeperiod=ContextInfo.rsi_period)
# 移除 NaN 值以便处理索引
valid_start_idx = ContextInfo.rsi_period
if np.isnan(rsi_values[valid_start_idx]):
return
# --- 检测底背离 (Bullish Divergence) ---
# 1. 寻找 RSI 的局部低点
_, rsi_troughs = get_peaks_and_troughs(rsi_values, window=ContextInfo.lookback_window)
buy_signal = False
if len(rsi_troughs) >= 2:
# 获取最近的两个 RSI 低点索引
curr_trough_idx = rsi_troughs[-1] # 最近的低点
prev_trough_idx = rsi_troughs[-2] # 前一个低点
# 确保最近的低点发生在近期 (例如最近 10 根 K 线内)，避免信号过时
if len(rsi_values) - curr_trough_idx <= 10:
# 获取对应时间点的 RSI 值和 价格(Low) 值
curr_rsi_val = rsi_values[curr_trough_idx]
prev_rsi_val = rsi_values[prev_trough_idx]
curr_price_low = low_prices[curr_trough_idx]
prev_price_low = low_prices[prev_trough_idx]
# 判断底背离条件：
# 1. 价格创新低 (当前价格低点 < 前次价格低点)
# 2. RSI 抬高 (当前 RSI 低点 > 前次 RSI 低点)
# 3. (可选) 当前 RSI 处于超卖区附近 (例如 < 30 或 < 40)
if (curr_price_low < prev_price_low) and \
(curr_rsi_val > prev_rsi_val) and \
(curr_rsi_val < 40): # 放宽一点超卖限制以便更容易触发
buy_signal = True
print(f"[{ContextInfo.barpos}] 触发底背离买入信号: {stock_code}")
print(f" 价格: {prev_price_low} -> {curr_price_low} (创新低)")
print(f" RSI : {prev_rsi_val:.2f} -> {curr_rsi_val:.2f} (抬高)")
# --- 检测顶背离 (Bearish Divergence) ---
# 1. 寻找 RSI 的局部高点
rsi_peaks, _ = get_peaks_and_troughs(rsi_values, window=ContextInfo.lookback_window)
sell_signal = False
if len(rsi_peaks) >= 2:
# 获取最近的两个 RSI 高点索引
curr_peak_idx = rsi_peaks[-1]
prev_peak_idx = rsi_peaks[-2]
if len(rsi_values) - curr_peak_idx <= 10:
curr_rsi_val = rsi_values[curr_peak_idx]
prev_rsi_val = rsi_values[prev_peak_idx]
curr_price_high = high_prices[curr_peak_idx]
prev_price_high = high_prices[prev_peak_idx]
# 判断顶背离条件：
# 1. 价格创新高
# 2. RSI 走低
# 3. (可选) RSI 处于超买区
if (curr_price_high > prev_price_high) and \
(curr_rsi_val < prev_rsi_val) and \
(curr_rsi_val > 60):
sell_signal = True
print(f"[{ContextInfo.barpos}] 触发顶背离卖出信号: {stock_code}")
print(f" 价格: {prev_price_high} -> {curr_price_high} (创新高)")
print(f" RSI : {prev_rsi_val:.2f} -> {curr_rsi_val:.2f} (走低)")
# --- 执行交易 ---
# 获取当前持仓
position = ContextInfo.get_position(stock_code)
# 买入逻辑
if buy_signal:
# 全仓买入 (目标仓位 100%)
order_target_percent(stock_code, 1.0, ContextInfo, ContextInfo.accID)
# 卖出逻辑
elif sell_signal:
# 清仓卖出 (目标仓位 0%)
order_target_percent(stock_code, 0.0, ContextInfo, ContextInfo.accID)
代码关键点解析
-
数据获取 (
get_market_data_ex):- 我们获取了
high,low,close价格。 - 注意：背离判断中，底背离通常比较最低价(Low)，顶背离通常比较最高价(High)，这样比单纯使用收盘价更准确。
- 我们获取了
-
RSI 计算:
- 使用了 QMT 内置的
talib库，计算效率高。
- 使用了 QMT 内置的
-
极值检测 (
get_peaks_and_troughs):- 这是一个自定义函数。因为背离是基于“波峰”和“波谷”的比较，而不是简单的逐日比较。
window参数决定了判断极值的范围。例如window=5意味着一个点必须比它左右各 5 个点都高/低，才算作极值点。
-
背离逻辑:
- 底背离：最近的 RSI 波谷比上一个波谷高，但对应的股价波谷比上一个股价波谷低。
- 顶背离：最近的 RSI 波峰比上一个波峰低，但对应的股价波峰比上一个股价波峰高。
-
交易执行:
- 使用了
order_target_percent。这是一种简便的调仓函数。 1.0代表买入至满仓，0.0代表清仓。
- 使用了
如何在 QMT 中使用
- 打开 QMT 客户端，进入 “模型研究” -> “策略编辑器”。
- 新建一个 Python 策略。
- 将上述代码复制粘贴进去。
- 修改账号：将
init函数中的YOUR_ACCOUNT_ID替换为您真实的资金账号。 - 回测：点击“回测”按钮，选择标的（如 600000.SH）和周期（建议日线或 60 分钟线），查看策略表现。
- 实盘：如果用于实盘，请务必先在模拟盘充分测试，并确保
ContextInfo.accID设置正确。
Q&A
Q: 为什么代码中要自己写 get_peaks_and_troughs 函数？
A: RSI 背离的核心在于比较“波峰”和“波谷”。标准的 API 通常只提供当前值。为了找到“上一个波峰”，我们需要遍历历史数据来识别局部极值点。
Q: 为什么有时候肉眼看着背离了，策略没交易？
A: 这通常取决于 ContextInfo.lookback_window（回溯窗口）的设置。如果设置为 5，意味着极值点确立需要该点之后的 5 根 K 线都比它低/高。这会带来一定的滞后性，但能过滤掉很多假信号。如果希望信号更灵敏，可以将 lookback_window 改为 3，但假信号可能会增加。
Q: 这个策略适合什么周期？
A: RSI 背离在震荡市或趋势末端较有效。建议在 日线 或 1小时线 上使用。分钟级（如 1分钟）噪音太大，背离信号频繁且不可靠。