From cf0c6d2677746aef5f4c1df2b1e691562d5f9c53 Mon Sep 17 00:00:00 2001 From: Beast Trader Date: Wed, 10 Jun 2026 20:45:00 +0800 Subject: [PATCH] =?UTF-8?q?v3.0=20(Swing):=20=E7=BA=AF=E9=9C=87=E8=8D=A1?= =?UTF-8?q?=E6=B3=A2=E6=AE=B5=E7=AD=96=E7=95=A5=20-=204H=E6=A1=86=E6=9E=B6?= =?UTF-8?q?=E4=BE=9B=E9=9C=80=E5=8C=BA=E6=B3=A2=E6=AE=B5=E4=BA=A4=E6=98=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategy.py | 614 +++++++++++++++++++++++++--------------------------- 1 file changed, 294 insertions(+), 320 deletions(-) diff --git a/strategy.py b/strategy.py index a33e08b..1b41667 100644 --- a/strategy.py +++ b/strategy.py @@ -1,9 +1,24 @@ """ -Structure Flow Strategy v2.2c — 冷却期修复版 -============================================== -变更记录： - v2.2c (2026-06-11): 1H S/R 替代 4H S/R - v2.2c-coolfix (2026-06-11): 修复冷却期无限阻止下单 bug +Structure Flow Swing Strategy v3.0 +================================== +波段交易策略 — 基于4H震荡区间，保守参数 + +核心思路（冯总指示）： + 1. 在4H级别识别震荡区间 + 2. 只在确认震荡时交易（区间宽度稳定、价格测试过边界、无突破） + 3. 止损设在支撑/阻力外侧，确保几乎不被噪音触发 + 4. 止损被触发 = 结构已坏，离场正确 + 5. 止盈：区间高度的70% + +保守参数： + - 杠杆：1x（无杠杆） + - 止损安全边际：ATR(4H, 14) * 1.5 + - 区间宽度稳定阈值：15% + - 止盈：区间70% + - 入场范围：支撑/阻力2%以内 + +版本历史： + v3.0 (2026-06-10): 初版，基于冯总波段交易新思路 """ from datetime import datetime @@ -14,25 +29,31 @@ from freqtrade.strategy import IStrategy, IntParameter, informative from freqtrade.persistence import Trade -class StructureFlowStrategyV22d(IStrategy): +class StructureFlowSwingV30(IStrategy): + """ + Structure Flow Swing Strategy v3.0 + 4H震荡区间波段交易 + """ + can_short = True - stoploss = -0.15 + stoploss = -0.20 use_custom_stoploss = True minimal_roi = {"0": 100} max_open_trades = 1 - timeframe = "1h" + timeframe = "4h" # ===================== - # 可优化参数 + # 可优化参数（保守默认值） # ===================== + swing_lookback = IntParameter(4, 8, default=5, space="buy") + zone_stability_threshold = IntParameter(10, 25, default=15, space="buy") + entry_zone_pct = IntParameter(1, 3, default=2, space="buy") + atr_stop_mult = IntParameter(10, 25, default=15, space="buy") # /10, e.g. 15 = 1.5x + take_profit_pct = IntParameter(50, 80, default=70, space="sell") - swing_lookback_d1 = IntParameter(8, 14, default=10, space="buy") - swing_lookback_h4 = IntParameter(5, 10, default=8, space="buy") - swing_lookback_1h = IntParameter(3, 7, default=5, space="buy") # 新增：1H swing参数 - pin_bar_wick_ratio = IntParameter(50, 70, default=60, space="buy") - max_stop_dist = IntParameter(20, 50, default=50, space="buy") - cooldown_bars = IntParameter(3, 12, default=6, space="buy") - trend_strength_min = IntParameter(-50, 20, default=-20, space="buy") + # 固定参数 + zone_touch_lookback = 10 + breakout_bars = 2 # ===================== # 工具：Swing Point 检测 @@ -47,352 +68,250 @@ class StructureFlowStrategyV22d(IStrategy): n = len(high) sh = pd.Series(np.nan, index=high.index, dtype=float) sl = pd.Series(np.nan, index=low.index, dtype=float) - 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(): + 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(): + 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] - return sh, sl # ===================== - # 工具：结构分析 + # 工具：区间震荡检测 # ===================== - def _build_structure( + def _detect_range( self, + sh: pd.Series, + sl: pd.Series, high: pd.Series, low: pd.Series, close: pd.Series, - swing_high: pd.Series, - swing_low: pd.Series, ) -> DataFrame: n = len(high) - - trend_up_arr = np.full(n, False) - trend_down_arr = np.full(n, False) - nearest_support = np.full(n, np.nan) - nearest_resistance = np.full(n, np.nan) - in_demand_zone = np.full(n, False) - in_supply_zone = np.full(n, False) + is_ranging = np.full(n, False) + support_arr = np.full(n, np.nan) + resistance_arr = np.full(n, np.nan) + zone_width_arr = np.full(n, np.nan) sh_prices = [] sl_prices = [] for i in range(n): - if pd.notna(swing_high.iloc[i]): - sh_prices.append(swing_high.iloc[i]) - if len(sh_prices) > 4: - sh_prices.pop(0) - - if pd.notna(swing_low.iloc[i]): - sl_prices.append(swing_low.iloc[i]) - if len(sl_prices) > 4: - sl_prices.pop(0) - - if len(sh_prices) >= 2 and len(sl_prices) >= 2: - if sh_prices[-1] > sh_prices[-2] and sl_prices[-1] > sl_prices[-2]: - trend_up_arr[i] = True - elif sh_prices[-1] < sh_prices[-2] and sl_prices[-1] < sl_prices[-2]: - trend_down_arr[i] = True - elif i > 0: - trend_up_arr[i] = trend_up_arr[i - 1] - trend_down_arr[i] = trend_down_arr[i - 1] - 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] - if sh_prices: - nearest_resistance[i] = sh_prices[-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 - - 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( - 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]: - body = (close - open_).abs() - total_range = (high - low).replace(0, 0.0001) - - 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 = is_pin & (close > open_) & (lower_wick > upper_wick) - bearish_pin = is_pin & (close < open_) & (upper_wick > lower_wick) - - 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_) - - return bullish_pin, bearish_pin, bullish_engulf, bearish_engulf - - # ===================== - # 工具：冷却期正确实现（修复 bug） - # ===================== - - def _apply_cooldown(self, signal: pd.Series, cooldown_bars: int) -> pd.Series: - """ - 正确应用冷却期：入场后才冷却，而非条件满足就冷却。 - - 原逻辑 bug：long_base.rolling(cooldown).max().shift(1) == 0 - - 当市场持续满足入场条件时，rolling window 里永远有 True - - 导致冷却期无限阻止下单 - - 修复逻辑：遍历 K 线，模拟"入场 -> 冷却"过程。 - - 满足条件 + 距离上次入场 > cooldown -> 允许入场 - - 入场后 cooldown 根 K 线内不再入场 - """ - n = len(signal) - result = [False] * n - last_entry = -99999 # 上次入场的 bar 索引 - - # 遍历（对 numpy array 操作，O(n) 约几毫秒） - values = signal.values # numpy array，快速访问 - for i in range(n): - if values[i] and (i - last_entry) > cooldown_bars: - result[i] = True - last_entry = i - - return pd.Series(result, index=signal.index) - - # ================================================================ - # 信息时间框架 — D1 宏观结构 - # ================================================================ - - @informative("1d") - def populate_indicators_1d( - self, dataframe: DataFrame, metadata: dict - ) -> DataFrame: - sh, sl = self._detect_swing_points( - 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 - - # ================================================================ - # 信息时间框架 — 4H 趋势强度（原版保留） - # ================================================================ - - @informative("4h") - def populate_indicators_4h( - self, dataframe: DataFrame, metadata: dict - ) -> DataFrame: - sh, sl = self._detect_swing_points( - dataframe["high"], dataframe["low"], - self.swing_lookback_h4.value, - ) - structure = self._build_structure( - dataframe["high"], dataframe["low"], dataframe["close"], - sh, sl, - ) - - # 趋势强度计算（原版逻辑） - sh_prices = [] - sl_prices = [] - trend_strength_up = np.full(len(dataframe), np.nan) - trend_strength_down = np.full(len(dataframe), np.nan) - - for i in range(len(dataframe)): if pd.notna(sh.iloc[i]): sh_prices.append(sh.iloc[i]) - if len(sh_prices) > 4: + if len(sh_prices) > 5: sh_prices.pop(0) if pd.notna(sl.iloc[i]): sl_prices.append(sl.iloc[i]) - if len(sl_prices) > 4: + if len(sl_prices) > 5: sl_prices.pop(0) - if len(sh_prices) >= 2 and len(sl_prices) >= 2: - hh_dist = (sh_prices[-1] - sh_prices[-2]) / sh_prices[-2] if sh_prices[-2] > 0 else 0 - hl_dist = (sl_prices[-1] - sl_prices[-2]) / sl_prices[-2] if sl_prices[-2] > 0 else 0 - trend_strength_up[i] = hh_dist + hl_dist - trend_strength_down[i] = -(hh_dist + hl_dist) + if len(sh_prices) < 3 or len(sl_prices) < 3: + continue - dataframe["trend_strength_up"] = pd.Series(trend_strength_up, index=dataframe.index) - dataframe["trend_strength_down"] = pd.Series(trend_strength_down, index=dataframe.index) + current_sh = sh_prices[-1] + current_sl = sl_prices[-1] - min_strength = self.trend_strength_min.value / 100.0 - dataframe["strong_uptrend"] = dataframe["trend_strength_up"] > min_strength - dataframe["strong_downtrend"] = dataframe["trend_strength_down"] > min_strength + if current_sh <= current_sl: + continue - return dataframe + zone_width = (current_sh - current_sl) / current_sl + support_arr[i] = current_sl + resistance_arr[i] = current_sh + zone_width_arr[i] = zone_width - # ================================================================ - # 主时间框架 — 1H 指标（含 1H S/R + 活支撑/阻力） - # ================================================================ + # 条件1：区间宽度稳定性 + widths = [] + for j in range(min(len(sh_prices), len(sl_prices)) - 1, -1, -1): + w = (sh_prices[j] - sl_prices[j]) / sl_prices[j] + widths.append(w) + if len(widths) >= 3: + break - def populate_indicators( - self, dataframe: DataFrame, metadata: dict - ) -> DataFrame: - # ── K线形态 ── - 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 / 100.0, + if len(widths) >= 3: + mean_width = np.mean(widths) + if mean_width > 0: + max_dev = max(abs(w - mean_width) / mean_width for w in widths) + stability_threshold = self.zone_stability_threshold.value / 100.0 + is_stable = max_dev <= stability_threshold + else: + is_stable = False + else: + is_stable = False + + if not is_stable: + continue + + # 条件2：价格测试过边界 + start_idx = max(0, i - self.zone_touch_lookback) + support_zone_upper = current_sl * 1.01 + touched_support = any( + low.iloc[j] <= support_zone_upper + for j in range(start_idx, i + 1) + ) + resistance_zone_lower = current_sh * 0.99 + touched_resistance = any( + high.iloc[j] >= resistance_zone_lower + for j in range(start_idx, i + 1) ) - ) - 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 - # ── 1H级别 Swing Point + 结构（替代原4H S/R） ── - sh_1h, sl_1h = self._detect_swing_points( + if not (touched_support and touched_resistance): + continue + + # 条件3：无突破 + consecutive_outside = 0 + for j in range(i, max(0, i - self.breakout_bars) - 1, -1): + if close.iloc[j] > current_sh or close.iloc[j] < current_sl: + consecutive_outside += 1 + else: + break + + if consecutive_outside >= self.breakout_bars: + continue + + is_ranging[i] = True + + return DataFrame({ + "is_ranging": is_ranging, + "support": support_arr, + "resistance": resistance_arr, + "zone_width": zone_width_arr, + }, index=high.index) + + # ===================== + # 工具：ATR计算 + # ===================== + + @staticmethod + def _calc_atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: + tr = pd.DataFrame({ + "hl": high - low, + "hc": (high - close.shift(1)).abs(), + "lc": (low - close.shift(1)).abs(), + }).max(axis=1) + return tr.rolling(period).mean() + + # ================================================================ + # D1 信息时间框架 — 宏观趋势参考 + # ================================================================ + + @informative("1d") + def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + sh, sl = self._detect_swing_points( + dataframe["high"], dataframe["low"], window=5 + ) + sh_vals = sh.dropna() + sl_vals = sl.dropna() + + is_uptrend = pd.Series(False, index=dataframe.index) + is_downtrend = pd.Series(False, index=dataframe.index) + + if len(sh_vals) >= 2 and len(sl_vals) >= 2: + if sh_vals.iloc[-1] > sh_vals.iloc[-2] and sl_vals.iloc[-1] > sl_vals.iloc[-2]: + is_uptrend[:] = True + elif sh_vals.iloc[-1] < sh_vals.iloc[-2] and sl_vals.iloc[-1] < sl_vals.iloc[-2]: + is_downtrend[:] = True + + dataframe["d1_uptrend"] = is_uptrend + dataframe["d1_downtrend"] = is_downtrend + return dataframe + + # ================================================================ + # 主时间框架 — 4H 指标 + # ================================================================ + + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + sh, sl = self._detect_swing_points( dataframe["high"], dataframe["low"], - self.swing_lookback_1h.value, + self.swing_lookback.value, ) - structure_1h = self._build_structure( - dataframe["high"], dataframe["low"], dataframe["close"], - sh_1h, sl_1h, - ) - dataframe["trend_up_1h"] = structure_1h["trend_up"] - dataframe["trend_down_1h"] = structure_1h["trend_down"] - dataframe["support"] = structure_1h["support"] - dataframe["resistance"] = structure_1h["resistance"] - dataframe["in_demand"] = structure_1h["in_demand"] - dataframe["in_supply"] = structure_1h["in_supply"] - # ── 1H 活支撑/阻力检查 ── - touched_support = ( - (dataframe["low"] <= dataframe["support"] * 1.005) & - (dataframe["low"] >= dataframe["support"] * 0.995) - ) - held_support = dataframe["close"] > dataframe["support"] - support_tested_and_held = touched_support & held_support - dataframe["support_alive"] = support_tested_and_held.rolling(3, min_periods=1).max() > 0 + range_info = self._detect_range(sh, sl, dataframe["high"], dataframe["low"], dataframe["close"]) + dataframe["is_ranging"] = range_info["is_ranging"] + dataframe["range_support"] = range_info["support"] + dataframe["range_resistance"] = range_info["resistance"] + dataframe["zone_width_pct"] = range_info["zone_width"] - touched_resistance = ( - (dataframe["high"] >= dataframe["resistance"] * 0.995) & - (dataframe["high"] <= dataframe["resistance"] * 1.005) - ) - held_resistance = dataframe["close"] < dataframe["resistance"] - resistance_tested_and_held = touched_resistance & held_resistance - dataframe["resistance_alive"] = resistance_tested_and_held.rolling(3, min_periods=1).max() > 0 + dataframe["atr"] = self._calc_atr(dataframe["high"], dataframe["low"], dataframe["close"], 14) - # ── NaN 安全处理 ── - bool_cols = [ - "trend_up_1d", "trend_down_1d", - "trend_up_4h", "trend_down_4h", - "in_demand", "in_supply", - "support_alive", "resistance_alive", - "strong_uptrend_4h", "strong_downtrend_4h", - "bullish_signal", "bearish_signal", - ] - for col in bool_cols: + # 价格在区间内的位置 + denom = dataframe["range_resistance"] - dataframe["range_support"] + dataframe["zone_position"] = np.where( + denom > 0, + (dataframe["close"] - dataframe["range_support"]) / denom, + np.nan, + ) + + # 距离边界百分比 + dataframe["dist_to_support"] = np.where( + dataframe["range_support"] > 0, + (dataframe["close"] - dataframe["range_support"]) / dataframe["close"], + np.nan, + ) + dataframe["dist_to_resistance"] = np.where( + dataframe["range_resistance"] > 0, + (dataframe["range_resistance"] - dataframe["close"]) / dataframe["close"], + np.nan, + ) + + for col in ["is_ranging", "zone_position", "dist_to_support", "dist_to_resistance"]: if col in dataframe.columns: - dataframe[col] = dataframe[col].fillna(False) + dataframe[col] = dataframe[col].fillna(False if col == "is_ranging" else 999) return dataframe - # ===================== - # 入场信号（修复冷却期逻辑） - # ===================== + # ================================================================ + # 入场信号 + # ================================================================ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: - max_dist = self.max_stop_dist.value / 100.0 - cooldown = self.cooldown_bars.value + entry_zone = self.entry_zone_pct.value / 100.0 - bool_cols = [ - "trend_up_1d", "trend_down_1d", - "trend_up_4h", "trend_down_4h", - "in_demand", "in_supply", - "support_alive", "resistance_alive", - "strong_uptrend_4h", "strong_downtrend_4h", - "bullish_signal", "bearish_signal", - ] - for col in bool_cols: + # freqtrade adds _1d suffix to informative columns + d1_downtrend_col = "d1_downtrend_1d" + d1_uptrend_col = "d1_uptrend_1d" + + for col in ["is_ranging", d1_uptrend_col, d1_downtrend_col]: if col in dataframe.columns: dataframe[col] = dataframe[col].fillna(False) + else: + dataframe[col] = False - # ── 做多（使用1H S/R） ── - long_stop_dist = (dataframe["open"] - dataframe["support"]) / dataframe["open"] - - long_base = ( - dataframe["trend_up_1d"] - & dataframe["in_demand"] - & (long_stop_dist <= max_dist) - & (long_stop_dist > 0.003) - & dataframe["support_alive"] - & dataframe["strong_uptrend_4h"] + # ── 做多：震荡市中，价格靠近支撑位 ── + long_conds = ( + dataframe["is_ranging"] + & (dataframe["dist_to_support"] <= entry_zone) + & (dataframe["dist_to_support"] > 0) + & (~dataframe[d1_downtrend_col]) ) - # ✅ 修复：正确应用冷却期（基于实际入场，而非条件满足） - long_entries = self._apply_cooldown(long_base, cooldown) - dataframe.loc[long_entries, "enter_long"] = 1 + cooldown = 3 + long_recent = long_conds.rolling(cooldown, min_periods=1).max().shift(1) == 0 + dataframe.loc[long_conds & long_recent, "enter_long"] = 1 - # ── 做空（使用1H S/R） ── - short_stop_dist = (dataframe["resistance"] - dataframe["open"]) / dataframe["open"] - - short_base = ( - dataframe["trend_down_1d"] - & dataframe["in_supply"] - & (short_stop_dist <= max_dist) - & (short_stop_dist > 0.003) - & dataframe["resistance_alive"] - & dataframe["strong_downtrend_4h"] + # ── 做空：震荡市中，价格靠近阻力位 ── + short_conds = ( + dataframe["is_ranging"] + & (dataframe["dist_to_resistance"] <= entry_zone) + & (dataframe["dist_to_resistance"] > 0) + & (~dataframe[d1_uptrend_col]) ) - # ✅ 修复：正确应用冷却期（基于实际入场，而非条件满足） - short_entries = self._apply_cooldown(short_base, cooldown) - dataframe.loc[short_entries, "enter_short"] = 1 + short_recent = short_conds.rolling(cooldown, min_periods=1).max().shift(1) == 0 + dataframe.loc[short_conds & short_recent, "enter_short"] = 1 return dataframe - # ===================== + # ================================================================ # 出场信号 - # ===================== + # ================================================================ 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 - - exit_short = dataframe["trend_up_1d"].fillna(False) - dataframe.loc[exit_short, "exit_short"] = 1 - return dataframe - # ===================== - # 动态止损（基于1H S/R） - # ===================== + # ================================================================ + # 自定义止损：支撑/阻力外侧，ATR*1.5 缓冲 + # ================================================================ def custom_stoploss( self, @@ -409,43 +328,98 @@ class StructureFlowStrategyV22d(IStrategy): return -0.02 if not trade.is_short else 0.02 last = dataframe.iloc[-1] + atr_mult = self.atr_stop_mult.value / 10.0 if not trade.is_short: - support = last.get("support", np.nan) + support = last.get("range_support", np.nan) + atr = last.get("atr", np.nan) + if pd.isna(support) or support <= 0: return -0.02 - sl_price = support * 0.999 + + if pd.notna(atr) and atr > 0: + sl_price = support - atr * atr_mult + else: + sl_price = support * 0.985 + sl_ratio = (sl_price / current_rate) - 1.0 - return max(sl_ratio, -0.15) + return max(sl_ratio, -0.20) else: - resistance = last.get("resistance", np.nan) + resistance = last.get("range_resistance", np.nan) + atr = last.get("atr", np.nan) + if pd.isna(resistance) or resistance <= 0: return 0.02 - sl_price = resistance * 1.001 - sl_ratio = 1.0 - (sl_price / current_rate) - return min(sl_ratio, 0.15) - # ===================== + if pd.notna(atr) and atr > 0: + sl_price = resistance + atr * atr_mult + else: + sl_price = resistance * 1.015 + + sl_ratio = 1.0 - (sl_price / current_rate) + return min(sl_ratio, 0.20) + + # ================================================================ + # 自定义止盈：区间70% + # ================================================================ + + def custom_exit( + self, + pair: str, + trade: Trade, + current_time: datetime, + current_rate: float, + current_profit: float, + **kwargs, + ) -> str | None: + tp_pct = self.take_profit_pct.value / 100.0 + + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + if dataframe is None or len(dataframe) == 0: + return None + + last = dataframe.iloc[-1] + + if not trade.is_short: + support = last.get("range_support", np.nan) + resistance = last.get("range_resistance", np.nan) + + if pd.notna(support) and pd.notna(resistance) and resistance > support: + zone_height = (resistance - support) / support + tp_target = zone_height * tp_pct + if current_profit >= tp_target: + return "take_profit" + else: + support = last.get("range_support", np.nan) + resistance = last.get("range_resistance", np.nan) + + if pd.notna(support) and pd.notna(resistance) and resistance > support: + zone_height = (resistance - support) / resistance + tp_target = zone_height * tp_pct + if current_profit >= tp_target: + return "take_profit" + + return None + + # ================================================================ # Plot config - # ===================== + # ================================================================ @staticmethod def plot_config() -> dict: return { "main_plot": { - "support": {"color": "green", "type": "line"}, - "resistance": {"color": "red", "type": "line"}, + "range_support": {"color": "green", "type": "line"}, + "range_resistance": {"color": "red", "type": "line"}, }, "subplots": { - "signals": { - "bullish_pinbar": {"color": "green", "type": "scatter"}, - "bearish_pinbar": {"color": "red", "type": "scatter"}, + "range": { + "is_ranging": {"color": "blue", "type": "line"}, + "zone_width_pct": {"color": "purple", "type": "line"}, }, - "filters": { - "support_alive": {"color": "green", "type": "line"}, - "resistance_alive": {"color": "red", "type": "line"}, - "strong_uptrend_4h": {"color": "blue", "type": "line"}, - "strong_downtrend_4h": {"color": "orange", "type": "line"}, + "position": { + "dist_to_support": {"color": "green", "type": "line"}, + "dist_to_resistance": {"color": "red", "type": "line"}, }, }, }