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

Compare commits

base: beast:v2.2b
Branches Tags
beast:main
beast:v3.1 beast:v2.2d beast:v2.2c beast:v2.2b
...
compare: beast:v2.2d
Branches Tags
beast:main
beast:v3.1 beast:v2.2d beast:v2.2c beast:v2.2b

2 Commits
v2.2b ... v2.2d

Author SHA1 Message Date
Beast Trader
688fe36e3b v2.2d: 冷却期修复版 - 震荡市自动休眠 + 三层趋势共振保护 2026-06-11 21:14:16 +08:00
Beast Trader
34d61cfa43 v2.2c: D1趋势总闸门 + 三层共振(1H/4H/D1) + 冷却期机制 2026-06-11 21:14:12 +08:00
1 changed files with 101 additions and 105 deletions
Expand all files Collapse all files

206
strategy.py
View File

@ -1,14 +1,9 @@
"""
Structure Flow Strategy v2.2b
=======================
Structure Flow Strategy v2.2c — 冷却期修复版
==============================================
变更记录：
v1.6 (2026-06-07): 最优基线 — +3659.63%, 190笔, 69.3% trailing胜率
v2.0 (2026-06-08): B1 入场延迟确认 — 方向正确但降频严重
v2.2b (2026-06-09): ===== 只移除 bullish_signal/bearish_signal =====
在4H级别评估趋势强度：最近2个Swing Point的间距变化。
如果趋势在扩张（HH/HL间距增大），允许入场；
如果趋势在收缩（HH/HL间距缩小或震荡），过滤信号。
目的：只在趋势明确时交易，避免震荡市反复止损。
v2.2c (2026-06-11): 1H S/R 替代 4H S/R
v2.2c-coolfix (2026-06-11): 修复冷却期无限阻止下单 bug
"""
from datetime import datetime
@ -19,15 +14,7 @@ from freqtrade.strategy import IStrategy, IntParameter, informative
from freqtrade.persistence import Trade
class StructureFlowStrategyV22b(IStrategy):
"""
Structure Flow Strategy v2.2b — D1: 趋势强度过滤
v2.2b改动（相对于v2.1）：
在4H级别计算趋势强度：最近2个Swing High间距 + Swing Low间距的变化。
只有趋势在扩张（或至少不收缩）时才允许入场。
"""
class StructureFlowStrategyV22d(IStrategy):
can_short = True
stoploss = -0.15
use_custom_stoploss = True
@ -41,12 +28,11 @@ class StructureFlowStrategyV22b(IStrategy):
swing_lookback_d1 = IntParameter(8, 14, default=10, space="buy")
swing_lookback_h4 = IntParameter(5, 10, default=8, space="buy")
swing_lookback_1h = IntParameter(3, 7, default=5, space="buy") # 新增：1H swing参数
pin_bar_wick_ratio = IntParameter(50, 70, default=60, space="buy")
max_stop_dist = IntParameter(20, 50, default=50, space="buy")
cooldown_bars = IntParameter(3, 12, default=6, space="buy")
# v2.1 新增：趋势强度最小扩张比例（x/100 = 0%~50%）
# 0 = 只要不收缩就行；越大要求趋势扩张越强
trend_strength_min = IntParameter(-50, 20, default=-20, space="buy") # 扫描更宽范围
trend_strength_min = IntParameter(-50, 20, default=-20, space="buy")
# =====================
# 工具：Swing Point 检测
@ -168,6 +154,35 @@ class StructureFlowStrategyV22b(IStrategy):
return bullish_pin, bearish_pin, bullish_engulf, bearish_engulf
# =====================
# 工具：冷却期正确实现（修复 bug）
# =====================
def _apply_cooldown(self, signal: pd.Series, cooldown_bars: int) -> pd.Series:
"""
正确应用冷却期：入场后才冷却，而非条件满足就冷却。
原逻辑 bug：long_base.rolling(cooldown).max().shift(1) == 0
- 当市场持续满足入场条件时，rolling window 里永远有 True
- 导致冷却期无限阻止下单
修复逻辑：遍历 K 线，模拟"入场 -> 冷却"过程。
- 满足条件 + 距离上次入场 > cooldown -> 允许入场
- 入场后 cooldown 根 K 线内不再入场
"""
n = len(signal)
result = [False] * n
last_entry = -99999 # 上次入场的 bar 索引
# 遍历（对 numpy array 操作，O(n) 约几毫秒）
values = signal.values # numpy array，快速访问
for i in range(n):
if values[i] and (i - last_entry) > cooldown_bars:
result[i] = True
last_entry = i
return pd.Series(result, index=signal.index)
# ================================================================
# 信息时间框架 — D1 宏观结构
# ================================================================
@ -189,7 +204,7 @@ class StructureFlowStrategyV22b(IStrategy):
return dataframe
# ================================================================
# 信息时间框架 — 4H 中期结构
# 信息时间框架 — 4H 趋势强度（原版保留）
# ================================================================
@informative("4h")
@ -204,40 +219,8 @@ class StructureFlowStrategyV22b(IStrategy):
dataframe["high"], dataframe["low"], dataframe["close"],
sh, sl,
)
dataframe["trend_up"] = structure["trend_up"]
dataframe["trend_down"] = structure["trend_down"]
dataframe["support"] = structure["support"]
dataframe["resistance"] = structure["resistance"]
dataframe["in_demand"] = structure["in_demand"]
dataframe["in_supply"] = structure["in_supply"]
# ================================
# v1.6 活支撑/阻力检查（保留）
# ================================
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(3, min_periods=1).max() > 0
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(3, min_periods=1).max() > 0
# ================================
# v2.1 新增：趋势强度评估
# ================================
# 计算最近2个Swing Point之间的间距变化
# 上升趋势：HH间距 + HL间距都在扩大 → 趋势强
# 下降趋势：LH间距 + LL间距都在扩大 → 趋势强
# 间距缩小 → 趋势减弱/震荡
# 趋势强度计算（原版逻辑）
sh_prices = []
sl_prices = []
trend_strength_up = np.full(len(dataframe), np.nan)
@ -253,36 +236,29 @@ class StructureFlowStrategyV22b(IStrategy):
if len(sl_prices) > 4:
sl_prices.pop(0)
# 上升趋势强度：HH[-1] vs HH[-2], HL[-1] vs HL[-2]
if len(sh_prices) >= 2 and len(sl_prices) >= 2:
# HH间距：最近两个Swing High的差值百分比
hh_dist = (sh_prices[-1] - sh_prices[-2]) / sh_prices[-2] if sh_prices[-2] > 0 else 0
# HL间距：最近两个Swing Low的差值百分比
hl_dist = (sl_prices[-1] - sl_prices[-2]) / sl_prices[-2] if sl_prices[-2] > 0 else 0
# 上升趋势强度 = HH间距 + HL间距（都正=扩张，一正一负=不确定，都负=收缩）
trend_strength_up[i] = hh_dist + hl_dist
# 下降趋势强度（取反：间距缩小是负值）
trend_strength_down[i] = -(hh_dist + hl_dist)
dataframe["trend_strength_up"] = pd.Series(trend_strength_up, index=dataframe.index)
dataframe["trend_strength_down"] = pd.Series(trend_strength_down, index=dataframe.index)
# 趋势强度是否足够（扩张中）
min_strength = self.trend_strength_min.value / 100.0 # 0~0.30
min_strength = self.trend_strength_min.value / 100.0
dataframe["strong_uptrend"] = dataframe["trend_strength_up"] > min_strength
dataframe["strong_downtrend"] = dataframe["trend_strength_down"] > min_strength
return dataframe
# ================================================================
# 主时间框架 — 1H 指标
# 主时间框架 — 1H 指标（含 1H S/R + 活支撑/阻力）
# ================================================================
def populate_indicators(
self, dataframe: DataFrame, metadata: dict
) -> DataFrame:
"""1H 级别：K线形态（零指标）。"""
# ── K线形态 ──
bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = (
self._detect_candle_patterns(
dataframe["open"],
@ -299,12 +275,45 @@ class StructureFlowStrategyV22b(IStrategy):
dataframe["bullish_signal"] = bullish_pin | bullish_engulf
dataframe["bearish_signal"] = bearish_pin | bearish_engulf
# NaN 安全处理
# ── 1H级别 Swing Point + 结构（替代原4H S/R） ──
sh_1h, sl_1h = self._detect_swing_points(
dataframe["high"], dataframe["low"],
self.swing_lookback_1h.value,
)
structure_1h = self._build_structure(
dataframe["high"], dataframe["low"], dataframe["close"],
sh_1h, sl_1h,
)
dataframe["trend_up_1h"] = structure_1h["trend_up"]
dataframe["trend_down_1h"] = structure_1h["trend_down"]
dataframe["support"] = structure_1h["support"]
dataframe["resistance"] = structure_1h["resistance"]
dataframe["in_demand"] = structure_1h["in_demand"]
dataframe["in_supply"] = structure_1h["in_supply"]
# ── 1H 活支撑/阻力检查 ──
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(3, min_periods=1).max() > 0
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(3, min_periods=1).max() > 0
# ── NaN 安全处理 ──
bool_cols = [
"trend_up_1d", "trend_down_1d",
"trend_up_4h", "trend_down_4h",
"in_demand_4h", "in_supply_4h",
"support_alive_4h", "resistance_alive_4h",
"in_demand", "in_supply",
"support_alive", "resistance_alive",
"strong_uptrend_4h", "strong_downtrend_4h",
"bullish_signal", "bearish_signal",
]
@ -315,25 +324,18 @@ class StructureFlowStrategyV22b(IStrategy):
return dataframe
# =====================
# 入场信号
# 入场信号（修复冷却期逻辑）
# =====================
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
入场逻辑（1H 时间框架）。
v2.2b 改动：只移除 bullish_signal/bearish_signal（1H K线过滤）
消融实验变体3：移除后收益 +19.4%，是三个可移除条件中收益提升最大的
"""
max_dist = self.max_stop_dist.value / 100.0
cooldown = self.cooldown_bars.value
# NaN 安全处理
bool_cols = [
"trend_up_1d", "trend_down_1d",
"trend_up_4h", "trend_down_4h",
"in_demand_4h", "in_supply_4h",
"support_alive_4h", "resistance_alive_4h",
"in_demand", "in_supply",
"support_alive", "resistance_alive",
"strong_uptrend_4h", "strong_downtrend_4h",
"bullish_signal", "bearish_signal",
]
@ -341,39 +343,37 @@ class StructureFlowStrategyV22b(IStrategy):
if col in dataframe.columns:
dataframe[col] = dataframe[col].fillna(False)
# ── 做多 ──
long_stop_dist = (dataframe["open"] - dataframe["support_4h"]) / dataframe["open"]
# ── 做多（使用1H S/R） ──
long_stop_dist = (dataframe["open"] - dataframe["support"]) / dataframe["open"]
long_base = (
dataframe["trend_up_1d"]
& dataframe["in_demand_4h"]
# v2.2b: 已移除 bullish_signal（消融变体3）
& dataframe["in_demand"]
& (long_stop_dist <= max_dist)
& (long_stop_dist > 0.003)
& dataframe["support_alive_4h"]
# v2.1: 趋势强度 — 4H上升趋势必须在扩张
& dataframe["support_alive"]
& dataframe["strong_uptrend_4h"]
)
long_recent = long_base.rolling(cooldown, min_periods=1).max().shift(1) == 0
dataframe.loc[long_base & long_recent, "enter_long"] = 1
# ✅ 修复：正确应用冷却期（基于实际入场，而非条件满足）
long_entries = self._apply_cooldown(long_base, cooldown)
dataframe.loc[long_entries, "enter_long"] = 1
# ── 做空 ──
short_stop_dist = (dataframe["resistance_4h"] - dataframe["open"]) / dataframe["open"]
# ── 做空（使用1H S/R） ──
short_stop_dist = (dataframe["resistance"] - dataframe["open"]) / dataframe["open"]
short_base = (
dataframe["trend_down_1d"]
& dataframe["in_supply_4h"]
# v2.2b: 已移除 bearish_signal（消融变体3）
& dataframe["in_supply"]
& (short_stop_dist <= max_dist)
& (short_stop_dist > 0.003)
& dataframe["resistance_alive_4h"]
# v2.1: 趋势强度 — 4H下降趋势必须在扩张
& dataframe["resistance_alive"]
& dataframe["strong_downtrend_4h"]
)
short_recent = short_base.rolling(cooldown, min_periods=1).max().shift(1) == 0
dataframe.loc[short_base & short_recent, "enter_short"] = 1
# ✅ 修复：正确应用冷却期（基于实际入场，而非条件满足）
short_entries = self._apply_cooldown(short_base, cooldown)
dataframe.loc[short_entries, "enter_short"] = 1
return dataframe
@ -382,7 +382,6 @@ class StructureFlowStrategyV22b(IStrategy):
# =====================
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""出场逻辑 — 由结构反转触发。"""
exit_long = ~dataframe["trend_up_1d"].fillna(True)
dataframe.loc[exit_long, "exit_long"] = 1
@ -392,7 +391,7 @@ class StructureFlowStrategyV22b(IStrategy):
return dataframe
# =====================
# 动态止损 — 纯价格结构（基于Swing Point）
# 动态止损（基于1H S/R）
# =====================
def custom_stoploss(
@ -405,9 +404,6 @@ class StructureFlowStrategyV22b(IStrategy):
after_fill: bool,
**kwargs,
) -> float:
"""
止损逻辑：完全基于价格结构，零指标（与v1.6相同）。
"""
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
@ -415,14 +411,14 @@ class StructureFlowStrategyV22b(IStrategy):
last = dataframe.iloc[-1]
if not trade.is_short:
support = last.get("support_4h", np.nan)
support = last.get("support", np.nan)
if pd.isna(support) or support <= 0:
return -0.02
sl_price = support * 0.999
sl_ratio = (sl_price / current_rate) - 1.0
return max(sl_ratio, -0.15)
else:
resistance = last.get("resistance_4h", np.nan)
resistance = last.get("resistance", np.nan)
if pd.isna(resistance) or resistance <= 0:
return 0.02
sl_price = resistance * 1.001
@ -437,8 +433,8 @@ class StructureFlowStrategyV22b(IStrategy):
def plot_config() -> dict:
return {
"main_plot": {
"support_4h": {"color": "green", "type": "line"},
"resistance_4h": {"color": "red", "type": "line"},
"support": {"color": "green", "type": "line"},
"resistance": {"color": "red", "type": "line"},
},
"subplots": {
"signals": {
@ -446,8 +442,8 @@ class StructureFlowStrategyV22b(IStrategy):
"bearish_pinbar": {"color": "red", "type": "scatter"},
},
"filters": {
"support_alive_4h": {"color": "green", "type": "line"},
"resistance_alive_4h": {"color": "red", "type": "line"},
"support_alive": {"color": "green", "type": "line"},
"resistance_alive": {"color": "red", "type": "line"},
"strong_uptrend_4h": {"color": "blue", "type": "line"},
"strong_downtrend_4h": {"color": "orange", "type": "line"},
},