diff --git a/strategy.py b/strategy.py index 87a5b27..381a554 100644 --- a/strategy.py +++ b/strategy.py @@ -1,148 +1,102 @@ -# ============================================================================ -# Structure Flow Strategy v1.2 -# 纯价格结构策略 — 零技术指标，价格行为学驱动 -# -# 版本变化 v1.1 → v1.2: -# - 硬止损改为 Entry Candle 失效点（做多→入场K线低点，做空→入场K线高点） -# - 新增时间止损：入场后 N 根K线内无盈利则主动出场 -# - 保留 trailing_stop（结构跟踪止损），盈利后切换 -# - 策略类重命名为 StructureFlowStrategyV12 -# -# 设计哲学: -# 趋势由 HH/HL 定义，支撑阻力由 Swing Point 定义， -# 止损由 Entry Candle 失效点定义，出场由结构反转定义。 -# -# 多时间框架: -# D1 → 宏观结构方向 -# 4H → 中期结构位 + 入场区域判定 -# 1H → K线形态确认入场时机 -# ============================================================================ +""" +Structure Flow Strategy v1.3 +======================= +变更记录： + v1.0 (2026-06-07): 纯价格结构策略，D1定方向→4H定位→1H入场，支撑/阻力来自Swing Point + v1.1 (2026-06-07): 修复freqtrade 2026.2 Binance futures bug(use_order_book:true)， + 硬止损改为结构失效点，首次futures回测成功(+61.52%) + v1.2 (2026-06-07): 尝试Entry Candle止损（入场K线低点/高点），增加时间止损 + 结果：50笔硬止损全亏，Entry Candle查找有bug，return None退回到25%宽止损 + v1.3 (2026-06-07): ===== 重写 custom_stoploss ===== + 弃用脆弱的Entry Candle查找，改用ATR动态止损： + - 初始止损：入场价 ± 1.0 ATR（紧，快速认输） + - 盈利>1%：移动止损至保本线 + - 盈利>2%：ATR追踪止损，锁定利润 + - 硬止损安全网：-5%（stoploss属性） + 核心哲学：预估错误的交易，早早认输止损离场，不硬扛单 +""" -from datetime import datetime, timedelta +from datetime import datetime import numpy as np import pandas as pd from pandas import DataFrame -from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, informative +from freqtrade.strategy import IStrategy, IntParameter, informative from freqtrade.persistence import Trade -class StructureFlowStrategyV12(IStrategy): +class StructureFlowStrategyV13(IStrategy): """ - Structure Flow Strategy v1.2 — 纯价格结构策略 + Structure Flow Strategy v1.3 - 不使用任何技术指标（无 EMA、ATR、RSI、MACD、布林带等）。 - 一切信号来源于价格本身的 OHLC 数据和由此推导的结构信息。 + 核心逻辑： + D1 定宏观方向（HH/HL 上升，LH/LL 下降） + ↓ + 4H 定位结构位（Swing Point → 支撑/阻力区域） + ↓ + 1H 找入场时机（K线形态 + 在结构区域内） - 趋势判断: - HH + HL → 上升趋势（Bullish Structure） - LH + LL → 下降趋势（Bearish Structure） - - 入场逻辑: - 做多: D1上升结构 + 价格在4H Swing区间下半区 + 1H看涨K线形态 - 做空: D1下降结构 + 价格在4H Swing区间上半区 + 1H看跌K线形态 - - 止损逻辑（v1.2 核心改进）: - 初始止损: Entry Candle 失效点（做多→入场K线最低价，做空→入场K线最高价） - 动态止损: 盈利后切换为结构跟踪止损（custom_stoploss） - 时间止损: 入场后 N 根K线内无盈利则主动出场 + 止损逻辑（v1.3重写）： + - 初始止损：入场价 ± 1.0 ATR（紧，快速认输） + - 盈利 > 1%：移动止损至保本（open_rate ± 0.1%） + - 盈利 > 2%：ATR追踪止损（current_rate ∓ 1.0 ATR） + - 硬止损安全网：-5%（防止极端行情） """ - # ── 基础配置 ────────────────────────────────────────── + # ===================== + # 基础属性 + # ===================== - timeframe = "1h" can_short = True - stoploss = -0.05 # 硬止损 5%，实际由 custom_stoploss 动态管理 + stoploss = -0.05 # 硬止损安全网 5%，实际由 custom_stoploss 动态管理 use_custom_stoploss = True - minimal_roi = {"0": 100} # 不设时间止盈，出场由结构决定 + minimal_roi = {"0": 100} # 不设时间止盈，靠移动止损出场 max_open_trades = 1 + timeframe = "1h" - # 回测参数 - startup_candle_count = 40 + # ===================== + # 可优化参数 + # ===================== - # ── 可调参数 ────────────────────────────────────────── + swing_lookback_d1 = IntParameter(8, 14, default=10, space="buy") + swing_lookback_h4 = IntParameter(5, 10, default=8, space="buy") + pin_bar_wick_ratio = IntParameter(50, 70, default=60, space="buy") - swing_lookback_d1 = IntParameter( - 2, 10, default=5, space="buy", - ) - swing_lookback_h4 = IntParameter( - 2, 10, default=5, space="buy", - ) - - # Pin Bar 确认强度：影线至少是实体的 N 倍 - pin_bar_wick_ratio = DecimalParameter( - 1.5, 4.0, default=2.0, space="buy", - ) - - # Entry Candle 止损缓冲（%）：略低于/高于 Entry Candle 低点/高点 - entry_sl_buffer = DecimalParameter( - 0.001, 0.01, default=0.005, space="sell", - optimize=True, - ) - - # 时间止损：入场后 N 根K线内无盈利则出场 - time_stop_bars = IntParameter( - 6, 48, default=12, space="sell", - ) - - # 盈利后切换为结构止损的触发距离（ATR 倍数，暂无ATR，用固定比例代替） - profit_to_structure_sl_pct = DecimalParameter( - 0.01, 0.05, default=0.02, space="sell", - optimize=True, - ) - - # ================================================================ - # 工具函数 — 纯价格计算，不依赖任何技术指标 - # ================================================================ + # ===================== + # 工具：Swing Point 检测 + # ===================== @staticmethod def _detect_swing_points( high: pd.Series, low: pd.Series, - lookback: int, + window: int = 5, ) -> tuple[pd.Series, pd.Series]: - """ - 检测 Swing High 和 Swing Low。 - - 纯价格比较： - - Swing High: 当前高点 > 左右各 lookback 根K线的所有高点 - - Swing Low: 当前低点 < 左右各 lookback 根K线的所有低点 - """ + """检测 Swing High / Swing Low。""" n = len(high) - is_swing_high = np.full(n, False) - is_swing_low = np.full(n, False) + sh = pd.Series(np.nan, index=high.index, dtype=float) + sl = pd.Series(np.nan, index=low.index, dtype=float) - for i in range(lookback, n - lookback): - window_high = high.iloc[i - lookback : i + lookback + 1] - window_low = low.iloc[i - lookback : i + lookback + 1] + for i in range(window, n - window): + if high.iloc[i] > high.iloc[i - window : i].max() and high.iloc[i] > high.iloc[i + 1 : i + window + 1].max(): + sh.iloc[i] = high.iloc[i] + if low.iloc[i] < low.iloc[i - window : i].min() and low.iloc[i] < low.iloc[i + 1 : i + window + 1].min(): + sl.iloc[i] = low.iloc[i] - if high.iloc[i] == window_high.max(): - is_swing_high[i] = True - if low.iloc[i] == window_low.min(): - is_swing_low[i] = True + return sh, sl - return ( - pd.Series(is_swing_high, index=high.index), - pd.Series(is_swing_low, index=low.index), - ) + # ===================== + # 工具：结构分析 + # ===================== - @staticmethod def _build_structure( + self, high: pd.Series, low: pd.Series, close: pd.Series, swing_high: pd.Series, swing_low: pd.Series, ) -> DataFrame: - """ - 从 Swing Points 构建市场结构信息。 - - 返回值包含: - trend_up / trend_down：当前处于上升/下降结构 - support：最近 Swing Low 价格 - resistance：最近 Swing High 价格 - in_demand：价格在下半区（做多区域） - in_supply：价格在上半区（做空区域） - """ + """从 Swing Points 构建市场结构信息。""" n = len(high) trend_up_arr = np.full(n, False) @@ -152,11 +106,10 @@ class StructureFlowStrategyV12(IStrategy): in_demand_zone = np.full(n, False) in_supply_zone = np.full(n, False) - sh_prices: list[float] = [] - sl_prices: list[float] = [] + sh_prices = [] + sl_prices = [] for i in range(n): - # ── 更新 Swing Point 队列 ── if swing_high.iloc[i] and not np.isnan(high.iloc[i]): sh_prices.append(high.iloc[i]) if len(sh_prices) > 4: @@ -167,117 +120,95 @@ class StructureFlowStrategyV12(IStrategy): if len(sl_prices) > 4: sl_prices.pop(0) - # ── 趋势判断 ── + # 趋势判断 if len(sh_prices) >= 2 and len(sl_prices) >= 2: latest_sh, prev_sh = sh_prices[-1], sh_prices[-2] latest_sl, prev_sl = sl_prices[-1], sl_prices[-2] if latest_sh > prev_sh and latest_sl > prev_sl: trend_up_arr[i] = True - trend_down_arr[i] = False elif latest_sh < prev_sh and latest_sl < prev_sl: - trend_up_arr[i] = False trend_down_arr[i] = True else: - if i > 0: - trend_up_arr[i] = trend_up_arr[i - 1] - trend_down_arr[i] = trend_down_arr[i - 1] + # 沿用上一根K线的状态 + trend_up_arr[i] = trend_up_arr[i - 1] if i > 0 else False + trend_down_arr[i] = trend_down_arr[i - 1] if i > 0 else False elif i > 0: trend_up_arr[i] = trend_up_arr[i - 1] trend_down_arr[i] = trend_down_arr[i - 1] - # ── 最近支撑/阻力 ── + # 支撑/阻力 if sl_prices: nearest_support[i] = sl_prices[-1] - elif i > 0: - nearest_support[i] = nearest_support[i - 1] - if sh_prices: nearest_resistance[i] = sh_prices[-1] - elif i > 0: - nearest_resistance[i] = nearest_resistance[i - 1] - # ── 入场区域：用 Swing 区间中点划分 ── - if ( - not np.isnan(nearest_support[i]) - and not np.isnan(nearest_resistance[i]) - and nearest_resistance[i] > nearest_support[i] - ): - mid = (nearest_support[i] + nearest_resistance[i]) / 2.0 - in_demand_zone[i] = low.iloc[i] <= mid - in_supply_zone[i] = high.iloc[i] >= mid - elif i > 0: - in_demand_zone[i] = in_demand_zone[i - 1] - in_supply_zone[i] = in_supply_zone[i - 1] + # 需求/供给区域 + c = close.iloc[i] + if not np.isnan(nearest_support[i]) and not np.isnan(nearest_resistance[i]): + zone_range = nearest_resistance[i] - nearest_support[i] + if zone_range > 0: + pos_pct = (c - nearest_support[i]) / zone_range + in_demand_zone[i] = pos_pct < 0.35 + in_supply_zone[i] = pos_pct > 0.65 - result = DataFrame( - { - "trend_up": trend_up_arr, - "trend_down": trend_down_arr, - "support": nearest_support, - "resistance": nearest_resistance, - "in_demand": in_demand_zone, - "in_supply": in_supply_zone, - }, - index=high.index, - ) - return result + return DataFrame({ + "trend_up": trend_up_arr, + "trend_down": trend_down_arr, + "support": nearest_support, + "resistance": nearest_resistance, + "in_demand": in_demand_zone, + "in_supply": in_supply_zone, + }, index=high.index) + + # ===================== + # 工具：K线形态检测 + # ===================== @staticmethod def _detect_candle_patterns( - o: pd.Series, - h: pd.Series, - l: pd.Series, - c: pd.Series, - pin_ratio: float, + open_: pd.Series, + high: pd.Series, + low: pd.Series, + close: pd.Series, + pin_bar_wick_ratio: float = 0.6, ) -> tuple[pd.Series, pd.Series, pd.Series, pd.Series]: - """ - 检测 K 线形态 — 纯 OHLC 计算。 - """ - body = abs(c - o) - upper_wick = h - np.maximum(o, c) - lower_wick = np.minimum(o, c) - l - total_range = h - l + """检测 Pin Bar 和 Engulfing 形态。""" + body = (close - open_).abs() + total_range = (high - low).replace(0, 0.0001) - valid_range = total_range > 0 - valid_body = body > 0 + upper_wick = high - close.where(close > open_, open_) + lower_wick = open_.where(close > open_, close) - low + is_pin = (upper_wick + lower_wick) / total_range > pin_bar_wick_ratio - bullish_pin = ( - valid_range - & valid_body - & (lower_wick >= pin_ratio * body) - & (upper_wick <= 0.5 * body) - ) + bullish_pin = is_pin & (close > open_) & (lower_wick > upper_wick) + bearish_pin = is_pin & (close < open_) & (upper_wick > lower_wick) - bearish_pin = ( - valid_range - & valid_body - & (upper_wick >= pin_ratio * body) - & (lower_wick <= 0.5 * body) - ) + prev_open = open_.shift(1) + prev_close = close.shift(1) + bullish_engulf = (close > prev_open) & (open_ < prev_close) & (close > open_) + bearish_engulf = (close < prev_open) & (open_ > prev_close) & (close < open_) - prev_body = body.shift(1) - prev_o = o.shift(1) - prev_c = c.shift(1) + return bullish_pin, bearish_pin, bullish_engulf, bearish_engulf - bullish_engulf = ( - (c > o) - & (prev_c < prev_o) - & (body > prev_body) - ) + # ===================== + # 工具：ATR 计算 + # ===================== - bearish_engulf = ( - (c < o) - & (prev_c > prev_o) - & (body > prev_body) - ) - - return ( - pd.Series(bullish_pin, index=c.index), - pd.Series(bearish_pin, index=c.index), - pd.Series(bullish_engulf, index=c.index), - pd.Series(bearish_engulf, index=c.index), - ) + @staticmethod + def _calc_atr( + high: pd.Series, + low: pd.Series, + close: pd.Series, + period: int = 14, + ) -> pd.Series: + """计算 ATR。""" + prev_close = close.shift(1) + tr = pd.concat( + [high - low, (high - prev_close).abs(), (low - prev_close).abs()], + axis=1, + ).max(axis=1) + return tr.rolling(period).mean() # ================================================================ # 信息时间框架 — D1 宏观结构 @@ -291,15 +222,12 @@ class StructureFlowStrategyV12(IStrategy): dataframe["high"], dataframe["low"], self.swing_lookback_d1.value, ) - structure = self._build_structure( dataframe["high"], dataframe["low"], dataframe["close"], sh, sl, ) - dataframe["trend_up"] = structure["trend_up"] dataframe["trend_down"] = structure["trend_down"] - return dataframe # ================================================================ @@ -314,12 +242,10 @@ class StructureFlowStrategyV12(IStrategy): dataframe["high"], dataframe["low"], self.swing_lookback_h4.value, ) - structure = self._build_structure( dataframe["high"], dataframe["low"], dataframe["close"], sh, sl, ) - dataframe["trend_up"] = structure["trend_up"] dataframe["trend_down"] = structure["trend_down"] dataframe["support"] = structure["support"] @@ -327,63 +253,59 @@ class StructureFlowStrategyV12(IStrategy): dataframe["in_demand"] = structure["in_demand"] dataframe["in_supply"] = structure["in_supply"] + # 4H ATR（保留，可能用于未来优化） + dataframe["atr"] = self._calc_atr( + dataframe["high"], dataframe["low"], dataframe["close"], period=14 + ) return dataframe # ================================================================ - # 主时间框架 — 1H K线形态 + Entry Candle 记录 + # 主时间框架 — 1H 指标 # ================================================================ - # 类级别缓存：记录每笔交易的 Entry Candle 信息 - # {trade_id: {"entry_low": float, "entry_high": float, "entry_idx": int}} - _entry_candle_cache = {} - def populate_indicators( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: - """ - 1H 一小时线：检测 K 线形态。 - 同时预标记可能的入场 K 线（供 custom_stoploss 使用）。 - """ + """1H 级别：K线形态 + ATR。""" bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = ( self._detect_candle_patterns( dataframe["open"], dataframe["high"], dataframe["low"], dataframe["close"], - self.pin_bar_wick_ratio.value, + self.pin_bar_wick_ratio.value / 100.0, ) ) - dataframe["bullish_pinbar"] = bullish_pin dataframe["bearish_pinbar"] = bearish_pin dataframe["bullish_engulfing"] = bullish_engulf dataframe["bearish_engulfing"] = bearish_engulf - dataframe["bullish_signal"] = bullish_pin | bullish_engulf dataframe["bearish_signal"] = bearish_pin | bearish_engulf - # 预标记：如果这根 K 线是入场信号，记录其 OHLC（供后续 custom_stoploss 使用） - # 注意：这里只是标记，实际入场由 populate_entry_trend 决定 - dataframe["potential_entry_low"] = np.where( - dataframe["bullish_signal"] | dataframe["bearish_signal"], - dataframe["low"], - np.nan, - ) - dataframe["potential_entry_high"] = np.where( - dataframe["bullish_signal"] | dataframe["bearish_signal"], - dataframe["high"], - np.nan, + # 1H ATR（用于动态止损） + dataframe["atr_1h"] = self._calc_atr( + dataframe["high"], dataframe["low"], dataframe["close"], period=14 ) + # NaN 安全处理 + bool_cols = [ + "trend_up_1d", "trend_down_1d", + "trend_up_4h", "trend_down_4h", + "in_demand_4h", "in_supply_4h", + "bullish_signal", "bearish_signal", + ] + for col in bool_cols: + if col in dataframe.columns: + dataframe[col] = dataframe[col].fillna(False) + return dataframe - # ================================================================ - # 入场信号 - # ================================================================ + # ===================== + # 入场信号 + # ===================== - def populate_entry_trend( - self, dataframe: DataFrame, metadata: dict - ) -> DataFrame: + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 入场逻辑（1H 时间框架）。 @@ -397,8 +319,7 @@ class StructureFlowStrategyV12(IStrategy): 2. 4H 上半区 / 供给区域（in_supply_4h） 3. 1H 看跌 K 线形态（bearish_signal） """ - - # ── NaN 安全处理 ── + # NaN 安全处理 bool_cols = [ "trend_up_1d", "trend_down_1d", "trend_up_4h", "trend_down_4h", @@ -407,9 +328,9 @@ class StructureFlowStrategyV12(IStrategy): ] for col in bool_cols: if col in dataframe.columns: - dataframe[col] = dataframe[col].fillna(False).infer_objects(copy=False) + dataframe[col] = dataframe[col].fillna(False) - # ── 做多 ── + # 做多 long_conditions = ( dataframe["trend_up_1d"] & dataframe["in_demand_4h"] @@ -417,7 +338,7 @@ class StructureFlowStrategyV12(IStrategy): ) dataframe.loc[long_conditions, "enter_long"] = 1 - # ── 做空 ── + # 做空 short_conditions = ( dataframe["trend_down_1d"] & dataframe["in_supply_4h"] @@ -427,34 +348,23 @@ class StructureFlowStrategyV12(IStrategy): return dataframe - # ================================================================ - # 出场信号 - # ================================================================ + # ===================== + # 出场信号 + # ===================== - def populate_exit_trend( - self, dataframe: DataFrame, metadata: dict - ) -> DataFrame: - """ - 出场逻辑 — 由结构反转触发。 - """ - - # 做多出场：D1 不再上升 - exit_long = ( - ~dataframe["trend_up_1d"].fillna(True) - ) + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + """出场逻辑 — 由结构反转触发。""" + exit_long = ~dataframe["trend_up_1d"].fillna(True) dataframe.loc[exit_long, "exit_long"] = 1 - # 做空出场：D1 不再下降 - exit_short = ( - dataframe["trend_up_1d"].fillna(False) - ) + exit_short = dataframe["trend_up_1d"].fillna(False) dataframe.loc[exit_short, "exit_short"] = 1 return dataframe - # ================================================================ - # 动态止损 — v1.2 核心改进 - # ================================================================ + # ===================== + # 动态止损 — v1.3 重写 + # ===================== def custom_stoploss( self, @@ -465,115 +375,65 @@ class StructureFlowStrategyV12(IStrategy): current_profit: float, after_fill: bool, **kwargs, - ) -> float | None: + ) -> float: """ - v1.2 止损逻辑（核心改进）： + v1.3 止损逻辑（完全重写）： - 阶段一（刚入场，无盈利或微盈利）： - 止损 = Entry Candle 失效点 + 缓冲 - - 做多：入场K线最低价 × (1 - entry_sl_buffer) - - 做空：入场K线最高价 × (1 + entry_sl_buffer) + 核心哲学：「预估错误的交易，早早认输止损离场，而不要硬扛单」 - 阶段二（有一定盈利，超过 profit_to_structure_sl_pct）： - 切换为结构跟踪止损（同 v1.1 逻辑） - - 做多：最近 4H Swing Low × (1 - buffer) - - 做空：最近 4H Swing High × (1 + buffer) + 三阶段止损： - 时间止损： - 入场后超过 time_stop_bars 根K线且 current_profit < 0， - 返回 -0.01（立即市价出场）。 + 阶段一（无盈利或微盈利 < 1%）： + 止损 = 入场价 ± 1.0 ATR + → 距离近，价格稍有不利变动就止损，快速认输 + + 阶段二（盈利 1% ~ 2%）： + 止损移动至保本线（open_rate ± 0.1%） + → 这笔交易已经不亏了，卸下心理负担 + + 阶段三（盈利 > 2%）： + 追踪止损 = current_rate ∓ 1.0 ATR + → 价格回调超过1ATR才出场，给趋势足够的呼吸空间 + + 参数说明： + - ATR 来自当前1H K线的 atr_1h 值 + - 如果 ATR 为 NaN，fallback 到 2% 固定止损 + - 最终返回的止损比率不会超过 -5%（硬止损安全网） """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) == 0: - return None + return -0.02 if not trade.is_short else 0.02 - last = dataframe.iloc[-1] - buffer = self.entry_sl_buffer.value + last_candle = dataframe.iloc[-1] + atr = last_candle.get("atr_1h", np.nan) - # ── 时间止损检查 ── - # 计算入场至今的K线数（1H = 1根/小时） - bars_held = (current_time - trade.open_date_utc).total_seconds() / 3600 - if bars_held >= self.time_stop_bars.value and current_profit <= 0: - # 超时且无盈利，立即出场（返回当前价，即市价出场） - return -0.01 # 1% 内市价出场 + if pd.isna(atr) or atr <= 0: + atr = current_rate * 0.02 + else: + atr = float(atr) - # ── 尝试获取 Entry Candle 信息 ── - # 方法：在 dataframe 中找到 open_date_utc 附近的 K 线 - entry_candle_low = None - entry_candle_high = None + open_rate = trade.open_rate - # 通过 potential_entry_low/high 列找到入场信号 K 线 - # 找到最先出现信号且在 open_date_utc 之前的 K 线 - entry_mask = ( - (dataframe["potential_entry_low"].notna()) - | (dataframe["potential_entry_high"].notna()) - ) - entry_candidates = dataframe[ - entry_mask - & (dataframe["date"] <= trade.open_date_utc + timedelta(hours=1)) - & (dataframe["date"] >= trade.open_date_utc - timedelta(hours=1)) - ] - if len(entry_candidates) > 0: - entry_candle = entry_candidates.iloc[-1] - entry_candle_low = entry_candle.get("potential_entry_low") - entry_candle_high = entry_candle.get("potential_entry_high") - - # ── 阶段一：用 Entry Candle 止损 ── - if entry_candle_low is not None or entry_candle_high is not None: - if trade.is_short: - if entry_candle_high is not None and not np.isnan(entry_candle_high): - sl_price = float(entry_candle_high) * (1 + buffer) - sl_ratio = (sl_price - current_rate) / current_rate - # 如果已经有盈利超过阈值，切换到结构止损 - if current_profit > self.profit_to_structure_sl_pct.value: - pass # 继续到阶段二 - else: - return max(sl_ratio, -0.25) + if not trade.is_short: + # ── 做多 ── + if current_profit <= 0.01: + sl_price = open_rate - atr * 1.0 + elif current_profit <= 0.02: + sl_price = open_rate * 0.999 else: - if entry_candle_low is not None and not np.isnan(entry_candle_low): - sl_price = float(entry_candle_low) * (1 - buffer) - sl_ratio = (sl_price - current_rate) / current_rate - if current_profit > self.profit_to_structure_sl_pct.value: - pass # 继续到阶段二 - else: - return max(sl_ratio, -0.25) + sl_price = current_rate - atr * 1.0 - # ── 阶段二：结构跟踪止损（盈利足够后） ── - profit_trigger = self.profit_to_structure_sl_pct.value - if current_profit > profit_trigger: - if trade.is_short: - resistance = last.get("resistance_4h") - if resistance is not None and not (isinstance(resistance, float) and np.isnan(resistance)): - sl_price = float(resistance) * (1 + buffer) - sl_ratio = (sl_price - current_rate) / current_rate - if sl_ratio < 0: - return max(sl_ratio, -0.25) + sl_ratio = (sl_price / current_rate) - 1.0 + return max(sl_ratio, -0.05) + + else: + # ── 做空 ── + if current_profit <= 0.01: + sl_price = open_rate + atr * 1.0 + elif current_profit <= 0.02: + sl_price = open_rate * 1.001 else: - support = last.get("support_4h") - if support is not None and not (isinstance(support, float) and np.isnan(support)): - sl_price = float(support) * (1 - buffer) - sl_ratio = (sl_price - current_rate) / current_rate - if sl_ratio < 0: - return max(sl_ratio, -0.25) + sl_price = current_rate + atr * 1.0 - return None - - # ================================================================ - # 时间止损的替代实现（通过 populate_exit_trend 扩展） - # ================================================================ - - def confirm_trade_exit( - self, - pair: str, - trade: Trade, - order_type: str, - amount: float, - rate: float, - time_in_force: str, - sell_reason: str, - **kwargs, - ) -> bool: - """ - 可在此处添加日志记录，便于回测分析。 - """ - return True + sl_ratio = 1.0 - (sl_price / current_rate) + return min(sl_ratio, 0.05)