diff --git a/strategy.py b/strategy.py index c4685be..87a5b27 100644 --- a/strategy.py +++ b/strategy.py @@ -1,17 +1,16 @@ # ============================================================================ -# Structure Flow Strategy v1.1 +# Structure Flow Strategy v1.2 # 纯价格结构策略 — 零技术指标，价格行为学驱动 # -# 版本变化 v1.0 → v1.1: -# - 主时间框架：5M → 1H（匹配用户实际交易尺度） -# - 信息时间框架：D1 → 4H → 1H -# - 启用做空（can_short=True），适配无杠杆合约交易 -# - 硬止损改为结构失效点 + 1% 缓冲（不再用固定百分比） -# - 修复 custom_stoploss 动态跟踪结构位 +# 版本变化 v1.1 → v1.2: +# - 硬止损改为 Entry Candle 失效点（做多→入场K线低点，做空→入场K线高点） +# - 新增时间止损：入场后 N 根K线内无盈利则主动出场 +# - 保留 trailing_stop（结构跟踪止损），盈利后切换 +# - 策略类重命名为 StructureFlowStrategyV12 # # 设计哲学: # 趋势由 HH/HL 定义，支撑阻力由 Swing Point 定义， -# 止损由结构失效点定义，出场由结构反转定义。 +# 止损由 Entry Candle 失效点定义，出场由结构反转定义。 # # 多时间框架: # D1 → 宏观结构方向 @@ -19,7 +18,7 @@ # 1H → K线形态确认入场时机 # ============================================================================ -from datetime import datetime +from datetime import datetime, timedelta import numpy as np import pandas as pd from pandas import DataFrame @@ -27,9 +26,9 @@ from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, inform from freqtrade.persistence import Trade -class StructureFlowStrategy(IStrategy): +class StructureFlowStrategyV12(IStrategy): """ - Structure Flow Strategy v1.1 — 纯价格结构策略 + Structure Flow Strategy v1.2 — 纯价格结构策略 不使用任何技术指标（无 EMA、ATR、RSI、MACD、布林带等）。 一切信号来源于价格本身的 OHLC 数据和由此推导的结构信息。 @@ -42,22 +41,23 @@ class StructureFlowStrategy(IStrategy): 做多: D1上升结构 + 价格在4H Swing区间下半区 + 1H看涨K线形态 做空: D1下降结构 + 价格在4H Swing区间上半区 + 1H看跌K线形态 - 止损逻辑: - 初始止损: 4H 最近 Swing Low（做多）/ Swing High（做空）+ 1% 缓冲 - 动态止损: custom_stoploss 随新 Swing Point 形成而跟踪 + 止损逻辑（v1.2 核心改进）: + 初始止损: Entry Candle 失效点（做多→入场K线最低价，做空→入场K线最高价） + 动态止损: 盈利后切换为结构跟踪止损（custom_stoploss） + 时间止损: 入场后 N 根K线内无盈利则主动出场 """ # ── 基础配置 ────────────────────────────────────────── timeframe = "1h" - can_short = True # v1.1 启用做空，适配无杠杆合约 - stoploss = -0.05 # 硬止损 5%，实际由 custom_stoploss 动态管理 + can_short = True + stoploss = -0.05 # 硬止损 5%，实际由 custom_stoploss 动态管理 use_custom_stoploss = True - minimal_roi = {"0": 100} # 不设时间止盈，出场由结构决定 + minimal_roi = {"0": 100} # 不设时间止盈，出场由结构决定 max_open_trades = 1 # 回测参数 - startup_candle_count = 40 # 需要更多历史数据（1H 级别） + startup_candle_count = 40 # ── 可调参数 ────────────────────────────────────────── @@ -73,9 +73,20 @@ class StructureFlowStrategy(IStrategy): 1.5, 4.0, default=2.0, space="buy", ) - # 结构止损缓冲（%）：止损设在结构位之外一点，避免被噪音扫损 - sl_buffer_pct = DecimalParameter( - 0.005, 0.03, default=0.01, space="sell", + # Entry Candle 止损缓冲（%）：略低于/高于 Entry Candle 低点/高点 + entry_sl_buffer = DecimalParameter( + 0.001, 0.01, default=0.005, space="sell", + optimize=True, + ) + + # 时间止损：入场后 N 根K线内无盈利则出场 + time_stop_bars = IntParameter( + 6, 48, default=12, space="sell", + ) + + # 盈利后切换为结构止损的触发距离（ATR 倍数，暂无ATR，用固定比例代替） + profit_to_structure_sl_pct = DecimalParameter( + 0.01, 0.05, default=0.02, space="sell", optimize=True, ) @@ -319,14 +330,19 @@ class StructureFlowStrategy(IStrategy): return dataframe # ================================================================ - # 主时间框架 — 1H K线形态 + # 主时间框架 — 1H K线形态 + Entry Candle 记录 # ================================================================ + # 类级别缓存：记录每笔交易的 Entry Candle 信息 + # {trade_id: {"entry_low": float, "entry_high": float, "entry_idx": int}} + _entry_candle_cache = {} + def populate_indicators( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: """ 1H 一小时线：检测 K 线形态。 + 同时预标记可能的入场 K 线（供 custom_stoploss 使用）。 """ bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = ( self._detect_candle_patterns( @@ -346,6 +362,19 @@ class StructureFlowStrategy(IStrategy): dataframe["bullish_signal"] = bullish_pin | bullish_engulf dataframe["bearish_signal"] = bearish_pin | bearish_engulf + # 预标记：如果这根 K 线是入场信号，记录其 OHLC（供后续 custom_stoploss 使用） + # 注意：这里只是标记，实际入场由 populate_entry_trend 决定 + dataframe["potential_entry_low"] = np.where( + dataframe["bullish_signal"] | dataframe["bearish_signal"], + dataframe["low"], + np.nan, + ) + dataframe["potential_entry_high"] = np.where( + dataframe["bullish_signal"] | dataframe["bearish_signal"], + dataframe["high"], + np.nan, + ) + return dataframe # ================================================================ @@ -409,13 +438,13 @@ class StructureFlowStrategy(IStrategy): 出场逻辑 — 由结构反转触发。 """ - # 做多出场：D1 不再上升 或 4H 不再上升 + # 做多出场：D1 不再上升 exit_long = ( ~dataframe["trend_up_1d"].fillna(True) ) dataframe.loc[exit_long, "exit_long"] = 1 - # 做空出场：D1 不再下降 或 4H 不再下降 + # 做空出场：D1 不再下降 exit_short = ( dataframe["trend_up_1d"].fillna(False) ) @@ -424,7 +453,7 @@ class StructureFlowStrategy(IStrategy): return dataframe # ================================================================ - # 动态止损 — 基于结构失效 + # 动态止损 — v1.2 核心改进 # ================================================================ def custom_stoploss( @@ -438,33 +467,113 @@ class StructureFlowStrategy(IStrategy): **kwargs, ) -> float | None: """ - 结构止损：止损位设在最近的 4H Swing Low（做多）或 Swing High（做空）， - 加上缓冲距离（sl_buffer_pct）。 + v1.2 止损逻辑（核心改进）： - 随着行情发展，新的 Swing Point 形成，止损自动跟随。 + 阶段一（刚入场，无盈利或微盈利）： + 止损 = Entry Candle 失效点 + 缓冲 + - 做多：入场K线最低价 × (1 - entry_sl_buffer) + - 做空：入场K线最高价 × (1 + entry_sl_buffer) + + 阶段二（有一定盈利，超过 profit_to_structure_sl_pct）： + 切换为结构跟踪止损（同 v1.1 逻辑） + - 做多：最近 4H Swing Low × (1 - buffer) + - 做空：最近 4H Swing High × (1 + buffer) + + 时间止损： + 入场后超过 time_stop_bars 根K线且 current_profit < 0， + 返回 -0.01（立即市价出场）。 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) == 0: return None last = dataframe.iloc[-1] - buffer = self.sl_buffer_pct.value + buffer = self.entry_sl_buffer.value - if trade.is_short: - resistance = last.get("resistance_4h") - if resistance is not None and not (isinstance(resistance, float) and np.isnan(resistance)): - # 做空止损：resistance × (1 + buffer)，在当前价格上方 - sl_price = float(resistance) * (1 + buffer) - sl_ratio = (current_rate - sl_price) / current_rate - if sl_ratio < 0: # 止损在当前价格上方（做空方向正确） - return max(sl_ratio, -0.25) # 不超过 25% - else: - support = last.get("support_4h") - if support is not None and not (isinstance(support, float) and np.isnan(support)): - # 做多止损：support × (1 - buffer)，在当前价格下方 - sl_price = float(support) * (1 - buffer) - sl_ratio = (sl_price - current_rate) / current_rate - if sl_ratio < 0: # 止损在当前价格下方（做多方向正确） - return max(sl_ratio, -0.25) + # ── 时间止损检查 ── + # 计算入场至今的K线数（1H = 1根/小时） + bars_held = (current_time - trade.open_date_utc).total_seconds() / 3600 + if bars_held >= self.time_stop_bars.value and current_profit <= 0: + # 超时且无盈利，立即出场（返回当前价，即市价出场） + return -0.01 # 1% 内市价出场 + + # ── 尝试获取 Entry Candle 信息 ── + # 方法：在 dataframe 中找到 open_date_utc 附近的 K 线 + entry_candle_low = None + entry_candle_high = None + + # 通过 potential_entry_low/high 列找到入场信号 K 线 + # 找到最先出现信号且在 open_date_utc 之前的 K 线 + entry_mask = ( + (dataframe["potential_entry_low"].notna()) + | (dataframe["potential_entry_high"].notna()) + ) + entry_candidates = dataframe[ + entry_mask + & (dataframe["date"] <= trade.open_date_utc + timedelta(hours=1)) + & (dataframe["date"] >= trade.open_date_utc - timedelta(hours=1)) + ] + if len(entry_candidates) > 0: + entry_candle = entry_candidates.iloc[-1] + entry_candle_low = entry_candle.get("potential_entry_low") + entry_candle_high = entry_candle.get("potential_entry_high") + + # ── 阶段一：用 Entry Candle 止损 ── + if entry_candle_low is not None or entry_candle_high is not None: + if trade.is_short: + if entry_candle_high is not None and not np.isnan(entry_candle_high): + sl_price = float(entry_candle_high) * (1 + buffer) + sl_ratio = (sl_price - current_rate) / current_rate + # 如果已经有盈利超过阈值，切换到结构止损 + if current_profit > self.profit_to_structure_sl_pct.value: + pass # 继续到阶段二 + else: + return max(sl_ratio, -0.25) + else: + if entry_candle_low is not None and not np.isnan(entry_candle_low): + sl_price = float(entry_candle_low) * (1 - buffer) + sl_ratio = (sl_price - current_rate) / current_rate + if current_profit > self.profit_to_structure_sl_pct.value: + pass # 继续到阶段二 + else: + return max(sl_ratio, -0.25) + + # ── 阶段二：结构跟踪止损（盈利足够后） ── + profit_trigger = self.profit_to_structure_sl_pct.value + if current_profit > profit_trigger: + if trade.is_short: + resistance = last.get("resistance_4h") + if resistance is not None and not (isinstance(resistance, float) and np.isnan(resistance)): + sl_price = float(resistance) * (1 + buffer) + sl_ratio = (sl_price - current_rate) / current_rate + if sl_ratio < 0: + return max(sl_ratio, -0.25) + else: + support = last.get("support_4h") + if support is not None and not (isinstance(support, float) and np.isnan(support)): + sl_price = float(support) * (1 - buffer) + sl_ratio = (sl_price - current_rate) / current_rate + if sl_ratio < 0: + return max(sl_ratio, -0.25) return None + + # ================================================================ + # 时间止损的替代实现（通过 populate_exit_trend 扩展） + # ================================================================ + + def confirm_trade_exit( + self, + pair: str, + trade: Trade, + order_type: str, + amount: float, + rate: float, + time_in_force: str, + sell_reason: str, + **kwargs, + ) -> bool: + """ + 可在此处添加日志记录，便于回测分析。 + """ + return True