熊市回调做空策略(Regime + 反弹强度双过滤)
Timeframe
1h
Direction
Long & Short
Stoploss
-15.0%
Trailing Stop
No
ROI
0m: 10000.0%
Interface Version
3
Startup Candles
N/A
Indicators
6
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# -*- coding: utf-8 -*-
"""
BearPullbackShortBearV4 - 熊市回调做空策略(强化 Regime + 反弹强度过滤)
在 V3 的基础上增加:
- pullback_strength(反弹强度)过滤,只做“反弹足够强”的高概率做空:
pullback_strength = close / rolling_low(N) - 1
- 通过可调参数 pullback_min 控制最小反弹幅度(默认 5%)。
整体结构:
- 只在“真正的熊市 Regime”做空:
- EMA20 < EMA50 < EMA200
- close < EMA200
- MACD < 0
- ADX > adx_threshold
- 以上条件连续 regime_persistence 根 K 成立才视为熊市
- 熊市中:布林上轨突破 + RSI 过热 + pullback_strength > 阈值 才允许开空;
- 止损:ATR * mult + 全局上限(1%~2.2%)+ soft_stop(短期走反 -1% 认错);
- 止盈:超卖信号 + 盈利 >= 3% / 6% 分两档;
- 移动止盈:3%、5%、8%、12% 四档动态止盈。
"""
from datetime import datetime
import numpy as np
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from freqtrade.persistence import Trade
class BearPullbackShortBearV4(IStrategy):
"""熊市回调做空策略(Regime + 反弹强度双过滤)"""
INTERFACE_VERSION = 3
timeframe = "1h"
can_short = True
# 兜底硬止损(真正逻辑由 custom_exit / custom_stoploss 接管)
stoploss = -0.15
# ROI 完全交给 custom_exit
minimal_roi = {"0": 100}
# === 参数区(后续可用 hyperopt 调优) ===
# 判趋势用 ADX 阈值(熊市要求适中,不必极强)
adx_threshold = IntParameter(15, 30, default=20, space="buy")
# 入场过热条件:RSI 阈值(熊市反弹一般 55~70 区间)
rsi_entry = IntParameter(55, 75, default=60, space="buy")
# 布林带标准差
bb_std = DecimalParameter(1.8, 2.6, default=2.0, space="buy")
# 超时(小时):超过这个时间,盈利或小亏就平仓
timeout_hours = IntParameter(12, 48, default=24, space="sell")
# ATR 止损倍数(最大允许亏损 = min(ATR_pct * mult, cap))
atr_mult = DecimalParameter(2.0, 4.0, default=3.0, space="sell")
# Regime 持续性要求:熊市条件至少连续多少根 K 后才算真正进入熊市
regime_persistence = IntParameter(12, 72, default=24, space="buy")
# 反弹强度过滤:最小 pullback_strength(例如 0.05 = 5%)
pullback_min = DecimalParameter(0.03, 0.12, default=0.05, space="buy")
# === 保护机制 ===
@property
def protections(self):
return [
{"method": "CooldownPeriod", "stop_duration_candles": 1},
{
"method": "StoplossGuard",
"lookback_period_candles": 24,
"trade_limit": 5,
"stop_duration_candles": 6,
"only_per_pair": False,
},
]
# === 指标计算 ===
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 布林带
bb = ta.BBANDS(
dataframe,
timeperiod=20,
nbdevup=float(self.bb_std.value),
nbdevdn=float(self.bb_std.value),
)
dataframe["bb_upper"] = bb["upperband"]
dataframe["bb_middle"] = bb["middleband"]
dataframe["bb_lower"] = bb["lowerband"]
# RSI / ADX
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
dataframe["adx"] = ta.ADX(dataframe)
# EMA 趋势线
dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)
dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
# MACD(用于确认长期处于 0 轴下方)
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
# ATR 用于动态止损
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
# --- 熊市 Regime 检测 ---
adx_thr = int(self.adx_threshold.value)
pers = int(self.regime_persistence.value)
# 基础熊市条件:价格与均线空头排列 + MACD < 0
base_bear = (
(dataframe["close"] < dataframe["ema_200"])
& (dataframe["ema_20"] < dataframe["ema_50"])
& (dataframe["ema_50"] < dataframe["ema_200"])
& (dataframe["macd"] < 0)
)
# 趋势强度:ADX > 阈值
strong_trend = dataframe["adx"] > adx_thr
# 原始熊市判定(未考虑持续性)
raw_bear = base_bear & strong_trend
dataframe["bear_raw"] = np.where(raw_bear, 1, 0)
# 持续性过滤:要求 raw_bear 连续 pers 根 K 都为 True
if pers > 1:
bear_regime = (
raw_bear.astype("int").rolling(window=pers, min_periods=pers).sum()
== pers
)
else:
bear_regime = raw_bear
dataframe["bear_regime"] = np.where(bear_regime, 1, 0)
# --- 反弹强度:当前价相对过去 N 根最低价的反弹比例 ---
# 使用 low 的 rolling min 更稳健
lookback = 20
rolling_low = dataframe["low"].rolling(lookback).min()
dataframe["pullback_strength"] = (dataframe["close"] / rolling_low - 1.0).clip(
lower=-0.5, upper=1.0
)
# 预备列
if "enter_tag" not in dataframe.columns:
dataframe["enter_tag"] = ""
if "exit_short" not in dataframe.columns:
dataframe["exit_short"] = 0
return dataframe
# === 入场:熊市中的“强反弹 + 过热” ===
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 只做首次突破上轨的K线,避免上轨附近连环抄顶
cross_upper = (
(dataframe["close"] > dataframe["bb_upper"])
& (dataframe["close"].shift(1) <= dataframe["bb_upper"].shift(1))
)
rsi_thr = int(self.rsi_entry.value)
pb_min = float(self.pullback_min.value)
cond = (
(dataframe["bear_regime"] == 1)
& cross_upper
& (dataframe["rsi"] > rsi_thr)
& (dataframe["pullback_strength"] > pb_min)
& (dataframe["volume"] > 0)
)
dataframe.loc[cond, ["enter_short", "enter_tag"]] = (1, "bear_strong")
return dataframe
# === 出场信号基准(超卖位置) ===
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""给出一个“超卖”信号,供 custom_exit 使用:
价格跌破布林下轨 或 RSI < 30。
"""
if "exit_short" not in dataframe.columns:
dataframe["exit_short"] = 0
oversold = (
(dataframe["close"] < dataframe["bb_lower"]) | (dataframe["rsi"] < 30)
) & (dataframe["volume"] > 0)
dataframe.loc[oversold, "exit_short"] = 1
return dataframe
# === 工具:ATR 动态止损(返回最大允许亏损百分比,正数) ===
def _calc_max_loss_pct(self, candle) -> float:
atr = float(candle.get("atr", 0) or 0)
close = float(candle.get("close", 0) or 1)
atr_pct = atr / close if close > 0 else 0.01
mult = float(self.atr_mult.value)
raw = atr_pct * mult
# 根据 enter_tag 设置不同的上限(预留扩展位,当前只用 bear_strong)
enter_tag = candle.get("enter_tag", "") or "bear_strong"
if enter_tag == "bear_strong":
cap = 0.022 # 强熊市允许稍宽
else:
cap = 0.018 # 预留:弱熊市更窄
floor = 0.010 # 至少 1%
return max(floor, min(raw, cap))
# === 核心:自定义出场逻辑 ===
def custom_exit(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
):
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
try:
df = dataframe.loc[dataframe["date"] <= current_time]
last = df.iloc[-1] if not df.empty else dataframe.iloc[-1]
except Exception:
last = dataframe.iloc[-1]
is_bear = int(last.get("bear_regime", 0)) == 1
exit_signal = int(last.get("exit_short", 0))
max_loss_pct = self._calc_max_loss_pct(last)
# 交易存活时长(小时)
duration_hrs = (current_time - trade.open_date_utc).total_seconds() / 3600
timeout = int(self.timeout_hours.value)
# 1) 硬止损:动态 ATR 止损
if current_profit <= -max_loss_pct:
return "bear_stop_loss"
# 2) soft_stop:开仓后不久就走反,-1% 附近直接认错
# 避免拖到最大止损
if is_bear and duration_hrs < 6: # 开仓 6 小时内
if current_profit <= -0.01:
return "soft_stop"
# 3) Regime Flip:不再是熊市时,盈利或小亏就走
if not is_bear:
if current_profit > -0.005: # -0.5% 以内
return "regime_flip"
# 更深亏损交由硬止损处理
# 4) 正常熊市中,根据盈利和超卖信号出场
if exit_signal == 1:
# 第二目标:盈利 >= 6%
if current_profit >= 0.06:
return "tp2_exit"
# 第一目标:盈利 >= 3%
if current_profit >= 0.03:
return "tp1_exit"
# 5) 超时处理:超过 timeout,小亏/小赢也离场
if duration_hrs > timeout:
tag = f"timeout_exit_({timeout}h)"
if current_profit > 0:
return tag
if current_profit > -0.005:
return tag
# 其它情况不强制退出
return None
# === 动态移动止盈:在大盈利时保护仓位 ===
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
) -> float:
"""根据浮盈动态上调止损价位:
- 盈利 > 3% -> 锁定 0.5%
- 盈利 > 5% -> 锁定 2%
- 盈利 > 8% -> 锁定 4%
- 盈利 > 12% -> 锁定 6%
其它情况返回 1,让 custom_exit / 兜底止损处理。
"""
if current_profit > 0.12:
return 0.06
if current_profit > 0.08:
return 0.04
if current_profit > 0.05:
return 0.02
if current_profit > 0.03:
return 0.005
return 1