""" Structure Flow Strategy v1.4 ======================= 变更记录： v1.0 (2026-06-07): 纯价格结构策略，D1定方向→4H定位→1H入场 v1.1 (2026-06-07): 1H futures，结构止损，首次回测成功(+61.52%) v1.2 (2026-06-07): Entry Candle止损，bug导致50笔硬止损全亏 v1.3 (2026-06-07): ATR动态止损，结果-63.72%，胜率20.2% v1.4 (2026-06-07): ===== 回归纯价格结构止损 ===== - 完全移除ATR（违背价格行为内核） - 止损 = support_4h(resistance_4h) ± 缓冲 - support_4h随新Swing Low自动更新 → 天然追踪止损 - 新增入场过滤：止损距离>3%则跳过（赔率太差） 核心哲学：止损必须在价格结构位，不在指标计算结果 """ from datetime import datetime import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, IntParameter, informative from freqtrade.persistence import Trade class StructureFlowStrategyV14(IStrategy): """ Structure Flow Strategy v1.4 — 纯价格结构，零指标 止损逻辑（v1.4重写，完全移除ATR）： - 做多止损 = support_4h - 0.1%缓冲 - 做空止损 = resistance_4h + 0.1%缓冲 - support_4h / resistance_4h 随时间更新 → 天然追踪止损 - 硬止损安全网：-5%（stoploss属性） """ can_short = True stoploss = -0.05 use_custom_stoploss = True minimal_roi = {"0": 100} max_open_trades = 1 timeframe = "1h" # ===================== # 可优化参数 # ===================== 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") # 最大可接受止损距离（超过则跳过入场） max_stop_dist = IntParameter(20, 50, default=30, space="buy") # ===================== # 工具：Swing Point 检测 # ===================== @staticmethod def _detect_swing_points( high: pd.Series, low: pd.Series, window: int = 5, ) -> tuple[pd.Series, pd.Series]: 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(): 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] return sh, sl # ===================== # 工具：结构分析 # ===================== def _build_structure( self, 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) 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 # ================================================================ # 信息时间框架 — 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, ) dataframe["trend_up"] = structure["trend_up"] dataframe["trend_down"] = structure["trend_down"] dataframe["support"] = structure["support"] dataframe["resistance"] = structure["resistance"] dataframe["in_demand"] = structure["in_demand"] dataframe["in_supply"] = structure["in_supply"] return dataframe # ================================================================ # 主时间框架 — 1H 指标 # ================================================================ def populate_indicators( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: """1H 级别：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, ) ) 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 # 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: """ 入场逻辑（1H 时间框架）。 做多条件: 1. D1 上升结构（trend_up_1d）— 宏观方向 2. 4H 需求区域（in_demand_4h）— 在支撑附近 3. 1H 看涨 K 线形态（bullish_signal） 4. 止损距离 ≤ max_stop_dist% — 赔率过滤 做空条件: 1. D1 下降结构（trend_down_1d） 2. 4H 供给区域（in_supply_4h） 3. 1H 看跌 K 线形态（bearish_signal） 4. 止损距离 ≤ max_stop_dist% """ max_dist = self.max_stop_dist.value / 100.0 # 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) # ── 做多 ── # 止损距离 = (入场价 - support_4h) / 入场价 # support_4h 已 ffilled，取当前值 long_stop_dist = (dataframe["open"] - dataframe["support_4h"]) / dataframe["open"] long_conditions = ( dataframe["trend_up_1d"] & dataframe["in_demand_4h"] & dataframe["bullish_signal"] & (long_stop_dist <= max_dist) & (long_stop_dist > 0.003) # 至少0.3%距离（避免support就在眼前） ) dataframe.loc[long_conditions, "enter_long"] = 1 # ── 做空 ── short_stop_dist = (dataframe["resistance_4h"] - dataframe["open"]) / dataframe["open"] short_conditions = ( dataframe["trend_down_1d"] & dataframe["in_supply_4h"] & dataframe["bearish_signal"] & (short_stop_dist <= max_dist) & (short_stop_dist > 0.003) ) dataframe.loc[short_conditions, "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 # ===================== # 动态止损 — v1.4 重写：纯价格结构 # ===================== def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> float: """ v1.4 止损逻辑：完全基于价格结构，零指标。 止损位： 做多 → support_4h - 0.1%缓冲（最近4H Swing Low下方） 做空 → resistance_4h + 0.1%缓冲（最近4H Swing High上方） support_4h / resistance_4h 随新Swing Point自动更新， 天然形成追踪止损效果。 永不返回 None，始终返回显式止损比率。 最终截断在 -5% / +5% 安全网内。 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) == 0: # 极端情况：返回2%固定止损 return -0.02 if not trade.is_short else 0.02 last = dataframe.iloc[-1] if not trade.is_short: support = last.get("support_4h", np.nan) if pd.isna(support) or support <= 0: return -0.02 # fallback # 止损 = support_4h 下方 0.1% sl_price = support * 0.999 sl_ratio = (sl_price / current_rate) - 1.0 return max(sl_ratio, -0.05) else: resistance = last.get("resistance_4h", np.nan) if pd.isna(resistance) or resistance <= 0: return 0.02 # fallback # 止损 = resistance_4h 上方 0.1% sl_price = resistance * 1.001 sl_ratio = 1.0 - (sl_price / current_rate) return min(sl_ratio, 0.05) # ===================== # Plot config # ===================== @staticmethod def plot_config() -> dict: return { "main_plot": { "support_4h": {"color": "green", "type": "line"}, "resistance_4h": {"color": "red", "type": "line"}, }, "subplots": { "signals": { "bullish_pinbar": {"color": "green", "type": "scatter"}, "bearish_pinbar": {"color": "red", "type": "scatter"}, }, }, }