v0.1: 初始价格行为策略 - 基础S/R + 结构突破

strategy.py

@@ -0,0 +1,612 @@
+"""
+多时间框架价格行为策略 — ETH/USDT 中低频交易
+==============================================
+
+设计理念 (v0.3)：
+
+1. 反转大多会失败 → 不做反转预测，只做趋势延续。
+   在 S/R 位入场不是赌反弹，是赌"回调结束、趋势恢复"。
+
+2. 移动止损优先 → 放弃固定止盈，用 ATR 追踪止损让利润在趋势中奔跑。
+
+3. 多时间框架自上而下分析：
+   D1 → 判断宏观方向（能不能做）
+   1H → 识别中期结构 + S/R 区域（在哪做）
+   5M → 确认入场时机（什么时候做）
+
+核心原则：只在大趋势方向上，在关键位置，等确认信号入场。
+
+版本：v0.3.0 — v0.2 回测后优化
+"""
+
+from functools import reduce
+from typing import Optional
+
+import numpy as np
+import pandas as pd
+import talib.abstract as ta
+from pandas import DataFrame
+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()
+
+    df["is_swing_high"] = (
+        (df[col_high] == roll_max)
+        & (df[col_high] > df[col_high].shift(1))
+        & (df[col_high] > df[col_high].shift(-1))
+    )
+    df["is_swing_low"] = (
+        (df[col_low] == roll_min)
+        & (df[col_low] < df[col_low].shift(1))
+        & (df[col_low] < df[col_low].shift(-1))
+    )
+
+    df["last_swing_high"] = df[col_high].where(df["is_swing_high"]).ffill()
+    df["last_swing_low"] = df[col_low].where(df["is_swing_low"]).ffill()
+
+    return df
+
+
+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)
+        & (upper_wick > lower_wick)
+        & (df["close"] < df["open"])
+    )
+
+    # 看涨吞没
+    prev_open = df["open"].shift(1)
+    prev_close = df["close"].shift(1)
+    prev_body = abs(prev_close - prev_open)
+
+    df["bullish_engulfing"] = (
+        (prev_close < prev_open)
+        & (df["close"] > df["open"])
+        & (df["open"] < prev_close)
+        & (df["close"] > prev_open)
+        & (body > engulf_ratio * prev_body)
+    )
+
+    # 看跌吞没
+    df["bearish_engulfing"] = (
+        (prev_close > prev_open)
+        & (df["close"] < df["open"])
+        & (df["open"] > prev_close)
+        & (df["close"] < prev_open)
+        & (body > engulf_ratio * prev_body)
+    )
+
+    return df
+
+
+# ── 策略类 ──────────────────────────────────────────────────────
+
+
+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
+    startup_candle_count = 200
+    process_only_new_candles = True
+    use_exit_signal = True
+
+    # ── 运行时强制属性（回测配置补齐） ─────────────────────────
+    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.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 = []
+        for pair in pairs:
+            informative_pairs.append((pair, "1h"))
+            informative_pairs.append((pair, "1d"))
+        return informative_pairs
+
+    # ── 指标计算 ──────────────────────────────────────────────
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        三层时间框架的指标计算流水线：
+
+        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"])
+            )
+            daily["trend_down"] = (
+                (daily["ema_fast"] < daily["ema_slow"])
+                & (daily["close"] < daily["ema_fast"])
+            )
+        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",
+            ]:
+                daily[col] = np.nan
+
+        dataframe = merge_informative_pair(
+            dataframe, daily, self.timeframe, "1d", ffill=True,
+        )
+
+        # ============================================================
+        # Layer 2: 1H —— 中期结构 + S/R 区域
+        # ============================================================
+
+        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"])
+            )
+            hourly["trend_down"] = (
+                (hourly["ema_fast"] < hourly["ema_slow"])
+                & (hourly["close"] < hourly["ema_fast"])
+            )
+        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",
+            ]:
+                hourly[col] = np.nan
+
+        dataframe = merge_informative_pair(
+            dataframe, hourly, self.timeframe, "1h", ffill=True,
+        )
+
+        # ============================================================
+        # Layer 3: 5M —— 入场执行信号
+        # ============================================================
+
+        # ATR
+        dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
+        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"]
+        )
+
+        # ============================================================
+        # 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) / support * 100,
+            np.nan,
+        )
+        dataframe["dist_to_resistance_pct"] = np.where(
+            resistance > 0,
+            (resistance - dataframe["close"]) / dataframe["close"] * 100,
+            np.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",
+            "bullish_pinbar", "bearish_pinbar",
+            "bullish_engulfing", "bearish_engulfing",
+            "volume_surge",
+            "bullish_pattern_prev", "bearish_pattern_prev",
+        ]
+        for col in bool_cols:
+            if col in dataframe.columns:
+                dataframe[col] = dataframe[col].fillna(False).infer_objects(copy=False)
+
+        return dataframe
+
+    # ── 入场信号 ──────────────────────────────────────────────
+
+    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        入场逻辑 —— 四层确认（v0.3 强化版）：
+
+        做多：
+          D1: 上升趋势
+          1H: 也必须上升趋势（v0.2 只要求"非下降"→ v0.3 要求同向）
+          5M: 价格在支撑附近(<1.5%) + 看涨形态 + 成交量放大 + 连续确认
+          风控: ATR 波动率充足（不在沉闷市场中交易）
+        """
+
+        # ── 宏观环境 ──
+
+        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
+
+        # 连续确认：当前和前一根 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
+
+        if conditions_short:
+            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:
+        """
+        信号出场（v0.3 增强）：
+
+        主要出场仍由 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
+
+        if conditions_exit_short:
+            dataframe.loc[
+                reduce(lambda a, b: a | b, conditions_exit_short),
+                "exit_short",
+            ] = 1
+
+        return dataframe
+
+    # ── 动态移动止损 ──────────────────────────────────────────
+
+    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
+
+        last_candle = dataframe.iloc[-1]
+        atr = last_candle.get("atr", current_rate * 0.005)
+        entry_price = trade.open_rate
+        atr_ratio = atr / entry_price
+
+        if trade.is_short:
+            profit_ratio = -current_profit
+
+            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 * 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: 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) or last_candle.get("trend_up_1h", False):
+                return "trend_reversed"
+        else:
+            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: 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
+
+        last_candle = dataframe.iloc[-1]
+        atr = last_candle.get("atr", current_rate * 0.005)
+        stop_distance = atr * self.atr_stop_multiplier.value
+
+        available_balance = self.wallets.get_total_stake_amount()
+        risk_amount = available_balance * 0.01
+
+        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: str,
+        order_type: str,
+        amount: float,
+        rate: float,
+        time_in_force: str,
+        current_time,
+        entry_tag: Optional[str],
+        side: str,
+        **kwargs,
+    ) -> bool:
+        return True