# 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 Optional, Union, Dict, Any from functools import reduce from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, # Strategy helper functions merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) # -------------------------------- # Add your lib to import here import pandas_ta as ta from technical import qtpylib class QuantumPandasTAStrategy(IStrategy): """ Quantum Pandas TA Strategy - Advanced multi-timeframe cryptocurrency trading strategy Features: - Multi-timeframe analysis (5m, 15m, 1h, 4h) - Market regime detection using ADX and moving averages - Volume analysis and confirmation - Multiple momentum and trend indicators - Dynamic risk management - Position sizing based on market volatility OPTIMIZED VERSION - Generated from Hyperopt on 2025-11-08 """ INTERFACE_VERSION = 3 # Strategy metadata can_short: bool = False timeframe = "5m" process_only_new_candles = True # ROI configuration - optimized for crypto volatility minimal_roi = { # Aim for higher average profit per trade by avoiding near-breakeven exits. "0": 0.02, # 2% immediate target "60": 0.015, # 1.5% after 1 hour "120": 0.012, # 1.2% after 2 hours "240": 0.008, # 0.8% after 4 hours (floor) } # Stoploss and risk management stoploss = -0.02 # tighter 2% stoploss to reduce tail losses trailing_stop = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.02 trailing_only_offset_is_reached = False # Exit behavior: Only exit on ROI when in profit (avoid tiny/breakeven exits) use_exit_signal = True exit_profit_only = True ignore_roi_if_entry_signal = False # Enable custom stoploss for adaptive risk control use_custom_stoploss = True # OPTIMIZED PARAMETERS FROM HYPEROPT # Buy parameters (optimized for maximum profitability) buy_params = { "adx_strong": 34, "adx_threshold": 10, "atr_multiplier": 2.528, "bb_std": 1.514, "bb_window": 18, "buy_rsi_high": 49, "buy_rsi_low": 29, "ema_fast_period": 12, "ema_slow_period": 15, "ema_trend_period": 46, "regime_ema_period": 121, "stoch_k_low": 35, "volume_factor": 0.941, } # Sell parameters (optimized for maximum profitability) sell_params = { "sell_rsi_high": 75, "sell_rsi_low": 71, "stoch_k_high": 63, } # Strategy parameters with hyperopt optimization # RSI parameters (optimized) buy_rsi_low = IntParameter(20, 45, default=buy_params["buy_rsi_low"], space="buy", optimize=True, load=True) buy_rsi_high = IntParameter(45, 70, default=buy_params["buy_rsi_high"], space="buy", optimize=True, load=True) sell_rsi_low = IntParameter(55, 75, default=sell_params["sell_rsi_low"], space="sell", optimize=True, load=True) sell_rsi_high = IntParameter(75, 95, default=sell_params["sell_rsi_high"], space="sell", optimize=True, load=True) # ADX parameters for trend strength (optimized) adx_threshold = IntParameter(10, 25, default=buy_params["adx_threshold"], space="buy", optimize=True, load=True) adx_strong = IntParameter(25, 40, default=buy_params["adx_strong"], space="buy", optimize=True, load=True) # Volume parameters (optimized) volume_factor = DecimalParameter(0.6, 1.5, default=buy_params["volume_factor"], space="buy", optimize=True, load=True) # Bollinger Bands parameters (optimized) bb_window = IntParameter(15, 25, default=buy_params["bb_window"], space="buy", optimize=True, load=True) bb_std = DecimalParameter(1.2, 2.0, default=buy_params["bb_std"], space="buy", optimize=True, load=True) # ATR multiplier for volatility-based position sizing (optimized) atr_multiplier = DecimalParameter(1.5, 3.5, default=buy_params["atr_multiplier"], space="buy", optimize=True, load=True) # EMA periods for trend confirmation (optimized) ema_fast_period = IntParameter(8, 15, default=buy_params["ema_fast_period"], space="buy", optimize=True, load=True) ema_slow_period = IntParameter(12, 26, default=buy_params["ema_slow_period"], space="buy", optimize=True, load=True) ema_trend_period = IntParameter(30, 60, default=buy_params["ema_trend_period"], space="buy", optimize=True, load=True) # Stochastic parameters (optimized) stoch_k_low = IntParameter(20, 40, default=buy_params["stoch_k_low"], space="buy", optimize=True, load=True) stoch_k_high = IntParameter(60, 80, default=sell_params["stoch_k_high"], space="sell", optimize=True, load=True) # Market regime parameters (optimized) regime_ema_period = IntParameter(80, 150, default=buy_params["regime_ema_period"], space="buy", optimize=True, load=True) # Startup candle count for indicator calculation startup_candle_count: int = 400 # Order configuration order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False, } order_time_in_force = {"entry": "GTC", "exit": "GTC"} # Plot configuration plot_config = { "main_plot": { "ema_fast": {"color": "blue"}, "ema_slow": {"color": "orange"}, "ema_trend": {"color": "red"}, "bb_upper": {"color": "gray"}, "bb_lower": {"color": "gray"}, }, "subplots": { "RSI": { "rsi": {"color": "purple"}, }, "Stochastic": { "stoch_k": {"color": "green"}, "stoch_d": {"color": "red"}, }, "ADX": { "adx": {"color": "blue"}, }, "Volume": { "volume_ratio": {"color": "orange"}, }, "ATR": { "atr": {"color": "brown"}, }, }, } def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. """ return [ ("BTC/USDT", "15m"), ("BTC/USDT", "1h"), ("ETH/USDT", "15m"), ("ETH/USDT", "1h"), ] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds multiple technical indicators using pandas_ta for comprehensive market analysis OPTIMIZED VERSION """ # Calculate RSI for different timeframes dataframe['rsi'] = ta.rsi(dataframe['close'], length=14) # Calculate EMAs for trend analysis dataframe['ema_fast'] = ta.ema(dataframe['close'], length=self.ema_fast_period.value) dataframe['ema_slow'] = ta.ema(dataframe['close'], length=self.ema_slow_period.value) dataframe['ema_trend'] = ta.ema(dataframe['close'], length=self.ema_trend_period.value) # Calculate Bollinger Bands bb = ta.bbands(dataframe['close'], length=self.bb_window.value, std=self.bb_std.value) dataframe['bb_upper'] = bb[f'BBU_{self.bb_window.value}_{self.bb_std.value}'] dataframe['bb_middle'] = bb[f'BBM_{self.bb_window.value}_{self.bb_std.value}'] dataframe['bb_lower'] = bb[f'BBL_{self.bb_window.value}_{self.bb_std.value}'] dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle'] dataframe['bb_position'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) # Calculate ADX for trend strength adx = ta.adx(dataframe['high'], dataframe['low'], dataframe['close'], length=14) dataframe['adx'] = adx['ADX_14'] dataframe['plus_di'] = adx['DMP_14'] dataframe['minus_di'] = adx['DMN_14'] # Calculate Stochastic stoch = ta.stoch(dataframe['high'], dataframe['low'], dataframe['close'], k=14, d=3) dataframe['stoch_k'] = stoch['STOCHk_14_3_3'] dataframe['stoch_d'] = stoch['STOCHd_14_3_3'] # Calculate ATR for volatility and position sizing dataframe['atr'] = ta.atr(dataframe['high'], dataframe['low'], dataframe['close'], length=14) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] * 100 # Volume analysis dataframe['volume_sma'] = ta.sma(dataframe['volume'], length=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # Market regime detection dataframe['regime_ema'] = ta.ema(dataframe['close'], length=self.regime_ema_period.value) dataframe['market_regime'] = np.where(dataframe['close'] > dataframe['regime_ema'], 1, -1) # Calculate MACD for momentum macd = ta.macd(dataframe['close'], fast=12, slow=26, signal=9) dataframe['macd'] = macd['MACD_12_26_9'] dataframe['macd_signal'] = macd['MACDs_12_26_9'] dataframe['macd_hist'] = macd['MACDh_12_26_9'] # Add informative pair indicators (simplified approach) try: # Get informative pairs for market context btc_15m = self.dp.get_pair_dataframe("BTC/USDT", "15m") if not btc_15m.empty: btc_15m_rsi = ta.rsi(btc_15m['close'], length=14) dataframe['btc_rsi_15m'] = btc_15m_rsi.reindex(dataframe.index, method='ffill') except Exception: dataframe['btc_rsi_15m'] = 50 # Default neutral value try: btc_1h = self.dp.get_pair_dataframe("BTC/USDT", "1h") if not btc_1h.empty: btc_1h_adx = ta.adx(btc_1h['high'], btc_1h['low'], btc_1h['close'], length=14) dataframe['btc_adx_1h'] = btc_1h_adx['ADX_14'].reindex(dataframe.index, method='ffill') except Exception: dataframe['btc_adx_1h'] = 25 # Default neutral value try: eth_15m = self.dp.get_pair_dataframe("ETH/USDT", "15m") if not eth_15m.empty: eth_15m_rsi = ta.rsi(eth_15m['close'], length=14) dataframe['eth_rsi_15m'] = eth_15m_rsi.reindex(dataframe.index, method='ffill') except Exception: dataframe['eth_rsi_15m'] = 50 # Default neutral value return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Generate entry signals based on OPTIMIZED technical conditions """ # Long entry conditions - OPTIMIZED for maximum profitability long_conditions = [ # RSI conditions - OPTIMIZED range (dataframe['rsi'] >= self.buy_rsi_low.value) & (dataframe['rsi'] <= self.buy_rsi_high.value), # Trend confirmation - OPTIMIZED flexibility (dataframe['ema_fast'] >= dataframe['ema_slow'] * 0.99), # ADX - OPTIMIZED minimum trend strength (lower = more trades) (dataframe['adx'] >= self.adx_threshold.value), # Bollinger Bands - OPTIMIZED positioning (dataframe['bb_position'] <= 0.6), # Volume confirmation - OPTIMIZED (lower = more trades) (dataframe['volume_ratio'] >= self.volume_factor.value), # Stochastic - OPTIMIZED oversold condition (dataframe['stoch_k'] <= self.stoch_k_low.value * 1.1), # MACD - OPTIMIZED bullish condition (dataframe['macd'] >= dataframe['macd_signal'] * 0.98), # Market regime - OPTIMIZED (neutral to bullish) (dataframe['market_regime'] >= 0), # BTC market context - OPTIMIZED (dataframe['btc_rsi_15m'] <= 75), # Basic safety checks (dataframe['volume'] > 0), ] # Combine all long conditions dataframe.loc[ reduce(lambda x, y: x & y, long_conditions), ['enter_long', 'enter_tag'] ] = (1, 'quantum_pandas_long') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Generate exit signals based on OPTIMIZED technical conditions """ # Long exit conditions - OPTIMIZED exit_long_conditions = [ # RSI conditions - OPTIMIZED range (dataframe['rsi'] >= self.sell_rsi_low.value) & (dataframe['rsi'] <= self.sell_rsi_high.value), # Trend reversal - OPTIMIZED flexibility (dataframe['ema_fast'] <= dataframe['ema_slow'] * 1.01), # Bollinger Bands - OPTIMIZED positioning (dataframe['bb_position'] >= 0.4), # Stochastic - OPTIMIZED overbought condition (dataframe['stoch_k'] >= self.stoch_k_high.value * 0.9), # MACD - OPTIMIZED bearish condition (dataframe['macd'] <= dataframe['macd_signal'] * 1.02), # Volume confirmation - OPTIMIZED (dataframe['volume_ratio'] >= 0.8), # Basic safety checks (dataframe['volume'] > 0), ] # Combine all exit conditions dataframe.loc[ reduce(lambda x, y: x & y, exit_long_conditions), ['exit_long', 'exit_tag'] ] = (1, 'quantum_pandas_exit') return dataframe def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ Custom position sizing based on market volatility using OPTIMIZED ATR """ # Get the latest ATR data for volatility assessment dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return proposed_stake latest = dataframe.iloc[-1] atr_percent = latest['atr_percent'] # OPTIMIZED position sizing for volatility if atr_percent > 5.0: # High volatility (>5%) return proposed_stake * 0.5 elif atr_percent > 3.0: # Medium volatility (>3%) return proposed_stake * 0.75 else: # Low volatility (<=3%) return proposed_stake def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ Additional confirmation before entering a trade - OPTIMIZED """ # Get latest data for final confirmation dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return True latest = dataframe.iloc[-1] # OPTIMIZED safety checks # Avoid trading in extreme volatility if latest['atr_percent'] > 8.0: return False # Avoid trading if volume is too low if latest['volume_ratio'] < 0.5: return False # Avoid trading if spread is too high (indicating low liquidity) if self.dp.runmode.value in ('live', 'dry_run'): try: orderbook = self.dp.orderbook(pair, 1) spread = (orderbook['asks'][0][0] - orderbook['bids'][0][0]) / orderbook['bids'][0][0] if spread > 0.01: # >1% spread return False except: pass return True def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ Dynamic leverage based on market conditions - OPTIMIZED """ # Get latest data for leverage decision dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return 1.0 latest = dataframe.iloc[-1] # OPTIMIZED leverage based on volatility if latest['atr_percent'] > 4.0: return 1.0 # No leverage in high volatility elif latest['atr_percent'] > 2.5: return min(2.0, max_leverage) # Low leverage in medium volatility else: return min(3.0, max_leverage) # Higher leverage in low volatility def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Adaptive stoploss using ATR and current profit. - In high volatility, cut losers earlier (~1% to 1.5%). - If already in profit, tighten aggressively to protect gains. Returns negative ratio (e.g., -0.01 for -1%). """ try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or dataframe.empty: return self.stoploss latest = dataframe.iloc[-1] atr_percent = float(latest.get('atr_percent', 3.0)) except Exception: atr_percent = 3.0 # Base stoploss from volatility if atr_percent > 4.0: base_sl = -0.01 # very high vol -> cut early elif atr_percent > 3.0: base_sl = -0.012 # high vol else: base_sl = -0.015 # moderate/low vol # If trade is in profit, tighten further if current_profit > 0.01: # >1% in profit return min(base_sl, -0.008) return base_sl