beast/beast-trader-strategies
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1.3: 精简过滤器 + 结构突破确认优化

Browse Source
This commit is contained in:
Beast Trader
2026-06-07 23:40:00 +08:00
parent ffbb646a1f
commit e59aa09c49
1 changed files with 210 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files

560
strategy.py
View File

@ -1,148 +1,102 @@
# ============================================================================ """
# Structure Flow Strategy v1.2 Structure Flow Strategy v1.3
# 纯价格结构策略 — 零技术指标，价格行为学驱动 =======================
# 变更记录：
# 版本变化 v1.1 → v1.2: v1.0 (2026-06-07): 纯价格结构策略，D1定方向→4H定位→1H入场，支撑/阻力来自Swing Point
# - 硬止损改为 Entry Candle 失效点（做多→入场K线低点，做空→入场K线高点） v1.1 (2026-06-07): 修复freqtrade 2026.2 Binance futures bug(use_order_book:true)，
# - 新增时间止损：入场后 N 根K线内无盈利则主动出场 硬止损改为结构失效点，首次futures回测成功(+61.52%)
# - 保留 trailing_stop（结构跟踪止损），盈利后切换 v1.2 (2026-06-07): 尝试Entry Candle止损（入场K线低点/高点），增加时间止损
# - 策略类重命名为 StructureFlowStrategyV12 结果：50笔硬止损全亏，Entry Candle查找有bug，return None退回到25%宽止损
# v1.3 (2026-06-07): ===== 重写 custom_stoploss =====
# 设计哲学: 弃用脆弱的Entry Candle查找，改用ATR动态止损：
# 趋势由 HH/HL 定义，支撑阻力由 Swing Point 定义， - 初始止损：入场价 ± 1.0 ATR（紧，快速认输）
# 止损由 Entry Candle 失效点定义，出场由结构反转定义。 - 盈利>1%：移动止损至保本线
# - 盈利>2%：ATR追踪止损，锁定利润
# 多时间框架: - 硬止损安全网：-5%（stoploss属性）
# D1 → 宏观结构方向 核心哲学：预估错误的交易，早早认输止损离场，不硬扛单
# 4H → 中期结构位 + 入场区域判定 """
# 1H → K线形态确认入场时机
# ============================================================================
from datetime import datetime, timedelta from datetime import datetime
import numpy as np import numpy as np
import pandas as pd import pandas as pd
from pandas import DataFrame from pandas import DataFrame
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, informative from freqtrade.strategy import IStrategy, IntParameter, informative
from freqtrade.persistence import Trade from freqtrade.persistence import Trade
class StructureFlowStrategyV12(IStrategy): class StructureFlowStrategyV13(IStrategy):
""" """
Structure Flow Strategy v1.2 — 纯价格结构策略 Structure Flow Strategy v1.3
不使用任何技术指标（无 EMA、ATR、RSI、MACD、布林带等）。 核心逻辑：
一切信号来源于价格本身的 OHLC 数据和由此推导的结构信息。 D1 定宏观方向（HH/HL 上升，LH/LL 下降）
↓
4H 定位结构位（Swing Point → 支撑/阻力区域）
↓
1H 找入场时机（K线形态 + 在结构区域内）
趋势判断: 止损逻辑（v1.3重写）：
HH + HL → 上升趋势（Bullish Structure） - 初始止损：入场价 ± 1.0 ATR（紧，快速认输）
LH + LL → 下降趋势（Bearish Structure） - 盈利 > 1%：移动止损至保本（open_rate ± 0.1%）
- 盈利 > 2%：ATR追踪止损（current_rate ∓ 1.0 ATR）
入场逻辑: - 硬止损安全网：-5%（防止极端行情）
做多: D1上升结构 + 价格在4H Swing区间下半区 + 1H看涨K线形态
做空: D1下降结构 + 价格在4H Swing区间上半区 + 1H看跌K线形态
止损逻辑（v1.2 核心改进）:
初始止损: Entry Candle 失效点（做多→入场K线最低价，做空→入场K线最高价）
动态止损: 盈利后切换为结构跟踪止损（custom_stoploss）
时间止损: 入场后 N 根K线内无盈利则主动出场
""" """
# ── 基础配置 ────────────────────────────────────────── # =====================
# 基础属性
# =====================
timeframe = "1h"
can_short = True can_short = True
stoploss = -0.05 # 硬止损 5%，实际由 custom_stoploss 动态管理 stoploss = -0.05 # 硬止损安全网 5%，实际由 custom_stoploss 动态管理
use_custom_stoploss = True use_custom_stoploss = True
minimal_roi = {"0": 100} # 不设时间止盈，出场由结构决定 minimal_roi = {"0": 100} # 不设时间止盈，靠移动止损出场
max_open_trades = 1 max_open_trades = 1
timeframe = "1h"
# 回测参数 # =====================
startup_candle_count = 40 # 可优化参数
# =====================
# ── 可调参数 ────────────────────────────────────────── swing_lookback_d1 = IntParameter(8, 14, default=10, space="buy")
swing_lookback_h4 = IntParameter(5, 10, default=8, space="buy")
pin_bar_wick_ratio = IntParameter(50, 70, default=60, space="buy")
swing_lookback_d1 = IntParameter( # =====================
2, 10, default=5, space="buy", # 工具：Swing Point 检测
) # =====================
swing_lookback_h4 = IntParameter(
2, 10, default=5, space="buy",
)
# Pin Bar 确认强度：影线至少是实体的 N 倍
pin_bar_wick_ratio = DecimalParameter(
1.5, 4.0, default=2.0, space="buy",
)
# 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,
)
# ================================================================
# 工具函数 — 纯价格计算，不依赖任何技术指标
# ================================================================
@staticmethod @staticmethod
def _detect_swing_points( def _detect_swing_points(
high: pd.Series, high: pd.Series,
low: pd.Series, low: pd.Series,
lookback: int, window: int = 5,
) -> tuple[pd.Series, pd.Series]: ) -> tuple[pd.Series, pd.Series]:
""" """检测 Swing High / Swing Low。"""
检测 Swing High 和 Swing Low。
纯价格比较：
- Swing High: 当前高点 > 左右各 lookback 根K线的所有高点
- Swing Low: 当前低点 < 左右各 lookback 根K线的所有低点
"""
n = len(high) n = len(high)
is_swing_high = np.full(n, False) sh = pd.Series(np.nan, index=high.index, dtype=float)
is_swing_low = np.full(n, False) sl = pd.Series(np.nan, index=low.index, dtype=float)
for i in range(lookback, n - lookback): for i in range(window, n - window):
window_high = high.iloc[i - lookback : i + lookback + 1] if high.iloc[i] > high.iloc[i - window : i].max() and high.iloc[i] > high.iloc[i + 1 : i + window + 1].max():
window_low = low.iloc[i - lookback : i + lookback + 1] 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]
if high.iloc[i] == window_high.max(): return sh, sl
is_swing_high[i] = True
if low.iloc[i] == window_low.min():
is_swing_low[i] = True
return ( # =====================
pd.Series(is_swing_high, index=high.index), # 工具：结构分析
pd.Series(is_swing_low, index=low.index), # =====================
)
@staticmethod
def _build_structure( def _build_structure(
self,
high: pd.Series, high: pd.Series,
low: pd.Series, low: pd.Series,
close: pd.Series, close: pd.Series,
swing_high: pd.Series, swing_high: pd.Series,
swing_low: pd.Series, swing_low: pd.Series,
) -> DataFrame: ) -> DataFrame:
""" """从 Swing Points 构建市场结构信息。"""
从 Swing Points 构建市场结构信息。
返回值包含:
trend_up / trend_down：当前处于上升/下降结构
support：最近 Swing Low 价格
resistance：最近 Swing High 价格
in_demand：价格在下半区（做多区域）
in_supply：价格在上半区（做空区域）
"""
n = len(high) n = len(high)
trend_up_arr = np.full(n, False) trend_up_arr = np.full(n, False)
@ -152,11 +106,10 @@ class StructureFlowStrategyV12(IStrategy):
in_demand_zone = np.full(n, False) in_demand_zone = np.full(n, False)
in_supply_zone = np.full(n, False) in_supply_zone = np.full(n, False)
sh_prices: list[float] = [] sh_prices = []
sl_prices: list[float] = [] sl_prices = []
for i in range(n): for i in range(n):
# ── 更新 Swing Point 队列 ──
if swing_high.iloc[i] and not np.isnan(high.iloc[i]): if swing_high.iloc[i] and not np.isnan(high.iloc[i]):
sh_prices.append(high.iloc[i]) sh_prices.append(high.iloc[i])
if len(sh_prices) > 4: if len(sh_prices) > 4:
@ -167,117 +120,95 @@ class StructureFlowStrategyV12(IStrategy):
if len(sl_prices) > 4: if len(sl_prices) > 4:
sl_prices.pop(0) sl_prices.pop(0)
# ── 趋势判断 ── # 趋势判断
if len(sh_prices) >= 2 and len(sl_prices) >= 2: if len(sh_prices) >= 2 and len(sl_prices) >= 2:
latest_sh, prev_sh = sh_prices[-1], sh_prices[-2] latest_sh, prev_sh = sh_prices[-1], sh_prices[-2]
latest_sl, prev_sl = sl_prices[-1], sl_prices[-2] latest_sl, prev_sl = sl_prices[-1], sl_prices[-2]
if latest_sh > prev_sh and latest_sl > prev_sl: if latest_sh > prev_sh and latest_sl > prev_sl:
trend_up_arr[i] = True trend_up_arr[i] = True
trend_down_arr[i] = False
elif latest_sh < prev_sh and latest_sl < prev_sl: elif latest_sh < prev_sh and latest_sl < prev_sl:
trend_up_arr[i] = False
trend_down_arr[i] = True trend_down_arr[i] = True
else: else:
if i > 0: # 沿用上一根K线的状态
trend_up_arr[i] = trend_up_arr[i - 1] trend_up_arr[i] = trend_up_arr[i - 1] if i > 0 else False
trend_down_arr[i] = trend_down_arr[i - 1] trend_down_arr[i] = trend_down_arr[i - 1] if i > 0 else False
elif i > 0: elif i > 0:
trend_up_arr[i] = trend_up_arr[i - 1] trend_up_arr[i] = trend_up_arr[i - 1]
trend_down_arr[i] = trend_down_arr[i - 1] trend_down_arr[i] = trend_down_arr[i - 1]
# ── 最近支撑/阻力 ── # 支撑/阻力
if sl_prices: if sl_prices:
nearest_support[i] = sl_prices[-1] nearest_support[i] = sl_prices[-1]
elif i > 0:
nearest_support[i] = nearest_support[i - 1]
if sh_prices: if sh_prices:
nearest_resistance[i] = sh_prices[-1] nearest_resistance[i] = sh_prices[-1]
elif i > 0:
nearest_resistance[i] = nearest_resistance[i - 1]
# ── 入场区域：用 Swing 区间中点划分 ── # 需求/供给区域
if ( c = close.iloc[i]
not np.isnan(nearest_support[i]) if not np.isnan(nearest_support[i]) and not np.isnan(nearest_resistance[i]):
and not np.isnan(nearest_resistance[i]) zone_range = nearest_resistance[i] - nearest_support[i]
and nearest_resistance[i] > nearest_support[i] if zone_range > 0:
): pos_pct = (c - nearest_support[i]) / zone_range
mid = (nearest_support[i] + nearest_resistance[i]) / 2.0 in_demand_zone[i] = pos_pct < 0.35
in_demand_zone[i] = low.iloc[i] <= mid in_supply_zone[i] = pos_pct > 0.65
in_supply_zone[i] = high.iloc[i] >= mid
elif i > 0:
in_demand_zone[i] = in_demand_zone[i - 1]
in_supply_zone[i] = in_supply_zone[i - 1]
result = DataFrame( return DataFrame({
{ "trend_up": trend_up_arr,
"trend_up": trend_up_arr, "trend_down": trend_down_arr,
"trend_down": trend_down_arr, "support": nearest_support,
"support": nearest_support, "resistance": nearest_resistance,
"resistance": nearest_resistance, "in_demand": in_demand_zone,
"in_demand": in_demand_zone, "in_supply": in_supply_zone,
"in_supply": in_supply_zone, }, index=high.index)
},
index=high.index, # =====================
) # 工具：K线形态检测
return result # =====================
@staticmethod @staticmethod
def _detect_candle_patterns( def _detect_candle_patterns(
o: pd.Series, open_: pd.Series,
h: pd.Series, high: pd.Series,
l: pd.Series, low: pd.Series,
c: pd.Series, close: pd.Series,
pin_ratio: float, pin_bar_wick_ratio: float = 0.6,
) -> tuple[pd.Series, pd.Series, pd.Series, pd.Series]: ) -> tuple[pd.Series, pd.Series, pd.Series, pd.Series]:
""" """检测 Pin Bar 和 Engulfing 形态。"""
检测 K 线形态 — 纯 OHLC 计算。 body = (close - open_).abs()
""" total_range = (high - low).replace(0, 0.0001)
body = abs(c - o)
upper_wick = h - np.maximum(o, c)
lower_wick = np.minimum(o, c) - l
total_range = h - l
valid_range = total_range > 0 upper_wick = high - close.where(close > open_, open_)
valid_body = body > 0 lower_wick = open_.where(close > open_, close) - low
is_pin = (upper_wick + lower_wick) / total_range > pin_bar_wick_ratio
bullish_pin = ( bullish_pin = is_pin & (close > open_) & (lower_wick > upper_wick)
valid_range bearish_pin = is_pin & (close < open_) & (upper_wick > lower_wick)
& valid_body
& (lower_wick >= pin_ratio * body)
& (upper_wick <= 0.5 * body)
)
bearish_pin = ( prev_open = open_.shift(1)
valid_range prev_close = close.shift(1)
& valid_body bullish_engulf = (close > prev_open) & (open_ < prev_close) & (close > open_)
& (upper_wick >= pin_ratio * body) bearish_engulf = (close < prev_open) & (open_ > prev_close) & (close < open_)
& (lower_wick <= 0.5 * body)
)
prev_body = body.shift(1) return bullish_pin, bearish_pin, bullish_engulf, bearish_engulf
prev_o = o.shift(1)
prev_c = c.shift(1)
bullish_engulf = ( # =====================
(c > o) # 工具：ATR 计算
& (prev_c < prev_o) # =====================
& (body > prev_body)
)
bearish_engulf = ( @staticmethod
(c < o) def _calc_atr(
& (prev_c > prev_o) high: pd.Series,
& (body > prev_body) low: pd.Series,
) close: pd.Series,
period: int = 14,
return ( ) -> pd.Series:
pd.Series(bullish_pin, index=c.index), """计算 ATR。"""
pd.Series(bearish_pin, index=c.index), prev_close = close.shift(1)
pd.Series(bullish_engulf, index=c.index), tr = pd.concat(
pd.Series(bearish_engulf, index=c.index), [high - low, (high - prev_close).abs(), (low - prev_close).abs()],
) axis=1,
).max(axis=1)
return tr.rolling(period).mean()
# ================================================================ # ================================================================
# 信息时间框架 — D1 宏观结构 # 信息时间框架 — D1 宏观结构
@ -291,15 +222,12 @@ class StructureFlowStrategyV12(IStrategy):
dataframe["high"], dataframe["low"], dataframe["high"], dataframe["low"],
self.swing_lookback_d1.value, self.swing_lookback_d1.value,
) )
structure = self._build_structure( structure = self._build_structure(
dataframe["high"], dataframe["low"], dataframe["close"], dataframe["high"], dataframe["low"], dataframe["close"],
sh, sl, sh, sl,
) )
dataframe["trend_up"] = structure["trend_up"] dataframe["trend_up"] = structure["trend_up"]
dataframe["trend_down"] = structure["trend_down"] dataframe["trend_down"] = structure["trend_down"]
return dataframe return dataframe
# ================================================================ # ================================================================
@ -314,12 +242,10 @@ class StructureFlowStrategyV12(IStrategy):
dataframe["high"], dataframe["low"], dataframe["high"], dataframe["low"],
self.swing_lookback_h4.value, self.swing_lookback_h4.value,
) )
structure = self._build_structure( structure = self._build_structure(
dataframe["high"], dataframe["low"], dataframe["close"], dataframe["high"], dataframe["low"], dataframe["close"],
sh, sl, sh, sl,
) )
dataframe["trend_up"] = structure["trend_up"] dataframe["trend_up"] = structure["trend_up"]
dataframe["trend_down"] = structure["trend_down"] dataframe["trend_down"] = structure["trend_down"]
dataframe["support"] = structure["support"] dataframe["support"] = structure["support"]
@ -327,63 +253,59 @@ class StructureFlowStrategyV12(IStrategy):
dataframe["in_demand"] = structure["in_demand"] dataframe["in_demand"] = structure["in_demand"]
dataframe["in_supply"] = structure["in_supply"] dataframe["in_supply"] = structure["in_supply"]
# 4H ATR（保留，可能用于未来优化）
dataframe["atr"] = self._calc_atr(
dataframe["high"], dataframe["low"], dataframe["close"], period=14
)
return dataframe return dataframe
# ================================================================ # ================================================================
# 主时间框架 — 1H K线形态 + Entry Candle 记录 # 主时间框架 — 1H 指标
# ================================================================ # ================================================================
# 类级别缓存：记录每笔交易的 Entry Candle 信息
# {trade_id: {"entry_low": float, "entry_high": float, "entry_idx": int}}
_entry_candle_cache = {}
def populate_indicators( def populate_indicators(
self, dataframe: DataFrame, metadata: dict self, dataframe: DataFrame, metadata: dict
) -> DataFrame: ) -> DataFrame:
""" """1H 级别：K线形态 + ATR。"""
1H 一小时线：检测 K 线形态。
同时预标记可能的入场 K 线（供 custom_stoploss 使用）。
"""
bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = ( bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = (
self._detect_candle_patterns( self._detect_candle_patterns(
dataframe["open"], dataframe["open"],
dataframe["high"], dataframe["high"],
dataframe["low"], dataframe["low"],
dataframe["close"], dataframe["close"],
self.pin_bar_wick_ratio.value, self.pin_bar_wick_ratio.value / 100.0,
) )
) )
dataframe["bullish_pinbar"] = bullish_pin dataframe["bullish_pinbar"] = bullish_pin
dataframe["bearish_pinbar"] = bearish_pin dataframe["bearish_pinbar"] = bearish_pin
dataframe["bullish_engulfing"] = bullish_engulf dataframe["bullish_engulfing"] = bullish_engulf
dataframe["bearish_engulfing"] = bearish_engulf dataframe["bearish_engulfing"] = bearish_engulf
dataframe["bullish_signal"] = bullish_pin | bullish_engulf dataframe["bullish_signal"] = bullish_pin | bullish_engulf
dataframe["bearish_signal"] = bearish_pin | bearish_engulf dataframe["bearish_signal"] = bearish_pin | bearish_engulf
# 预标记：如果这根 K 线是入场信号，记录其 OHLC（供后续 custom_stoploss 使用） # 1H ATR（用于动态止损）
# 注意：这里只是标记，实际入场由 populate_entry_trend 决定 dataframe["atr_1h"] = self._calc_atr(
dataframe["potential_entry_low"] = np.where( dataframe["high"], dataframe["low"], dataframe["close"], period=14
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,
) )
# NaN 安全处理
bool_cols = [
"trend_up_1d", "trend_down_1d",
"trend_up_4h", "trend_down_4h",
"in_demand_4h", "in_supply_4h",
"bullish_signal", "bearish_signal",
]
for col in bool_cols:
if col in dataframe.columns:
dataframe[col] = dataframe[col].fillna(False)
return dataframe return dataframe
# ================================================================ # =====================
# 入场信号 # 入场信号
# ================================================================ # =====================
def populate_entry_trend( def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
self, dataframe: DataFrame, metadata: dict
) -> DataFrame:
""" """
入场逻辑（1H 时间框架）。 入场逻辑（1H 时间框架）。
@ -397,8 +319,7 @@ class StructureFlowStrategyV12(IStrategy):
2. 4H 上半区 / 供给区域（in_supply_4h） 2. 4H 上半区 / 供给区域（in_supply_4h）
3. 1H 看跌 K 线形态（bearish_signal） 3. 1H 看跌 K 线形态（bearish_signal）
""" """
# NaN 安全处理
# ── NaN 安全处理 ──
bool_cols = [ bool_cols = [
"trend_up_1d", "trend_down_1d", "trend_up_1d", "trend_down_1d",
"trend_up_4h", "trend_down_4h", "trend_up_4h", "trend_down_4h",
@ -407,9 +328,9 @@ class StructureFlowStrategyV12(IStrategy):
] ]
for col in bool_cols: for col in bool_cols:
if col in dataframe.columns: if col in dataframe.columns:
dataframe[col] = dataframe[col].fillna(False).infer_objects(copy=False) dataframe[col] = dataframe[col].fillna(False)
# ── 做多 ── # 做多
long_conditions = ( long_conditions = (
dataframe["trend_up_1d"] dataframe["trend_up_1d"]
& dataframe["in_demand_4h"] & dataframe["in_demand_4h"]
@ -417,7 +338,7 @@ class StructureFlowStrategyV12(IStrategy):
) )
dataframe.loc[long_conditions, "enter_long"] = 1 dataframe.loc[long_conditions, "enter_long"] = 1
# ── 做空 ── # 做空
short_conditions = ( short_conditions = (
dataframe["trend_down_1d"] dataframe["trend_down_1d"]
& dataframe["in_supply_4h"] & dataframe["in_supply_4h"]
@ -427,34 +348,23 @@ class StructureFlowStrategyV12(IStrategy):
return dataframe return dataframe
# ================================================================ # =====================
# 出场信号 # 出场信号
# ================================================================ # =====================
def populate_exit_trend( def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
self, dataframe: DataFrame, metadata: dict """出场逻辑 — 由结构反转触发。"""
) -> DataFrame: exit_long = ~dataframe["trend_up_1d"].fillna(True)
"""
出场逻辑 — 由结构反转触发。
"""
# 做多出场：D1 不再上升
exit_long = (
~dataframe["trend_up_1d"].fillna(True)
)
dataframe.loc[exit_long, "exit_long"] = 1 dataframe.loc[exit_long, "exit_long"] = 1
# 做空出场：D1 不再下降 exit_short = dataframe["trend_up_1d"].fillna(False)
exit_short = (
dataframe["trend_up_1d"].fillna(False)
)
dataframe.loc[exit_short, "exit_short"] = 1 dataframe.loc[exit_short, "exit_short"] = 1
return dataframe return dataframe
# ================================================================ # =====================
# 动态止损 — v1.2 核心改进 # 动态止损 — v1.3 重写
# ================================================================ # =====================
def custom_stoploss( def custom_stoploss(
self, self,
@ -465,115 +375,65 @@ class StructureFlowStrategyV12(IStrategy):
current_profit: float, current_profit: float,
after_fill: bool, after_fill: bool,
**kwargs, **kwargs,
) -> float | None: ) -> float:
""" """
v1.2 止损逻辑（核心改进）： v1.3 止损逻辑（完全重写）：
阶段一（刚入场，无盈利或微盈利）： 核心哲学：「预估错误的交易，早早认输止损离场，而不要硬扛单」
止损 = 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)
时间止损： 阶段一（无盈利或微盈利 < 1%）：
入场后超过 time_stop_bars 根K线且 current_profit < 0， 止损 = 入场价 ± 1.0 ATR
返回 -0.01（立即市价出场）。 → 距离近，价格稍有不利变动就止损，快速认输
阶段二（盈利 1% ~ 2%）：
止损移动至保本线（open_rate ± 0.1%）
→ 这笔交易已经不亏了，卸下心理负担
阶段三（盈利 > 2%）：
追踪止损 = current_rate ∓ 1.0 ATR
→ 价格回调超过1ATR才出场，给趋势足够的呼吸空间
参数说明：
- ATR 来自当前1H K线的 atr_1h 值
- 如果 ATR 为 NaN，fallback 到 2% 固定止损
- 最终返回的止损比率不会超过 -5%（硬止损安全网）
""" """
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or len(dataframe) == 0: if dataframe is None or len(dataframe) == 0:
return None return -0.02 if not trade.is_short else 0.02
last = dataframe.iloc[-1] last_candle = dataframe.iloc[-1]
buffer = self.entry_sl_buffer.value atr = last_candle.get("atr_1h", np.nan)
# ── 时间止损检查 ── if pd.isna(atr) or atr <= 0:
# 计算入场至今的K线数（1H = 1根/小时） atr = current_rate * 0.02
bars_held = (current_time - trade.open_date_utc).total_seconds() / 3600 else:
if bars_held >= self.time_stop_bars.value and current_profit <= 0: atr = float(atr)
# 超时且无盈利，立即出场（返回当前价，即市价出场）
return -0.01 # 1% 内市价出场
# ── 尝试获取 Entry Candle 信息 ── open_rate = trade.open_rate
# 方法：在 dataframe 中找到 open_date_utc 附近的 K 线
entry_candle_low = None
entry_candle_high = None
# 通过 potential_entry_low/high 列找到入场信号 K 线 if not trade.is_short:
# 找到最先出现信号且在 open_date_utc 之前的 K 线 # ── 做多 ──
entry_mask = ( if current_profit <= 0.01:
(dataframe["potential_entry_low"].notna()) sl_price = open_rate - atr * 1.0
| (dataframe["potential_entry_high"].notna()) elif current_profit <= 0.02:
) sl_price = open_rate * 0.999
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: else:
if entry_candle_low is not None and not np.isnan(entry_candle_low): sl_price = current_rate - atr * 1.0
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)
# ── 阶段二：结构跟踪止损（盈利足够后） ── sl_ratio = (sl_price / current_rate) - 1.0
profit_trigger = self.profit_to_structure_sl_pct.value return max(sl_ratio, -0.05)
if current_profit > profit_trigger:
if trade.is_short: else:
resistance = last.get("resistance_4h") # ── 做空 ──
if resistance is not None and not (isinstance(resistance, float) and np.isnan(resistance)): if current_profit <= 0.01:
sl_price = float(resistance) * (1 + buffer) sl_price = open_rate + atr * 1.0
sl_ratio = (sl_price - current_rate) / current_rate elif current_profit <= 0.02:
if sl_ratio < 0: sl_price = open_rate * 1.001
return max(sl_ratio, -0.25)
else: else:
support = last.get("support_4h") sl_price = current_rate + atr * 1.0
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 sl_ratio = 1.0 - (sl_price / current_rate)
return min(sl_ratio, 0.05)
# ================================================================
# 时间止损的替代实现（通过 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