v1.8 (Scalp): 反向S/R剥头皮 - 全线失败/0%胜率
This commit is contained in:
621
strategy.py
621
strategy.py
@ -1,49 +1,97 @@
|
||||
"""
|
||||
Structure Flow Swing Strategy v4.2
|
||||
==================================
|
||||
纯价格行为学震荡策略 — 借鉴 v2.2b 的 swing point + K线形态框架
|
||||
Structure Flow Scalp — 震荡市剥头皮策略
|
||||
==========================================
|
||||
基于Al Brooks价格行为学：
|
||||
- 在已识别的震荡区间内，支撑位做多、阻力位做空
|
||||
- 15m级别支撑/阻力决定交易区间，5m级别入场
|
||||
- 100x全仓杠杆，每次10%仓位
|
||||
- 区间高度40%止盈，15m支撑/阻力外侧0.3%止损
|
||||
|
||||
核心逻辑（模拟手工交易）：
|
||||
1. 用 swing point 识别近期高低点，自动形成交易区间
|
||||
2. 价格到支撑 + K线止跌形态（bullish pinbar/engulfing）→ 做多
|
||||
3. 价格到阻力 + K线滞涨 → 平多（反向开空同理）
|
||||
4. 不在震荡判定上设严苛门槛，价格够到边界+形态确认就做
|
||||
5. 趋势突破时，突破边界导致亏损，但可控
|
||||
|
||||
v4.1 系列教训（2026-06-10）：
|
||||
用滚动window + 碰壁验证 + ATR比例止损 → 全是技术指标思维，不是价格行为学
|
||||
唯一正收益的是最简单的版本（1H + 形态 + 固定-3%止损）
|
||||
变更记录：
|
||||
v1 (2026-06-10): 初版，基于v2.2b核心逻辑重构
|
||||
v1.1 (2026-06-10): 支撑阻力从4H改为15m
|
||||
v1.2 (2026-06-10): 去掉4H趋势强度判断（冗余）；启用100x全仓杠杆，10%仓位
|
||||
v1.3 (2026-06-10): 代码审查修复——移除populate_exit_trend死循环，NaN安全，杠杆上限
|
||||
v1.4 (2026-06-10): EMA动态S/R + 入场锁定S/R——止损止盈使用入场时的锁定值，不追最新
|
||||
v1.5 (2026-06-10): 扩展入场信号 + 追踪止损保护 + 延长活S/R窗口
|
||||
v1.6 (2026-06-10): 止损改为ATR动态计算——绑入场价，不绑支撑位；追踪改为ATR×0.5自适应
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from pandas import DataFrame
|
||||
from freqtrade.strategy import IStrategy, IntParameter
|
||||
from freqtrade.strategy import IStrategy, IntParameter, informative
|
||||
from freqtrade.persistence import Trade
|
||||
|
||||
|
||||
class StructureFlowSwingV42(IStrategy):
|
||||
class StructureFlowScalp(IStrategy):
|
||||
"""
|
||||
震荡市剥头皮策略 — 5m框架，100x全仓杠杆。
|
||||
去掉4H趋势强度判断——15m支撑阻力本身就是最好的过滤器。
|
||||
"""
|
||||
|
||||
can_short = True
|
||||
stoploss = -0.20
|
||||
stoploss = -0.15
|
||||
use_custom_stoploss = True
|
||||
use_custom_exit = True
|
||||
minimal_roi = {"0": 100}
|
||||
max_open_trades = 1
|
||||
timeframe = "1h"
|
||||
timeframe = "5m"
|
||||
|
||||
# ── 价格行为学参数 ──
|
||||
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%）
|
||||
# =====================
|
||||
# 杠杆设置 - 全仓 100x
|
||||
# =====================
|
||||
|
||||
# 固定参数
|
||||
cooldown = 6 # 冷却6根1H（6小时）
|
||||
def leverage(self, pair: str, current_time: datetime, current_rate: float,
|
||||
proposed_leverage: float, max_leverage: float, side: str,
|
||||
**kwargs) -> float:
|
||||
"""返回固定 100x 杠杆，不超过交易所允许的最大值"""
|
||||
return min(100.0, max_leverage)
|
||||
|
||||
# ================================================================
|
||||
# Swing Point 检测（v2.2b 同款）
|
||||
# ================================================================
|
||||
# =====================
|
||||
# 工具：查找入场K线（锁定S/R用）
|
||||
# =====================
|
||||
|
||||
def _get_entry_row(self, dataframe: DataFrame, trade: Trade) -> pd.Series | None:
|
||||
"""
|
||||
从 dataframe 中找到入场 trade 对应的 K 线行。
|
||||
兼容 live/dry_run（DatetimeIndex）和 backtesting（RangeIndex + date 列）两种模式。
|
||||
"""
|
||||
if 'date' in dataframe.columns:
|
||||
# Backtesting 模式：dataframe 有 date 列，index 是 int
|
||||
entry_mask = pd.to_datetime(dataframe['date']) <= trade.open_date
|
||||
if not entry_mask.any():
|
||||
return None
|
||||
return dataframe[entry_mask].iloc[-1]
|
||||
else:
|
||||
# Live/Dry-run 模式：index 是 DatetimeIndex
|
||||
try:
|
||||
entry_idx = dataframe.index.get_indexer([trade.open_date], method="pad")
|
||||
if entry_idx[0] < 0 or entry_idx[0] >= len(dataframe):
|
||||
return None
|
||||
return dataframe.iloc[entry_idx[0]]
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
# =====================
|
||||
# 可优化参数
|
||||
# =====================
|
||||
|
||||
# 15m支撑阻力计算窗口
|
||||
swing_lookback_15m = IntParameter(5, 15, default=10, space="buy")
|
||||
pin_bar_wick_ratio = IntParameter(50, 70, default=60, space="buy")
|
||||
cooldown_bars = IntParameter(2, 8, default=3, space="buy")
|
||||
|
||||
# 区间高度止盈比例（%）
|
||||
profit_zone_pct = IntParameter(20, 60, default=40, space="buy")
|
||||
|
||||
# =====================
|
||||
# 工具：Swing Point 检测
|
||||
# =====================
|
||||
|
||||
@staticmethod
|
||||
def _detect_swing_points(
|
||||
self,
|
||||
high: pd.Series,
|
||||
low: pd.Series,
|
||||
window: int = 5,
|
||||
@ -51,120 +99,335 @@ class StructureFlowSwingV42(IStrategy):
|
||||
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():
|
||||
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():
|
||||
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 同款）
|
||||
# ================================================================
|
||||
# =====================
|
||||
# 工具：结构分析
|
||||
# =====================
|
||||
|
||||
def _build_structure(
|
||||
self,
|
||||
high: pd.Series,
|
||||
low: pd.Series,
|
||||
close: pd.Series,
|
||||
swing_high: pd.Series,
|
||||
swing_low: pd.Series,
|
||||
) -> DataFrame:
|
||||
n = len(high)
|
||||
|
||||
trend_up_arr = np.full(n, False)
|
||||
trend_down_arr = np.full(n, False)
|
||||
nearest_support = np.full(n, np.nan)
|
||||
nearest_resistance = np.full(n, np.nan)
|
||||
|
||||
sh_prices = []
|
||||
sl_prices = []
|
||||
|
||||
for i in range(n):
|
||||
if pd.notna(swing_high.iloc[i]):
|
||||
sh_prices.append(swing_high.iloc[i])
|
||||
if len(sh_prices) > 4:
|
||||
sh_prices.pop(0)
|
||||
|
||||
if pd.notna(swing_low.iloc[i]):
|
||||
sl_prices.append(swing_low.iloc[i])
|
||||
if len(sl_prices) > 4:
|
||||
sl_prices.pop(0)
|
||||
|
||||
if len(sh_prices) >= 2 and len(sl_prices) >= 2:
|
||||
if sh_prices[-1] > sh_prices[-2] and sl_prices[-1] > sl_prices[-2]:
|
||||
trend_up_arr[i] = True
|
||||
elif sh_prices[-1] < sh_prices[-2] and sl_prices[-1] < sl_prices[-2]:
|
||||
trend_down_arr[i] = True
|
||||
elif i > 0:
|
||||
trend_up_arr[i] = trend_up_arr[i - 1]
|
||||
trend_down_arr[i] = trend_down_arr[i - 1]
|
||||
elif i > 0:
|
||||
trend_up_arr[i] = trend_up_arr[i - 1]
|
||||
trend_down_arr[i] = trend_down_arr[i - 1]
|
||||
|
||||
if sl_prices:
|
||||
# EMA平滑：不取最后一个，而是对最近swing lows做指数加权
|
||||
# alpha=0.3，每个新swing point向它移动30%，有"惯性"不跳变
|
||||
ema_s = sl_prices[0]
|
||||
for p in sl_prices[1:]:
|
||||
ema_s = 0.3 * p + 0.7 * ema_s
|
||||
nearest_support[i] = ema_s
|
||||
if sh_prices:
|
||||
ema_r = sh_prices[0]
|
||||
for p in sh_prices[1:]:
|
||||
ema_r = 0.3 * p + 0.7 * ema_r
|
||||
nearest_resistance[i] = ema_r
|
||||
|
||||
return DataFrame({
|
||||
"trend_up": trend_up_arr,
|
||||
"trend_down": trend_down_arr,
|
||||
"support": nearest_support,
|
||||
"resistance": nearest_resistance,
|
||||
}, index=high.index)
|
||||
|
||||
# =====================
|
||||
# 工具：K线形态检测
|
||||
# =====================
|
||||
|
||||
@staticmethod
|
||||
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
|
||||
def _detect_candle_patterns(
|
||||
open_: pd.Series,
|
||||
high: pd.Series,
|
||||
low: pd.Series,
|
||||
close: pd.Series,
|
||||
pin_bar_wick_ratio: float = 0.6,
|
||||
) -> tuple[pd.Series, pd.Series, pd.Series, pd.Series]:
|
||||
body = (close - open_).abs()
|
||||
total_range = (high - low).replace(0, 0.0001)
|
||||
|
||||
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
|
||||
upper_wick = high - close.where(close > open_, open_)
|
||||
lower_wick = open_.where(close > open_, close) - low
|
||||
is_pin = (upper_wick + lower_wick) / total_range > pin_bar_wick_ratio
|
||||
|
||||
@staticmethod
|
||||
def _detect_engulfing(open: pd.Series, close: pd.Series) -> tuple[pd.Series, pd.Series]:
|
||||
prev_open = open.shift(1)
|
||||
bullish_pin = is_pin & (close > open_) & (lower_wick > upper_wick)
|
||||
bearish_pin = is_pin & (close < open_) & (upper_wick > lower_wick)
|
||||
|
||||
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
|
||||
bullish_engulf = (close > prev_open) & (open_ < prev_close) & (close > open_)
|
||||
bearish_engulf = (close < prev_open) & (open_ > prev_close) & (close < open_)
|
||||
|
||||
return bullish_pin, bearish_pin, bullish_engulf, bearish_engulf
|
||||
|
||||
# ================================================================
|
||||
# 主指标
|
||||
# 信息时间框架 — 15m 短期支撑阻力（核心过滤器）
|
||||
# ================================================================
|
||||
|
||||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||||
sw = self.swing_window.value
|
||||
|
||||
# ── Swing Points ──
|
||||
sh, sl = self._detect_swing_points(dataframe["high"], dataframe["low"], sw)
|
||||
|
||||
# 向前填充最近的 swing high / low 作为动态 S/R
|
||||
dataframe["swing_high"] = sh.ffill()
|
||||
dataframe["swing_low"] = sl.ffill()
|
||||
|
||||
# ── 区间宽度（用于止盈参考） ──
|
||||
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,
|
||||
@informative("15m")
|
||||
def populate_indicators_15m(
|
||||
self, dataframe: DataFrame, metadata: dict
|
||||
) -> DataFrame:
|
||||
sh, sl = self._detect_swing_points(
|
||||
dataframe["high"], dataframe["low"],
|
||||
self.swing_lookback_15m.value,
|
||||
)
|
||||
|
||||
# ── K线形态 ──
|
||||
bull_pin, bear_pin = self._detect_pinbar(
|
||||
dataframe["open"], dataframe["high"], dataframe["low"], dataframe["close"]
|
||||
structure = self._build_structure(
|
||||
dataframe["high"], dataframe["low"], dataframe["close"],
|
||||
sh, sl,
|
||||
)
|
||||
bull_eng, bear_eng = self._detect_engulfing(dataframe["open"], dataframe["close"])
|
||||
dataframe["support"] = structure["support"]
|
||||
dataframe["resistance"] = structure["resistance"]
|
||||
|
||||
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_swing_high"] = np.where(
|
||||
dataframe["swing_high"].notna() & (dataframe["swing_high"] > 0),
|
||||
(dataframe["swing_high"] - dataframe["close"]) / dataframe["close"],
|
||||
np.nan,
|
||||
# ── 活支撑检查（15根15m ≈ 3.75小时，震荡市中支撑可长期有效）──
|
||||
touched_support = (
|
||||
(dataframe["low"] <= dataframe["support"] * 1.005) &
|
||||
(dataframe["low"] >= dataframe["support"] * 0.995)
|
||||
)
|
||||
held_support = dataframe["close"] > dataframe["support"]
|
||||
support_tested_and_held = touched_support & held_support
|
||||
dataframe["support_alive"] = support_tested_and_held.rolling(15, min_periods=1).max() > 0
|
||||
|
||||
for col in ["dist_to_swing_low", "dist_to_swing_high"]:
|
||||
dataframe[col] = dataframe[col].fillna(999)
|
||||
# ── 活阻力检查（15根窗口）──
|
||||
touched_resistance = (
|
||||
(dataframe["high"] >= dataframe["resistance"] * 0.995) &
|
||||
(dataframe["high"] <= dataframe["resistance"] * 1.005)
|
||||
)
|
||||
held_resistance = dataframe["close"] < dataframe["resistance"]
|
||||
resistance_tested_and_held = touched_resistance & held_resistance
|
||||
dataframe["resistance_alive"] = resistance_tested_and_held.rolling(15, min_periods=1).max() > 0
|
||||
|
||||
# 区间高度（用于止盈计算）
|
||||
dataframe["zone_height"] = (dataframe["resistance"] - dataframe["support"]).fillna(0)
|
||||
|
||||
return dataframe
|
||||
|
||||
# ================================================================
|
||||
# 入场
|
||||
# 主时间框架 — 5m 指标
|
||||
# ================================================================
|
||||
|
||||
def populate_indicators(
|
||||
self, dataframe: DataFrame, metadata: dict
|
||||
) -> DataFrame:
|
||||
"""5m级别：ATR + K线形态 + 信号整合。"""
|
||||
|
||||
# ── ATR(14) — 用于动态止损，根据市场波动自适应 ──
|
||||
high = dataframe["high"]
|
||||
low = dataframe["low"]
|
||||
close = dataframe["close"]
|
||||
prev_close = close.shift(1)
|
||||
tr = pd.concat([
|
||||
high - low,
|
||||
(high - prev_close).abs(),
|
||||
(low - prev_close).abs(),
|
||||
], axis=1).max(axis=1)
|
||||
dataframe["atr"] = tr.rolling(14).mean()
|
||||
|
||||
bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = (
|
||||
self._detect_candle_patterns(
|
||||
dataframe["open"],
|
||||
dataframe["high"],
|
||||
dataframe["low"],
|
||||
dataframe["close"],
|
||||
self.pin_bar_wick_ratio.value / 100.0,
|
||||
)
|
||||
)
|
||||
dataframe["bullish_pinbar"] = bullish_pin
|
||||
dataframe["bearish_pinbar"] = bearish_pin
|
||||
dataframe["bullish_engulfing"] = bullish_engulf
|
||||
dataframe["bearish_engulfing"] = bearish_engulf
|
||||
|
||||
# ── 扩展信号：长下影线（比pinbar更宽松，只要下影线>总范围50%） ──
|
||||
total_range = (dataframe["high"] - dataframe["low"]).replace(0, 0.0001)
|
||||
body = (dataframe["close"] - dataframe["open"]).abs()
|
||||
# 下影线 = min(open, close) - low
|
||||
lower_wick = (
|
||||
dataframe[["open", "close"]].min(axis=1) - dataframe["low"]
|
||||
)
|
||||
# 上影线 = high - max(open, close)
|
||||
upper_wick = (
|
||||
dataframe["high"] - dataframe[["open", "close"]].max(axis=1)
|
||||
)
|
||||
# 长下影线：下影线>总范围50% 且 下影线>上影线
|
||||
long_lower_wick = (
|
||||
(lower_wick / total_range > 0.5) &
|
||||
(lower_wick > upper_wick)
|
||||
)
|
||||
dataframe["long_lower_wick"] = long_lower_wick
|
||||
|
||||
# ── 扩展信号：支撑位附近的强力反弹阳线 ──
|
||||
# 条件：价格在支撑0.5%范围内 + 阳线 + 实体>0.2%
|
||||
if "support_15m" in dataframe.columns:
|
||||
near_support = (
|
||||
(dataframe["low"] <= dataframe["support_15m"] * 1.005) &
|
||||
(dataframe["low"] >= dataframe["support_15m"] * 0.995)
|
||||
)
|
||||
is_bullish = dataframe["close"] > dataframe["open"]
|
||||
body_pct = body / dataframe["open"]
|
||||
strong_recovery = near_support & is_bullish & (body_pct > 0.002)
|
||||
else:
|
||||
strong_recovery = pd.Series(False, index=dataframe.index)
|
||||
dataframe["strong_recovery"] = strong_recovery
|
||||
|
||||
# ── 综合止跌/止涨信号（扩展后） ──
|
||||
dataframe["bullish_signal"] = (
|
||||
bullish_pin | bullish_engulf | long_lower_wick | strong_recovery
|
||||
)
|
||||
dataframe["bearish_signal"] = (
|
||||
bearish_pin | bearish_engulf
|
||||
)
|
||||
# 做空对称：阻力位附近的强力下跌阴线
|
||||
if "resistance_15m" in dataframe.columns:
|
||||
near_resistance = (
|
||||
(dataframe["high"] >= dataframe["resistance_15m"] * 0.995) &
|
||||
(dataframe["high"] <= dataframe["resistance_15m"] * 1.005)
|
||||
)
|
||||
is_bearish = dataframe["close"] < dataframe["open"]
|
||||
body_pct = body / dataframe["open"]
|
||||
strong_rejection = near_resistance & is_bearish & (body_pct > 0.002)
|
||||
else:
|
||||
strong_rejection = pd.Series(False, index=dataframe.index)
|
||||
dataframe["strong_rejection"] = strong_rejection
|
||||
dataframe["bearish_signal"] = (
|
||||
bearish_pin | bearish_engulf | strong_rejection
|
||||
)
|
||||
|
||||
# NaN 安全处理
|
||||
bool_cols = [
|
||||
"support_alive_15m", "resistance_alive_15m",
|
||||
"bullish_signal", "bearish_signal",
|
||||
]
|
||||
for col in bool_cols:
|
||||
if col in dataframe.columns:
|
||||
dataframe[col] = dataframe[col].fillna(False)
|
||||
|
||||
# ATR fillna（前14根无ATR值用均值填补）
|
||||
if "atr" in dataframe.columns:
|
||||
atr_mean = dataframe["atr"].mean()
|
||||
dataframe["atr"] = dataframe["atr"].fillna(atr_mean)
|
||||
|
||||
return dataframe
|
||||
|
||||
# =====================
|
||||
# 入场信号
|
||||
# =====================
|
||||
|
||||
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||||
entry_zone = self.entry_zone_pct.value / 1000.0
|
||||
"""
|
||||
入场逻辑（5m 时间框架）。
|
||||
|
||||
# ── 做多：价格在 swing low 附近 + 止跌形态 ──
|
||||
long_conds = (
|
||||
(dataframe["dist_to_swing_low"] < entry_zone)
|
||||
& (dataframe["dist_to_swing_low"] > 0)
|
||||
不做4H趋势判断——15m支撑阻力本身就是过滤器：
|
||||
- 趋势强时价格直接突破15m S/R，不会在支撑/阻力附近停留
|
||||
- 在支撑/阻力附近停留 = 震荡市
|
||||
|
||||
入场条件（3个，去掉了冗余的4H趋势判断）：
|
||||
- 做多：价格贴近15m支撑 + 支撑有效 + K线止跌信号
|
||||
- 做空：价格贴近15m阻力 + 阻力有效 + K线止涨信号
|
||||
|
||||
出场只依赖 custom_stoploss 和 custom_exit，不需要 D1 结构反转退出。
|
||||
（去掉 populate_exit_trend：震荡市入场 → D1 非上升趋势 → 立即出场 的死循环）
|
||||
"""
|
||||
cooldown = self.cooldown_bars.value
|
||||
|
||||
# NaN 安全处理 — 如果 15m informative 列还没对齐，直接跳过本根 K 线
|
||||
required_cols = ["support_15m", "resistance_15m",
|
||||
"support_alive_15m", "resistance_alive_15m"]
|
||||
for col in required_cols:
|
||||
if col not in dataframe.columns:
|
||||
return dataframe # 数据尚未就绪，跳过
|
||||
|
||||
for col in ["bullish_signal", "bearish_signal",
|
||||
"support_alive_15m", "resistance_alive_15m"]:
|
||||
dataframe[col] = dataframe[col].fillna(False)
|
||||
|
||||
# ── 做多 ──
|
||||
# 条件：价格贴近15m支撑（0.5%范围内）- 使用 low 而非 open
|
||||
# 因为支撑测试看的是价格是否到达支撑位，不是开盘在哪
|
||||
near_support = (
|
||||
(dataframe["low"] <= dataframe["support_15m"] * 1.005) &
|
||||
(dataframe["low"] >= dataframe["support_15m"] * 0.995)
|
||||
)
|
||||
|
||||
long_conditions = (
|
||||
near_support
|
||||
& dataframe["support_alive_15m"]
|
||||
& 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["dist_to_swing_high"] < entry_zone)
|
||||
& (dataframe["dist_to_swing_high"] > 0)
|
||||
long_recent = long_conditions.rolling(cooldown, min_periods=1).max().shift(1) == 0
|
||||
dataframe.loc[long_conditions & long_recent, "enter_long"] = 1
|
||||
|
||||
# ── 做空 ──
|
||||
# 条件：价格贴近15m阻力（0.5%范围内）- 使用 high 而非 open
|
||||
near_resistance = (
|
||||
(dataframe["high"] >= dataframe["resistance_15m"] * 0.995) &
|
||||
(dataframe["high"] <= dataframe["resistance_15m"] * 1.005)
|
||||
)
|
||||
|
||||
short_conditions = (
|
||||
near_resistance
|
||||
& dataframe["resistance_alive_15m"]
|
||||
& 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
|
||||
|
||||
short_recent = short_conditions.rolling(cooldown, min_periods=1).max().shift(1) == 0
|
||||
dataframe.loc[short_conditions & short_recent, "enter_short"] = 1
|
||||
|
||||
return dataframe
|
||||
|
||||
# ================================================================
|
||||
# 出场
|
||||
# ================================================================
|
||||
# =====================
|
||||
# exit_trend（freqtrade 2025.11 要求必须实现，即使 use_custom_exit=True）
|
||||
# =====================
|
||||
|
||||
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||||
"""退出逻辑完全由 custom_stoploss + custom_exit 管理。"""
|
||||
return dataframe
|
||||
|
||||
# ================================================================
|
||||
# 止损：区间宽度 × 0.5（自适应）
|
||||
# ================================================================
|
||||
# =====================
|
||||
# 动态止损 — 入场价 - ATR×2.0（基于市场波动，非固定比例）
|
||||
# =====================
|
||||
|
||||
def custom_stoploss(
|
||||
self,
|
||||
@ -176,24 +439,88 @@ class StructureFlowSwingV42(IStrategy):
|
||||
after_fill: bool,
|
||||
**kwargs,
|
||||
) -> float:
|
||||
"""
|
||||
止损锚定入场价，宽度根据市场波动（ATR）动态计算，而非固定比例。
|
||||
|
||||
核心逻辑：
|
||||
- 做多止损 = entry_price - ATR_5m × 2.0
|
||||
- 做空止损 = entry_price + ATR_5m × 2.0
|
||||
- ATR值从入场时的K线锁定，持仓期间不漂移
|
||||
|
||||
为什么用ATR不用固定比例：
|
||||
- ATR自动适应市场：波动大时止损放宽免误扫，波动小时收紧控风险
|
||||
- 固定比例是拍脑袋，ATR是算出来的
|
||||
|
||||
追踪保护（v1.6 ATR自适应版）：
|
||||
- 利润达止盈目标50%：上移到保本（入场价）
|
||||
- 利润达止盈目标80%：启动ATR×0.5窄追踪
|
||||
"""
|
||||
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:
|
||||
stop_pct = 0.015
|
||||
# 查找入场时的 K 线，锁定当时的 ATR 值
|
||||
entry_row = self._get_entry_row(dataframe, trade)
|
||||
if entry_row is None:
|
||||
return -0.02 if not trade.is_short else 0.02
|
||||
|
||||
# 锁定入场时的 ATR 值，用于全程止损/追踪计算（不追最新，防止漂移）
|
||||
atr_value = entry_row.get("atr", np.nan)
|
||||
if pd.isna(atr_value) or atr_value <= 0:
|
||||
return -0.02 if not trade.is_short else 0.02
|
||||
|
||||
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)
|
||||
# 做多：止损 = 入场价 - ATR × 2.0
|
||||
base_sl_price = trade.open_rate - (atr_value * 2.0)
|
||||
base_sl = (base_sl_price / trade.open_rate) - 1.0
|
||||
base_sl = max(base_sl, -0.15)
|
||||
|
||||
# ================================================================
|
||||
# 止盈：到对侧边界 + K线形态确认 → 平仓
|
||||
# ================================================================
|
||||
# 追踪保护：需要入场行计算止盈目标
|
||||
support = entry_row.get("support_15m", np.nan)
|
||||
resistance = entry_row.get("resistance_15m", np.nan)
|
||||
if (not pd.isna(support) and not pd.isna(resistance)
|
||||
and resistance > support and current_profit > 0):
|
||||
zone_height = resistance - support
|
||||
tp_target = (zone_height * self.profit_zone_pct.value / 100.0) / trade.open_rate
|
||||
|
||||
if current_profit >= tp_target * 0.8:
|
||||
# 利润达止盈80%：ATR自适应窄追踪
|
||||
trail_price = current_rate - (atr_value * 0.5)
|
||||
trail_ratio = (trail_price / trade.open_rate) - 1.0
|
||||
return max(trail_ratio, base_sl)
|
||||
elif current_profit >= tp_target * 0.5:
|
||||
# 利润达止盈50%：保本
|
||||
return max(0.0, base_sl)
|
||||
|
||||
return base_sl
|
||||
else:
|
||||
# 做空：止损 = 入场价 + ATR × 2.0
|
||||
base_sl_price = trade.open_rate + (atr_value * 2.0)
|
||||
base_sl = 1.0 - (base_sl_price / trade.open_rate)
|
||||
base_sl = min(base_sl, 0.15)
|
||||
|
||||
# 追踪保护（做空对称）
|
||||
support = entry_row.get("support_15m", np.nan)
|
||||
resistance = entry_row.get("resistance_15m", np.nan)
|
||||
if (not pd.isna(support) and not pd.isna(resistance)
|
||||
and resistance > support and current_profit > 0):
|
||||
zone_height = resistance - support
|
||||
tp_target = (zone_height * self.profit_zone_pct.value / 100.0) / trade.open_rate
|
||||
|
||||
if current_profit >= tp_target * 0.8:
|
||||
# ATR自适应窄追踪（做空对称）
|
||||
trail_price = current_rate + (atr_value * 0.5)
|
||||
trail_ratio = (trail_price / trade.open_rate) - 1.0
|
||||
return min(trail_ratio, base_sl)
|
||||
elif current_profit >= tp_target * 0.5:
|
||||
# 保本
|
||||
return min(0.0, base_sl)
|
||||
|
||||
return base_sl
|
||||
|
||||
# =====================
|
||||
# 区间高度止盈
|
||||
# =====================
|
||||
|
||||
def custom_exit(
|
||||
self,
|
||||
@ -204,31 +531,59 @@ class StructureFlowSwingV42(IStrategy):
|
||||
current_profit: float,
|
||||
**kwargs,
|
||||
) -> str | None:
|
||||
"""
|
||||
当利润达到入场时锁定的15m区间高度的设定比例时止盈。
|
||||
|
||||
使用入场时锁定的S/R值计算区间高度（zone_height），而非最新的值：
|
||||
- 入场后如果区间收缩，止盈目标不会跟着变小
|
||||
- 让入场时确定的止盈逻辑"钉死"
|
||||
- profit_zone_pct 默认40%，即锁定区间高度的40%
|
||||
"""
|
||||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||||
if dataframe is None or len(dataframe) == 0:
|
||||
return None
|
||||
|
||||
last = dataframe.iloc[-1]
|
||||
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)
|
||||
# 查找入场时的 K 线，锁定当时的 S/R 值
|
||||
entry_row = self._get_entry_row(dataframe, trade)
|
||||
if entry_row is None:
|
||||
return None
|
||||
|
||||
# ── 做多：到阻力附近 + 滞涨形态 → 平仓 ──
|
||||
if not trade.is_short:
|
||||
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(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"
|
||||
support = entry_row.get("support_15m", np.nan)
|
||||
resistance = entry_row.get("resistance_15m", np.nan)
|
||||
|
||||
if pd.isna(support) or pd.isna(resistance) or resistance <= support:
|
||||
return None
|
||||
|
||||
# 用锁定的区间高度计算止盈目标（不随市场漂移）
|
||||
locked_zone_height = resistance - support
|
||||
target_pct = (locked_zone_height * self.profit_zone_pct.value / 100.0) / trade.open_rate
|
||||
|
||||
if current_profit >= target_pct:
|
||||
return "zone_tp"
|
||||
|
||||
return None
|
||||
|
||||
# =====================
|
||||
# Plot config
|
||||
# =====================
|
||||
|
||||
@staticmethod
|
||||
def plot_config() -> dict:
|
||||
return {
|
||||
"main_plot": {
|
||||
"support_15m": {"color": "green", "type": "line"},
|
||||
"resistance_15m": {"color": "red", "type": "line"},
|
||||
},
|
||||
"subplots": {
|
||||
"signals": {
|
||||
"bullish_pinbar": {"color": "green", "type": "scatter"},
|
||||
"bearish_pinbar": {"color": "red", "type": "scatter"},
|
||||
"bullish_signal": {"color": "lime", "type": "scatter"},
|
||||
"bearish_signal": {"color": "orange", "type": "scatter"},
|
||||
},
|
||||
"filters": {
|
||||
"support_alive_15m": {"color": "green", "type": "line"},
|
||||
"resistance_alive_15m": {"color": "red", "type": "line"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user