# source: https://raw.githubusercontent.com/camster91/crypto-bots/af74d6ceb208bb150cde9021c9fef130fea8ad16/bot-instance/user_data/strategies/MeanReversionStrategyV4.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Dict, Optional, Union, Tuple from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, merge_informative_pair, stoploss_from_absolute, stoploss_from_open, AnnotationType, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta from technical import qtpylib from functools import reduce class Github_camster91_crypto_bots__MeanReversionStrategyV4__20260701_171251(IStrategy): """ Mean Reversion Strategy with Short-Selling Capability Long entries: Price oversold (zscore < -threshold), expecting reversion to mean Short entries: Price overbought (zscore > +threshold), expecting reversion down to mean Both directions use: - Volume confirmation (spike required) - Trend filter (SMA slope) - RSI confirmation - Maximum hold time - Dynamic stop-loss based on ATR """ # Strategy interface version INTERFACE_VERSION = 3 # Can this strategy go short? can_short: bool = False # Minimal ROI designed for the strategy. minimal_roi = { "0": 0.11, # 11% immediate (from hyperopt) "37": 0.045, # 4.5% after 37 candles "71": 0.013, # 1.3% after 71 candles "141": 0, # 0% after 141 candles } # Optimal stoploss designed for the strategy. stoploss = -0.161 # -16.1% stop-loss (from hyperopt) # Trailing stoploss - DISABLED for mean reversion trailing_stop = False # Optimal timeframe for the strategy. timeframe = "5m" # Run "populate_indicators()" only for new candle. process_only_new_candles = True # These values can be overridden in the config. use_exit_signal = True exit_profit_only = True ignore_roi_if_entry_signal = True use_custom_stoploss = False # Use fixed stop-loss # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 200 # ===== LONG PARAMETERS ===== # Hyperoptable parameters for LONG entries buy_zscore = DecimalParameter(-3.0, -1.0, default=-2.0, space="buy", optimize=True, decimals=1) volume_threshold_long = DecimalParameter(1.2, 3.0, default=2.9, space="buy", optimize=True, decimals=1) rsi_buy = IntParameter(20, 35, default=24, space="buy", optimize=True) # ===== SHORT PARAMETERS ===== # Hyperoptable parameters for SHORT entries short_zscore = DecimalParameter(1.0, 3.0, default=1.5, space="sell", optimize=True, decimals=1) volume_threshold_short = DecimalParameter(1.2, 3.0, default=2.9, space="sell", optimize=True, decimals=1) rsi_short = IntParameter(65, 85, default=76, space="sell", optimize=True) # ===== COMMON PARAMETERS ===== # Moving average period for mean calculation ma_period = IntParameter(50, 150, default=92, space="buy", optimize=True) # Exit parameters sell_zscore_long = DecimalParameter(-0.5, 1.5, default=0.3, space="sell", optimize=True, decimals=1) rsi_sell_long = IntParameter(50, 80, default=80, space="sell", optimize=True) exit_zscore_short = DecimalParameter(-1.5, 0.5, default=-0.3, space="buy", optimize=True, decimals=1) rsi_exit_short = IntParameter(20, 50, default=24, space="buy", optimize=True) # Maximum hold time (candles) max_hold_candles = IntParameter(20, 100, default=86, space="sell", optimize=True) # SMA slope filter - different for long and short sma_slope_max_long = DecimalParameter(-1.0, 0.5, default=-0.09, space="buy", optimize=True, decimals=2) sma_slope_min_short = DecimalParameter(-0.5, 1.0, default=0.09, space="sell", optimize=True, decimals=2) # Volatility filter max_atr_pct = DecimalParameter(1.0, 5.0, default=3.0, space="buy", optimize=True, decimals=1) def populate_indicators(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ Adds several technical indicators to the dataframe. """ # Calculate Simple Moving Average (mean) dataframe['sma'] = ta.SMA(dataframe, timeperiod=self.ma_period.value) # Calculate standard deviation for Z-score dataframe['std'] = dataframe['close'].rolling(window=self.ma_period.value).std() # Calculate Z-score: how many standard deviations from the mean dataframe['zscore'] = (dataframe['close'] - dataframe['sma']) / dataframe['std'] # Calculate SMA slope (percentage change over last 12 candles) dataframe['sma_slope'] = (dataframe['sma'] - dataframe['sma'].shift(12)) / dataframe['sma'].shift(12) * 100 # Bollinger Bands for additional confirmation bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe['bb_lower'] = bb['lowerband'] dataframe['bb_middle'] = bb['middleband'] dataframe['bb_upper'] = bb['upperband'] dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle'] # Volume indicators dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # Calculate distance from SMA as percentage dataframe['sma_distance_pct'] = (dataframe['close'] - dataframe['sma']) / dataframe['sma'] * 100 # For visualization: flag oversold/overbought conditions dataframe['oversold'] = dataframe['zscore'] < self.buy_zscore.value dataframe['overbought'] = dataframe['zscore'] > self.short_zscore.value # Calculate average true range for volatility dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] * 100 return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ Entry conditions for both LONG and SHORT trades. """ # ===== LONG ENTRY CONDITIONS ===== long_conditions = [] # Core condition: Z-score indicates oversold long_conditions.append(dataframe['zscore'] < self.buy_zscore.value) # Safety: price should be below SMA (we're buying low) long_conditions.append(dataframe['close'] < dataframe['sma']) # Limit distance to avoid extreme deviations (max -5%) long_conditions.append(dataframe['sma_distance_pct'] > -5.0) # SMA slope filter: avoid strong downtrends for longs long_conditions.append(dataframe['sma_slope'] > self.sma_slope_max_long.value) # Volume confirmation: need volume spike long_conditions.append(dataframe['volume_ratio'] > self.volume_threshold_long.value) # RSI confirmation long_conditions.append(dataframe['rsi'] < self.rsi_buy.value) # Volatility filter: avoid extreme volatility long_conditions.append(dataframe['atr_pct'] < self.max_atr_pct.value) # Combine long conditions if long_conditions: dataframe.loc[ reduce(lambda x, y: x & y, long_conditions), 'enter_long'] = 1 # ===== SHORT ENTRY CONDITIONS ===== short_conditions = [] # Core condition: Z-score indicates overbought short_conditions.append(dataframe['zscore'] > self.short_zscore.value) # Safety: price should be above SMA (we're shorting high) short_conditions.append(dataframe['close'] > dataframe['sma']) # Limit distance to avoid extreme deviations (max +5%) short_conditions.append(dataframe['sma_distance_pct'] < 5.0) # SMA slope filter: avoid strong uptrends for shorts short_conditions.append(dataframe['sma_slope'] < self.sma_slope_min_short.value) # Volume confirmation: need volume spike short_conditions.append(dataframe['volume_ratio'] > self.volume_threshold_short.value) # RSI confirmation (overbought) short_conditions.append(dataframe['rsi'] > self.rsi_short.value) # Volatility filter: avoid extreme volatility short_conditions.append(dataframe['atr_pct'] < self.max_atr_pct.value) # Combine short conditions if short_conditions: dataframe.loc[ reduce(lambda x, y: x & y, short_conditions), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ Exit conditions for both LONG and SHORT trades. """ # ===== LONG EXIT CONDITIONS ===== long_exit_conditions = [] # Core condition: Z-score returned to mean long_exit_conditions.append(dataframe['zscore'] > self.sell_zscore_long.value) # Alternative: price crossed above SMA (returned to mean) long_exit_conditions.append(dataframe['close'] > dataframe['sma']) # RSI exit (overbought) long_exit_conditions.append(dataframe['rsi'] > self.rsi_sell_long.value) # Price reached Bollinger Band middle band long_exit_conditions.append(dataframe['close'] > dataframe['bb_middle']) # Combine long exit conditions with OR logic if long_exit_conditions: dataframe.loc[ reduce(lambda x, y: x | y, long_exit_conditions), 'exit_long'] = 1 # ===== SHORT EXIT CONDITIONS ===== short_exit_conditions = [] # Core condition: Z-score returned to mean short_exit_conditions.append(dataframe['zscore'] < self.exit_zscore_short.value) # Alternative: price crossed below SMA (returned to mean) short_exit_conditions.append(dataframe['close'] < dataframe['sma']) # RSI exit (oversold) short_exit_conditions.append(dataframe['rsi'] < self.rsi_exit_short.value) # Price reached Bollinger Band middle band short_exit_conditions.append(dataframe['close'] < dataframe['bb_middle']) # Combine short exit conditions with OR logic if short_exit_conditions: dataframe.loc[ reduce(lambda x, y: x | y, short_exit_conditions), 'exit_short'] = 1 return dataframe def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[str]: """ Custom exit logic for maximum hold time. """ # Calculate how many candles we've held trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60 candle_minutes = timeframe_to_minutes(self.timeframe) candles_held = trade_duration / candle_minutes # Maximum hold period check if candles_held >= self.max_hold_candles.value: return 'max_hold_time_reached' # If we have small profit and have held enough candles, take it if candles_held >= 10 and current_profit > 0.005: # 0.5% profit return 'quick_profit' return None def populate_buy(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: # For compatibility with older freqtrade versions dataframe = self.populate_entry_trend(dataframe, metadata) dataframe.loc[dataframe['enter_long'] == 1, 'buy'] = 1 return dataframe def populate_sell(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: # For compatibility with older freqtrade versions dataframe = self.populate_exit_trend(dataframe, metadata) dataframe.loc[dataframe['exit_long'] == 1, 'sell'] = 1 return dataframe