# source: https://raw.githubusercontent.com/tradingstrategy-ai/gmx-ccxt-freqtrade/32865acf8d19c7e8d51cf1e1c63dbfd5e3617e7d/user_data/strategies/IchiV2_LS_Live.py # Github_tradingstrategy_ai_gmx_ccxt_freqtrade__IchiV2_LS_Live__20251230_182600 - Dual Long/Short Ichimoku Strategy (1h timeframe) # Author: Gui # Version: 2.0 # Description: Combines cloud crossover (long) with senkou B retest breakdown (short) from freqtrade.strategy.interface import IStrategy from pandas import DataFrame import talib.abstract as ta import pandas as pd pd.options.mode.chained_assignment = None import technical.indicators as ftt import freqtrade.vendor.qtpylib.indicators as qtpylib from datetime import datetime from freqtrade.persistence import Trade from freqtrade.enums import RunMode from freqtrade.strategy import informative, DecimalParameter, IntParameter from freqtrade.strategy.strategy_helper import stoploss_from_absolute import logging import json from freqtrade.optimize.space import SKDecimal class Github_tradingstrategy_ai_gmx_ccxt_freqtrade__IchiV2_LS_Live__20251230_182600(IStrategy): """ Github_tradingstrategy_ai_gmx_ccxt_freqtrade__IchiV2_LS_Live__20251230_182600 - for live config bots Dual long/short Ichimoku strategy on 1h timeframe. Long entries (checked at 4h boundary): - 4h close crosses above senkou_a_4h - Cloud is green (senkou_a_4h > senkou_b_4h) - SAR is below close (sar_4h < close_4h) Long exits: - 1h close crosses below sar_4h (exit on SAR cross) - Base stoploss at -0.2 as fallback - No ATR stop Short entries (checked on any 1h candle): - Cloud is bearish (senkou_a < senkou_b) - Price retested senkou_b (high >= senkou_b) within lookback window - Current price breaks down below senkou_b Short exits: - RSI 4h oversold (< 25) - OR price velocity exhaustion (< -10%) - ATR-based stop loss (3.5x ATR from entry, locked at entry) - No SAR exit for shorts Standard Freqtrade position sizing (no market cap scaling). """ def __init__(self, config): super().__init__(config) self.logger = logging.getLogger(__name__) self._pairs_analyzed_count = 0 def _log_jsonl(self, signal_type: str, pair: str = None, conditions: dict = None, indicators: dict = None, trade_data: dict = None, **extra_kwargs): """ Universal JSONL logger for strategy analysis. :param signal_type: Type of signal/analysis (e.g., 'long_entry_check', 'short_exit_check') :param pair: Trading pair :param conditions: Dict of conditions being evaluated {name: bool} :param indicators: Dict of indicator values {name: float} :param trade_data: Dict of trade-specific data (trade_id, profit, etc.) :param extra_kwargs: Any additional key-value pairs to log """ import numpy as np def convert_to_serializable(obj, max_decimals=4): """Convert numpy/pandas types to native Python types with decimal precision limit.""" if isinstance(obj, (np.integer, np.floating)): return round(float(obj), max_decimals) elif isinstance(obj, np.bool_): return bool(obj) elif isinstance(obj, float): return round(obj, max_decimals) elif isinstance(obj, dict): return {k: convert_to_serializable(v, max_decimals) for k, v in obj.items()} elif isinstance(obj, (list, tuple)): return [convert_to_serializable(item, max_decimals) for item in obj] elif pd.isna(obj): return None return obj log_entry = { "timestamp": datetime.now().isoformat(), "signal_type": signal_type, } # Add pair if provided if pair: log_entry["pair"] = pair # Add conditions if provided if conditions: log_entry["conditions"] = convert_to_serializable(conditions) # Add condition summary log_entry["conditions_passed"] = sum(1 for v in conditions.values() if v) log_entry["conditions_total"] = len(conditions) log_entry["all_conditions_met"] = all(conditions.values()) # Add indicators if provided if indicators: log_entry["indicators"] = convert_to_serializable(indicators) # Add trade data if provided if trade_data: log_entry["trade"] = convert_to_serializable(trade_data) # Add any extra kwargs log_entry.update(convert_to_serializable(extra_kwargs)) # Log as single-line JSON (no prefix for easy parsing) self.logger.info(json.dumps(log_entry)) def _calculate_atr_stop_price(self, current_price: float, atr: float, is_short: bool) -> dict: """ Calculate ATR-based stop loss price and related metrics. :param current_price: Current market price :param atr: Current ATR value :param is_short: True if short position, False if long :return: Dict with stop price and distance info """ atr_multiplier = self.atr_stop_multiplier.value atr_distance = atr_multiplier * atr if is_short: # Short: stop is ABOVE entry (price + distance) stop_price = current_price + atr_distance stop_pct = (atr_distance / current_price) * 100 else: # Long: stop is BELOW entry (price - distance) stop_price = current_price - atr_distance stop_pct = (atr_distance / current_price) * 100 return { 'atr_stop_price': stop_price, 'atr_stop_distance': atr_distance, 'atr_stop_pct': stop_pct, 'atr_multiplier': atr_multiplier, } INTERFACE_VERSION = 3 can_short = True # Set to False to test long-only performance # Hyperoptable parameters buy_params = { # Short entry (RetestSenkou) 'retest_lookback_window': 24, # Max candles to look for retest (24h) 'rsi_tp_threshold': 25, # RSI oversold threshold for shorts 'velocity_tp_threshold': -0.1, # Velocity threshold for shorts 'price_velocity_window': 10, # Velocity calculation window } # Short entry parameters retest_lookback_window = IntParameter(4, 96, default=24, space='buy', optimize=False) # Short exit parameters rsi_tp_threshold = IntParameter(15, 35, default=25, space='sell', optimize=False) use_rsi_exit = IntParameter(0, 1, default=1, space='sell', optimize=False) # ATR stop loss for shorts atr_stop_multiplier = DecimalParameter(1.5, 4.0, default=3.5, decimals=1, space='sell', optimize=False) atr_timeperiod = IntParameter(2, 20, default=14, space='sell', optimize=False) # Price velocity window velocity_tp_threshold = DecimalParameter(-0.13, -0.06, default=-0.1, decimals=2, space='sell', optimize=True) price_velocity_window = IntParameter(4, 20, default=20, space='sell', optimize=False) sell_params = {} custom_info = {} # ROI table - minimal minimal_roi = { "0": 10.0 # Disabled } # Stop loss # -0.2 for longs (only SAR exit matters) # Shorts use ATR-based custom stoploss stoploss = -0.129 use_custom_stoploss = True # Only applies to shorts (returns None for longs) # Timeframe timeframe = '1h' startup_candle_count = 500 process_only_new_candles = True # Trailing stop - DISABLED trailing_stop = False trailing_stop_positive = None trailing_stop_positive_offset = None trailing_only_offset_is_reached = False # Exit settings use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True ICHI_PARAMS = dict( conversion_line_period=20, base_line_periods=60, laggin_span=120, displacement=1, ) plot_config = { 'main_plot': { 'sar_4h': {'color': 'purple', 'marker': '.', 'linestyle': 'none'}, 'senkou_a_4h': {'color': 'lightgreen', 'linestyle': '--'}, 'senkou_b_4h': {'color': 'lightcoral', 'linestyle': '--'}, # 4h Bollinger Bands # 'bb_upper_4h': {'color': 'lightblue'}, # 'bb_middle_4h': {'color': 'lightgray'}, # 'bb_lower_4h': {'color': 'lightblue'}, }, 'subplots': { 'RSI 4h': { 'rsi_4h': {'color': 'purple'}, }, 'Price Velocity': { 'price_velocity': {'color': 'magenta'}, }, }, } def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float | None: """ ATR-based stop loss for SHORTS only. Returns the TIGHTER of ATR-based or base stoploss. """ if not trade.is_short: return None entry_atr = trade.get_custom_data(key='entry_atr') if entry_atr is None: custom_info_pair = self.custom_info.get(pair) if custom_info_pair is not None: try: entry_candle = custom_info_pair[custom_info_pair.index <= trade.open_date_utc] if not entry_candle.empty: entry_atr = entry_candle.iloc[-1]['atr'] if pd.isna(entry_atr): return None else: return None except (KeyError, IndexError): return None else: return None entry_rate = trade.open_rate atr_distance = self.atr_stop_multiplier.value * float(entry_atr) atr_stop_price = entry_rate + atr_distance fixed_stop_price = entry_rate * (1 + abs(self.stoploss)) # Use the TIGHTER stop loss between atr and fixed stoploss final_stop_price = min(atr_stop_price, fixed_stop_price) # Convert absolute price to stoploss distance using Freqtrade utility return stoploss_from_absolute( stop_rate=final_stop_price, current_rate=current_rate, is_short=trade.is_short, leverage=trade.leverage ) def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> str | bool | None: """ Exit logic - COMPLETELY SEPARATE for longs vs shorts: LONGS (from IchiV3_LS_1h): - SAR exit: 1h close crosses below sar_4h → exit immediately - This is the PRIMARY and ONLY exit for longs (base stoploss is just fallback) SHORTS (from IchiV2_Short_RetestSenkou): - RSI 4h oversold (< 25) - OR velocity exhaustion (< -10%) - ATR-based stop loss (via custom_stoploss) """ custom_info_pair = self.custom_info.get(pair) if custom_info_pair is None: return None current_mode = self.config['runmode'] is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN) try: completed_candles = custom_info_pair.loc[custom_info_pair.index < current_time] if completed_candles.empty: return None last_row = completed_candles.iloc[-1] last_close = last_row['close'] last_sar = last_row.get('sar_4h') # LONG EXIT: SAR if not trade.is_short: # Cloud flip check (to prevent concurrent trades between long and short) cloud_flip = last_row['senkou_a_4h'] < last_row['senkou_b_4h'] if cloud_flip: if is_trading_mode: self._log_jsonl( signal_type="long_exit_check", pair=pair, conditions={'cloud_flip_bearish': cloud_flip}, indicators={ 'close': float(last_close), 'sar_4h': float(last_sar) if not pd.isna(last_sar) else None, 'senkou_a_4h': float(last_row['senkou_a_4h']), 'senkou_b_4h': float(last_row['senkou_b_4h']), }, trade_data={ 'trade_id': trade.id, 'current_profit': current_profit, 'current_profit_pct': current_profit * 100, }, exit_triggered=True, exit_reason='senkou_a_4h_exit_long' ) return 'senkou_a_4h_exit_long' # SAR cross down sar_cross_down = not pd.isna(last_sar) and last_close <= last_sar if is_trading_mode: self._log_jsonl( signal_type="long_exit_check", pair=pair, conditions={'sar_cross_down': sar_cross_down}, indicators={ 'close': float(last_close), 'sar_4h': float(last_sar) if not pd.isna(last_sar) else None, }, trade_data={ 'trade_id': trade.id, 'current_profit': current_profit, 'current_profit_pct': current_profit * 100, }, exit_triggered=sar_cross_down, exit_reason='sar_exit_long' if sar_cross_down else None ) if sar_cross_down: return 'sar_exit_long' # SHORT EXIT: RSI + Velocity if trade.is_short: last_rsi_4h = last_row.get('rsi_4h') last_velocity = last_row.get('price_velocity') # Cloud flip check (to prevent concurrent trades between long and short) cloud_flip = last_row['senkou_a_4h'] > last_row['senkou_b_4h'] if cloud_flip: if is_trading_mode: self._log_jsonl( signal_type="short_exit_check", pair=pair, conditions={'cloud_flip_bullish': cloud_flip}, indicators={ 'close': float(last_close), 'senkou_a_4h': float(last_row['senkou_a_4h']), 'senkou_b_4h': float(last_row['senkou_b_4h']), 'rsi_4h': float(last_rsi_4h) if not pd.isna(last_rsi_4h) else None, 'price_velocity': float(last_velocity) if not pd.isna(last_velocity) else None, 'atr': float(last_row.get('atr', 0)), }, trade_data={ 'trade_id': trade.id, 'current_profit': current_profit, 'current_profit_pct': current_profit * 100, }, exit_triggered=True, exit_reason='senkou_a_4h_exit_short' ) return 'senkou_a_4h_exit_short' # Check exit conditions rsi_oversold = False velocity_exhausted = False if self.use_rsi_exit.value == 1: rsi_oversold = not pd.isna(last_rsi_4h) and last_rsi_4h < self.rsi_tp_threshold.value velocity_exhausted = not pd.isna(last_velocity) and last_velocity < self.velocity_tp_threshold.value if is_trading_mode: self._log_jsonl( signal_type="short_exit_check", pair=pair, conditions={ 'rsi_oversold': rsi_oversold, 'velocity_exhausted': velocity_exhausted, }, indicators={ 'close': float(last_close), 'rsi_4h': float(last_rsi_4h) if not pd.isna(last_rsi_4h) else None, 'price_velocity': float(last_velocity) if not pd.isna(last_velocity) else None, 'atr': float(last_row.get('atr', 0)), 'rsi_threshold': self.rsi_tp_threshold.value, 'velocity_threshold': self.velocity_tp_threshold.value, }, trade_data={ 'trade_id': trade.id, 'current_profit': current_profit, 'current_profit_pct': current_profit * 100, }, exit_triggered=rsi_oversold or velocity_exhausted, exit_reason='tp_rsi_4h_oversold' if rsi_oversold else ('tp_velocity_achieved' if velocity_exhausted else None) ) if rsi_oversold: return 'tp_rsi_4h_oversold' if velocity_exhausted: return 'tp_velocity_achieved' except (KeyError, IndexError) as e: self.logger.warning(f"Error in custom_exit for {pair}: {e}") return None 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: """Store ATR at entry time for shorts and log detailed entry information.""" trade = kwargs.get('trade') custom_info_pair = self.custom_info.get(pair) # Store ATR for shorts if trade is not None and side == 'short' and custom_info_pair is not None: try: entry_candles = custom_info_pair[custom_info_pair.index <= current_time] if not entry_candles.empty: entry_row = entry_candles.iloc[-1] entry_atr = entry_row.get('atr') if not pd.isna(entry_atr): trade.set_custom_data(key='entry_atr', value=float(entry_atr)) except (KeyError, IndexError) as e: self.logger.warning(f"Failed to store entry ATR for {pair}: {e}") current_mode = self.config['runmode'] is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN) if is_trading_mode: self.logger.info(f"ENTRY | {pair} | {side} | amount={amount} | rate={rate} | time={current_time}") return True def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: """Log detailed trade exit information.""" current_mode = self.config['runmode'] is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN) if is_trading_mode: profit_ratio = trade.calc_profit_ratio(rate) duration = current_time - trade.open_date_utc side = 'short' if trade.is_short else 'long' self.logger.info(f"EXIT | {pair} | {side} | reason={exit_reason} | profit={profit_ratio:.2%} | duration={duration} | time={current_time}") return True def bot_loop_start(self, current_time: datetime, **kwargs) -> None: """ Called at the start of each trading cycle. Logs balance and open positions summary. """ current_mode = self.config['runmode'] is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN) if is_trading_mode: self.logger.info(f"🔄 NEW TRADING CYCLE: {current_time}") self._log_portfolio_summary(current_time, **kwargs) # Reset counter for new cycle self._pairs_analyzed_count = 0 def _log_portfolio_summary(self, cycle_time: datetime, **kwargs) -> None: """ Log portfolio summary with balance and open positions details in JSONL format. """ open_trades = Trade.get_open_trades() # Get balance information balance_info = {} stake_currency = self.config.get('stake_currency', 'USDC') wallets = kwargs.get('wallets') or self.wallets if wallets: try: balance_info = { 'total_balance': float(wallets.get_total(stake_currency)), 'total_stake_amount': float(wallets.get_total_stake_amount()), 'free_balance': float(wallets.get_free(stake_currency)), 'used_balance': float(wallets.get_used(stake_currency)), 'stake_currency': stake_currency, } except Exception as e: self.logger.warning(f"Failed to fetch balance info: {e}") balance_info = {'error': str(e)} # Collect position details positions = [] total_unrealized_pnl = 0 winning = 0 losing = 0 for trade in open_trades: # Get current rate from dataprovider try: dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if not dataframe.empty: relevant_candles = dataframe[dataframe['date'] <= cycle_time] if not relevant_candles.empty: current_rate = relevant_candles.iloc[-1]['close'] else: current_rate = trade.open_rate else: current_rate = trade.open_rate except: current_rate = trade.open_rate profit_ratio = trade.calc_profit_ratio(current_rate) profit_pct = profit_ratio * 100 entry_date = trade.open_date_utc duration = cycle_time - entry_date duration_seconds = duration.total_seconds() total_unrealized_pnl += profit_ratio if profit_ratio > 0: winning += 1 else: losing += 1 side_label = "short" if trade.is_short else "long" positions.append({ 'trade_id': trade.id, 'pair': trade.pair, 'side': side_label, 'entry_rate': float(trade.open_rate), 'current_rate': float(current_rate), 'entry_time': entry_date.isoformat(), 'profit_pct': float(profit_pct), 'profit_ratio': float(profit_ratio), 'duration_seconds': int(duration_seconds), 'duration_str': str(duration), 'is_winning': profit_ratio > 0, }) # Log portfolio summary as JSONL self._log_jsonl( signal_type="portfolio_summary", cycle_time=cycle_time.isoformat(), balance=balance_info, positions_count=len(open_trades), positions=positions, summary={ 'winning_positions': winning, 'losing_positions': losing, 'total_unrealized_pnl_pct': float(total_unrealized_pnl * 100), 'total_unrealized_pnl_ratio': float(total_unrealized_pnl), } ) def _add_ichimoku(self, df: DataFrame) -> None: ichi = ftt.ichimoku(df, **self.ICHI_PARAMS) df['senkou_a'] = ichi['senkou_span_a'] df['senkou_b'] = ichi['senkou_span_b'] df['chikou_span'] = ichi['chikou_span'] df['tenkan_sen'] = ichi['tenkan_sen'] df['kijun_sen'] = ichi['kijun_sen'] df['cloud_green'] = ichi['cloud_green'] df['cloud_red'] = ichi['cloud_red'] @informative('4h') def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Calculate 4h Ichimoku, SAR, and RSI.""" self._add_ichimoku(dataframe) dataframe['sar'] = ta.SAR(dataframe) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) # # Bollinger Bands for analysis (using qtpylib) # bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=3.0) # dataframe['bb_upper'] = bollinger['upper'] # dataframe['bb_middle'] = bollinger['mid'] # dataframe['bb_lower'] = bollinger['lower'] return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Calculate 1h indicators.""" # Mark 4h boundaries for long entries dataframe['is_4h_boundary'] = dataframe['date'].dt.hour % 4 == 0 # Cloud status for short entries dataframe['cloud_bearish'] = dataframe['senkou_a_4h'] < dataframe['senkou_b_4h'] # Price velocity for short exits dataframe['price_velocity'] = ( (dataframe['close'] - dataframe['close'].shift(self.price_velocity_window.value)) / dataframe['close'].shift(self.price_velocity_window.value) ) # ATR for short stop loss dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) # Store for custom functions self.custom_info[metadata['pair']] = dataframe[[ 'date', 'close', 'sar_4h', 'rsi_4h', 'price_velocity', 'atr', 'senkou_a_4h', 'senkou_b_4h' # 'bb_upper_4h' ]].copy().set_index('date') return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Dual entry system: 1. LONG: Ichimoku cloud crossover at 4h boundaries 2. SHORT: Senkou B retest breakdown """ pair = metadata.get('pair', 'unknown') current_mode = self.config['runmode'] is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN) # Initialize dataframe['enter_long'] = 0 dataframe['enter_short'] = 0 dataframe['buy'] = 0 dataframe['enter_tag'] = '' # === LONG ENTRY (IchiV3_LS_1h logic) === # Only at 4h boundaries at_4h_boundary = dataframe['is_4h_boundary'] effective_close = dataframe['close_4h'] # Long: prev 4h close was below senkou_a, current 4h close is above prev_close_4h = effective_close.shift(4) prev_senkou_a = dataframe['senkou_a_4h'].shift(4) prev_below_cloud = prev_close_4h < prev_senkou_a curr_above_cloud = effective_close > dataframe['senkou_a_4h'] cloud_green = dataframe['senkou_a_4h'] > dataframe['senkou_b_4h'] sar_below_close = dataframe['sar_4h'] < effective_close long_entry = at_4h_boundary & prev_below_cloud & curr_above_cloud & cloud_green & sar_below_close # === SHORT ENTRY (RetestSenkou logic) === window = self.retest_lookback_window.value # Retest detection dataframe['retested_senkou_b'] = dataframe['high'] >= dataframe['senkou_b_4h'] dataframe['had_senkou_b_retest'] = dataframe['retested_senkou_b'].rolling( window=window, min_periods=1 ).max().astype(bool) # Breakdown detection dataframe['cross_down_senkou_b'] = ( (dataframe['close'].shift(1) >= dataframe['senkou_b_4h'].shift(1)) & (dataframe['close'] < dataframe['senkou_b_4h']) ) # Short entry: cloud bearish + retest + breakdown short_entry = ( dataframe['cloud_bearish'] & dataframe['had_senkou_b_retest'] & dataframe['cross_down_senkou_b'] ) # Apply signals dataframe.loc[long_entry, 'enter_long'] = 1 dataframe.loc[long_entry, 'enter_tag'] = 'long_entry' dataframe.loc[short_entry, 'enter_short'] = 1 dataframe.loc[short_entry, 'enter_tag'] = 'short_entry' dataframe.loc[long_entry | short_entry, 'buy'] = 1 # Signal logging (only in trading mode) if is_trading_mode and len(dataframe) > 0: latest = dataframe.iloc[-1] # LONG signal analysis if latest['is_4h_boundary']: long_conditions = { 'prev_below_cloud': bool(prev_below_cloud.iloc[-1]) if len(prev_below_cloud) > 0 else False, 'curr_above_cloud': bool(curr_above_cloud.iloc[-1]) if len(curr_above_cloud) > 0 else False, 'cloud_green': bool(cloud_green.iloc[-1]) if len(cloud_green) > 0 else False, 'sar_below_close': bool(sar_below_close.iloc[-1]) if len(sar_below_close) > 0 else False, } long_signal = latest['enter_long'] == 1 # Calculate ATR stop for longs (informational - longs use SAR exit) atr_stop_info = self._calculate_atr_stop_price( current_price=float(latest['close']), atr=float(latest.get('atr', 0)), is_short=False ) # Standardized JSONL logging self._log_jsonl( signal_type="long_entry_check", pair=pair, conditions=long_conditions, indicators={ 'close': float(latest['close']), 'close_4h': float(latest['close_4h']), 'senkou_a_4h': float(latest['senkou_a_4h']), 'senkou_b_4h': float(latest['senkou_b_4h']), 'sar_4h': float(latest['sar_4h']), 'rsi_4h': float(latest.get('rsi_4h', 0)), 'atr': float(latest.get('atr', 0)), 'atr_stop_price': atr_stop_info['atr_stop_price'], 'atr_stop_distance': atr_stop_info['atr_stop_distance'], 'atr_stop_pct': atr_stop_info['atr_stop_pct'], }, signal_triggered=long_signal, is_4h_boundary=True, action='ENTER_LONG' if long_signal else 'REJECTED', note='longs_use_sar_exit_not_atr' ) # SHORT signal analysis short_conditions = { 'cloud_bearish': bool(latest['cloud_bearish']), 'had_retest': bool(latest['had_senkou_b_retest']), 'cross_down': bool(latest['cross_down_senkou_b']), } short_signal = latest['enter_short'] == 1 # Only log if at least one condition is met or signal triggered (reduce noise) should_log_short = short_signal or latest['cloud_bearish'] or latest['had_senkou_b_retest'] or latest['cross_down_senkou_b'] if should_log_short: # Calculate ATR stop for shorts (this is the actual stop used) atr_stop_info = self._calculate_atr_stop_price( current_price=float(latest['close']), atr=float(latest.get('atr', 0)), is_short=True ) # Standardized JSONL logging self._log_jsonl( signal_type="short_entry_check", pair=pair, conditions=short_conditions, indicators={ 'close': float(latest['close']), 'senkou_a_4h': float(latest['senkou_a_4h']), 'senkou_b_4h': float(latest['senkou_b_4h']), 'rsi_4h': float(latest.get('rsi_4h', 0)), 'price_velocity': float(latest.get('price_velocity', 0)), 'atr': float(latest.get('atr', 0)), 'atr_stop_price': atr_stop_info['atr_stop_price'], 'atr_stop_distance': atr_stop_info['atr_stop_distance'], 'atr_stop_pct': atr_stop_info['atr_stop_pct'], }, signal_triggered=short_signal, action='ENTER_SHORT' if short_signal else 'REJECTED', note='shorts_use_atr_stop' ) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Exit signals disabled - using custom_exit.""" dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 dataframe['exit_tag'] = '' dataframe['sell'] = 0 return dataframe