海龟法则：20日突破入场，10日反向突破出场，ATR动态止损
Timeframe
1h
Direction
Long & Short
Stoploss
-4.0%
Trailing Stop
No
ROI
0m: 20.0%, 480m: 10.0%, 1440m: 5.0%, 2880m: 0.0%
Interface Version
3
Startup Candles
100
Indicators
12
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
Classic Trading Strategies — 5 different approaches
Turtle, Donchian, RSI-Larry, Bollinger reversion, Dual Thrust
"""
from pandas import DataFrame
import talib.abstract as ta
import numpy as np
from freqtrade.strategy import IStrategy
# ==============================
# 1. 海龟交易法则 Turtle Trading
# ==============================
class Turtle(IStrategy):
"""海龟法则：20日突破入场，10日反向突破出场，ATR动态止损"""
INTERFACE_VERSION = 3; timeframe = '1h'; can_short = True
stoploss = -0.04; trailing_stop = False; use_custom_stoploss = True
minimal_roi = {"0": 0.20, "480": 0.10, "1440": 0.05, "2880": 0}
max_open_trades = 4; startup_candle_count = 100
process_only_new_candles = True; use_exit_signal = True
def populate_indicators(self, d, m):
d['dc_upper'] = d['high'].rolling(20).max().shift(1)
d['dc_lower'] = d['low'].rolling(20).min().shift(1)
d['dc_exit_u'] = d['high'].rolling(10).max().shift(1)
d['dc_exit_l'] = d['low'].rolling(10).min().shift(1)
d['atr'] = ta.ATR(d, timeperiod=20)
d['ema20'] = ta.EMA(d, timeperiod=20)
d['ema50'] = ta.EMA(d, timeperiod=50)
return d
def populate_entry_trend(self, d, m):
# 海龟做多：价格突破20日高点，且EMA20>EMA50（趋势向上）
d.loc[(d['close'] > d['dc_upper']) & (d['ema20'] > d['ema50']) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'turtle_L')
# 海龟做空：价格跌破20日低点，且EMA20<EMA50（趋势向下）
d.loc[(d['close'] < d['dc_lower']) & (d['ema20'] < d['ema50']) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'turtle_S')
return d
def populate_exit_trend(self, d, m):
# 10日反向突破出场
d.loc[d['close'] < d['dc_exit_l'], ['exit_long', 'exit_tag']] = (1, 'turtle_exit')
d.loc[d['close'] > d['dc_exit_u'], ['exit_short', 'exit_tag']] = (1, 'turtle_exit')
return d
def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
"""海龟止损：2倍ATR"""
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or len(df) == 0: return -0.02
atr = float(df.iloc[-1].get('atr', df['close'].iloc[-1]*0.01))
return -(atr * 2.0 / current_rate)
# ==============================
# 2. 唐奇安通道 + ADX 趋势过滤
# ==============================
class Donchian(IStrategy):
"""Donchian Channel breakout with ADX strength filter"""
INTERFACE_VERSION = 3; timeframe = '30m'; can_short = True
stoploss = -0.025; trailing_stop = True
trailing_stop_positive = 0.008; trailing_stop_positive_offset = 0.018
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.06, "480": 0.04, "1440": 0.025, "4320": 0}
max_open_trades = 6; startup_candle_count = 100
process_only_new_candles = True; use_exit_signal = False
def populate_indicators(self, d, m):
d['dc_h'] = d['high'].rolling(20).max().shift(1)
d['dc_l'] = d['low'].rolling(20).min().shift(1)
d['dc_m'] = (d['dc_h'] + d['dc_l']) / 2
d['adx'] = ta.ADX(d, timeperiod=14)
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
d['atr'] = ta.ATR(d, timeperiod=14) / d['close'] * 100
return d
def populate_entry_trend(self, d, m):
d.loc[(d['close'] > d['dc_h']) & (d['adx'] > 25) & (d['vr'] > 1.2) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'dc_L')
d.loc[(d['close'] < d['dc_l']) & (d['adx'] > 25) & (d['vr'] > 1.2) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'dc_S')
return d
def populate_exit_trend(self, d, m): return d
# ==============================
# 3. Larry Williams RSI 均值回归
# ==============================
class RsiLarry(IStrategy):
"""Larry Connors RSI-2: extreme RSI pullback with trend filter"""
INTERFACE_VERSION = 3; timeframe = '5m'; can_short = True
stoploss = -0.015; trailing_stop = True
trailing_stop_positive = 0.005; trailing_stop_positive_offset = 0.012
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.03, "240": 0.02, "960": 0.01, "2880": 0}
max_open_trades = 8; startup_candle_count = 50
process_only_new_candles = True; use_exit_signal = False
def populate_indicators(self, d, m):
d['rsi2'] = ta.RSI(d, timeperiod=2)
d['ema100'] = ta.EMA(d, timeperiod=100)
d['ema200'] = ta.EMA(d, timeperiod=200)
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
return d
def populate_entry_trend(self, d, m):
# RSI(2) < 10 超卖 + 长期趋势向上 → 反弹做多
d.loc[(d['rsi2'] < 10) & (d['close'] > d['ema200']) & (d['ema100'] > d['ema200']) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'rsi_L')
# RSI(2) > 90 超买 + 长期趋势向下 → 回调做空
d.loc[(d['rsi2'] > 90) & (d['close'] < d['ema200']) & (d['ema100'] < d['ema200']) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'rsi_S')
return d
def populate_exit_trend(self, d, m): return d
# ==============================
# 4. 布林带均值回归
# ==============================
class BollingerMR(IStrategy):
"""Bollinger Band Mean Reversion: 2.5σ extreme + momentum confirmation"""
INTERFACE_VERSION = 3; timeframe = '15m'; can_short = True
stoploss = -0.02; trailing_stop = True
trailing_stop_positive = 0.006; trailing_stop_positive_offset = 0.014
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.04, "480": 0.025, "1440": 0.015, "2880": 0}
max_open_trades = 6; startup_candle_count = 60
process_only_new_candles = True; use_exit_signal = True
def populate_indicators(self, d, m):
bb = ta.BBANDS(d, timeperiod=20, nbdevup=2.5, nbdevdn=2.5)
d['bb_u'] = bb['upperband']; d['bb_l'] = bb['lowerband']; d['bb_m'] = bb['middleband']
d['bb_p'] = (d['close'] - d['bb_l']) / (d['bb_u'] - d['bb_l'] + 0.0001)
d['rsi'] = ta.RSI(d, timeperiod=14)
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
st = ta.STOCH(d); d['sk'] = st['slowk']; d['sd'] = st['slowd']
return d
def populate_entry_trend(self, d, m):
# 触及下轨 + RSI<30 + 随机指标金叉 → 均值回归做多
d.loc[(d['bb_p'] < 0.05) & (d['rsi'] < 30) & (d['sk'] > d['sd']) & (d['vr'] > 1.2) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'bb_L')
# 触及上轨 + RSI>70 + 随机指标死叉 → 均值回归做空
d.loc[(d['bb_p'] > 0.95) & (d['rsi'] > 70) & (d['sk'] < d['sd']) & (d['vr'] > 1.2) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'bb_S')
return d
def populate_exit_trend(self, d, m):
# 回到中轨出场
d.loc[d['bb_p'] > 0.45, ['exit_long', 'exit_tag']] = (1, 'bb_exit')
d.loc[d['bb_p'] < 0.55, ['exit_short', 'exit_tag']] = (1, 'bb_exit')
return d
# ==============================
# 5. Dual Thrust 双推力突破
# ==============================
class DualThrust(IStrategy):
"""Dual Thrust: Range breakout with adaptive range"""
INTERFACE_VERSION = 3; timeframe = '30m'; can_short = True
stoploss = -0.02; trailing_stop = True
trailing_stop_positive = 0.006; trailing_stop_positive_offset = 0.015
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.05, "480": 0.03, "1440": 0.02, "2880": 0}
max_open_trades = 6; startup_candle_count = 80
process_only_new_candles = True; use_exit_signal = False
def populate_indicators(self, d, m):
# Dual Thrust: 用前4根K线的范围
lookback = 4; k1 = 0.5; k2 = 0.5
d['hh'] = d['high'].rolling(lookback).max()
d['hc'] = d['close'].rolling(lookback).max()
d['lc'] = d['close'].rolling(lookback).min()
d['ll'] = d['low'].rolling(lookback).min()
# Range
d['range'] = np.maximum(d['hh'] - d['lc'], d['hc'] - d['ll'])
# 上轨 = 开盘 + k1 * range
d['open_shift'] = d['open']
d['upper'] = d['open_shift'] + k1 * d['range'].shift(1)
d['lower'] = d['open_shift'] - k2 * d['range'].shift(1)
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
d['adx'] = ta.ADX(d, timeperiod=14)
return d
def populate_entry_trend(self, d, m):
d.loc[(d['close'] > d['upper']) & (d['adx'] > 20) & (d['vr'] > 1.1) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'dt_L')
d.loc[(d['close'] < d['lower']) & (d['adx'] > 20) & (d['vr'] > 1.1) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'dt_S')
return d
def populate_exit_trend(self, d, m): return d
# ==============================
# 6. Keltner Channel squeeze
# ==============================
class KeltnerSqueeze(IStrategy):
"""Keltner Channel + BB squeeze breakout"""
INTERFACE_VERSION = 3; timeframe = '15m'; can_short = True
stoploss = -0.02; trailing_stop = True
trailing_stop_positive = 0.005; trailing_stop_positive_offset = 0.015
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.05, "480": 0.03, "1440": 0.02, "2880": 0}
max_open_trades = 6; startup_candle_count = 60
process_only_new_candles = True; use_exit_signal = False
def populate_indicators(self, d, m):
# Keltner Channel
d['kc_mid'] = ta.EMA(d, timeperiod=20)
d['atr'] = ta.ATR(d, timeperiod=10)
d['kc_u'] = d['kc_mid'] + 1.5 * d['atr']
d['kc_l'] = d['kc_mid'] - 1.5 * d['atr']
# Bollinger Bands
bb = ta.BBANDS(d, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
d['bb_u'] = bb['upperband']; d['bb_l'] = bb['lowerband']
# Squeeze indicator: BB inside KC = squeeze
d['squeeze'] = (d['bb_u'] < d['kc_u']) & (d['bb_l'] > d['kc_l'])
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
d['mom'] = ta.ROC(d, timeperiod=3)
return d
def populate_entry_trend(self, d, m):
# Squeeze fires when BB breaks OUT of KC after being inside
squeeze_fire = (~d['squeeze']) & d['squeeze'].shift(1)
d.loc[squeeze_fire & (d['close'] > d['kc_u']) & (d['mom'] > 0) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'sq_L')
d.loc[squeeze_fire & (d['close'] < d['kc_l']) & (d['mom'] < 0) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'sq_S')
return d
def populate_exit_trend(self, d, m): return d
# ==============================
# 7. Parabolic SAR + ADX trend riding
# ==============================
class ParabolicTrend(IStrategy):
"""Parabolic SAR stop-and-reverse with ADX filter"""
INTERFACE_VERSION = 3; timeframe = '15m'; can_short = True
stoploss = -0.03; trailing_stop = True
trailing_stop_positive = 0.005; trailing_stop_positive_offset = 0.02
trailing_only_offset_is_reached = True
minimal_roi = {"0": 0.06, "480": 0.04, "1440": 0.02, "2880": 0}
max_open_trades = 6; startup_candle_count = 80
process_only_new_candles = True; use_exit_signal = True
def populate_indicators(self, d, m):
d['sar'] = ta.SAR(d, acceleration=0.02, maximum=0.2)
d['adx'] = ta.ADX(d, timeperiod=14)
d['vr'] = d['volume'] / ta.SMA(d['volume'], timeperiod=20)
d['ema20'] = ta.EMA(d, timeperiod=20)
return d
def populate_entry_trend(self, d, m):
# SAR在价格下方 + ADX>22 + EMA向上 → 强趋势做多
d.loc[(d['sar'] < d['close']) & (d['adx'] > 22) & (d['close'] > d['ema20']) & (d['volume'] > 0),
['enter_long', 'enter_tag']] = (1, 'sar_L')
d.loc[(d['sar'] > d['close']) & (d['adx'] > 22) & (d['close'] < d['ema20']) & (d['volume'] > 0),
['enter_short', 'enter_tag']] = (1, 'sar_S')
return d
def populate_exit_trend(self, d, m):
d.loc[d['sar'] > d['close'], ['exit_long', 'exit_tag']] = (1, 'sar_exit')
d.loc[d['sar'] < d['close'], ['exit_short', 'exit_tag']] = (1, 'sar_exit')
return d