docs: 添加 ablation 消融实验 + README 版本概览

2026-06-11 21:15:00 +08:00
parent 89174352e2
commit bd598f0203
11 changed files with 4564 additions and 0 deletions
--- a/ablation/ablation_3.py
+++ b/ablation/ablation_3.py
@ -0,0 +1,454 @@
+"""
+Structure Flow Strategy v2.1
+=======================
+变更记录：
+  v1.6 (2026-06-07): 最优基线 — +3659.63%, 190笔, 69.3% trailing胜率
+  v2.0 (2026-06-08): B1 入场延迟确认 — 方向正确但降频严重
+  v2.1 (2026-06-08): ===== D1: 趋势强度过滤 =====
+                      在4H级别评估趋势强度：最近2个Swing Point的间距变化。
+                      如果趋势在扩张（HH/HL间距增大），允许入场；
+                      如果趋势在收缩（HH/HL间距缩小或震荡），过滤信号。
+                      目的：只在趋势明确时交易，避免震荡市反复止损。
+"""
+
+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 StructureFlowStrategyV21_Abl3(IStrategy):
+    """
+    Ablation Variant 3: 移除条件 3
+
+    v2.1改动（相对于v1.6）：
+      在4H级别计算趋势强度：最近2个Swing High间距 + Swing Low间距的变化。
+      只有趋势在扩张（或至少不收缩）时才允许入场。
+    """
+
+    can_short = True
+    stoploss = -0.15
+    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=50, space="buy")
+    cooldown_bars = IntParameter(3, 12, default=6, space="buy")
+    # v2.1 新增：趋势强度最小扩张比例（x/100 = 0%~50%）
+    # 0 = 只要不收缩就行；越大要求趋势扩张越强
+    trend_strength_min = IntParameter(-50, 20, default=-20, space="buy")  # -20=允许SP轻微收缩, 最佳值
+
+    # =====================
+    # 工具：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"]
+
+        # ================================
+        # v1.6 活支撑/阻力检查（保留）
+        # ================================
+        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
+
+        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
+
+        # ================================
+        # v2.1 新增：趋势强度评估
+        # ================================
+        # 计算最近2个Swing Point之间的间距变化
+        # 上升趋势：HH间距 + HL间距都在扩大 → 趋势强
+        # 下降趋势：LH间距 + LL间距都在扩大 → 趋势强
+        # 间距缩小 → 趋势减弱/震荡
+
+        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:
+                    sh_prices.pop(0)
+            if pd.notna(sl.iloc[i]):
+                sl_prices.append(sl.iloc[i])
+                if len(sl_prices) > 4:
+                    sl_prices.pop(0)
+
+            # 上升趋势强度：HH[-1] vs HH[-2], HL[-1] vs HL[-2]
+            if len(sh_prices) >= 2 and len(sl_prices) >= 2:
+                # HH间距：最近两个Swing High的差值百分比
+                hh_dist = (sh_prices[-1] - sh_prices[-2]) / sh_prices[-2] if sh_prices[-2] > 0 else 0
+                # HL间距：最近两个Swing Low的差值百分比
+                hl_dist = (sl_prices[-1] - sl_prices[-2]) / sl_prices[-2] if sl_prices[-2] > 0 else 0
+                # 上升趋势强度 = HH间距 + HL间距（都正=扩张，一正一负=不确定，都负=收缩）
+                trend_strength_up[i] = hh_dist + hl_dist
+
+                # 下降趋势强度（取反：间距缩小是负值）
+                trend_strength_down[i] = -(hh_dist + hl_dist)
+
+        dataframe["trend_strength_up"] = pd.Series(trend_strength_up, index=dataframe.index)
+        dataframe["trend_strength_down"] = pd.Series(trend_strength_down, index=dataframe.index)
+
+        # 趋势强度是否足够（扩张中）
+        min_strength = self.trend_strength_min.value / 100.0  # 0~0.30
+        dataframe["strong_uptrend"] = dataframe["trend_strength_up"] > min_strength
+        dataframe["strong_downtrend"] = dataframe["trend_strength_down"] > min_strength
+
+        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",
+            "support_alive_4h", "resistance_alive_4h",
+            "strong_uptrend_4h", "strong_downtrend_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 时间框架）。
+
+        v2.1 核心改动：D1 — 趋势强度过滤
+          做多额外条件：4H上升趋势在扩张（strong_uptrend_4h）
+          做空额外条件：4H下降趋势在扩张（strong_downtrend_4h）
+        """
+        max_dist = self.max_stop_dist.value / 100.0
+        cooldown = self.cooldown_bars.value
+
+        # NaN 安全处理
+        bool_cols = [
+            "trend_up_1d", "trend_down_1d",
+            "trend_up_4h", "trend_down_4h",
+            "in_demand_4h", "in_supply_4h",
+            "support_alive_4h", "resistance_alive_4h",
+            "strong_uptrend_4h", "strong_downtrend_4h",
+            "bullish_signal", "bearish_signal",
+        ]
+        for col in bool_cols:
+            if col in dataframe.columns:
+                dataframe[col] = dataframe[col].fillna(False)
+
+        # ── 做多 ──
+        long_stop_dist = (dataframe["open"] - dataframe["support_4h"]) / dataframe["open"]
+
+        long_base = (
+            dataframe["trend_up_1d"]
+            & dataframe["in_demand_4h"]
+            & (long_stop_dist <= max_dist)
+            & (long_stop_dist > 0.003)
+            & dataframe["support_alive_4h"]
+            # v2.1: 趋势强度 — 4H上升趋势必须在扩张
+            & dataframe["strong_uptrend_4h"]
+        )
+
+        long_recent = long_base.rolling(cooldown, min_periods=1).max().shift(1) == 0
+        dataframe.loc[long_base & long_recent, "enter_long"] = 1
+
+        # ── 做空 ──
+        short_stop_dist = (dataframe["resistance_4h"] - dataframe["open"]) / dataframe["open"]
+
+        short_base = (
+            dataframe["trend_down_1d"]
+            & dataframe["in_supply_4h"]
+            & (short_stop_dist <= max_dist)
+            & (short_stop_dist > 0.003)
+            & dataframe["resistance_alive_4h"]
+            # v2.1: 趋势强度 — 4H下降趋势必须在扩张
+            & dataframe["strong_downtrend_4h"]
+        )
+
+        short_recent = short_base.rolling(cooldown, min_periods=1).max().shift(1) == 0
+        dataframe.loc[short_base & 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
+
+    # =====================
+    # 动态止损 — 纯价格结构（基于Swing Point）
+    # =====================
+
+    def custom_stoploss(
+        self,
+        pair: str,
+        trade: Trade,
+        current_time: datetime,
+        current_rate: float,
+        current_profit: float,
+        after_fill: bool,
+        **kwargs,
+    ) -> float:
+        """
+        止损逻辑：完全基于价格结构，零指标（与v1.6相同）。
+        """
+        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+        if dataframe is None or len(dataframe) == 0:
+            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
+            sl_price = support * 0.999
+            sl_ratio = (sl_price / current_rate) - 1.0
+            return max(sl_ratio, -0.15)
+        else:
+            resistance = last.get("resistance_4h", 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)
+
+    # =====================
+    # 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"},
+                },
+                "filters": {
+                    "support_alive_4h": {"color": "green", "type": "line"},
+                    "resistance_alive_4h": {"color": "red", "type": "line"},
+                    "strong_uptrend_4h": {"color": "blue", "type": "line"},
+                    "strong_downtrend_4h": {"color": "orange", "type": "line"},
+                },
+            },
+        }