"""
均值回归策略
基于布林带的均值回归交易策略
"""
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import talib.abstract as ta
from technical import qtpylib
import numpy as np
import logging
logger = logging.getLogger(name)
class MeanReversionStrategy(IStrategy):
"""
均值回归策略 - 基于布林带的反转交易
策略逻辑:
1. 使用布林带识别超买超卖状态
2. 当价格触及下轨时买入（低买）
3. 当价格触及上轨时卖出（高卖）
4. 结合RSI确认反转信号
5. 使用ATR动态调整止损
"""
INTERFACE_VERSION = 3
# 基础策略参数
minimal_roi = {
"0": 0.08, # 8%收益立即止盈
"30": 0.04, # 30分钟后4%收益
"60": 0.02, # 1小时后2%收益
"120": 0.01 # 2小时后1%收益
}
stoploss = -0.05 # 5%止损
timeframe = '15m' # 15分钟时间框架
# 策略控制参数
can_short = False
startup_candle_count = 50
process_only_new_candles = True
use_exit_signal = True
use_custom_stoploss = True
# 布林带参数
bb_period = IntParameter(15, 25, default=20, space="buy")
bb_std = DecimalParameter(1.5, 2.5, default=2.0, space="buy")
# RSI参数
rsi_period = IntParameter(10, 20, default=14, space="buy")
rsi_oversold = IntParameter(35, 50, default=45, space="buy") # 放宽RSI条件
rsi_overbought = IntParameter(55, 75, default=65, space="sell") # 放宽RSI条件
# 均值回归确认参数
price_deviation_threshold = DecimalParameter(0.001, 0.01, default=0.005, space="buy") # 大幅放宽偏离度
volume_threshold = DecimalParameter(0.8, 1.5, default=1.0, space="buy") # 放宽成交量要求
# ATR参数（用于动态止损）
atr_period = IntParameter(10, 20, default=14, space="buy")
atr_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="sell")
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算技术指标
"""
# 布林带
bollinger = qtpylib.bollinger_bands(
dataframe['close'],
window=self.bb_period.value,
stds=self.bb_std.value
)
dataframe['bb_lower'] = bollinger['lower']
dataframe['bb_middle'] = bollinger['mid']
dataframe['bb_upper'] = bollinger['upper']
# 布林带宽度和位置
dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle']
dataframe['bb_position'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower'])
# RSI指标
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
# 均线系统
dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20)
dataframe['ema_12'] = ta.EMA(dataframe, timeperiod=12)
dataframe['ema_26'] = ta.EMA(dataframe, timeperiod=26)
# 价格偏离度
dataframe['price_deviation'] = abs(dataframe['close'] - dataframe['sma_20']) / dataframe['sma_20']
# 成交量指标
dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
# ATR（平均真实波动率）
dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
# MACD（作为趋势确认）
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macd_signal'] = macd['macdsignal']
dataframe['macd_hist'] = macd['macdhist']
# 威廉指标（额外的超买超卖确认）
dataframe['williams_r'] = ta.WILLR(dataframe, timeperiod=14)
# 价格动量
dataframe['momentum'] = ta.MOM(dataframe, timeperiod=10)
# 支撑阻力位（简化版）
dataframe['resistance'] = dataframe['high'].rolling(window=20).max()
dataframe['support'] = dataframe['low'].rolling(window=20).min()
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
均值回归买入信号
买入条件（低买策略）：
1. 价格触及或突破布林带下轨
2. RSI处于超卖状态
3. 成交量放大确认
4. 价格偏离均线足够远
"""
conditions = [
# 主要条件：价格接近布林带下轨
(dataframe['close'] <= dataframe['bb_lower'] * 1.005) | # 允许价格接近下轨
(dataframe['low'] <= dataframe['bb_lower']),
# RSI相对较低（放宽条件）
dataframe['rsi'] < self.rsi_oversold.value,
# 威廉指标相对较低（放宽条件）
dataframe['williams_r'] < -50, # 从-80放宽到-50
# 价格偏离均线（大幅放宽）
dataframe['price_deviation'] > self.price_deviation_threshold.value,
# 成交量确认（放宽）
dataframe['volume_ratio'] > self.volume_threshold.value,
# 趋势不要太强（放宽条件）
dataframe['ema_12'] > dataframe['ema_26'] * 0.95, # 允许更多下跌趋势
# 布林带宽度足够（大幅放宽）
dataframe['bb_width'] > 0.005, # 从2%降到0.5%
# 动量转正（可选条件）
dataframe['momentum'] > dataframe['momentum'].shift(1),
]
# 组合主要条件（简化逻辑）
dataframe.loc[
(
conditions[0] & # 布林带下轨
conditions[1] & # RSI相对较低
conditions[2] & # 威廉指标相对较低
conditions[3] & # 价格偏离（已大幅放宽）
conditions[4] & # 成交量（已放宽）
conditions[5] # 趋势条件（已放宽）
# 暂时移除布林带宽度和动量条件以增加信号
),
'enter_long'
] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
均值回归卖出信号
卖出条件（高卖策略）：
1. 价格触及或突破布林带上轨
2. RSI进入超买区域
3. 价格回归至均线附近
"""
conditions = [
# 主要条件：价格接近布林带上轨
(dataframe['close'] >= dataframe['bb_upper'] * 0.998) | # 允许接近上轨
(dataframe['high'] >= dataframe['bb_upper']),
# RSI相对较高
dataframe['rsi'] > self.rsi_overbought.value,
# 威廉指标相对较高
dataframe['williams_r'] > -30, # 从-20放宽到-30
# 价格回归至布林带中上部（放宽）
dataframe['bb_position'] > 0.7, # 从0.8降到0.7
]
# 替代卖出条件：趋势转弱
trend_weak_conditions = [
# MACD转弱
qtpylib.crossed_below(dataframe['macd'], dataframe['macd_signal']),
# 短期均线下穿长期均线
qtpylib.crossed_below(dataframe['ema_12'], dataframe['ema_26']),
# 动量转负
dataframe['momentum'] < 0,
# RSI从高位回落
(dataframe['rsi'] < dataframe['rsi'].shift(1)) &
(dataframe['rsi'] > 60),
]
dataframe.loc[
(
# 主要卖出条件（均值回归完成）
(conditions[0] & conditions[1] & conditions[2] & conditions[3]) |
# 替代卖出条件（趋势转弱的任一信号）
trend_weak_conditions[0] |
trend_weak_conditions[1] |
(trend_weak_conditions[2] & trend_weak_conditions[3])
),
'exit_long'
] = 1
return dataframe
def custom_stoploss(self, pair: str, trade: 'Trade', current_time,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
动态止损基于ATR
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# 基于ATR的动态止损
atr_stop = last_candle['atr'] * self.atr_multiplier.value
atr_stop_pct = atr_stop / current_rate
# 根据持仓时间调整止损
if current_profit > 0.02: # 盈利超过2%时收紧止损
return max(-atr_stop_pct * 0.5, -0.02)
elif current_profit > 0.01: # 盈利超过1%时适度收紧
return max(-atr_stop_pct * 0.7, -0.03)
else:
# 正常ATR止损，但不超过最大止损
return max(-atr_stop_pct, self.stoploss)
def confirm_trade_entry(self, pair: str, order_type: str, amount: float,
rate: float, time_in_force: str, current_time,
entry_tag: str, side: str, **kwargs) -> bool:
"""
交易确认 - 最后的风险检查
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return False
last_candle = dataframe.iloc[-1].squeeze()
# 确保布林带宽度足够（避免在极度窄幅震荡中交易）
if last_candle['bb_width'] < 0.015: # 小于1.5%
return False
# 确保不是在强烈下跌趋势中
if last_candle['ema_12'] < last_candle['ema_26'] * 0.95: # 短期均线低于长期5%以上
return False
# 确保RSI不会过度超卖（可能继续下跌）
if last_candle['rsi'] < 15: # 极度超卖可能继续下跌
return False
return True
def custom_entry_price(self, pair: str, current_time, proposed_rate: float,
entry_tag: str, side: str, **kwargs) -> float:
"""
自定义入场价格 - 尝试获得更好的入场点
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# 如果当前价格高于布林带下轨，尝试以更低价格入场
if proposed_rate > last_candle['bb_lower']:
# 在布林带下轨和当前价格之间设置限价单
target_price = (proposed_rate + last_candle['bb_lower']) / 2
return min(target_price, proposed_rate * 0.999) # 最多降低0.1%
return proposed_rate
def informative_pairs(self):
"""
定义需要的额外数据对
"""
pairs = self.dp.current_whitelist()
informative_pairs = []
# 添加1小时时间框架数据用于趋势确认
for pair in pairs:
informative_pairs.append((pair, '1h'))
return informative_pairs
def leverage(self, pair: str, current_time, current_rate: float,
proposed_leverage: float, max_leverage: float, entry_tag: str,
side: str, **kwargs) -> float:
"""
杠杆设置 - 均值回归策略使用保守杠杆
"""
return 1.0 # 不使用杠杆，降低风险
def test_strategy():
"""
测试策略的基本功能
"""
import pandas as pd
import numpy as np
print("测试均值回归策略...")
# 创建测试数据
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=200, freq='15min')
# 模拟有均值回归特征的价格数据
price = 50000 # BTC起始价格
prices = [price]
for i in range(199):
# 添加均值回归特性：价格偏离均值时有回归倾向
mean_price = np.mean(prices[-20:]) if len(prices) >= 20 else price
deviation = (price - mean_price) / mean_price
# 均值回归力度
reversion_force = -deviation * 0.1
random_walk = np.random.normal(0, 0.005)
change = reversion_force + random_walk
price *= (1 + change)
prices.append(price)
# 生成测试数据
data = pd.DataFrame({
'timestamp': dates,
'open': prices,
'high': [p * (1 + abs(np.random.normal(0, 0.002))) for p in prices],
'low': [p * (1 - abs(np.random.normal(0, 0.002))) for p in prices],
'close': prices,
'volume': np.random.randint(100, 1000, 200)
})
# 测试策略
strategy = MeanReversionStrategy()
# 计算指标
data_with_indicators = strategy.populate_indicators(data, {'pair': 'BTC/USDT'})
# 生成信号
data_with_signals = strategy.populate_entry_trend(data_with_indicators, {'pair': 'BTC/USDT'})
data_with_signals = strategy.populate_exit_trend(data_with_signals, {'pair': 'BTC/USDT'})
# 统计信号
buy_signals = data_with_signals['enter_long'].sum() if 'enter_long' in data_with_signals.columns else 0
sell_signals = data_with_signals['exit_long'].sum() if 'exit_long' in data_with_signals.columns else 0
print(f"测试结果:")
print(f"数据点数: {len(data)}")
print(f"买入信号: {buy_signals}")
print(f"卖出信号: {sell_signals}")
print(f"价格范围: {min(prices):.2f} - {max(prices):.2f}")
# 显示一些关键指标
print(f"\n关键指标示例（最后5行）:")
key_columns = ['close', 'bb_lower', 'bb_upper', 'rsi', 'bb_position']
available_columns = [col for col in key_columns if col in data_with_signals.columns]
if available_columns:
print(data_with_signals[available_columns].tail())
return data_with_signals
if name == "main":
test_strategy()
Top comments (0)