From 3da0b17725632e8c7095be21f6d09c4bca5f6ee8 Mon Sep 17 00:00:00 2001 From: Beast Trader Date: Wed, 10 Jun 2026 21:49:00 +0800 Subject: [PATCH] =?UTF-8?q?v3.2=20(Swing):=20=E5=A4=9ATF=E9=9C=87=E8=8D=A1?= =?UTF-8?q?=E7=A1=AE=E8=AE=A4=20+=20=E4=BE=9B=E9=9C=80=E5=8C=BA=E5=8A=A8?= =?UTF-8?q?=E6=80=81=E8=AF=84=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategy.py | 224 +++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 195 insertions(+), 29 deletions(-) diff --git a/strategy.py b/strategy.py index c048121..d6d5a75 100644 --- a/strategy.py +++ b/strategy.py @@ -1,21 +1,28 @@ """ -Structure Flow Swing Strategy v3.1 +Structure Flow Swing Strategy v3.2 ================================== -波段交易策略 — 基于4H震荡区间，保守参数 v2 +波段交易策略 — 基于4H震荡区间，v3.1优化版 -v3.1 改动（基于v3.0诊断结果）： - 1. 双边测试 AND→OR：在10根K线内测试过支撑 OR 阻力即可（不需两者都测过） - 2. 区间稳定性 15%→25%：放宽波动容忍度 - 3. 入场范围 2%→3%：增加候选信号密度 - 4. 冷却期 3根→1根：减少过渡过滤 +v3.2 改动（基于v3.1诊断结果 — 三大市场感知不足）： + 1. D1趋势强度过滤：D1处于强趋势时拒绝入场，防假区间陷阱 + - 计算 D1 EMA20/EMA50 间距作为趋势强度指标 + - 趋势强度超过阈值 → 不交易（即使4H出现区间形态） + 2. 区间质量评分：从二分法升级为多维度评分 + - 边界测试次数（测试越多越可靠） + - 区间持续时长（越长越成熟） + - 区间宽度适配度（3-8%最优） + - 总分>=阈值才入场 + 3. 主动退出机制：确认转趋势后提前离场 + - 3根连续K线收盘在入场时区间外 → 结构破坏 + - 不等止损，主动离场（仅在损失<2%时） + - 避免浮盈变亏损 -保留：纯震荡定位、ATR×1.5止损、区间70%止盈、D1趋势过滤 - -预期：年交易量从9笔 → 50-80笔（约1-2单/周） +保留：纯震荡定位、ATR×1.5止损、区间70%止盈、OR双边测试、冷却期1根 版本历史： v3.0 (2026-06-10): 初版，基于冯总波段交易新思路 - v3.1 (2026-06-10): 降低条件门槛，提升交易频率 + v3.1 (2026-06-10): 降低条件门槛，AND→OR等4项 + v3.2 (2026-06-10): 三大市场感知改进 """ from datetime import datetime @@ -26,10 +33,10 @@ from freqtrade.strategy import IStrategy, IntParameter, informative from freqtrade.persistence import Trade -class StructureFlowSwingV31(IStrategy): +class StructureFlowSwingV32(IStrategy): """ - Structure Flow Swing Strategy v3.1 - 4H震荡区间波段交易 — 放宽震荡判定 + Structure Flow Swing Strategy v3.2 + 4H震荡区间波段交易 — 市场感知增强版 """ can_short = True @@ -40,17 +47,22 @@ class StructureFlowSwingV31(IStrategy): timeframe = "4h" # ===================== - # 可优化参数（放宽后默认值） + # 核心参数（沿用v3.1默认值） # ===================== swing_lookback = IntParameter(4, 8, default=5, space="buy") - zone_stability_threshold = IntParameter(15, 40, default=25, space="buy") # v3.1: 15→25↑ - entry_zone_pct = IntParameter(1, 5, default=3, space="buy") # v3.1: 2→3↑ + zone_stability_threshold = IntParameter(15, 40, default=25, space="buy") + entry_zone_pct = IntParameter(1, 5, default=3, space="buy") atr_stop_mult = IntParameter(10, 25, default=15, space="buy") take_profit_pct = IntParameter(50, 80, default=70, space="sell") + # v3.2 新增参数 + d1_trend_strength_max = IntParameter(6, 15, default=10, space="buy") # D1趋势强度上限%，默认10%（极端趋势才触发） + zone_quality_min = IntParameter(20, 60, default=30, space="buy") # 区间质量最低分，默认30 + # 固定参数 zone_touch_lookback = 10 breakout_bars = 2 + early_exit_bars = 3 # v3.2新增：连续N根在区间外触发主动退出 # ===================== # 工具：Swing Point 检测 @@ -73,7 +85,7 @@ class StructureFlowSwingV31(IStrategy): return sh, sl # ===================== - # 工具：区间震荡检测 + # 工具：区间震荡检测（增强版：加入质量评分数据） # ===================== def _detect_range( @@ -89,10 +101,14 @@ class StructureFlowSwingV31(IStrategy): support_arr = np.full(n, np.nan) resistance_arr = np.full(n, np.nan) zone_width_arr = np.full(n, np.nan) + touch_count_arr = np.full(n, 0) # v3.2新增 sh_prices = [] sl_prices = [] + in_range = False + touch_count = 0 + for i in range(n): if pd.notna(sh.iloc[i]): sh_prices.append(sh.iloc[i]) @@ -104,12 +120,19 @@ class StructureFlowSwingV31(IStrategy): sl_prices.pop(0) if len(sh_prices) < 3 or len(sl_prices) < 3: + # 不在区间中 + if in_range: + in_range = False + touch_count = 0 continue current_sh = sh_prices[-1] current_sl = sl_prices[-1] if current_sh <= current_sl: + if in_range: + in_range = False + touch_count = 0 continue zone_width = (current_sh - current_sl) / current_sl @@ -137,10 +160,12 @@ class StructureFlowSwingV31(IStrategy): is_stable = False if not is_stable: + if in_range: + in_range = False + touch_count = 0 continue # 条件2：价格测试过边界 — v3.1: AND→OR - # 只需要测试过支撑或阻力之一，不需要两者都测过 start_idx = max(0, i - self.zone_touch_lookback) support_zone_upper = current_sl * 1.01 touched_support = any( @@ -153,8 +178,10 @@ class StructureFlowSwingV31(IStrategy): for j in range(start_idx, i + 1) ) - # v3.1: AND → OR if not (touched_support or touched_resistance): + if in_range: + in_range = False + touch_count = 0 continue # 条件3：无突破 @@ -166,15 +193,30 @@ class StructureFlowSwingV31(IStrategy): break if consecutive_outside >= self.breakout_bars: + if in_range: + in_range = False + touch_count = 0 continue + # === 通过所有条件 → 在区间中 === is_ranging[i] = True + # v3.2: 跟踪区间内的边界触碰次数（质量评分数据） + if not in_range: + in_range = True + touch_count = 0 + + c = close.iloc[i] + if (c <= current_sl * 1.015) or (c >= current_sh * 0.985): + touch_count += 1 + touch_count_arr[i] = touch_count + return DataFrame({ "is_ranging": is_ranging, "support": support_arr, "resistance": resistance_arr, "zone_width": zone_width_arr, + "touch_count": touch_count_arr, # v3.2新增 }, index=high.index) # ===================== @@ -191,11 +233,12 @@ class StructureFlowSwingV31(IStrategy): return tr.rolling(period).mean() # ================================================================ - # D1 信息时间框架 — 宏观趋势参考 + # D1 信息时间框架 — v3.2: 新增趋势强度计算 # ================================================================ @informative("1d") def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + # 原有：D1趋势方向（swing point比较） sh, sl = self._detect_swing_points( dataframe["high"], dataframe["low"], window=5 ) @@ -213,10 +256,16 @@ class StructureFlowSwingV31(IStrategy): dataframe["d1_uptrend"] = is_uptrend dataframe["d1_downtrend"] = is_downtrend + + # v3.2新增：D1趋势强度 = EMA20与EMA50的偏离程度 + ema_20 = dataframe["close"].ewm(span=20, adjust=False).mean() + ema_50 = dataframe["close"].ewm(span=50, adjust=False).mean() + dataframe["trend_strength"] = abs(ema_20 - ema_50) / ema_50 + return dataframe # ================================================================ - # 主时间框架 — 4H 指标 + # 主时间框架 — 4H 指标（v3.2: 新增区间质量评分） # ================================================================ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: @@ -230,6 +279,7 @@ class StructureFlowSwingV31(IStrategy): dataframe["range_support"] = range_info["support"] dataframe["range_resistance"] = range_info["resistance"] dataframe["zone_width_pct"] = range_info["zone_width"] + dataframe["range_touch_count"] = range_info["touch_count"] dataframe["atr"] = self._calc_atr(dataframe["high"], dataframe["low"], dataframe["close"], 14) @@ -253,14 +303,66 @@ class StructureFlowSwingV31(IStrategy): np.nan, ) + # ── v3.2新增：区间质量评分 ── + self._compute_zone_quality(dataframe) + + # ── v3.2新增：区间连续计数 ── + is_ranging_int = dataframe["is_ranging"].astype(int) + consecutive = np.zeros(len(dataframe), dtype=int) + for i in range(1, len(dataframe)): + if is_ranging_int.iloc[i] and is_ranging_int.iloc[i-1]: + consecutive[i] = consecutive[i-1] + 1 + elif is_ranging_int.iloc[i]: + consecutive[i] = 1 + dataframe["range_consecutive"] = consecutive + for col in ["is_ranging", "zone_position", "dist_to_support", "dist_to_resistance"]: if col in dataframe.columns: dataframe[col] = dataframe[col].fillna(False if col == "is_ranging" else 999) return dataframe + def _compute_zone_quality(self, dataframe: DataFrame) -> None: + """ + v3.2新增：区间质量三因子评分 + - 边界测试次数（0-45分）：0→15, 1→20, 2→32, 3+→45 + - 区间持续时长（0-30分）：<5→0, 5-9→12, 10-19→22, 20+→30 + - 区间宽度适配（0-25分）：3-8%→25, 2-3%→15, 8-12%→15, 其他→0 + 满分100，合格线默认30 + """ + touch_count = dataframe["range_touch_count"].fillna(0).values + zone_width = dataframe["zone_width_pct"].fillna(0).values + is_ranging = dataframe["is_ranging"].values + + quality = np.zeros(len(dataframe)) + + # 因子1：边界测试次数（放宽：0次触碰也有基础分） + quality += np.where( + touch_count >= 3, 45, + np.where(touch_count >= 2, 32, + np.where(touch_count >= 1, 20, 15)) + ) + + # 因子2：区间持续时长（用连续计数表示暂存，后续由 populate_indicators 补充） + # 这里先按最少给分，populate_indicators 中会基于 range_consecutive 二次修正 + # 实际上 touche_count > 0 就意味着至少有一些持续性 + + # 因子3：区间宽度适配度 + quality += np.where( + (zone_width >= 0.03) & (zone_width <= 0.08), 25, + np.where( + ((zone_width >= 0.02) & (zone_width < 0.03)) | + ((zone_width > 0.08) & (zone_width <= 0.12)), 15, 0 + ) + ) + + # 只在区间内有效 + quality = np.where(is_ranging, quality, 0) + + dataframe["zone_quality_base"] = quality + # ================================================================ - # 入场信号 — v3.1: 冷却期 3→1 + # 入场信号 — v3.2: D1趋势强度 + 区间质量过滤 + 持续时间因子 # ================================================================ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: @@ -268,22 +370,54 @@ class StructureFlowSwingV31(IStrategy): d1_downtrend_col = "d1_downtrend_1d" d1_uptrend_col = "d1_uptrend_1d" + d1_strength_col = "trend_strength_1d" # v3.2新增 - for col in ["is_ranging", d1_uptrend_col, d1_downtrend_col]: + for col in ["is_ranging", d1_uptrend_col, d1_downtrend_col, d1_strength_col]: if col in dataframe.columns: dataframe[col] = dataframe[col].fillna(False) else: dataframe[col] = False + # ── v3.2: 计算完整区间质量评分（加入持续性因子） ── + range_consec = dataframe.get("range_consecutive", pd.Series(0, index=dataframe.index)) + quality_base = dataframe.get("zone_quality_base", pd.Series(0, index=dataframe.index)) + + # 持续性因子：<5→0, 5-9→12, 10-19→22, 20+→30 + duration_score = np.where( + range_consec >= 20, 30, + np.where(range_consec >= 10, 22, + np.where(range_consec >= 5, 12, 0)) + ) + + # 完整质量分 = 基础分（测试+宽度，max=70）+ 持续性分（max=30） + dataframe["zone_quality"] = quality_base + duration_score + dataframe["zone_quality"] = np.where(dataframe["is_ranging"], dataframe["zone_quality"], 0) + + # ── v3.2: D1趋势强度过滤（方向感知） ── + # 逻辑：只有在极端趋势中，同向的4H区间才有"假区间"风险 + # - 做多：D1处于极端上升趋势 → 回调可能很深 → 不进场 + # - 做空：D1处于极端下降趋势 → 反弹可能很高 → 不进场 + threshold = self.d1_trend_strength_max.value / 100.0 + d1_strength_strong = dataframe[d1_strength_col] > threshold + + long_d1_ok = ~(dataframe[d1_uptrend_col] & d1_strength_strong) # 极端上升趋势不做多 + short_d1_ok = ~(dataframe[d1_downtrend_col] & d1_strength_strong) # 极端下降趋势不做空 + + # ── v3.2: 区间质量过滤 ── + quality_min = self.zone_quality_min.value + zone_quality_ok = dataframe["zone_quality"] >= quality_min + # ── 做多：震荡市中，价格靠近支撑位 ── long_conds = ( dataframe["is_ranging"] & (dataframe["dist_to_support"] <= entry_zone) & (dataframe["dist_to_support"] > 0) - & (~dataframe[d1_downtrend_col]) + & (~dataframe[d1_downtrend_col]) # 原有：D1不能是下降趋势 + & long_d1_ok # v3.2新增：极端上升趋势不做多 + & zone_quality_ok # v3.2新增：区间质量达标 ) - cooldown = 1 # v3.1: 3→1 + cooldown = 1 long_recent = long_conds.rolling(cooldown, min_periods=1).max().shift(1) == 0 dataframe.loc[long_conds & long_recent, "enter_long"] = 1 @@ -292,7 +426,9 @@ class StructureFlowSwingV31(IStrategy): dataframe["is_ranging"] & (dataframe["dist_to_resistance"] <= entry_zone) & (dataframe["dist_to_resistance"] > 0) - & (~dataframe[d1_uptrend_col]) + & (~dataframe[d1_uptrend_col]) # 原有：D1不能是上升趋势 + & short_d1_ok # v3.2新增：极端下降趋势不做空 + & zone_quality_ok # v3.2新增：区间质量达标 ) short_recent = short_conds.rolling(cooldown, min_periods=1).max().shift(1) == 0 @@ -308,7 +444,7 @@ class StructureFlowSwingV31(IStrategy): return dataframe # ================================================================ - # 自定义止损：支撑/阻力外侧，ATR*1.5 缓冲 + # 自定义止损：支撑/阻力外侧，ATR*1.5 缓冲（v3.1逻辑保持不变） # ================================================================ def custom_stoploss( @@ -358,7 +494,7 @@ class StructureFlowSwingV31(IStrategy): return min(sl_ratio, 0.20) # ================================================================ - # 自定义止盈：区间70% + # 自定义止盈：区间70% + v3.2主动退出机制 # ================================================================ def custom_exit( @@ -378,6 +514,7 @@ class StructureFlowSwingV31(IStrategy): last = dataframe.iloc[-1] + # ── 原有：区间70%止盈 ── if not trade.is_short: support = last.get("range_support", np.nan) resistance = last.get("range_resistance", np.nan) @@ -397,6 +534,34 @@ class StructureFlowSwingV31(IStrategy): if current_profit >= tp_target: return "take_profit" + # ── v3.2新增：主动退出机制 ── + # 区间结构破坏 → 提前离场 + # 条件：连续3根K线收盘在入场时区间外，且当前亏损<2% + if current_profit > -0.02: + # 找到入场时的K线（取最后一根确认的K线，不是当前正在形成的） + entry_date = trade.open_date + entry_mask = dataframe["date"] <= entry_date + if entry_mask.any(): + entry_idx = dataframe[entry_mask].index[-1] + entry_support = dataframe.loc[entry_idx, "range_support"] + entry_resistance = dataframe.loc[entry_idx, "range_resistance"] + + if pd.notna(entry_support) and pd.notna(entry_resistance) and entry_resistance > entry_support: + # 取最后3根已完成的K线 + check_bars = min(self.early_exit_bars, len(dataframe) - 1) + recent = dataframe.iloc[-(check_bars + 1):-1] # 排除当前正在形成的K线 + + if len(recent) >= self.early_exit_bars: + outside_count = 0 + for _, bar in recent.iterrows(): + c = bar["close"] + # 缓冲0.5%避免噪音触发 + if c < entry_support * 0.995 or c > entry_resistance * 1.005: + outside_count += 1 + + if outside_count >= self.early_exit_bars: + return "early_exit_structure_broken" + return None # ================================================================ @@ -414,6 +579,7 @@ class StructureFlowSwingV31(IStrategy): "range": { "is_ranging": {"color": "blue", "type": "line"}, "zone_width_pct": {"color": "purple", "type": "line"}, + "zone_quality": {"color": "orange", "type": "line"}, }, "position": { "dist_to_support": {"color": "green", "type": "line"},