保守型做多策略,适合大市值币种(BTC/ETH等) 时间周期:4h
Timeframe
4h
Direction
Long Only
Stoploss
-5.0%
Trailing Stop
No
ROI
0m: 10000.0%
Interface Version
N/A
Startup Candles
N/A
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
Freqtrade策略文件 - 包含两个策略
1. UnifiedLong4hBTC - 保守型做多策略
2. RSIOversold4hBTC - RSI超卖反弹策略
使用方法:
freqtrade backtesting --strategy UnifiedLong4hBTC ...
freqtrade backtesting --strategy RSIOversold4hBTC ...
"""
import talib.abstract as ta
import pandas as pd
from pandas import DataFrame
from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
from datetime import datetime, timedelta
# ============================================================
# 策略1: UnifiedLong4hBTC - 保守型做多策略
# ============================================================
class UnifiedLong4hBTC(IStrategy):
"""
保守型做多策略,适合大市值币种(BTC/ETH等)
时间周期:4h
买入信号:
- 成交量过滤:平均成交量 >= 阈值
- MACD金叉:DIF > DEA
- KDJ金叉:K > D,且满足上限过滤
- 市场状态过滤:ATR比率、ADX、价格波动范围、RSI、StochRSI等
卖出信号:
- 固定止损:5%
- ATR移动止损:盈利达到1×ATR后,止损线为最高价-3×ATR
"""
# 策略参数
timeframe = '4h'
# ROI 配置(禁用,使用移动止损)
minimal_roi = {
"0": 100 # 不使用ROI,完全依靠止损
}
# 固定止损
stoploss = -0.05 # -5%
# 关闭百分比移动止损(使用ATR移动止损)
trailing_stop = False
# 使用自定义止损
use_custom_stoploss = True
# 参数配置
min_volume_threshold = DecimalParameter(1000, 2000, default=1500, space='buy', optimize=True)
long_kdj_max_k = DecimalParameter(80, 90, default=88, space='buy', optimize=True)
long_kdj_max_d = DecimalParameter(75, 85, default=83, space='buy', optimize=True)
long_kdj_max_j = DecimalParameter(100, 120, default=110, space='buy', optimize=True)
long_min_atr_ratio = DecimalParameter(0.5, 0.8, default=0.6, space='buy', optimize=True)
long_min_adx = DecimalParameter(10, 20, default=16, space='buy', optimize=True)
long_min_price_range_30 = DecimalParameter(3, 6, default=4.8, space='buy', optimize=True)
long_max_price_range_30 = DecimalParameter(14, 18, default=16, space='buy', optimize=True)
long_max_rsi = DecimalParameter(70, 80, default=76, space='buy', optimize=True)
long_min_stochrsi = DecimalParameter(40, 60, default=52, space='buy', optimize=True)
long_min_macd_dif = DecimalParameter(-300, -100, default=-200, space='buy', optimize=True)
long_min_macd_dea = DecimalParameter(-500, -200, default=-400, space='buy', optimize=True)
# ATR移动止损参数
atr_trailing_period = IntParameter(10, 20, default=14, space='buy', optimize=False)
atr_trailing_multiplier = DecimalParameter(2.0, 4.0, default=3.0, space='sell', optimize=True)
atr_trailing_trigger = DecimalParameter(0.5, 2.0, default=1.0, space='sell', optimize=True) # 盈利达到1×ATR时激活
def __init__(self, config: dict) -> None:
super().__init__(config)
self.entry_atr = {} # pair -> entry_atr (入场时的ATR值)
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""计算技术指标"""
# MACD (12, 26, 9)
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe['macd'] = macd['macd']
dataframe['macd_signal'] = macd['macdsignal']
dataframe['macd_hist'] = macd['macdhist']
# RSI (14)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
# Stochastic RSI
stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=14, fastk_period=14, fastd_period=3)
dataframe['stochrsi_k'] = stoch_rsi['fastk']
dataframe['stochrsi_d'] = stoch_rsi['fastd']
# KDJ (基于Stochastic)
stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3)
dataframe['kdj_k'] = stoch['slowk']
dataframe['kdj_d'] = stoch['slowd']
dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d']
# ATR (14)
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
# ATR比率 (ATR(14) / SMA(ATR, 60))
dataframe['atr_sma60'] = ta.SMA(dataframe['atr'], timeperiod=60)
dataframe['atr_ratio'] = dataframe['atr'] / dataframe['atr_sma60']
# ADX (14)
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
# 30根K线价格波动范围百分比
dataframe['price_high_30'] = dataframe['high'].rolling(window=30).max()
dataframe['price_low_30'] = dataframe['low'].rolling(window=30).min()
dataframe['price_range_30'] = (
(dataframe['price_high_30'] - dataframe['price_low_30']) / dataframe['price_low_30'] * 100
)
# EMA (40) - 用于可视化
dataframe['ema40'] = ta.EMA(dataframe, timeperiod=40)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""买入信号"""
dataframe.loc[
(
# 成交量过滤
(dataframe['volume'] >= self.min_volume_threshold.value) &
# MACD金叉
(dataframe['macd'] > dataframe['macd_signal']) &
# KDJ金叉及上限
(dataframe['kdj_k'] > dataframe['kdj_d']) &
(dataframe['kdj_j'] < self.long_kdj_max_j.value) &
(dataframe['kdj_k'] < self.long_kdj_max_k.value) &
(dataframe['kdj_d'] < self.long_kdj_max_d.value) &
# ATR比率(避免震荡期)
(dataframe['atr_ratio'] >= self.long_min_atr_ratio.value) &
# ADX(趋势强度)
(dataframe['adx'] >= self.long_min_adx.value) &
# 价格波动范围
(dataframe['price_range_30'] >= self.long_min_price_range_30.value) &
(dataframe['price_range_30'] <= self.long_max_price_range_30.value) &
# RSI过滤
(dataframe['rsi'] <= self.long_max_rsi.value) &
# StochRSI过滤
(dataframe['stochrsi_k'] >= self.long_min_stochrsi.value) &
# MACD DIF/DEA过滤
(dataframe['macd'] >= self.long_min_macd_dif.value) &
(dataframe['macd_signal'] >= self.long_min_macd_dea.value) &
# 有效数据
(dataframe['volume'] > 0)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""卖出信号(主要依靠止损,这里可以添加其他退出条件)"""
# 不设置主动退出信号,完全依靠止损
dataframe['exit_long'] = 0
return dataframe
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str, **kwargs) -> bool:
"""
确认交易入场时,存储入场时的ATR值
注意:使用前一根已完成K线的ATR(因为入场信号基于前一根K线计算)
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) >= 2:
# 使用前一根已完成K线的ATR(-2位置),而不是最新K线(-1位置)
# 因为入场信号是基于前一根K线的指标计算的
prev_candle = dataframe.iloc[-2].squeeze()
if 'atr' in prev_candle:
self.entry_atr[pair] = float(prev_candle['atr'])
return True
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
自定义止损逻辑 - ATR移动止损
逻辑:
1. 触发条件:盈利金额 >= EntryATR × atr_trailing_trigger
2. 止损价 = 最高价 - EntryATR × atr_trailing_multiplier
3. 止损只能向上调整(保护利润)
Returns:
float: 止损百分比(负数,相对于入场价)
例如:-0.05 = -5% 止损
"""
# 如果没有存储入场ATR,使用固定止损
if pair not in self.entry_atr:
return self.stoploss # -0.05
entry_atr = self.entry_atr[pair]
entry_price = trade.open_rate
highest_price = trade.max_rate if hasattr(trade, 'max_rate') else current_rate
# 计算当前盈利金额(基于当前价格)
current_profit_amount = current_rate - entry_price
# ATR触发金额:EntryATR × atr_trailing_trigger
atr_trigger_amount = entry_atr * self.atr_trailing_trigger.value
# 检查是否达到触发条件
if current_profit_amount >= atr_trigger_amount:
# 计算ATR止损价:最高价 - (EntryATR × atr_trailing_multiplier)
atr_stop_price = highest_price - (entry_atr * self.atr_trailing_multiplier.value)
# 计算止损百分比(相对于入场价)
stop_loss_pct = (atr_stop_price - entry_price) / entry_price
# 确保止损不会低于固定止损
if stop_loss_pct > self.stoploss:
return stop_loss_pct
# 未达到触发条件或计算出的止损低于固定止损,使用固定止损
return self.stoploss
def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float,
rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool:
"""确认退出时清理entry_atr记录"""
if pair in self.entry_atr:
del self.entry_atr[pair]
return True
# ============================================================
# 策略2: RSIOversold4hBTC - RSI超卖反弹策略
# ============================================================
class RSIOversold4hBTC(IStrategy):
"""
RSI超卖反弹策略
时间周期:4h
买入信号:
- RSI < 30 (超卖)
- 额外过滤:StochRSI <= 10, KDJ K <= 35, KDJ D <= 35, -3 <= KDJ J <= 1
卖出信号:
- 固定持仓3根K线(12小时)
- 止损:2%
"""
# 策略参数
timeframe = '4h'
# ROI 配置(禁用)
minimal_roi = {
"0": 100
}
# 固定止损
stoploss = -0.02 # -2%
# 禁用移动止损
trailing_stop = False
# 参数配置
rsi_period = IntParameter(10, 20, default=14, space='buy', optimize=True)
rsi_oversold_threshold = DecimalParameter(20, 35, default=30, space='buy', optimize=True)
holding_bars = IntParameter(2, 5, default=3, space='sell', optimize=True)
# 额外过滤条件
enable_filters = True
stochrsi_threshold = DecimalParameter(5, 15, default=10, space='buy', optimize=True)
kdj_k_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True)
kdj_d_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True)
kdj_j_max_threshold = DecimalParameter(0, 5, default=1, space='buy', optimize=True)
kdj_j_min_threshold = DecimalParameter(-5, 0, default=-3, space='buy', optimize=True)
# 用于追踪持仓时间
def __init__(self, config: dict) -> None:
super().__init__(config)
self.entry_candle_time = {} # pair -> entry candle timestamp
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""计算技术指标"""
# RSI
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
# Stochastic RSI
stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=self.rsi_period.value,
fastk_period=14, fastd_period=3)
dataframe['stochrsi_k'] = stoch_rsi['fastk']
dataframe['stochrsi_d'] = stoch_rsi['fastd']
# KDJ
stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3)
dataframe['kdj_k'] = stoch['slowk']
dataframe['kdj_d'] = stoch['slowd']
dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d']
# MACD (用于记录)
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe['macd'] = macd['macd']
dataframe['macd_signal'] = macd['macdsignal']
# 标记买入信号出现的位置(前一根K线RSI超卖)
dataframe['rsi_oversold_prev'] = (dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""买入信号 - 前一根K线RSI超卖,当前K线开盘买入"""
conditions = [
# 前一根K线RSI超卖
(dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value),
(dataframe['rsi'].shift(1) > 0), # 确保RSI有效
]
# 额外过滤条件
if self.enable_filters:
conditions.extend([
# StochRSI超卖
(dataframe['stochrsi_k'].shift(1) <= self.stochrsi_threshold.value),
# KDJ超卖
(dataframe['kdj_k'].shift(1) <= self.kdj_k_threshold.value),
(dataframe['kdj_d'].shift(1) <= self.kdj_d_threshold.value),
(dataframe['kdj_j'].shift(1) >= self.kdj_j_min_threshold.value),
(dataframe['kdj_j'].shift(1) <= self.kdj_j_max_threshold.value),
])
# 合并所有条件
dataframe.loc[
(
conditions[0] &
(conditions[1] if len(conditions) > 1 else True) &
(conditions[2] if len(conditions) > 2 else True) &
(conditions[3] if len(conditions) > 3 else True) &
(conditions[4] if len(conditions) > 4 else True) &
(conditions[5] if len(conditions) > 5 else True) &
(conditions[6] if len(conditions) > 6 else True) &
(dataframe['volume'] > 0)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""卖出信号 - 持仓固定K线数后卖出(通过custom_exit处理)"""
# 不在这里设置退出信号,使用 custom_exit_long
dataframe['exit_long'] = 0
return dataframe
def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> bool:
"""
自定义退出逻辑 - 持仓固定K线数后卖出
Returns:
bool: True表示退出
"""
# 记录入场时的K线时间
if pair not in self.entry_candle_time:
self.entry_candle_time[pair] = trade.open_date_utc
# 计算持仓时间(按4h K线数量)
time_diff = current_time - self.entry_candle_time[pair]
hours_held = time_diff.total_seconds() / 3600
bars_held = int(hours_held / 4) # 4h周期
# 持仓达到目标K线数后退出
if bars_held >= self.holding_bars.value:
return True
return False
def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float,
rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool:
"""确认退出时清理记录"""
if pair in self.entry_candle_time:
del self.entry_candle_time[pair]
return True
# ============================================================
# 策略3: CombinedStrategy - 组合策略(可选)
# ============================================================
class CombinedStrategy(IStrategy):
"""
组合策略 - 同时使用UnifiedLong和RSIOversold的信号
买入信号:满足任一策略的买入条件
卖出信号:根据入场策略选择对应的退出条件
注意:此策略仅作为示例,实际使用时需要仔细测试和调优
"""
timeframe = '4h'
minimal_roi = {
"0": 100
}
stoploss = -0.05
trailing_stop = False
def __init__(self, config: dict) -> None:
super().__init__(config)
self.unified = UnifiedLong4hBTC(config)
self.rsi_oversold = RSIOversold4hBTC(config)
self.entry_strategy = {} # pair -> strategy_name
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""计算两个策略的所有指标"""
dataframe = self.unified.populate_indicators(dataframe, metadata)
dataframe = self.rsi_oversold.populate_indicators(dataframe, metadata)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""组合买入信号"""
# 计算两个策略的信号
df_unified = self.unified.populate_entry_trend(dataframe.copy(), metadata)
df_rsi = self.rsi_oversold.populate_entry_trend(dataframe.copy(), metadata)
# 优先级:RSI超卖策略 > Unified策略
dataframe.loc[
(df_rsi['enter_long'] == 1),
'enter_long'] = 1
dataframe.loc[
((df_unified['enter_long'] == 1) & (dataframe['enter_long'] != 1)),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""组合卖出信号"""
dataframe['exit_long'] = 0
return dataframe
def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> bool:
"""根据入场策略选择退出逻辑"""
# 这里可以根据需要实现组合退出逻辑
# 简化处理:使用RSI策略的退出条件
return self.rsi_oversold.custom_exit(pair, trade, current_time,
current_rate, current_profit, **kwargs)