# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime, timedelta from typing import Optional, Union import logging from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, IStrategy, merge_informative_pair) import talib.abstract as ta from functools import reduce logger = logging.getLogger(__name__) # ================================================================================================ # Multitimeframe v9.3-RSI36 Strategy - COMPLETE MIGRATION FROM JESSE # ================================================================================================ class Multitimeframe_v93_Complete(IStrategy): """ Multi-Timeframe Strategy v9.3-RSI36 - ELITE Performance COMPLETE MIGRATION from Jesse Trading Framework Optimized Parameters: - Break-Even: 1.35R (v9.2 breakthrough) - RSI Long: 36 (v9.3 breakthrough - early entries) - RSI Short: 64 (v9.3 - symmetry with LONG) - Take Profit: 3.0R (v9.1 optimal) Validated Results (2023-2025, 2.78 years): - Net Profit: +110.68% - Annual Return: 30.8% - Calmar Ratio: 1.55 (ELITE > 1.5) - Max DD: -19.93% - Win Rate: 25.14% - Sharpe: 1.09 WARNING: Temporal Overfitting - Works EXCELLENT in 2022-2025 (high volatility regime) - FAILS in 2020-2021 (parabolic bull regime) - Use circuit breakers and regime monitoring Complete Features: ✅ Score system with weighted signals ✅ Fair Value Gap (FVG) detection ✅ RSI Divergence detection (1H) ✅ Cooldown between trades (60min) ✅ Daily loss limit (3%) ✅ Multi-timeframe analysis (15m/1h/4h) ✅ Custom break-even logic (1.35R) ✅ Custom take-profit (3.0R) """ # Strategy interface version INTERFACE_VERSION = 3 # Can this strategy go short? can_short: bool = True # Minimal ROI designed for the strategy. minimal_roi = { "0": 10.0 # Effectively disabled, using custom_exit } # Optimal stoploss designed for the strategy. stoploss = -0.99 # Wide stoploss - actual SL managed in custom_exit() # Trailing stoploss trailing_stop = False # Enable custom stoploss use_custom_stoploss = False # Disabled - SL managed in custom_exit() # 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 = False ignore_roi_if_entry_signal = True # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 300 # Optional order type mapping. order_types = { 'entry': 'market', 'exit': 'market', 'stoploss': 'market', 'stoploss_on_exchange': False } # Optional order time in force. order_time_in_force = { 'entry': 'GTC', 'exit': 'GTC' } # ======================================================================== # OPTIMIZED PARAMETERS v9.3-RSI36 # ======================================================================== # v9.3-RSI36: RSI threshold for LONG (breakthrough #2) rsi_long_threshold = IntParameter(30, 40, default=36, space='buy') # v9.3-RSI36: RSI threshold for SHORT (symmetry with LONG) rsi_short_threshold = IntParameter(60, 70, default=64, space='sell') # v9.2: Break-even ratio (breakthrough #1) break_even_ratio = DecimalParameter(1.0, 2.0, default=1.35, decimals=2, space='sell') # v9.1: Take profit final ratio (already optimal) tp_final_ratio = DecimalParameter(2.0, 5.0, default=3.0, decimals=1, space='sell') # Risk management risk_percent = 1.5 # 1.5% risk per trade atr_period = 14 atr_multiplier = 1.0 # SL = entry ± 1.0*ATR # Filters min_atr_pct = DecimalParameter(0.001, 0.01, default=0.004, decimals=3, space='buy') signal_cooldown_minutes = IntParameter(30, 120, default=60, space='buy') minimum_score = IntParameter(2, 5, default=3, space='buy') # FVG parameters fvg_lookback = IntParameter(10, 30, default=20, space='buy') # ADX threshold (relaxed) adx_threshold = IntParameter(10, 25, default=15, space='buy') # Daily loss limit max_daily_loss_pct = DecimalParameter(1.0, 5.0, default=3.0, decimals=1, space='sell') # CRITICAL: Store entry ATR and break-even status per trade ID # This dict persists for the lifetime of the strategy instance trade_state = {} # Custom variables to track state between candles custom_info = {} def bot_start(self, **kwargs) -> None: """ Called only once after bot instantiation. Initialize custom state tracking. """ self.custom_info = { 'last_signal_time': {}, # Per pair 'daily_start_balance': {}, # Per pair 'daily_loss_check_day': {}, # Per pair } def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. """ pairs = self.dp.current_whitelist() informative_pairs = [] for pair in pairs: informative_pairs.append((pair, '1h')) informative_pairs.append((pair, '4h')) return informative_pairs def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame """ # 15m timeframe indicators dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period) dataframe['atr_pct'] = (dataframe['atr'] / dataframe['close']) * 100 # MACD 15m macd = ta.MACD(dataframe, fastperiod=15, slowperiod=30, signalperiod=9) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # EMA 15m dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200) # Volume dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) # ADX dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) # Fair Value Gaps dataframe['bullish_fvg'] = self.detect_bullish_fvg(dataframe) dataframe['bearish_fvg'] = self.detect_bearish_fvg(dataframe) # 1h timeframe inf_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='1h') if not inf_1h.empty: inf_1h['rsi_1h'] = ta.RSI(inf_1h, timeperiod=14) macd_1h = ta.MACD(inf_1h, fastperiod=15, slowperiod=30, signalperiod=9) inf_1h['macd_1h'] = macd_1h['macd'] inf_1h['macd_signal_1h'] = macd_1h['macdsignal'] # Divergences in 1h inf_1h['bullish_div_1h'] = self.detect_bullish_divergence_1h(inf_1h) inf_1h['bearish_div_1h'] = self.detect_bearish_divergence_1h(inf_1h) # Merge 1h indicators into 15m dataframe dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, '1h', ffill=True) # 4h timeframe inf_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='4h') if not inf_4h.empty: inf_4h['ema_50_4h'] = ta.EMA(inf_4h, timeperiod=50) inf_4h['ema_200_4h'] = ta.EMA(inf_4h, timeperiod=200) inf_4h['adx_4h'] = ta.ADX(inf_4h, timeperiod=14) # Merge 4h indicators into 15m dataframe dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, '4h', ffill=True) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the entry signal for the given dataframe """ pair = metadata['pair'] conditions_long = [] conditions_short = [] # ======= LONG CONDITIONS ======= # Cooldown filter if pair in self.custom_info.get('last_signal_time', {}): last_signal = self.custom_info['last_signal_time'][pair] dataframe['cooldown_ok_long'] = (dataframe['date'] - last_signal).dt.total_seconds() >= (self.signal_cooldown_minutes.value * 60) else: dataframe['cooldown_ok_long'] = True conditions_long.append(dataframe['cooldown_ok_long']) # Volatility filter conditions_long.append(dataframe['atr_pct'] >= self.min_atr_pct.value) # Calculate score dataframe['score_long'] = 0 # RSI oversold 15m = +1 point dataframe.loc[dataframe['rsi'] < self.rsi_long_threshold.value, 'score_long'] += 1 # MACD bullish 15m = +1 point dataframe.loc[dataframe['macd'] > dataframe['macd_signal'], 'score_long'] += 1 # FVG bullish = +2 points (PREMIUM SIGNAL) dataframe.loc[dataframe['bullish_fvg'] == True, 'score_long'] += 2 # MACD bullish 1h = +1 point (if available) if 'macd_1h' in dataframe.columns and 'macd_signal_1h' in dataframe.columns: dataframe.loc[dataframe['macd_1h'] > dataframe['macd_signal_1h'], 'score_long'] += 1 # Bullish divergence 1h = +2 points (PREMIUM SIGNAL) if 'bullish_div_1h' in dataframe.columns: dataframe.loc[dataframe['bullish_div_1h'] == True, 'score_long'] += 2 # Score threshold conditions_long.append(dataframe['score_long'] >= self.minimum_score.value) # Not in SHORT zone conditions_long.append(dataframe['rsi'] <= self.rsi_short_threshold.value) # Volume confirmation conditions_long.append(dataframe['volume'] > dataframe['volume_sma'] * 0.8) if conditions_long: long_signals = reduce(lambda x, y: x & y, conditions_long) # Log signal details when new signal is detected if long_signals.iloc[-1]: last_row = dataframe.iloc[-1] logger.info(f"🔵 LONG SIGNAL DETECTED - {pair}") logger.info(f" Score: {int(last_row['score_long'])}/{self.minimum_score.value} ✅") logger.info(f" 📊 Signal Breakdown:") if last_row['rsi'] < self.rsi_long_threshold.value: logger.info(f" + RSI 15m: {last_row['rsi']:.1f} < {self.rsi_long_threshold.value} (+1 point)") if last_row['macd'] > last_row['macd_signal']: logger.info(f" + MACD 15m: Bullish (+1 point)") if last_row.get('bullish_fvg', False): logger.info(f" + FVG Bullish detected (+2 points) 🎯") if 'macd_1h' in dataframe.columns and last_row.get('macd_1h', 0) > last_row.get('macd_signal_1h', 0): logger.info(f" + MACD 1h: Bullish (+1 point)") if last_row.get('bullish_div_1h', False): logger.info(f" + Divergence Bullish 1h detected (+2 points) 🎯") logger.info(f" Price: ${last_row['close']:.2f}") logger.info(f" ATR: ${last_row['atr']:.2f} ({last_row['atr_pct']:.2f}%)") dataframe.loc[long_signals, 'enter_long'] = 1 # ======= SHORT CONDITIONS ======= # Cooldown filter if pair in self.custom_info.get('last_signal_time', {}): last_signal = self.custom_info['last_signal_time'][pair] dataframe['cooldown_ok_short'] = (dataframe['date'] - last_signal).dt.total_seconds() >= (self.signal_cooldown_minutes.value * 60) else: dataframe['cooldown_ok_short'] = True conditions_short.append(dataframe['cooldown_ok_short']) # Volatility filter conditions_short.append(dataframe['atr_pct'] >= self.min_atr_pct.value) # Calculate score dataframe['score_short'] = 0 # RSI overbought 15m = +1 point dataframe.loc[dataframe['rsi'] > self.rsi_short_threshold.value, 'score_short'] += 1 # MACD bearish 15m = +1 point dataframe.loc[dataframe['macd'] < dataframe['macd_signal'], 'score_short'] += 1 # FVG bearish = +2 points (PREMIUM SIGNAL) dataframe.loc[dataframe['bearish_fvg'] == True, 'score_short'] += 2 # MACD bearish 1h = +1 point (if available) if 'macd_1h' in dataframe.columns and 'macd_signal_1h' in dataframe.columns: dataframe.loc[dataframe['macd_1h'] < dataframe['macd_signal_1h'], 'score_short'] += 1 # Bearish divergence 1h = +2 points (PREMIUM SIGNAL) if 'bearish_div_1h' in dataframe.columns: dataframe.loc[dataframe['bearish_div_1h'] == True, 'score_short'] += 2 # Score threshold conditions_short.append(dataframe['score_short'] >= self.minimum_score.value) # Not in LONG zone conditions_short.append(dataframe['rsi'] >= self.rsi_long_threshold.value) # Volume confirmation conditions_short.append(dataframe['volume'] > dataframe['volume_sma'] * 0.8) if conditions_short: short_signals = reduce(lambda x, y: x & y, conditions_short) # Log signal details when new signal is detected if short_signals.iloc[-1]: last_row = dataframe.iloc[-1] logger.info(f"🔴 SHORT SIGNAL DETECTED - {pair}") logger.info(f" Score: {int(last_row['score_short'])}/{self.minimum_score.value} ✅") logger.info(f" 📊 Signal Breakdown:") if last_row['rsi'] > self.rsi_short_threshold.value: logger.info(f" + RSI 15m: {last_row['rsi']:.1f} > {self.rsi_short_threshold.value} (+1 point)") if last_row['macd'] < last_row['macd_signal']: logger.info(f" + MACD 15m: Bearish (+1 point)") if last_row.get('bearish_fvg', False): logger.info(f" + FVG Bearish detected (+2 points) 🎯") if 'macd_1h' in dataframe.columns and last_row.get('macd_1h', 0) < last_row.get('macd_signal_1h', 0): logger.info(f" + MACD 1h: Bearish (+1 point)") if last_row.get('bearish_div_1h', False): logger.info(f" + Divergence Bearish 1h detected (+2 points) 🎯") logger.info(f" Price: ${last_row['close']:.2f}") logger.info(f" ATR: ${last_row['atr']:.2f} ({last_row['atr_pct']:.2f}%)") dataframe.loc[short_signals, 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ We use custom_exit for all exits, so no exit signals here """ dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 return dataframe 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: """ Called right before placing a entry order. Update last signal time for cooldown tracking. """ if 'last_signal_time' not in self.custom_info: self.custom_info['last_signal_time'] = {} self.custom_info['last_signal_time'][pair] = current_time # Check daily loss limit if not self.can_trade_today(pair, current_time): return False return True def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> Optional[float]: """ Custom stoploss logic EXACTLY matching Jesse's update_position(): - Initial SL at entry ± ATR (FIXED, set once) - Move to break-even at 1.35R (ONE TIME ONLY) - Returns ABSOLUTE DISTANCE from current_rate in ratio format CRITICAL: Freqtrade expects a NEGATIVE ratio for stoploss distance: - For LONG: return -(sl_price - current_rate) / current_rate - For SHORT: return (sl_price - current_rate) / current_rate """ trade_id = trade.id entry_price = trade.open_rate # Get dataframe dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None # ALWAYS recalculate entry ATR from historical data entry_candles = dataframe[dataframe['date'] <= pd.to_datetime(trade.open_date_utc)] if not entry_candles.empty: entry_atr = entry_candles.iloc[-1]['atr'] else: entry_atr = dataframe.iloc[-1]['atr'] logger.warning(f"Trade {trade_id}: Entry candle not found, using current ATR") # Initialize trade state on first call if trade_id not in self.trade_state: sl_distance_from_entry = abs(trade.stop_loss - entry_price) be_was_already_hit = sl_distance_from_entry < (entry_atr * 0.5) logger.info(f"Trade {trade_id} INIT: entry={entry_price}, entry_atr={entry_atr}") self.trade_state[trade_id] = { 'entry_atr': entry_atr, 'break_even_hit': be_was_already_hit, } # Use FIXED entry ATR atr = self.trade_state[trade_id]['entry_atr'] # Calculate FIXED initial SL price if trade.is_short: initial_sl_price = entry_price + atr be_sl_price = entry_price * 1.001 # 0.1% above entry else: # is_long initial_sl_price = entry_price - atr be_sl_price = entry_price * 0.999 # 0.1% below entry # Calculate R-ratio if trade.is_short: profit_distance = entry_price - current_rate else: profit_distance = current_rate - entry_price risk_distance = abs(initial_sl_price - entry_price) r_ratio = profit_distance / risk_distance if risk_distance > 0 else 0 # Determine which SL price to use if r_ratio >= self.break_even_ratio.value and not self.trade_state[trade_id]['break_even_hit']: # Move to break-even at 1.35R (ONE TIME ONLY) self.trade_state[trade_id]['break_even_hit'] = True sl_price = be_sl_price logger.info(f"🎯 Trade {trade_id} reached {r_ratio:.2f}R - Moving to BREAK-EVEN at {be_sl_price}") else: # Use initial SL sl_price = initial_sl_price # Return the stoploss as a ratio from ENTRY PRICE (to keep it fixed) # CRITICAL: Freqtrade ALWAYS expects NEGATIVE values from custom_stoploss # The absolute value determines the distance, sign is always negative if trade.is_short: # SHORT: SL = entry + ATR (price goes UP = loss) # Return NEGATIVE ratio: -(sl_price - entry) / entry sl_ratio = -abs((sl_price - entry_price) / entry_price) else: # LONG: SL = entry - ATR (price goes DOWN = loss) # Return NEGATIVE ratio: -(entry - sl_price) / entry sl_ratio = -abs((entry_price - sl_price) / entry_price) logger.info(f"Trade {trade_id}: SL={sl_price:.2f}, ratio={sl_ratio:.4f}, R={r_ratio:.2f}") return sl_ratio def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ Custom exit logic - EXACTLY like Jesse's update_position(): - SL at entry +/- ATR (FIXED) - Move SL to break-even at 1.35R - TP at 3.0R """ trade_id = trade.id entry_price = trade.open_rate # Get dataframe for ATR dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None # Initialize trade state on first call if trade_id not in self.trade_state: # Get ATR from entry candle entry_candles = dataframe[dataframe['date'] <= pd.to_datetime(trade.open_date_utc)] if not entry_candles.empty: entry_atr = entry_candles.iloc[-1]['atr'] else: entry_atr = dataframe.iloc[-1]['atr'] self.trade_state[trade_id] = { 'entry_atr': entry_atr, 'break_even_hit': False, 'sl_price': entry_price - entry_atr if not trade.is_short else entry_price + entry_atr } logger.info(f"Trade {trade_id} INIT: entry={entry_price}, ATR={entry_atr:.2f}, SL={self.trade_state[trade_id]['sl_price']:.2f}") # Get stored values atr = self.trade_state[trade_id]['entry_atr'] sl_price = self.trade_state[trade_id]['sl_price'] # Calculate R-ratio if trade.is_short: profit_distance = entry_price - current_rate risk_distance = atr else: profit_distance = current_rate - entry_price risk_distance = atr r_ratio = profit_distance / risk_distance if risk_distance > 0 else 0 # LOG current status logger.info(f"Trade {trade_id}: Price={current_rate:.2f}, SL={sl_price:.2f}, R={r_ratio:.2f}") # === STOP LOSS CHECK === if trade.is_short: # SHORT: SL triggers when price goes UP above sl_price if current_rate >= sl_price: logger.info(f"Trade {trade_id}: STOP LOSS HIT at {current_rate:.2f} (SL={sl_price:.2f})") if trade_id in self.trade_state: del self.trade_state[trade_id] return 'stop_loss_hit' else: # LONG: SL triggers when price goes DOWN below sl_price if current_rate <= sl_price: logger.info(f"Trade {trade_id}: STOP LOSS HIT at {current_rate:.2f} (SL={sl_price:.2f})") if trade_id in self.trade_state: del self.trade_state[trade_id] return 'stop_loss_hit' # === BREAK-EVEN at 1.35R === if r_ratio >= self.break_even_ratio.value and not self.trade_state[trade_id]['break_even_hit']: self.trade_state[trade_id]['break_even_hit'] = True # Move SL to entry price (break-even) self.trade_state[trade_id]['sl_price'] = entry_price logger.info(f"Trade {trade_id}: BREAK-EVEN activated at {r_ratio:.2f}R - SL moved to {entry_price:.2f}") # === TAKE PROFIT at 3.0R === if r_ratio >= self.tp_final_ratio.value: logger.info(f"Trade {trade_id}: TAKE PROFIT HIT at {r_ratio:.2f}R") if trade_id in self.trade_state: del self.trade_state[trade_id] return 'tp_3.0R_hit' return None # ======================================================================== # HELPER FUNCTIONS # ======================================================================== def can_trade_today(self, pair: str, current_time: datetime) -> bool: """ Check if daily loss limit has been reached """ if 'daily_start_balance' not in self.custom_info: self.custom_info['daily_start_balance'] = {} if 'daily_loss_check_day' not in self.custom_info: self.custom_info['daily_loss_check_day'] = {} day_key = current_time.date() last_day = self.custom_info['daily_loss_check_day'].get(pair, None) # New day: reset balance if last_day != day_key: self.custom_info['daily_loss_check_day'][pair] = day_key # Get current balance from wallet try: wallets = self.wallets.get_all_balances() stake_currency = self.config['stake_currency'] current_balance = wallets.get(stake_currency, {}).get('total', 10000) self.custom_info['daily_start_balance'][pair] = current_balance except: self.custom_info['daily_start_balance'][pair] = 10000 # Calculate daily loss daily_start = self.custom_info['daily_start_balance'].get(pair, 10000) if daily_start <= 0: return True try: wallets = self.wallets.get_all_balances() stake_currency = self.config['stake_currency'] current_balance = wallets.get(stake_currency, {}).get('total', 10000) except: current_balance = 10000 daily_loss_pct = ((daily_start - current_balance) / daily_start) * 100 # Allow trading if not exceeded limit return daily_loss_pct < self.max_daily_loss_pct.value def calculate_rsi_array(self, prices: pd.Series, period: int = 14) -> pd.Series: """ Calculate RSI manually for divergence detection """ if len(prices) < period + 1: return pd.Series([50.0] * len(prices)) # Calculate price changes deltas = prices.diff() # Separate gains and losses gains = deltas.where(deltas > 0, 0.0) losses = -deltas.where(deltas < 0, 0.0) # Calculate average gains/losses using EMA (Wilder's smoothing) avg_gains = gains.ewm(com=period-1, min_periods=period).mean() avg_losses = losses.ewm(com=period-1, min_periods=period).mean() # Calculate RS rs = avg_gains / (avg_losses + 1e-10) # Calculate RSI rsi = 100 - (100 / (1 + rs)) return rsi.fillna(50.0) def detect_bullish_fvg(self, dataframe: DataFrame) -> pd.Series: """ Detect Bullish Fair Value Gaps FVG occurs when candle[i-2].high < candle[i].low (gap between candles) """ bullish_fvg = pd.Series(False, index=dataframe.index) if len(dataframe) < self.fvg_lookback.value + 3: return bullish_fvg for i in range(len(dataframe) - 1, max(2, len(dataframe) - self.fvg_lookback.value), -1): if i < 2: continue high_old = dataframe.iloc[i-2]['high'] low_new = dataframe.iloc[i]['low'] # Gap detected? if high_old < low_new: gap_top = low_new gap_bottom = high_old gap_mid = (gap_top + gap_bottom) / 2 gap_size_pct = ((gap_top - gap_bottom) / gap_bottom) * 100 # Gap must be significant (at least 0.1%) if gap_size_pct < 0.1: continue # Check if current price is retesting the FVG current_close = dataframe.iloc[-1]['close'] price_to_gap_mid_pct = abs(current_close - gap_mid) / gap_mid * 100 # Within or near the gap (±0.5%) if price_to_gap_mid_pct < 0.5 or (current_close >= gap_bottom and current_close <= gap_top): bullish_fvg.iloc[-1] = True break return bullish_fvg def detect_bearish_fvg(self, dataframe: DataFrame) -> pd.Series: """ Detect Bearish Fair Value Gaps FVG occurs when candle[i-2].low > candle[i].high (gap between candles) """ bearish_fvg = pd.Series(False, index=dataframe.index) if len(dataframe) < self.fvg_lookback.value + 3: return bearish_fvg for i in range(len(dataframe) - 1, max(2, len(dataframe) - self.fvg_lookback.value), -1): if i < 2: continue low_old = dataframe.iloc[i-2]['low'] high_new = dataframe.iloc[i]['high'] # Gap detected? if low_old > high_new: gap_top = low_old gap_bottom = high_new gap_mid = (gap_top + gap_bottom) / 2 gap_size_pct = ((gap_top - gap_bottom) / gap_bottom) * 100 # Gap must be significant (at least 0.1%) if gap_size_pct < 0.1: continue # Check if current price is retesting the FVG current_close = dataframe.iloc[-1]['close'] price_to_gap_mid_pct = abs(current_close - gap_mid) / gap_mid * 100 # Within or near the gap (±0.5%) if price_to_gap_mid_pct < 0.5 or (current_close >= gap_bottom and current_close <= gap_top): bearish_fvg.iloc[-1] = True break return bearish_fvg def detect_bullish_divergence_1h(self, dataframe: DataFrame) -> pd.Series: """ Detect Bullish RSI Divergence in 1H timeframe Price makes lower low, but RSI makes higher low (reversal signal) """ bullish_div = pd.Series(False, index=dataframe.index) if len(dataframe) < 100: return bullish_div # Calculate RSI rsi = self.calculate_rsi_array(dataframe['close'], 14) # Find local lows in last 60 candles lookback = 60 price_lows = [] for i in range(len(dataframe) - 5, max(5, len(dataframe) - lookback), -1): # Local low (compare with 2 candles on each side) if (dataframe.iloc[i]['low'] <= dataframe.iloc[i-1]['low'] and dataframe.iloc[i]['low'] <= dataframe.iloc[i-2]['low'] and dataframe.iloc[i]['low'] <= dataframe.iloc[i+1]['low'] and dataframe.iloc[i]['low'] <= dataframe.iloc[i+2]['low']): price_lows.append((i, dataframe.iloc[i]['low'], rsi.iloc[i])) if len(price_lows) < 2: return bullish_div # Search for divergence: price lower, RSI higher for j in range(len(price_lows) - 1): idx_recent, price_recent, rsi_recent = price_lows[j] for k in range(j + 1, len(price_lows)): idx_old, price_old, rsi_old = price_lows[k] # Bullish divergence if price_recent < price_old and rsi_recent > rsi_old: price_diff_pct = abs((price_recent - price_old) / price_old) * 100 rsi_diff = abs(rsi_recent - rsi_old) # Thresholds: 0.2% price + 2 RSI points if price_diff_pct >= 0.2 and rsi_diff >= 2: bullish_div.iloc[-1] = True return bullish_div return bullish_div def detect_bearish_divergence_1h(self, dataframe: DataFrame) -> pd.Series: """ Detect Bearish RSI Divergence in 1H timeframe Price makes higher high, but RSI makes lower high (reversal signal) """ bearish_div = pd.Series(False, index=dataframe.index) if len(dataframe) < 100: return bearish_div # Calculate RSI rsi = self.calculate_rsi_array(dataframe['close'], 14) # Find local highs in last 60 candles lookback = 60 price_highs = [] for i in range(len(dataframe) - 5, max(5, len(dataframe) - lookback), -1): # Local high (compare with 2 candles on each side) if (dataframe.iloc[i]['high'] >= dataframe.iloc[i-1]['high'] and dataframe.iloc[i]['high'] >= dataframe.iloc[i-2]['high'] and dataframe.iloc[i]['high'] >= dataframe.iloc[i+1]['high'] and dataframe.iloc[i]['high'] >= dataframe.iloc[i+2]['high']): price_highs.append((i, dataframe.iloc[i]['high'], rsi.iloc[i])) if len(price_highs) < 2: return bearish_div # Search for divergence: price higher, RSI lower for j in range(len(price_highs) - 1): idx_recent, price_recent, rsi_recent = price_highs[j] for k in range(j + 1, len(price_highs)): idx_old, price_old, rsi_old = price_highs[k] # Bearish divergence if price_recent > price_old and rsi_recent < rsi_old: price_diff_pct = abs((price_recent - price_old) / price_old) * 100 rsi_diff = abs(rsi_recent - rsi_old) # Thresholds: 0.2% price + 2 RSI points if price_diff_pct >= 0.2 and rsi_diff >= 2: bearish_div.iloc[-1] = True return bearish_div return bearish_div 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: """ Calculate stake amount based on 1.5% risk (consistent with Jesse backtest) Jesse strategy uses: - Risk per trade: 1.5% of balance ($150 if balance = $10,000) - Position size calculated based on ATR and stop distance - This function replicates that logic in Freqtrade """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return proposed_stake # ATR for calculating stop distance (same as Jesse) atr = dataframe['atr'].iloc[-1] stop_distance_price = atr * 3.5 # Same SL multiplier as Jesse # Available balance balance = self.wallets.get_total(self.config['stake_currency']) # Risk target: 1.5% of balance (same as Jesse) risk_amount = balance * 0.015 # $150 if balance = $10,000 # Position value to achieve that risk # risk = position_value × (stop_distance / price) # → position_value = risk / (stop_distance / price) stop_distance_pct = stop_distance_price / current_rate # Prevent division by zero if stop_distance_pct <= 0: return proposed_stake stake_amount = risk_amount / stop_distance_pct # Limit to maximum allowed (with leverage consideration) max_stake_with_leverage = max_stake * leverage if leverage > 1 else max_stake stake_amount = min(stake_amount, max_stake_with_leverage) # Ensure minimum stake if min_stake and stake_amount < min_stake: stake_amount = min_stake return stake_amount