From 58ec67203fdeee7e79fc2c1047e5fff6dde45fbf Mon Sep 17 00:00:00 2001 From: Beast Trader Date: Wed, 10 Jun 2026 23:31:00 +0800 Subject: [PATCH] =?UTF-8?q?v4.2=20(Swing):=20=E6=9E=81=E7=AE=80=E7=89=88?= =?UTF-8?q?=20-=20=E6=9C=80=E5=B0=8F=E5=8C=96=E5=8F=82=E6=95=B0=20+=20?= =?UTF-8?q?=E7=BA=AF=E7=BB=93=E6=9E=84=E4=BF=A1=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategy.py | 332 +++++++++++++++++++++------------------------------- 1 file changed, 132 insertions(+), 200 deletions(-) diff --git a/strategy.py b/strategy.py index 001e1a8..0544167 100644 --- a/strategy.py +++ b/strategy.py @@ -1,10 +1,18 @@ """ -Structure Flow Swing Strategy v4.1 +Structure Flow Swing Strategy v4.2 ================================== -15m 震荡波段 — 基于"碰壁验证"价格聚集度检测（短窗口版） +纯价格行为学震荡策略 — 借鉴 v2.2b 的 swing point + K线形态框架 -v4.1b (2026-06-10): 1H + 3K线验证（正收益但频率太低） -v4.1c (2026-06-10): 回到15m + 缩短lookback至24根(6h) + 保留min_rejections=2 +核心逻辑（模拟手工交易）： +1. 用 swing point 识别近期高低点，自动形成交易区间 +2. 价格到支撑 + K线止跌形态（bullish pinbar/engulfing）→ 做多 +3. 价格到阻力 + K线滞涨 → 平多（反向开空同理） +4. 不在震荡判定上设严苛门槛，价格够到边界+形态确认就做 +5. 趋势突破时，突破边界导致亏损，但可控 + +v4.1 系列教训（2026-06-10）： + 用滚动window + 碰壁验证 + ATR比例止损 → 全是技术指标思维，不是价格行为学 + 唯一正收益的是最简单的版本（1H + 形态 + 固定-3%止损） """ from datetime import datetime @@ -15,158 +23,132 @@ from freqtrade.strategy import IStrategy, IntParameter from freqtrade.persistence import Trade -class StructureFlowSwingV41(IStrategy): - """ - Structure Flow Swing Strategy v4.1 - 15m 震荡区间波段交易 — 碰壁验证（短窗口） - """ - +class StructureFlowSwingV42(IStrategy): can_short = True stoploss = -0.20 use_custom_stoploss = True minimal_roi = {"0": 100} max_open_trades = 1 - timeframe = "15m" + timeframe = "1h" - # ===================== - # 可优化参数 - # ===================== - lookback = IntParameter(8, 48, default=12, space="buy") # 检测窗口：12根15m = 3h - min_rejections = IntParameter(1, 4, default=1, space="buy") # 碰壁验证1次即可 - rejection_window = IntParameter(1, 6, default=3, space="buy") # 碰壁验证窗口 - zone_width_atr_mult = IntParameter(2, 6, default=4, space="buy") - entry_zone_pct = IntParameter(2, 8, default=5, space="buy") - take_profit_pct = IntParameter(50, 80, default=70, space="sell") + # ── 价格行为学参数 ── + swing_window = IntParameter(5, 11, default=11, space="buy") # swing point 检测窗口 + entry_zone_pct = IntParameter(2, 8, default=5, space="buy") # 入场范围（距S/R 0.5%） # 固定参数 - breakout_bars = 2 - cooldown = 2 # 冷却 2 根15m（30分钟） + cooldown = 6 # 冷却6根1H（6小时） - # ===================== - # 工具：ATR计算 - # ===================== + # ================================================================ + # Swing Point 检测（v2.2b 同款） + # ================================================================ + + def _detect_swing_points( + self, + 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 + + # ================================================================ + # K线形态检测（v2.2b 同款） + # ================================================================ @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() + def _detect_pinbar(open: pd.Series, high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series: + body = (open - close).abs() + total_range = (high - low) + upper_wick = high - open.where(open > close, close) + lower_wick = open.where(open < close, close) - low - # ===================== - # 主时间框架 — 15m 指标 - # ===================== + bullish_pin = (lower_wick > body * 2) & (lower_wick > upper_wick * 2) & (close > open) + bearish_pin = (upper_wick > body * 2) & (upper_wick > lower_wick * 2) & (close < open) + return bullish_pin, bearish_pin + + @staticmethod + def _detect_engulfing(open: pd.Series, close: pd.Series) -> tuple[pd.Series, pd.Series]: + prev_open = open.shift(1) + prev_close = close.shift(1) + bullish_eng = (close > open) & (prev_close < prev_open) & (close > prev_open) & (open < prev_close) + bearish_eng = (close < open) & (prev_close > prev_open) & (close < prev_open) & (open > prev_close) + return bullish_eng, bearish_eng + + # ================================================================ + # 主指标 + # ================================================================ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: - lookback = self.lookback.value - rej_threshold = 0.005 # 边界 0.5% 范围内算"碰边" + sw = self.swing_window.value - # ── 价格聚集范围 ── - rolling_high = dataframe["high"].rolling(lookback).max() - rolling_low = dataframe["low"].rolling(lookback).min() - zone_width_raw = rolling_high - rolling_low + # ── Swing Points ── + sh, sl = self._detect_swing_points(dataframe["high"], dataframe["low"], sw) - dataframe["zone_high"] = rolling_high - dataframe["zone_low"] = rolling_low - dataframe["zone_width_raw"] = zone_width_raw - dataframe["zone_width_pct"] = zone_width_raw / rolling_low + # 向前填充最近的 swing high / low 作为动态 S/R + dataframe["swing_high"] = sh.ffill() + dataframe["swing_low"] = sl.ffill() - # ATR - dataframe["atr"] = self._calc_atr(dataframe["high"], dataframe["low"], dataframe["close"], 14) - - # ── 碰壁验证检测 ── - # 支撑验证：价格到了低点附近 → 随后 rejection_window 根K线内有反弹 - near_low = dataframe["low"] <= rolling_low * (1 + rej_threshold) - bounced_up = pd.Series(False, index=dataframe.index) - for i in range(1, self.rejection_window.value + 1): - future_close = dataframe["close"].shift(-i).fillna(dataframe["close"]) - bounced_up = bounced_up | (future_close > dataframe["close"]) - support_rejection = near_low & bounced_up - dataframe["support_rejection_count"] = support_rejection.rolling(lookback).sum() - - # 阻力验证：价格到了高点附近 → 随后 rejection_window 根K线内有回落 - near_high = dataframe["high"] >= rolling_high * (1 - rej_threshold) - bounced_down = pd.Series(False, index=dataframe.index) - for i in range(1, self.rejection_window.value + 1): - future_close = dataframe["close"].shift(-i).fillna(dataframe["close"]) - bounced_down = bounced_down | (future_close < dataframe["close"]) - resistance_rejection = near_high & bounced_down - dataframe["resistance_rejection_count"] = resistance_rejection.rolling(lookback).sum() - - # ── 震荡判定 ── - atr_mult = self.zone_width_atr_mult.value - min_rej = self.min_rejections.value - - # 条件1：区间宽度合理（不超过 ATR × N） - is_compact = zone_width_raw <= dataframe["atr"] * atr_mult - - # 条件2：上下边界都经过碰壁验证（各至少 min_rej 次） - support_ok = dataframe["support_rejection_count"] >= min_rej - resistance_ok = dataframe["resistance_rejection_count"] >= min_rej - - # 条件3：无突破 - no_break_high = True - no_break_low = True - for i in range(1, self.breakout_bars + 1): - if i <= len(dataframe): - no_break_high = no_break_high & (dataframe["close"].shift(i) <= rolling_high) - no_break_low = no_break_low & (dataframe["close"].shift(i) >= rolling_low) - - is_ranging = is_compact & support_ok & resistance_ok & no_break_high & no_break_low - dataframe["is_ranging"] = is_ranging - - # ── 价格在区间内的位置 ── - denom = rolling_high - rolling_low - dataframe["zone_position"] = np.where( - denom > 0, - (dataframe["close"] - rolling_low) / denom, + # ── 区间宽度（用于止盈参考） ── + dataframe["zone_width"] = np.where( + dataframe["swing_high"].notna() & dataframe["swing_low"].notna() & (dataframe["swing_high"] > dataframe["swing_low"]), + (dataframe["swing_high"] - dataframe["swing_low"]) / dataframe["swing_low"], np.nan, ) - dataframe["dist_to_low"] = np.where( - rolling_low > 0, - (dataframe["close"] - rolling_low) / dataframe["close"], + # ── K线形态 ── + bull_pin, bear_pin = self._detect_pinbar( + dataframe["open"], dataframe["high"], dataframe["low"], dataframe["close"] + ) + bull_eng, bear_eng = self._detect_engulfing(dataframe["open"], dataframe["close"]) + + dataframe["bullish_signal"] = bull_pin | bull_eng + dataframe["bearish_signal"] = bear_pin | bear_eng + + # ── 距边界的距离 ── + dataframe["dist_to_swing_low"] = np.where( + dataframe["swing_low"].notna() & (dataframe["swing_low"] > 0), + (dataframe["close"] - dataframe["swing_low"]) / dataframe["close"], np.nan, ) - dataframe["dist_to_high"] = np.where( - rolling_high > 0, - (rolling_high - dataframe["close"]) / dataframe["close"], + dataframe["dist_to_swing_high"] = np.where( + dataframe["swing_high"].notna() & (dataframe["swing_high"] > 0), + (dataframe["swing_high"] - dataframe["close"]) / dataframe["close"], np.nan, ) - # ── 填充 ── - for col in ["is_ranging", "zone_position", "dist_to_low", "dist_to_high"]: - if col in dataframe.columns: - dataframe[col] = dataframe[col].fillna(False if col == "is_ranging" else 999) + for col in ["dist_to_swing_low", "dist_to_swing_high"]: + dataframe[col] = dataframe[col].fillna(999) return dataframe # ================================================================ - # 入场信号 + # 入场 # ================================================================ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: entry_zone = self.entry_zone_pct.value / 1000.0 - if "is_ranging" not in dataframe.columns: - dataframe["is_ranging"] = False - - # ── 做多：震荡中，价格靠近下边界 ── + # ── 做多：价格在 swing low 附近 + 止跌形态 ── long_conds = ( - dataframe["is_ranging"] - & (dataframe["dist_to_low"] < entry_zone) - & (dataframe["dist_to_low"] > 0) + (dataframe["dist_to_swing_low"] < entry_zone) + & (dataframe["dist_to_swing_low"] > 0) + & dataframe["bullish_signal"] ) long_recent = long_conds.rolling(self.cooldown, min_periods=1).max().shift(1) == 0 dataframe.loc[long_conds & long_recent, "enter_long"] = 1 - # ── 做空：震荡中，价格靠近上边界 ── + # ── 做空：价格在 swing high 附近 + 止涨形态 ── short_conds = ( - dataframe["is_ranging"] - & (dataframe["dist_to_high"] < entry_zone) - & (dataframe["dist_to_high"] > 0) + (dataframe["dist_to_swing_high"] < entry_zone) + & (dataframe["dist_to_swing_high"] > 0) + & dataframe["bearish_signal"] ) short_recent = short_conds.rolling(self.cooldown, min_periods=1).max().shift(1) == 0 dataframe.loc[short_conds & short_recent, "enter_short"] = 1 @@ -174,14 +156,14 @@ class StructureFlowSwingV41(IStrategy): return dataframe # ================================================================ - # 出场信号 + # 出场 # ================================================================ def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe # ================================================================ - # 自定义止损：固定入场价下方-3%（真固定，不追踪） + # 止损：区间宽度 × 0.5（自适应） # ================================================================ def custom_stoploss( @@ -194,15 +176,23 @@ class StructureFlowSwingV41(IStrategy): after_fill: bool, **kwargs, ) -> float: - # 固定止损 = 入场价下方3%，不随当前价格移动 - # 不用current_rate，用trade.open_rate确保锚点固定 - if not trade.is_short: - return max((trade.open_rate * 0.98 / current_rate) - 1.0, -0.20) + 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 + + z_width = dataframe.iloc[-1].get("zone_width", np.nan) + if pd.notna(z_width) and z_width > 0.005: + stop_pct = min(max(z_width * 0.5, 0.005), 0.05) else: - return min(1.0 - (trade.open_rate * 1.02 / current_rate), 0.20) + stop_pct = 0.015 + + if not trade.is_short: + return max((trade.open_rate * (1 - stop_pct) / current_rate) - 1.0, -0.20) + else: + return min(1.0 - (trade.open_rate * (1 + stop_pct) / current_rate), 0.20) # ================================================================ - # 自定义止盈：区间高度 × TP% + # 止盈：到对侧边界 + K线形态确认 → 平仓 # ================================================================ def custom_exit( @@ -214,89 +204,31 @@ class StructureFlowSwingV41(IStrategy): 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] - z_low = last.get("zone_low", np.nan) - z_high = last.get("zone_high", np.nan) + swing_high = last.get("swing_high", np.nan) + swing_low = last.get("swing_low", np.nan) + bull_sig = last.get("bullish_signal", False) + bear_sig = last.get("bearish_signal", False) - if pd.notna(z_low) and pd.notna(z_high) and z_high > z_low: - base = z_low if not trade.is_short else z_high - zone_height = (z_high - z_low) / base - if current_profit >= zone_height * tp_pct: - return "take_profit" - - return None - - # ================================================================ - # 自定义退出：锁定入场边界 + 收盘确认止损 + 区间高度止盈 - # ================================================================ - - 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] - - # ── 锁定入场时的区间边界 ── - open_time = trade.open_date_utc.replace(tzinfo=None) - time_diff = (dataframe["date"] - open_time).abs() - open_idx = time_diff.idxmin() - open_row = dataframe.iloc[open_idx] - - z_low_open = open_row.get("zone_low", np.nan) - z_high_open = open_row.get("zone_high", np.nan) - - # ── 结构止损：收盘价确认跌破入场时的边界 ── + # ── 做多：到阻力附近 + 滞涨形态 → 平仓 ── if not trade.is_short: - if pd.notna(z_low_open) and z_low_open > 0 and last["close"] < z_low_open: - return "stop_loss" # 收盘跌破支撑 + if pd.notna(swing_high) and swing_high > 0: + near_high = current_rate >= swing_high * 0.99 + if near_high and bear_sig: + return "exit_signal" + if near_high and current_profit > 0: + return "exit_signal" + # ── 做空：到支撑附近 + 止跌形态 → 平仓 ── else: - if pd.notna(z_high_open) and z_high_open > 0 and last["close"] > z_high_open: - return "stop_loss" # 收盘涨破阻力 - - # ── 止盈：区间高度 × TP%（也用锁定边界） ── - if pd.notna(z_low_open) and pd.notna(z_high_open) and z_high_open > z_low_open: - base = z_low_open if not trade.is_short else z_high_open - zone_height = (z_high_open - z_low_open) / base - if current_profit >= zone_height * tp_pct: - return "take_profit" + if pd.notna(swing_low) and swing_low > 0: + near_low = current_rate <= swing_low * 1.01 + if near_low and bull_sig: + return "exit_signal" + if near_low and current_profit > 0: + return "exit_signal" return None - - # ================================================================ - # Plot config - # ================================================================ - - @staticmethod - def plot_config() -> dict: - return { - "main_plot": { - "zone_high": {"color": "red", "type": "line"}, - "zone_low": {"color": "green", "type": "line"}, - }, - "subplots": { - "rejections": { - "support_rejection_count": {"color": "green", "type": "line"}, - "resistance_rejection_count": {"color": "red", "type": "line"}, - }, - "zone": { - "is_ranging": {"color": "blue", "type": "line"}, - }, - }, - }