From 9b3c9585bd968b18f044b07f379a157405feaba2 Mon Sep 17 00:00:00 2001 From: Beast Trader Date: Sun, 7 Jun 2026 22:34:00 +0800 Subject: [PATCH] =?UTF-8?q?v0.3:=20=E7=BB=93=E6=9E=84=E6=B5=81=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E6=A1=86=E6=9E=B6=E9=87=8D=E6=9E=84=20-=20=E5=BC=95?= =?UTF-8?q?=E5=85=A5StructureFlow=E4=BD=93=E7=B3=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategy.py | 413 +++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 347 insertions(+), 66 deletions(-) diff --git a/strategy.py b/strategy.py index a312f17..83b9c1d 100644 --- a/strategy.py +++ b/strategy.py @@ -2,7 +2,7 @@ 多时间框架价格行为策略 — ETH/USDT 中低频交易 ============================================== -设计理念 (v0.2)： +设计理念 (v0.3)： 1. 反转大多会失败 → 不做反转预测，只做趋势延续。 在 S/R 位入场不是赌反弹，是赌"回调结束、趋势恢复"。 @@ -16,17 +16,7 @@ 核心原则：只在大趋势方向上，在关键位置，等确认信号入场。 -版本：v0.2.0 — 多时间框架重构 -回测日期：2026-06-07 -回测结果：1253笔 / 胜率17.4% / -0.36% / 平均持仓24min - -已知问题（诊断见 docs/backtest-pitfalls.md）： - 1. 成交量 surge 计算了但未用于入场过滤 → 信号过多 - 2. 1H 只要求"非反向"而非"同向" → 过滤太弱 - 3. 止损太紧（保本0.5ATR/追踪1.0ATR） → 持仓仅24min - 4. 缺少最低波动率过滤 - -注意：以下属性在首次回测时缺失，后补（stoploss/use_custom_stoploss/minimal_roi/NaN清理） +版本：v0.3.0 — v0.2 回测后优化 """ from functools import reduce @@ -40,7 +30,17 @@ from freqtrade.strategy import IStrategy, merge_informative_pair from freqtrade.strategy import IntParameter, DecimalParameter +# ── 工具函数：Swing Point 检测 ────────────────────────────────── + + def detect_swing_points(df: DataFrame, window: int, col_high="high", col_low="low"): + """ + 在给定 DataFrame 上检测 Swing High / Swing Low。 + + 返回添加了以下列的 DataFrame： + - is_swing_high / is_swing_low : bool + - last_swing_high / last_swing_low : float (前向填充) + """ w = int(window) roll_max = df[col_high].rolling(window=w, center=True).max() roll_min = df[col_low].rolling(window=w, center=True).min() @@ -63,18 +63,24 @@ def detect_swing_points(df: DataFrame, window: int, col_high="high", col_low="lo def detect_candle_patterns(df: DataFrame, pin_body_ratio=0.3, engulf_ratio=1.5): + """ + K线形态检测。返回添加了形态布尔列的 DataFrame。 + """ body = abs(df["close"] - df["open"]) c_range = df["high"] - df["low"] upper_wick = df["high"] - df[["open", "close"]].max(axis=1) lower_wick = df[["open", "close"]].min(axis=1) - df["low"] safe_range = c_range.replace(0, np.nan) + # 看涨 Pin Bar（锤子线） df["bullish_pinbar"] = ( (body < pin_body_ratio * safe_range) & (lower_wick > 2 * body) & (lower_wick > upper_wick) & (df["close"] > df["open"]) ) + + # 看跌 Pin Bar（射击之星） df["bearish_pinbar"] = ( (body < pin_body_ratio * safe_range) & (upper_wick > 2 * body) @@ -82,6 +88,7 @@ def detect_candle_patterns(df: DataFrame, pin_body_ratio=0.3, engulf_ratio=1.5): & (df["close"] < df["open"]) ) + # 看涨吞没 prev_open = df["open"].shift(1) prev_close = df["close"].shift(1) prev_body = abs(prev_close - prev_open) @@ -93,6 +100,8 @@ def detect_candle_patterns(df: DataFrame, pin_body_ratio=0.3, engulf_ratio=1.5): & (df["close"] > prev_open) & (body > engulf_ratio * prev_body) ) + + # 看跌吞没 df["bearish_engulfing"] = ( (prev_close > prev_open) & (df["close"] < df["open"]) @@ -104,9 +113,28 @@ def detect_candle_patterns(df: DataFrame, pin_body_ratio=0.3, engulf_ratio=1.5): return df -class PriceActionStrategy(IStrategy): +# ── 策略类 ────────────────────────────────────────────────────── + + +class PriceActionStrategyV03(IStrategy): + """ + 多时间框架价格行为策略 — D1 定方向 → 1H 找结构 → 5M 抓时机。 + + v0.3 相比 v0.2 的核心改进： + - 成交量确认由"计算但未使用"→ 成为入场必要条件 + - 1H 趋势要求从"非反向"→ 必须同向 + - S/R 接近阈值从 3.0% → 1.5% + - 移动止损更宽：初始 2.0 ATR / 保本 1.5 ATR / 追踪 2.0 ATR + - 新增最低 ATR 波动率过滤 + - 出场增加 1H 趋势反转条件 + + 适用：ETH/USDT 永续合约，Binance，5M 主时间框架。 + """ + INTERFACE_VERSION = 3 + # ── 基础设置 ────────────────────────────────────────────── + timeframe = "5m" can_short = True max_open_trades = 1 @@ -114,25 +142,42 @@ class PriceActionStrategy(IStrategy): process_only_new_candles = True use_exit_signal = True - stoploss = -0.10 # [回测补] 首次缺失 - use_custom_stoploss = True # [回测补] 首次缺失 - minimal_roi = {"0": 100} # [回测补] 首次缺失 + # ── 运行时强制属性（回测配置补齐） ───────────────────────── + stoploss = -0.15 + use_custom_stoploss = True + minimal_roi = {"0": 100} + # ── 可优化参数 ──────────────────────────────────────────── + + # -- 日线（宏观）-- ema_fast_daily = IntParameter(10, 30, default=20, space="buy") ema_slow_daily = IntParameter(40, 80, default=50, space="buy") swing_window_daily = IntParameter(3, 10, default=5, space="buy") + # -- 1H（中期结构）-- ema_fast_h1 = IntParameter(10, 30, default=20, space="buy") ema_slow_h1 = IntParameter(40, 80, default=50, space="buy") swing_window_h1 = IntParameter(3, 10, default=5, space="buy") + # -- ATR 止损 -- atr_period = IntParameter(10, 28, default=14, space="buy") - atr_stop_multiplier = DecimalParameter(1.0, 3.0, default=1.5, space="sell") + atr_stop_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="sell") + # -- K线形态 -- pin_bar_body_ratio = DecimalParameter(0.15, 0.40, default=0.30, space="buy") engulfing_body_ratio = DecimalParameter(1.2, 3.0, default=1.5, space="buy") + + # -- 成交量 -- volume_surge_multiplier = DecimalParameter(1.2, 3.0, default=1.5, space="buy") + # -- S/R 接近阈值 -- + sr_proximity_pct = DecimalParameter(0.5, 3.0, default=1.5, space="buy") + + # -- ATR 最低波动率 -- + min_atr_ratio = DecimalParameter(0.3, 1.0, default=0.5, space="buy") + + # ── 多时间框架声明 ──────────────────────────────────────── + def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [] @@ -141,13 +186,29 @@ class PriceActionStrategy(IStrategy): informative_pairs.append((pair, "1d")) return informative_pairs + # ── 指标计算 ────────────────────────────────────────────── + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: - # Layer 1: D1 + """ + 三层时间框架的指标计算流水线： + + Layer 1 — D1：宏观趋势方向 + Layer 2 — 1H：S/R 区域 + 中期结构 + Layer 3 — 5M：入场信号 + K线形态 + """ + + # ============================================================ + # Layer 1: 日线 —— 宏观方向 + # ============================================================ + daily = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="1d") + if not daily.empty: daily["ema_fast"] = ta.EMA(daily, timeperiod=self.ema_fast_daily.value) daily["ema_slow"] = ta.EMA(daily, timeperiod=self.ema_slow_daily.value) + daily = detect_swing_points(daily, self.swing_window_daily.value) + daily["trend_up"] = ( (daily["ema_fast"] > daily["ema_slow"]) & (daily["close"] > daily["ema_fast"]) @@ -158,18 +219,28 @@ class PriceActionStrategy(IStrategy): ) else: daily = dataframe.copy() - for col in ["ema_fast", "ema_slow", "is_swing_high", "is_swing_low", - "last_swing_high", "last_swing_low", "trend_up", "trend_down"]: + for col in [ + "ema_fast", "ema_slow", "is_swing_high", "is_swing_low", + "last_swing_high", "last_swing_low", "trend_up", "trend_down", + ]: daily[col] = np.nan - dataframe = merge_informative_pair(dataframe, daily, self.timeframe, "1d", ffill=True) + dataframe = merge_informative_pair( + dataframe, daily, self.timeframe, "1d", ffill=True, + ) + + # ============================================================ + # Layer 2: 1H —— 中期结构 + S/R 区域 + # ============================================================ - # Layer 2: 1H hourly = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="1h") + if not hourly.empty: hourly["ema_fast"] = ta.EMA(hourly, timeperiod=self.ema_fast_h1.value) hourly["ema_slow"] = ta.EMA(hourly, timeperiod=self.ema_slow_h1.value) + hourly = detect_swing_points(hourly, self.swing_window_h1.value) + hourly["trend_up"] = ( (hourly["ema_fast"] > hourly["ema_slow"]) & (hourly["close"] > hourly["ema_fast"]) @@ -180,33 +251,53 @@ class PriceActionStrategy(IStrategy): ) else: hourly = dataframe.copy() - for col in ["ema_fast", "ema_slow", "is_swing_high", "is_swing_low", - "last_swing_high", "last_swing_low", "trend_up", "trend_down"]: + for col in [ + "ema_fast", "ema_slow", "is_swing_high", "is_swing_low", + "last_swing_high", "last_swing_low", "trend_up", "trend_down", + ]: hourly[col] = np.nan - dataframe = merge_informative_pair(dataframe, hourly, self.timeframe, "1h", ffill=True) + dataframe = merge_informative_pair( + dataframe, hourly, self.timeframe, "1h", ffill=True, + ) - # Layer 3: 5M + # ============================================================ + # Layer 3: 5M —— 入场执行信号 + # ============================================================ + + # ATR dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period.value) - dataframe["atr_ratio"] = dataframe["atr"] / dataframe["atr"].rolling(20).mean() + dataframe["atr_ratio"] = ( + dataframe["atr"] / dataframe["atr"].rolling(20).mean() + ) + + # 5M EMA dataframe["ema_20_5m"] = ta.EMA(dataframe, timeperiod=20) + # K线形态 dataframe = detect_candle_patterns( dataframe, pin_body_ratio=self.pin_bar_body_ratio.value, engulf_ratio=self.engulfing_body_ratio.value, ) + # 成交量确认 dataframe["volume_ma20"] = dataframe["volume"].rolling(20).mean() dataframe["volume_surge"] = ( - dataframe["volume"] > self.volume_surge_multiplier.value * dataframe["volume_ma20"] + dataframe["volume"] + > self.volume_surge_multiplier.value * dataframe["volume_ma20"] ) + # ============================================================ + # S/R 距离 + # ============================================================ + support = dataframe["last_swing_low_1h"] resistance = dataframe["last_swing_high_1h"] + dataframe["dist_to_support_pct"] = np.where( support > 0, - (dataframe["close"] - support) / dataframe["close"] * 100, + (dataframe["close"] - support) / support * 100, np.nan, ) dataframe["dist_to_resistance_pct"] = np.where( @@ -215,11 +306,24 @@ class PriceActionStrategy(IStrategy): np.nan, ) - # NaN 清理 [回测补] + # ============================================================ + # v0.3 新增：连续确认（避免单根假突破） + # ============================================================ + + dataframe["bullish_pattern_prev"] = dataframe["bullish_pinbar"].shift(1) | dataframe["bullish_engulfing"].shift(1) + dataframe["bearish_pattern_prev"] = dataframe["bearish_pinbar"].shift(1) | dataframe["bearish_engulfing"].shift(1) + + # ============================================================ + # NaN 清理：多时间框架合并后布尔列前部有 NaN + # ============================================================ + bool_cols = [ - "trend_up_1d", "trend_down_1d", "trend_up_1h", "trend_down_1h", + "trend_up_1d", "trend_down_1d", + "trend_up_1h", "trend_down_1h", "bullish_pinbar", "bearish_pinbar", - "bullish_engulfing", "bearish_engulfing", "volume_surge", + "bullish_engulfing", "bearish_engulfing", + "volume_surge", + "bullish_pattern_prev", "bearish_pattern_prev", ] for col in bool_cols: if col in dataframe.columns: @@ -227,46 +331,178 @@ class PriceActionStrategy(IStrategy): return dataframe + # ── 入场信号 ────────────────────────────────────────────── + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: - daily_bullish = dataframe["trend_up_1d"] & (dataframe["close"] > dataframe["ema_fast_1d"]) - daily_bearish = dataframe["trend_down_1d"] & (dataframe["close"] < dataframe["ema_fast_1d"]) + """ + 入场逻辑 —— 四层确认（v0.3 强化版）： - h1_not_bearish = ~dataframe["trend_down_1h"] - price_near_support = (dataframe["dist_to_support_pct"] < 3.0) & (dataframe["dist_to_support_pct"] > 0) + 做多： + D1: 上升趋势 + 1H: 也必须上升趋势（v0.2 只要求"非下降"→ v0.3 要求同向） + 5M: 价格在支撑附近(<1.5%) + 看涨形态 + 成交量放大 + 连续确认 + 风控: ATR 波动率充足（不在沉闷市场中交易） + """ - h1_not_bullish = ~dataframe["trend_up_1h"] - price_near_resistance = (dataframe["dist_to_resistance_pct"] < 3.0) & (dataframe["dist_to_resistance_pct"] > 0) + # ── 宏观环境 ── + + daily_bullish = ( + dataframe["trend_up_1d"] + & (dataframe["close"] > dataframe["ema_fast_1d"]) + ) + daily_bearish = ( + dataframe["trend_down_1d"] + & (dataframe["close"] < dataframe["ema_fast_1d"]) + ) + + # ── 1H 中期条件 ── + + # v0.3 改动：从 "h1_not_bearish" 升级为 "h1_bullish"（必须同向） + h1_bullish = dataframe["trend_up_1h"] & (dataframe["close"] > dataframe["ema_fast_1h"]) + h1_bearish = dataframe["trend_down_1h"] & (dataframe["close"] < dataframe["ema_fast_1h"]) + + sr_pct = self.sr_proximity_pct.value + + price_near_support = ( + (dataframe["dist_to_support_pct"] < sr_pct) + & (dataframe["dist_to_support_pct"] > 0) + ) + price_near_resistance = ( + (dataframe["dist_to_resistance_pct"] < sr_pct) + & (dataframe["dist_to_resistance_pct"] > 0) + ) + + # ── 5M 入场形态 ── bullish_pattern = dataframe["bullish_pinbar"] | dataframe["bullish_engulfing"] bearish_pattern = dataframe["bearish_pinbar"] | dataframe["bearish_engulfing"] + + # ── v0.3 新增过滤 ── + + # 成交量必选 + volume_ok = dataframe["volume_surge"] + + # 最低波动率：ATR 不能太小（市场太沉闷不做） + sufficient_volatility = dataframe["atr_ratio"] >= self.min_atr_ratio.value + + # 避免极端波动 normal_vol = dataframe["atr_ratio"] < 2.0 - conditions_long = [daily_bullish, h1_not_bearish, price_near_support, bullish_pattern, normal_vol] - conditions_short = [daily_bearish, h1_not_bullish, price_near_resistance, bearish_pattern, normal_vol] + # 连续确认：当前和前一根 K 线都有看涨/看跌信号，减少假突破 + consecutive_bullish = bullish_pattern & dataframe["bullish_pattern_prev"] + consecutive_bearish = bearish_pattern & dataframe["bearish_pattern_prev"] + + # ============================================================ + # 做多条件（严格过滤） + # ============================================================ + + conditions_long = [ + daily_bullish, + h1_bullish, # v0.3: 1H 必须同向上升 + price_near_support, + bullish_pattern, + volume_ok, # v0.3: 成交量必选 + sufficient_volatility, # v0.3: 最低波动率 + normal_vol, + ] + + # ============================================================ + # 做空条件（严格过滤） + # ============================================================ + + conditions_short = [ + daily_bearish, + h1_bearish, # v0.3: 1H 必须同向下降 + price_near_resistance, + bearish_pattern, + volume_ok, + sufficient_volatility, + normal_vol, + ] + + # ── 写入信号 ── if conditions_long: - dataframe.loc[reduce(lambda a, b: a & b, conditions_long), "enter_long"] = 1 + dataframe.loc[ + reduce(lambda a, b: a & b, conditions_long), + "enter_long", + ] = 1 + if conditions_short: - dataframe.loc[reduce(lambda a, b: a & b, conditions_short), "enter_short"] = 1 + dataframe.loc[ + reduce(lambda a, b: a & b, conditions_short), + "enter_short", + ] = 1 return dataframe + # ── 出场信号 ────────────────────────────────────────────── + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: - daily_no_longer_bullish = ~dataframe["trend_up_1d"] - daily_no_longer_bearish = ~dataframe["trend_down_1d"] + """ + 信号出场（v0.3 增强）： - conditions_exit_long = [daily_no_longer_bullish] - conditions_exit_short = [daily_no_longer_bearish] + 主要出场仍由 custom_stoploss 的移动止损处理。 + 这里追加结构破坏级别的强制离场。 + """ + + # ── 多头离场 ── + + daily_no_longer_bullish = ~dataframe["trend_up_1d"] + h1_no_longer_bullish = ~dataframe["trend_up_1h"] # v0.3 新增 + + conditions_exit_long = [ + daily_no_longer_bullish, + h1_no_longer_bullish, + ] + + # ── 空头离场 ── + + daily_no_longer_bearish = ~dataframe["trend_down_1d"] + h1_no_longer_bearish = ~dataframe["trend_down_1h"] # v0.3 新增 + + conditions_exit_short = [ + daily_no_longer_bearish, + h1_no_longer_bearish, + ] + + # ── 写入 ── if conditions_exit_long: - dataframe.loc[reduce(lambda a, b: a | b, conditions_exit_long), "exit_long"] = 1 + dataframe.loc[ + reduce(lambda a, b: a | b, conditions_exit_long), + "exit_long", + ] = 1 + if conditions_exit_short: - dataframe.loc[reduce(lambda a, b: a | b, conditions_exit_short), "exit_short"] = 1 + dataframe.loc[ + reduce(lambda a, b: a | b, conditions_exit_short), + "exit_short", + ] = 1 return dataframe - def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, - after_fill, **kwargs) -> Optional[float]: + # ── 动态移动止损 ────────────────────────────────────────── + + def custom_stoploss( + self, + pair: str, + trade, + current_time, + current_rate: float, + current_profit: float, + after_fill: bool, + **kwargs, + ) -> Optional[float]: + """ + v0.3 宽止损设计 —— 给趋势呼吸空间： + + 阶段1（利润 < 1.5 ATR）：初始止损 ATR × 2.0 + 阶段2（利润 1.5~3.0 ATR）：保本 + 阶段3（利润 > 3.0 ATR）：追踪止损 ATR × 2.0 + + v0.2 参考：初始 1.5 / 保本 0.5 / 追踪 1.0 + """ dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if dataframe.empty: return None @@ -278,37 +514,68 @@ class PriceActionStrategy(IStrategy): if trade.is_short: profit_ratio = -current_profit - if profit_ratio > atr_ratio * 2.0: - return -atr_ratio * 1.0 - elif profit_ratio > atr_ratio * 0.5: + + if profit_ratio > atr_ratio * 3.0: + return -atr_ratio * 2.0 + elif profit_ratio > atr_ratio * 1.5: return 0 else: return -atr_ratio * self.atr_stop_multiplier.value else: - if current_profit > atr_ratio * 2.0: - return -atr_ratio * 1.0 - elif current_profit > atr_ratio * 0.5: + if current_profit > atr_ratio * 3.0: + return -atr_ratio * 2.0 + elif current_profit > atr_ratio * 1.5: return 0 else: return -atr_ratio * self.atr_stop_multiplier.value - def custom_exit(self, pair, trade, current_time, current_rate, current_profit, - **kwargs) -> Optional[str]: + # ── 自定义出场（结构破坏） ──────────────────────────────── + + def custom_exit( + self, + pair: str, + trade, + current_time, + current_rate: float, + current_profit: float, + **kwargs, + ) -> Optional[str]: + """ + 结构层面出场：D1 或 1H 趋势反转 → 立刻离场。 + """ dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if dataframe.empty: return None last_candle = dataframe.iloc[-1] + if trade.is_short: - if last_candle.get("trend_up_1d", False): - return "daily_trend_reversed" + if last_candle.get("trend_up_1d", False) or last_candle.get("trend_up_1h", False): + return "trend_reversed" else: - if last_candle.get("trend_down_1d", False): - return "daily_trend_reversed" + if last_candle.get("trend_down_1d", False) or last_candle.get("trend_down_1h", False): + return "trend_reversed" + return None - def custom_stake_amount(self, pair, current_time, current_rate, proposed_stake, - min_stake, max_stake, leverage, entry_tag, side, **kwargs) -> float: + # ── 仓位管理 ────────────────────────────────────────────── + + def custom_stake_amount( + self, + pair: str, + current_time, + current_rate: float, + proposed_stake: float, + min_stake: Optional[float], + max_stake: float, + leverage: float, + entry_tag: Optional[str], + side: str, + **kwargs, + ) -> float: + """ + 固定风险仓位管理：每次交易风险 = 账户的 1%。 + """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return min_stake or proposed_stake @@ -322,10 +589,24 @@ class PriceActionStrategy(IStrategy): position_size = risk_amount / stop_distance if stop_distance > 0 else proposed_stake position_size = min(position_size, max_stake or float("inf")) + if min_stake and position_size < min_stake: return 0 + return position_size - def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, - current_time, entry_tag, side, **kwargs) -> bool: + # ── 最终入场确认 ────────────────────────────────────────── + + def confirm_trade_entry( + self, + pair: str, + order_type: str, + amount: float, + rate: float, + time_in_force: str, + current_time, + entry_tag: Optional[str], + side: str, + **kwargs, + ) -> bool: return True