import talib.abstract as ta from pandas import DataFrame from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter from freqtrade.persistence import Trade from datetime import datetime import logging logger = logging.getLogger(__name__) class GridStrategyV2(IStrategy): """ 专业自适应网格 v2 —— 高胜率版 相比 v1 的改进： 1. EMA 金叉 + EMA200 斜率确认：只在趋势明确向上时入场 2. RSI 中性区过滤：只在 30~55 入场（不追高、不在恐慌区买） 3. 成交量确认：成交量高于20日均量，信号更可靠 4. 棘轮止损：盈利超过 1×ATR 时止损拉到保本位（消灭大亏损） 5. DCA 最多 3 档（v1 是 4 档），减少深度套牢敞口 6. 目标：胜率 85%+，Sharpe > 1.2 """ INTERFACE_VERSION = 3 timeframe = "1h" can_short = False position_adjustment_enable = True max_entry_position_adjustment = 4 # 保留 4 档 DCA，给价格回归留足空间 minimal_roi = { "0": 0.10, "2880": 0.02, "5760": 0.005, } stoploss = -0.25 # 宽止损：让 DCA 有充足空间摊成本 trailing_stop = False # 由 custom_stoploss 接管 use_custom_stoploss = True startup_candle_count = 210 # EMA200 需要预热 # ── Hyperopt 参数 ──────────────────────────────────────────────── # 网格间距：缩小上限，高胜率策略用更窄的格 atr_mult = DecimalParameter(0.5, 2.0, default=1.0, decimals=1, space="buy") adx_max = IntParameter(20, 45, default=32, space="buy") bb_width_max= DecimalParameter(0.05, 0.25, default=0.15, decimals=2, space="buy") rsi_min = IntParameter(25, 45, default=35, space="buy") rsi_max = IntParameter(45, 68, default=58, space="buy") # 止盈更快：高胜率的核心 —— 小利润快速兑现，让更多交易在价格反转前出场 tp_atr_mult = DecimalParameter(0.3, 1.5, default=0.6, decimals=1, space="sell") # 棘轮触发更早：尽快把止损拉到保本，消灭潜在大亏 breakeven_atr = DecimalParameter(0.2, 1.0, default=0.4, decimals=1, space="sell") _atr_cache: dict = {} # ── 指标计算 ───────────────────────────────────────────────────── def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema200"] = ta.EMA(dataframe, timeperiod=200) dataframe["vol_ma"] = dataframe["volume"].rolling(20).mean() upper, mid, lower = ta.BBANDS(dataframe["close"], timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe["bb_upper"] = upper dataframe["bb_mid"] = mid dataframe["bb_lower"] = lower dataframe["bb_width"] = (upper - lower) / mid # EMA200 斜率：当前值 vs 10根K线前的值 dataframe["ema200_slope"] = dataframe["ema200"] - dataframe["ema200"].shift(10) if len(dataframe) > 0: self._atr_cache[metadata["pair"]] = float(dataframe["atr"].iloc[-1]) return dataframe # ── 入场信号（比 v1 严格得多）──────────────────────────────────── def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( # 大方向：价格 > EMA200，EMA50 > EMA200（金叉），EMA200 本身向上 (dataframe["close"] > dataframe["ema200"]) & (dataframe["ema50"] > dataframe["ema200"]) & (dataframe["ema200_slope"] > 0) & # 震荡市：ADX 低 + BB 宽度不过大 (dataframe["adx"] < self.adx_max.value) & (dataframe["bb_width"] < self.bb_width_max.value) & # RSI 中性区：不追高（< rsi_max），不在极度恐慌区（> rsi_min） (dataframe["rsi"] > self.rsi_min.value) & (dataframe["rsi"] < self.rsi_max.value) & # 入场位置：价格在 BB 中轨以下（不在上半区追入） (dataframe["close"] < dataframe["bb_mid"]) & # 成交量确认 (dataframe["volume"] > dataframe["vol_ma"] * 0.8) & (dataframe["volume"] > 0) ), "enter_long", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe # ── 棘轮止损 ───────────────────────────────────────────────────── def custom_stoploss(self, pair, trade, current_time: datetime, current_rate, current_profit, after_fill, **kwargs): """ 棘轮机制： - 亏损中：维持原始止损 (-20%) - 盈利超过 breakeven_atr × ATR：止损拉到 -0.5%（近似保本） - 盈利超过 2 × breakeven_atr × ATR：止损追踪到当前盈利的 -50% """ atr = self._atr_cache.get(pair) if atr is None or trade.open_rate == 0: return self.stoploss be_threshold = (self.breakeven_atr.value * atr) / trade.open_rate if current_profit >= be_threshold * 2: # 追踪止损：盈利超过2倍阈值，止损锁在当前盈利的一半以上 return max(-current_profit * 0.5, -0.005) elif current_profit >= be_threshold: # 已盈利超过1个ATR，止损拉到保本 return -0.005 else: return self.stoploss # 还在初期，维持原止损 # ── 动态止盈 ───────────────────────────────────────────────────── # OKX 现货手续费（实际费率，按档位调整这个值） FEE_RATE = 0.001 # 0.10% 单边（挂单价） def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs): atr = self._atr_cache.get(pair) if atr is None or trade.open_rate == 0: return None filled = trade.nr_of_successful_entries # 动态最小止盈：必须覆盖所有入场手续费 + 出场手续费 + 0.2% 净利润缓冲 # current_profit 已由 Freqtrade 扣除手续费，这里额外加 0.2% 安全垫 # 确保每笔交易在任意 DCA 档位下都真正盈利 fee_buffer = self.FEE_RATE * filled + 0.002 # 已入场次数 × 入场费 + 净利缓冲 atr_target = (self.tp_atr_mult.value * atr) / trade.open_rate profit_target = max(atr_target, fee_buffer) if current_profit >= profit_target: return f"grid_tp_lvl{filled}_{current_profit:.3f}" return None # ── 网格加仓逻辑 ───────────────────────────────────────────────── def adjust_trade_entry( self, trade: Trade, current_time, current_rate: float, current_profit: float, min_stake, max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs, ): filled = trade.nr_of_successful_entries if filled >= (self.max_entry_position_adjustment + 1): return None atr = self._atr_cache.get(trade.pair) if atr: grid_step = (self.atr_mult.value * atr) / current_rate grid_step = max(grid_step, 0.012) grid_step = min(grid_step, 0.08) else: grid_step = 0.03 threshold = -(grid_step * filled) if current_profit > threshold: return None # 仓位递增：1x → 1.5x → 2x → 2.5x multiplier = 1.0 + max(filled - 1, 0) * 0.5 stake = trade.stake_amount * multiplier stake = max(stake, min_stake or 0) stake = min(stake, max_stake) logger.info( f"[GridV2] {trade.pair} 第{filled}次加仓 " f"间距={grid_step:.2%} 当前={current_profit:.2%} 加仓={stake:.2f}U ({multiplier}x)" ) return stake