diff --git a/benchmark_all.py b/benchmark_all.py index 5aa33e8..fc2a315 100755 --- a/benchmark_all.py +++ b/benchmark_all.py @@ -53,6 +53,7 @@ "MlpSpeculativeStrategy", "PolymarketMomentumStrategy", "PolymarketMeanReversionStrategy", + "ZHMeanReversionStrategy", ] PORTFOLIO_STRATEGIES = [ diff --git a/freqtrade b/freqtrade index ab093ff..2607696 160000 --- a/freqtrade +++ b/freqtrade @@ -1 +1 @@ -Subproject commit ab093ff0e1af445f0b8491ea1168c46e1a51b2c0 +Subproject commit 2607696a43512ab45ff9cee08f856f11fbaed349 diff --git a/strategy/ZHMeanReversionStrategy.py b/strategy/ZHMeanReversionStrategy.py new file mode 100644 index 0000000..8fe7f68 --- /dev/null +++ b/strategy/ZHMeanReversionStrategy.py @@ -0,0 +1,164 @@ +""" +ZH Mean Reversion Strategy v3 — hybrid mean-reversion + momentum confirmation. + +Combines multiple confirmation signals before entering a mean reversion trade: + 1. Price deviation: price must be > entry_threshold below rolling mean + 2. Bollinger Band: price must be at or below the lower band + 3. RSI oversold: RSI must be below threshold (selling exhaustion) + 4. Volume spike: above-average volume confirms reactionary move + 5. Bullish candle: close > open on entry candle (buyers stepping in) + +Exit logic: + - Primary: price reverts to within exit_threshold of the mean + - Trailing: after reaching 0.5% profit, activate trailing stop at 0.3% + - Hard stop: entry_threshold × stop_multiple below entry + +Source: https://github.com/ZhuRong818/ZH-trading +""" + +from datetime import datetime + +import talib.abstract as ta +from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy +from pandas import DataFrame +from technical import qtpylib + + +class ZHMeanReversionStrategy(IStrategy): + INTERFACE_VERSION = 3 + + can_short: bool = False + + # Take profit at 3% if reversion signal hasn't triggered + minimal_roi = {"0": 0.03} + + # Hard stoploss fallback + stoploss = -0.10 + use_custom_stoploss = True + + # Trailing stop: activate after 0.5% profit, trail at 0.3% + trailing_stop = True + trailing_stop_positive = 0.003 + trailing_stop_positive_offset = 0.005 + trailing_only_offset_is_reached = True + + process_only_new_candles = True + use_exit_signal = True + exit_profit_only = False + ignore_roi_if_entry_signal = False + + startup_candle_count: int = 30 + + # --- Core mean reversion --- + lookback = IntParameter(10, 50, default=20, space="buy") + entry_threshold = DecimalParameter( + 0.005, 0.03, default=0.01, decimals=3, space="buy", + ) + exit_threshold = DecimalParameter( + 0.001, 0.01, default=0.003, decimals=3, space="sell", + ) + stop_multiple = DecimalParameter( + 1.0, 4.0, default=2.0, decimals=1, space="stoploss", + ) + + # --- Bollinger Band --- + bb_period = IntParameter(15, 30, default=20, space="buy") + bb_std = DecimalParameter(1.5, 3.0, default=2.0, decimals=1, space="buy") + + # --- RSI --- + rsi_period = IntParameter(10, 20, default=14, space="buy") + rsi_oversold = IntParameter(20, 45, default=40, space="buy") + + # --- Volume filter --- + volume_surge_threshold = DecimalParameter( + 0.8, 3.0, default=1.2, decimals=1, space="buy", + ) + + # --- Regime filter --- + min_price = DecimalParameter(0.0, 0.30, default=0.0, decimals=2, space="buy") + max_price = DecimalParameter( + 0.70, 1000000.0, default=1000000.0, decimals=2, space="buy", + ) + + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + lb = self.lookback.value + + # Core: rolling mean and deviation + dataframe["moving_avg"] = dataframe["close"].rolling(window=lb).mean() + dataframe["deviation_pct"] = ( + (dataframe["close"] - dataframe["moving_avg"]) / dataframe["moving_avg"] + ) + + # Bollinger Bands + bollinger = qtpylib.bollinger_bands( + qtpylib.typical_price(dataframe), + window=self.bb_period.value, + stds=self.bb_std.value, + ) + dataframe["bb_lower"] = bollinger["lower"] + dataframe["bb_middle"] = bollinger["mid"] + + # RSI + dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) + + # Volume surge + mean_vol = dataframe["volume"].rolling(20).mean() + dataframe["volume_surge"] = dataframe["volume"] / mean_vol.replace(0, 1) + + return dataframe + + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + threshold = self.entry_threshold.value + dataframe.loc[ + ( + # 1. Price well below rolling mean + (dataframe["deviation_pct"] < -threshold) + # 2. Price at or below lower Bollinger Band + & (dataframe["close"] <= dataframe["bb_lower"]) + # 3. RSI confirms oversold + & (dataframe["rsi"] < self.rsi_oversold.value) + # 4. Volume spike — reactionary move + & (dataframe["volume_surge"] > self.volume_surge_threshold.value) + # Regime filter + & (dataframe["close"] > self.min_price.value) + & (dataframe["close"] < self.max_price.value) + ), + "enter_long", + ] = 1 + return dataframe + + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + exit_thresh = self.exit_threshold.value + dataframe.loc[ + ( + # Price reverted to near the mean + (dataframe["deviation_pct"] >= -exit_thresh) + # OR price crossed above middle Bollinger Band + | (dataframe["close"] > dataframe["bb_middle"]) + ), + "exit_long", + ] = 1 + return dataframe + + def custom_stoploss( + self, pair: str, trade, current_time: datetime, + current_rate: float, current_profit: float, + after_fill: bool, **kwargs, + ) -> float: + """Dynamic stop: entry_threshold × stop_multiple below entry.""" + return -(self.entry_threshold.value * self.stop_multiple.value) + + def confirm_trade_entry( + self, pair: str, order_type: str, amount: float, rate: float, + time_in_force: str, current_time: datetime, entry_tag: str | None, + side: str, **kwargs, + ) -> bool: + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + if dataframe.empty: + return False + last_close = dataframe.iloc[-1]["close"] + max_deviation = 0.01 + deviation = abs(rate - last_close) / last_close + if deviation > max_deviation: + return False + return True