均值回归盈利策略 - 专门在震荡市场中获利
Timeframe
15m
Direction
Long Only
Stoploss
-2.5%
Trailing Stop
No
ROI
0m: 3.0%, 30m: 2.0%, 60m: 1.0%, 120m: 0.0%
Interface Version
3
Startup Candles
N/A
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from pandas import DataFrame
from typing import Optional, Union
from collections import deque
from freqtrade.strategy import (
IStrategy,
Trade,
Order,
DecimalParameter,
IntParameter,
BooleanParameter,
)
import talib.abstract as ta
from technical import qtpylib
class MeanReversionProfitStrategy(IStrategy):
"""
均值回归盈利策略 - 专门在震荡市场中获利
核心思想:
1. 识别震荡市场环境
2. 在价格偏离均值时进场
3. 快速止盈,严格止损
4. 高频交易,积少成多
"""
INTERFACE_VERSION = 3
can_short: bool = False
# 激进盈利设置
minimal_roi = {
"0": 0.03, # 3%止盈
"30": 0.02, # 30分钟后2%止盈
"60": 0.01, # 1小时后1%止盈
"120": 0 # 2小时后平仓
}
stoploss = -0.025 # 2.5%止损
trailing_stop = False
timeframe = '15m'
use_custom_stoploss = False # 使用固定止损
# 策略参数
bb_period = IntParameter(15, 25, default=20, space="buy", optimize=True)
bb_std = DecimalParameter(1.8, 2.5, default=2.0, space="buy", optimize=True)
rsi_period = IntParameter(10, 21, default=14, space="buy", optimize=True)
rsi_oversold = IntParameter(25, 35, default=30, space="buy", optimize=True)
rsi_overbought = IntParameter(65, 80, default=70, space="buy", optimize=True)
# 均值回归参数
bb_entry_threshold = DecimalParameter(0.05, 0.25, default=0.1, space="buy", optimize=True) # 进入布林带下轨的程度
volume_threshold = DecimalParameter(0.8, 1.5, default=1.0, space="buy", optimize=True) # 成交量阈值
# 震荡市场识别参数
volatility_period = IntParameter(20, 50, default=30, space="buy", optimize=True)
trend_strength_threshold = DecimalParameter(15, 30, default=20, space="buy", optimize=True) # ADX阈值
# 快速出场参数
quick_profit_target = DecimalParameter(0.008, 0.025, default=0.015, space="sell", optimize=True) # 1.5%快速止盈
volume_exit_threshold = DecimalParameter(0.3, 0.8, default=0.5, space="sell", optimize=True) # 成交量萎缩出场
startup_candle_count: int = 200
def __init__(self, config: dict):
super().__init__(config)
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_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower'])
dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle']
# RSI - 超买超卖确认
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
# ADX - 趋势强度 (震荡市场识别)
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
dataframe['is_ranging'] = dataframe['adx'] < self.trend_strength_threshold.value
# EMA - 趋势过滤
dataframe['ema_20'] = ta.EMA(dataframe, timeperiod=20)
dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50)
# 成交量指标
dataframe['volume_sma'] = dataframe['volume'].rolling(20).mean()
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
# 波动率计算
dataframe['price_change'] = dataframe['close'].pct_change()
dataframe['volatility'] = dataframe['price_change'].rolling(self.volatility_period.value).std()
# 均值回归信号强度
dataframe['mean_reversion_strength'] = (
(dataframe['bb_percent'] < self.bb_entry_threshold.value).astype(int) +
(dataframe['rsi'] < self.rsi_oversold.value).astype(int) +
(dataframe['close'] < dataframe['ema_20'] * 0.99).astype(int) +
(dataframe['volume_ratio'] > self.volume_threshold.value).astype(int) +
(dataframe['is_ranging']).astype(int)
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
入场条件 - 均值回归信号
"""
# 强均值回归信号:多个条件同时满足
dataframe['enter_long'] = (
# 核心条件:价格远离布林带下轨
(dataframe['bb_percent'] < self.bb_entry_threshold.value) &
# RSI超卖确认
(dataframe['rsi'] < self.rsi_oversold.value) &
# 震荡市场确认 (非强趋势)
(dataframe['is_ranging'] == True) &
# 成交量确认
(dataframe['volume_ratio'] > self.volume_threshold.value) &
# 价格低于短期均线 (进一步确认超卖)
(dataframe['close'] < dataframe['ema_20']) &
# 布林带宽度适中 (避免极端波动)
(dataframe['bb_width'] > 0.02) & (dataframe['bb_width'] < 0.08)
)
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
出场条件 - 快速止盈
"""
dataframe['exit_long'] = (
# 快速反转信号
(dataframe['rsi'] > self.rsi_overbought.value) |
# 回到布林带中轨以上
(dataframe['close'] > dataframe['bb_middle'] * 1.01) |
# 成交量萎缩
(dataframe['volume_ratio'] < self.volume_exit_threshold.value) |
# 趋势转强 (不再是震荡市场)
(dataframe['adx'] > self.trend_strength_threshold.value * 1.5)
)
return dataframe
class TrendFollowProfitStrategy(IStrategy):
"""
趋势跟踪盈利策略 - 专门在趋势市场中获利
核心思想:
1. 识别趋势市场环境
2. 在趋势确认后进场
3. 金字塔加仓,长期持有
4. 追踪止盈,保护利润
"""
INTERFACE_VERSION = 3
can_short: bool = False
# 趋势跟踪设置
minimal_roi = {"0": 10} # 不使用固定ROI
stoploss = -0.08 # 8%止损
trailing_stop = True
trailing_stop_positive = 0.02 # 2%启动追踪
trailing_stop_positive_offset = 0.03 # 3%追踪距离
timeframe = '15m'
use_custom_stoploss = True
# 趋势识别参数
ema_fast = IntParameter(12, 21, default=15, space="buy", optimize=True)
ema_slow = IntParameter(26, 55, default=35, space="buy", optimize=True)
adx_period = IntParameter(10, 21, default=14, space="buy", optimize=True)
adx_threshold = IntParameter(25, 40, default=30, space="buy", optimize=True)
# 趋势确认参数
trend_confirmation_period = IntParameter(3, 8, default=5, space="buy", optimize=True)
momentum_threshold = DecimalParameter(0.003, 0.015, default=0.008, space="buy", optimize=True)
# 加仓参数
enable_pyramid = BooleanParameter(default=True, space="buy", optimize=True)
pyramid_distance = DecimalParameter(0.02, 0.05, default=0.03, space="buy", optimize=True)
max_pyramid_levels = IntParameter(2, 4, default=3, space="buy", optimize=True)
# 利润目标
profit_target_1 = DecimalParameter(0.05, 0.10, default=0.08, space="sell", optimize=True)
profit_target_2 = DecimalParameter(0.15, 0.25, default=0.20, space="sell", optimize=True)
profit_target_3 = DecimalParameter(0.30, 0.50, default=0.40, space="sell", optimize=True)
startup_candle_count: int = 200
def __init__(self, config: dict):
super().__init__(config)
self.pyramid_levels = {}
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪指标
"""
# EMA系统
dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
dataframe['ema_trend'] = dataframe['ema_fast'] > dataframe['ema_slow']
# ADX趋势强度
dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value)
dataframe['strong_trend'] = dataframe['adx'] > self.adx_threshold.value
# MACD动量确认
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macd_positive'] = dataframe['macd'] > dataframe['macdsignal']
# 动量计算
dataframe['momentum'] = dataframe['close'].pct_change(self.trend_confirmation_period.value)
dataframe['strong_momentum'] = dataframe['momentum'] > self.momentum_threshold.value
# 成交量确认
dataframe['volume_sma'] = dataframe['volume'].rolling(20).mean()
dataframe['volume_surge'] = dataframe['volume'] > dataframe['volume_sma'] * 1.2
# 综合趋势信号
dataframe['trend_score'] = (
dataframe['ema_trend'].astype(int) +
dataframe['strong_trend'].astype(int) +
dataframe['macd_positive'].astype(int) +
dataframe['strong_momentum'].astype(int) +
dataframe['volume_surge'].astype(int)
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪入场
"""
dataframe['enter_long'] = (
# 强趋势信号:至少4个确认
(dataframe['trend_score'] >= 4) &
# EMA排列确认
(dataframe['ema_fast'] > dataframe['ema_slow']) &
# 价格在EMA之上
(dataframe['close'] > dataframe['ema_fast']) &
# ADX确认强趋势
(dataframe['strong_trend'] == True) &
# 动量确认
(dataframe['strong_momentum'] == True)
)
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪出场
"""
dataframe['exit_long'] = (
# 趋势转弱
(dataframe['ema_fast'] < dataframe['ema_slow']) |
# ADX下降
(dataframe['adx'] < self.adx_threshold.value * 0.8) |
# MACD转负
(dataframe['macd'] < dataframe['macdsignal']) |
# 动量转负
(dataframe['momentum'] < -self.momentum_threshold.value)
)
return dataframe
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
动态追踪止损
"""
# 分阶段追踪止损
if current_profit >= self.profit_target_3.value:
return current_profit - 0.15 # 保护85%利润
elif current_profit >= self.profit_target_2.value:
return current_profit - 0.08 # 保护92%利润
elif current_profit >= self.profit_target_1.value:
return current_profit - 0.04 # 保护96%利润
elif current_profit >= 0.03:
return current_profit - 0.02 # 保护部分利润
else:
return self.stoploss
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: Optional[str] = None, side: str = 'long', **kwargs) -> bool:
"""
趋势确认入场
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return False
latest = dataframe.iloc[-1]
# 确认趋势信号强度
if latest['trend_score'] < 3:
return False
return True