# source: https://raw.githubusercontent.com/maxakak1998/trading_bot/d976656bdae438cd3ed7bca6c3c6464c56ab44bb/user_data/strategies/FreqAIStrategy.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- from datetime import datetime from typing import Optional import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter import logging from pandas import DataFrame import pandas_ta as pta # pandas_ta for advanced indicators import talib.abstract as ta # talib for basic indicators (required by FreqAI) import sys from pathlib import Path # Add strategies directory to path to import local modules sys.path.append(str(Path(__file__).parent)) from indicators.smc_indicators import SMCIndicators from indicators.data_enhancement import DataEnhancement # Phase 2 Features from indicators.feature_engineering import FeatureEngineering # Phase 3: Proper ML Features from indicators.chart_patterns import ChartPatterns # Phase 3: Chart Pattern Recognition from indicators.wave_indicators import WaveIndicators # Phase 3: Elliott Wave Lite (Fibonacci + AO) logger = logging.getLogger(__name__) import freqtrade.vendor.qtpylib.indicators as qtpylib class Github_maxakak1998_trading_bot__FreqAIStrategy__20251220_173046(IStrategy): """ FreqAI Strategy example. """ INTERFACE_VERSION = 3 # ===================================================== # HYPEROPT PARAMETERS - Tunable via `freqtrade hyperopt` # ===================================================== # Entry/Exit prediction thresholds # Entry Signal Optimization buy_adx_threshold = IntParameter(20, 50, default=25, space="buy", optimize=True) buy_rsi_high = IntParameter(60, 90, default=70, space="buy", optimize=True) buy_rsi_low = IntParameter(20, 40, default=30, space="buy", optimize=True) # AI Prediction Confidence # FIX v2: Threshold 0.002-0.015 too permissive (LOSS in Iter 22) # Moderate threshold for balanced quality/quantity buy_pred_threshold = DecimalParameter(0.005, 0.012, default=0.008, space="buy", optimize=True) sell_pred_threshold = DecimalParameter(-0.012, -0.005, default=-0.008, space="sell", optimize=True) # Entry Score Threshold for weighted scoring system # Higher = stricter (fewer but higher quality trades) # Lower = looser (more trades but lower quality) # Note: Max achievable score is ~0.25 when AI predictions are 0 entry_score_threshold = DecimalParameter(0.15, 0.35, default=0.20, space="buy", optimize=True) # Sell Signal Optimization sell_rsi_threshold = IntParameter(20, 80, default=50, space="sell", optimize=True) # ATR multiplier for dynamic stoploss (used in custom_stoploss) # MOVED to 'buy' space (safe space) to avoid KeyError atr_multiplier = DecimalParameter(1.5, 4.0, default=3.0, space="buy", optimize=True) # Confidence threshold for trade entries confidence_threshold = DecimalParameter(0.3, 0.7, default=0.5, space="buy", optimize=True) # ===================================================== # NEW INDICATOR HYPEROPT PARAMS # ===================================================== # Volatility Expansion thresholds # High = strong expansion (big moves), Mid = moderate expansion volatility_high_threshold = DecimalParameter(1.2, 2.0, default=1.5, space="buy", optimize=True) volatility_mid_threshold = DecimalParameter(1.0, 1.5, default=1.2, space="buy", optimize=True) # Mean Reversion Zone thresholds (RSI-based) # Overbought = avoid longs, Oversold = avoid shorts mr_overbought = IntParameter(65, 85, default=70, space="buy", optimize=True) mr_oversold = IntParameter(15, 35, default=30, space="buy", optimize=True) # ===================================================== # Minimal ROI designed for the strategy. # ITER 19: Moderate ROI 5-12% (evidence: Iter 17 failed with 8-25% - only 7 trades) # Balance between profit capture and achievable targets minimal_roi = { "180": 0.05, # 5% after 3 hours (achievable) "90": 0.08, # 8% after 1.5 hours "45": 0.10, # 10% after 45 min "0": 0.12 # 12% immediate - realistic target } # Risk Management - 1R = 10% stoploss stoploss = -0.10 # 1R = 10% = $5 on $50 stake # OPTION A: Enable trailing stop to capture profits # Activates at 5% profit, trails from 8% peak trailing_stop = True trailing_stop_positive = 0.05 # Activate at 5% profit trailing_stop_positive_offset = 0.08 # Trail from 8% peak trailing_only_offset_is_reached = True # Custom Trailing Parameters (Optimizable) - MOVED TO BUY SPACE # p_trail_start: Profit required to activate trailing p_trail_start = DecimalParameter(0.005, 0.05, default=0.01, space="buy", optimize=True) # p_trail_offset: Distance from CURRENT PRICE (e.g. 0.01 = 1% below current) p_trail_offset = DecimalParameter(0.002, 0.03, default=0.01, space="buy", optimize=True) # Use ROI and exit signals instead of trailing use_exit_signal = True use_custom_stoploss = True # Enable custom_stoploss function # Maximum risk per trade (% of margin) # 20% max loss per trade on margin # Example: With 4x leverage, actual price loss = 20%/4 = 5% max_risk_per_trade = 0.20 # 20% of stake (margin) # Leverage calculation: # Target Risk per Trade = max_risk_per_trade% of Stake (Margin) # Loss = Stake * Leverage * Stoploss_Price_Dist # max_risk_per_trade * Stake = Stake * Leverage * Stoploss_Price_Dist # Leverage = max_risk_per_trade / Stoploss_Price_Dist 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: """ Customize leverage for each new trade. Uses max_risk_per_trade to calculate appropriate leverage. """ risk_per_trade = self.max_risk_per_trade # Use configurable risk limit stoploss_dist = abs(self.stoploss) # Calculate leverage # Example: Stoploss 5% (0.05) -> Leverage = 0.20 / 0.05 = 4x # Example: Stoploss 1% (0.01) -> Leverage = 0.20 / 0.01 = 20x target_leverage = risk_per_trade / stoploss_dist # Cap leverage at max_leverage or a safe limit (e.g. 20x) final_leverage = min(target_leverage, max_leverage, 20.0) return final_leverage def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ UPGRADED: Structure-Based Stoploss with R-Based Trailing INITIAL SL (below 1R profit): For LONG: SL placed below nearest structure (Swing Low OR Bullish Order Block) For SHORT: SL placed above nearest structure (Swing High OR Bearish Order Block) This ensures SL has "proof" from market structure, not just arbitrary ATR distance. Tighter structure-based SL → Higher leverage → Amplified profits R-Based Trailing (after 1R profit): - Profit >= 1R (20%): Move SL to breakeven - Profit >= 2R (40%): Lock in 1R profit - Profit >= 3R (60%): Lock in 2R profit """ # Define R in terms of profit percentage one_r = self.max_risk_per_trade # 0.20 = 20% # ===================================================== # R-BASED TRAILING (when in profit) # ===================================================== # 3R or more: Lock in 2R profit if current_profit >= 3 * one_r: # >= 60% return -(current_profit - 2 * one_r) # 2R or more: Lock in 1R profit if current_profit >= 2 * one_r: # >= 40% return -(current_profit - one_r) # 1R or more: Move to breakeven if current_profit >= one_r: # >= 20% return -current_profit + 0.001 # ===================================================== # STRUCTURE-BASED INITIAL STOPLOSS (below 1R) # ===================================================== dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() current_leverage = getattr(trade, 'leverage', 1.0) or 1.0 max_risk_on_margin = self.max_risk_per_trade safe_sl_limit = -(max_risk_on_margin / current_leverage) # ATR fallback atr = last_candle.get('atr', 0) atr_based_sl = -self.atr_multiplier.value * (atr / trade.open_rate) if atr > 0 else -0.05 # Determine if LONG or SHORT trade is_long = trade.is_short is False # Helper to find column with pattern def find_col(pattern): cols = [c for c in dataframe.columns if pattern in c] if cols: # Prefer 5m or no suffix return next((c for c in cols if '5m' in c or c == f'%-{pattern}'), cols[0]) return None # Structure-based SL candidates structure_sl = None buffer = 0.005 # 0.5% buffer below structure if is_long: # LONG: SL below Swing Low or Bullish Order Block swing_low_col = find_col('dist_to_swing_low') ob_bull_col = find_col('dist_to_bull_ob') candidates = [] if swing_low_col: swing_low_dist = last_candle.get(swing_low_col, None) if swing_low_dist is not None and swing_low_dist < 0: candidates.append(swing_low_dist - buffer) if ob_bull_col: ob_bull_dist = last_candle.get(ob_bull_col, None) if ob_bull_dist is not None and ob_bull_dist < 0: candidates.append(ob_bull_dist - buffer) if candidates: structure_sl = max(candidates) # max of negatives = tightest SL else: # SHORT: SL above Swing High or Bearish Order Block swing_high_col = find_col('dist_to_swing_high') ob_bear_col = find_col('dist_to_bear_ob') candidates = [] if swing_high_col: swing_high_dist = last_candle.get(swing_high_col, None) if swing_high_dist is not None and swing_high_dist > 0: candidates.append(-(swing_high_dist + buffer)) if ob_bear_col: ob_bear_dist = last_candle.get(ob_bear_col, None) if ob_bear_dist is not None and ob_bear_dist > 0: candidates.append(-(ob_bear_dist + buffer)) if candidates: structure_sl = max(candidates) # Choose final SL: structure-based if available, else ATR if structure_sl is not None: # Use structure SL but respect safety limits final_sl = max(structure_sl, safe_sl_limit, -0.15) final_sl = min(final_sl, -0.01) # At least 1% SL else: # Fallback to ATR-based final_sl = max(atr_based_sl, safe_sl_limit, -0.15) final_sl = min(final_sl, -0.01) return final_sl 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: """ Customize stake amount for each trade based on AI confidence. Logic: - Low confidence (0.5-0.6): 50% of base stake (25 USDT) - Medium confidence (0.6-0.8): 100% of base stake (50 USDT) - High confidence (>0.8): 120% of base stake (60 USDT) """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Get AI confidence score (prediction mean) ai_confidence = last_candle.get('&s-up_or_down_mean', 0.5) # Scale stake based on confidence if ai_confidence > 0.8: stake_multiplier = 1.2 # High confidence: 120% elif ai_confidence > 0.6: stake_multiplier = 1.0 # Medium confidence: 100% else: stake_multiplier = 0.5 # Low confidence: 50% final_stake = proposed_stake * stake_multiplier # Ensure within min/max bounds if min_stake: final_stake = max(final_stake, min_stake) final_stake = min(final_stake, max_stake) return final_stake # Timeframe timeframe = '5m' process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # FreqAI attributes # BIDIRECTIONAL TRADING: Both long and short enabled # Shorts are filtered by market regime (only in bearish conditions) can_short = True def detect_market_regime(self, dataframe: DataFrame) -> DataFrame: """ Classify market regime: TREND, SIDEWAY, or VOLATILE Logic: - ADX > 25 + BB Width > 0.04 → TREND (strong directional movement) - ADX < 20 + BB Width < 0.02 → SIDEWAY (no clear direction) - Otherwise → VOLATILE (unpredictable, avoid trading) """ # Ensure ADX is calculated (using talib - uppercase function names) if 'adx' not in dataframe.columns: dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) # Ensure ATR is calculated if 'atr' not in dataframe.columns: dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] # Ensure BB width is calculated if 'bb_width' not in dataframe.columns: dataframe['bb_width'] = (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / dataframe['bb_middleband'] # Classify regime conditions = [ (dataframe['adx'] > 25) & (dataframe['bb_width'] > 0.04), # Strong trend (dataframe['adx'] < 20) & (dataframe['bb_width'] < 0.02), # Sideway/consolidation ] choices = ['TREND', 'SIDEWAY'] dataframe['market_regime'] = np.select(conditions, choices, default='VOLATILE') # ============================================== # TREND DIRECTION: BULLISH vs BEARISH # Used to filter long/short entries # ============================================== # Calculate EMAs for trend direction ema_50 = ta.EMA(dataframe['close'], timeperiod=50) ema_200 = ta.EMA(dataframe['close'], timeperiod=200) # Trend direction based on price vs EMAs and EMA alignment # BULLISH: price > EMA50 > EMA200 (strong uptrend) # BEARISH: price < EMA50 < EMA200 (strong downtrend) # NEUTRAL: otherwise trend_conditions = [ (dataframe['close'] > ema_50) & (ema_50 > ema_200), # Bullish alignment (dataframe['close'] < ema_50) & (ema_50 < ema_200), # Bearish alignment ] trend_choices = ['BULLISH', 'BEARISH'] dataframe['trend_direction'] = np.select(trend_conditions, trend_choices, default='NEUTRAL') # Store EMAs for later use dataframe['ema_50'] = ema_50 dataframe['ema_200'] = ema_200 return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame CRITICAL: Must call self.freqai.start() to trigger FreqAI training and prediction! Without this call, FreqAI will NOT train and you get 0 trades. """ # Calculate Bollinger Bands for market_regime detection (needed before detect_market_regime) # ta.BBANDS returns tuple: (upperband, middleband, lowerband) bb_upper, bb_middle, bb_lower = ta.BBANDS(dataframe['close'], timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe['bb_upperband'] = bb_upper dataframe['bb_lowerband'] = bb_lower dataframe['bb_middleband'] = bb_middle # This is THE critical line that triggers FreqAI training! # It calls feature_engineering_* methods and trains/predicts dataframe = self.freqai.start(dataframe, metadata, self) # Add market_regime for entry/exit decisions (after FreqAI processing) dataframe = self.detect_market_regime(dataframe) return dataframe def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: dict, **kwargs) -> DataFrame: """ *Only functional with FreqAI enabled strategies* expand_all: Features ở đây KHÔNG được tự động nhân bản cho các timeframes khác. Chỉ dùng cho các features cục bộ của base timeframe (5m). ĐẶT Ở ĐÂY (Chỉ 5m): - Chart Patterns (Double Top/Bottom, H&S, Wedge, Triangle, Flag) → Mô hình nến mang tính chất cục bộ, nhân bản 4 TF tạo dữ liệu rác - Data Enhancement (Fear & Greed, API-based) → Dữ liệu từ API, không cần đa khung - Legacy indicators for Market Regime KHÔNG ĐẶT Ở ĐÂY (Đã move sang expand_basic): - SMC Indicators → Order Block 4h có giá trị gấp 10 lần 5m - Wave Indicators → Fibonacci levels cần nhìn từ HTF """ # ==== Chart Pattern Recognition (5m only) ==== # Nhận dạng các mô hình giá: Double Top/Bottom, Head & Shoulders, Wedge, Triangle, Flag # Mang tính chất cục bộ - không cần expand cho multi-TF # Can be disabled via feature_flags.chart_patterns if self.config.get('freqai', {}).get('feature_flags', {}).get('chart_patterns', True): dataframe = ChartPatterns.add_all_patterns(dataframe) # ==== Data Enhancement (5m only) ==== # Fear & Greed Index, Volume Imbalance, Funding Proxy # API-based features, không cần đa khung # Can be disabled via feature_flags.data_enhancement if self.config.get('freqai', {}).get('feature_flags', {}).get('data_enhancement', True): dataframe = DataEnhancement.add_all_features(dataframe, period=period) # ==== Legacy indicators (cho Market Regime) ==== # Using talib (uppercase function names) dataframe['mfi'] = ta.MFI(dataframe['high'], dataframe['low'], dataframe['close'], dataframe['volume'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) # Bollinger Bands (cần cho market regime detection) dataframe['bb_upperband'], dataframe['bb_middleband'], dataframe['bb_lowerband'] = ta.BBANDS( dataframe['close'], timeperiod=20, nbdevup=2.0, nbdevdn=2.0 ) dataframe["bb_width"] = ( dataframe["bb_upperband"] - dataframe["bb_lowerband"] ) / dataframe["bb_middleband"] # ==== Market Regime Detection (cuối cùng) ==== dataframe = self.detect_market_regime(dataframe) return dataframe def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ *Only functional with FreqAI enabled strategies* expand_basic: Features ở đây SẼ ĐƯỢC TỰ ĐỘNG NHÂN BẢN cho các timeframes khác! FreqAI sẽ tạo ra: %-log_return_1_5m, %-log_return_1_1h, %-log_return_1_4h, v.v. ĐÂY LÀ NƠI ĐẶT CORE FEATURES + SMC/WAVE: 1. Core Features (Log Returns, EMA, Momentum, Volume, Volatility) → Nền tảng cho mọi phân tích, cần nhìn ở tất cả TF 2. SMC Indicators (Order Blocks, FVG, Structure, Liquidity) → Order Block ở 4H có giá trị GẤP 10 LẦN ở 5m → Structure ở HTF quyết định trend chính 3. Wave Indicators (Fibonacci Retracement/Extension, Awesome Oscillator) → Fibo levels từ swing 4H là key levels cho toàn bộ price action → AO divergence ở 1H xác nhận reversal mạnh hơn Kết quả: Bot sẽ học từ 5m + 15m + 1h + 4h (True Multi-timeframe SMC Analysis) Ví dụ AI sẽ học: "Nếu giá chạm vùng Fibo 0.618 của khung 4H (từ expand_basic) VÀ xuất hiện mẫu nến đảo chiều ở khung 5m (từ expand_all) → Vào lệnh Mua" """ # ==== CORE FEATURE ENGINEERING ==== # Features are now selected based on timeframe sensitivity: # - 5m: Momentum, Candles, Volume spikes (fast signals) # - 15m: Short-term structure, BB (bridge timeframe) # - 1h+: Full structure, VSA, Wave (direction signals) current_tf = metadata.get('tf', self.timeframe) dataframe = FeatureEngineering.add_all_features(dataframe, config=self.config, timeframe=current_tf) # ==== SMC INDICATORS (Multi-TF) ==== # Order Blocks, FVG, Structure Direction, Liquidity Zones # Order Block ở 4H có giá trị gấp 10 lần ở 5m # Can be disabled via feature_flags.smc_indicators if self.config.get('freqai', {}).get('feature_flags', {}).get('smc_indicators', True): dataframe = SMCIndicators.add_all_indicators(dataframe) # ==== WAVE INDICATORS (Multi-TF) ==== # Fibonacci Retracement/Extension, Awesome Oscillator, Wave Structure # Fibo levels từ swing 4H là key levels cho toàn bộ price action # Can be disabled via feature_flags.wave_indicators if self.config.get('freqai', {}).get('feature_flags', {}).get('wave_indicators', True): dataframe = WaveIndicators.add_all_features(dataframe) return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ *Only functional with FreqAI enabled strategies* This optional function will be called once with the dataframe of the base timeframe. This is the final chance to add features. The columns won't be modified. All features must be prepended with `%` to be recognized by FreqAI internals. """ return dataframe def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ *Only functional with FreqAI enabled strategies* Required function to set the targets for the model. All targets must be prepended with `&` to be recognized by the FreqAI internals. Three labeling methods (MUTUALLY EXCLUSIVE): 1. regression_labels: % price change in next N candles (for XGBoostRegressor) 2. classification_labels: UP/NEUTRAL/DOWN direction (for XGBoostClassifier) 3. trend_scanning: t-statistics based trend detection """ # Get flags flags = self.config.get('freqai', {}).get('feature_flags', {}) use_trend_scanning = flags.get('trend_scanning', False) use_regression = flags.get('regression_labels', False) use_classification = flags.get('classification_labels', True) # NEW DEFAULT # Validate: only one can be True enabled_count = sum([use_trend_scanning, use_regression, use_classification]) if enabled_count != 1: raise ValueError( f"ERROR: Exactly 1 labeling method must be enabled! " f"Current: trend_scanning={use_trend_scanning}, regression={use_regression}, " f"classification={use_classification}" ) if use_trend_scanning: logger.info("Using Trend Scanning labeling method") dataframe = self._trend_scanning_labels(dataframe, window=20, t_threshold=2.0) elif use_regression: logger.info("Using Regression labeling method") dataframe = self._regression_labels(dataframe, horizon=50) else: # CLASSIFICATION: Predict direction (UP/DOWN/NEUTRAL) # CRITICAL: Must set class_names for FreqAI XGBoostClassifier self.freqai.class_names = ['down', 'neutral', 'up'] logger.info(f"Using CLASSIFICATION labeling with classes: {self.freqai.class_names}") # Iter 33: Horizon 10 + Threshold 0.004 (0.4%) - The Sweet Spot dataframe = self._classification_labels(dataframe, horizon=10, threshold=0.004) return dataframe def _classification_labels(self, dataframe: DataFrame, horizon: int = 50, threshold: float = 0.01) -> DataFrame: """ Classification target: Predict direction of price movement using SMOOTHED target. FREQAI REQUIREMENT: Use STRING labels with &s- prefix! - Column name: &s-direction (not &-direction) - Labels: 'up', 'down', 'neutral' (not 0, 1, 2) This acts as low-pass filter, removing noise and improving label stability. """ # SMOOTHED TARGET: Rolling mean of future N candles vs current close future_mean = dataframe["close"].shift(-1).rolling(window=horizon).mean() smoothed_change = (future_mean - dataframe["close"]) / dataframe["close"] # Dynamic threshold based on ATR for volatility adjustment if 'atr' in dataframe.columns: atr_pct = dataframe['atr'] / dataframe['close'] dynamic_threshold = atr_pct.rolling(20).mean().fillna(threshold) dynamic_threshold = dynamic_threshold.clip(lower=0.005, upper=0.03) else: dynamic_threshold = threshold # FREQAI STRING LABELS: up/down/neutral # Using np.select for vectorized string assignment conditions = [ smoothed_change > dynamic_threshold, # UP smoothed_change < -dynamic_threshold, # DOWN ] choices = ['up', 'down'] # Column MUST use &s- prefix for FreqAI string classification dataframe["&s-direction"] = np.select(conditions, choices, default='neutral') # Handle NaN from rolling window - set to 'neutral' dataframe["&s-direction"] = dataframe["&s-direction"].fillna('neutral') # Log distribution direction_counts = dataframe["&s-direction"].value_counts() up_count = direction_counts.get('up', 0) neutral_count = direction_counts.get('neutral', 0) down_count = direction_counts.get('down', 0) logger.info(f"CLASSIFICATION LABELS (string): up={up_count}, " f"neutral={neutral_count}, down={down_count}") return dataframe def _regression_labels(self, dataframe: DataFrame, horizon: int = 20) -> DataFrame: """Simple regression target: % price change in next N candles.""" future_close = dataframe["close"].shift(-horizon) dataframe["&-price_change_pct"] = (future_close - dataframe["close"]) / dataframe["close"] return dataframe def _trend_scanning_labels(self, dataframe: DataFrame, window: int = 20, t_threshold: float = 2.0) -> DataFrame: """ Trend Scanning labeling using t-statistics. FIXED: Previously set label=0 for non-significant trends, causing 80%+ labels to be zero, which made the model overfit to always predict 0. NEW APPROACH: - Always use actual price change as base - Boost significant trends (multiply by confidence factor) - Dampen (but not zero) non-significant trends Returns: - &-price_change_pct: Weighted expected % change """ import numpy as np from scipy import stats close_prices = dataframe['close'].values n = len(close_prices) # Initialize arrays price_changes = np.zeros(n) trend_t_stats = np.zeros(n) x = np.arange(window) for i in range(n - window): y = close_prices[i:i + window] # Linear regression: y = slope * x + intercept slope, intercept, r_value, p_value, std_err = stats.linregress(x, y) # Calculate t-statistic for slope if std_err > 0: t_stat = slope / std_err else: t_stat = 0 # Calculate actual % price change (future price - current price) current_price = close_prices[i] future_price = close_prices[i + window - 1] if (i + window - 1) < n else close_prices[i] if current_price > 0: actual_pct_change = (future_price - current_price) / current_price else: actual_pct_change = 0 # Weight factor based on t-stat significance # Significant trends (|t| >= threshold): boost by 1.0-1.5x # Non-significant: dampen by 0.3-0.8x (but NOT zero) if abs(t_stat) >= t_threshold: # High confidence trend - use full or boosted value weight = min(1.0 + (abs(t_stat) - t_threshold) * 0.1, 1.5) else: # Low confidence - dampen but don't zero # Scale from 0.3 (t=0) to 0.8 (t=threshold) weight = 0.3 + (abs(t_stat) / t_threshold) * 0.5 price_changes[i] = actual_pct_change * weight trend_t_stats[i] = t_stat # Assign to dataframe dataframe["&-price_change_pct"] = price_changes dataframe["&-trend_t_stat"] = trend_t_stats # Optional: for analysis return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the entry signal for the given dataframe UPGRADED Entry conditions with Confluence system: LONG CONDITIONS: 1. AI Prediction > threshold (FreqAI output) 2. Market Regime = TREND (KER > 0.5 hoặc ADX > 25) 3. Trend Confluence > 0.5 (EMA alignment + ADX strong) 4. Momentum Confluence > 0.4 (RSI + MFI + CMF agree) 5. Money Pressure > 0 (buying pressure) 6. Pattern Net Score >= 0 (no bearish patterns) 7. Not in Extreme Greed (avoid FOMO) 8. Structure Direction > 0 (Higher Highs) SHORT CONDITIONS: 1. AI Prediction < -threshold 2. Market Regime = TREND 3. Trend Confluence < 0.5 (bearish alignment) 4. Momentum Confluence < 0.6 (bearish momentum) 5. Money Pressure < 0 (selling pressure) 6. Pattern Net Score <= 0 (no bullish patterns) 7. RSI > 75 (overbought) 8. Structure Direction < 0 (Lower Lows) """ # Debug logging logger.info(f"Columns available: {len(dataframe.columns)} columns") if self.config['freqai']['enabled']: # Support both classification and regression outputs classification_col = '&s-direction' # STRING labels for FreqAI classifier regression_col = '&-price_change_pct' flags = self.config.get('freqai', {}).get('feature_flags', {}) use_classification = flags.get('classification_labels', True) if use_classification and classification_col in dataframe.columns: # CLASSIFICATION MODE: direction is 'up'/'down'/'neutral' (STRING) direction = dataframe[classification_col] direction_counts = direction.value_counts() logger.info(f"CLASSIFICATION SIGNAL (string): up={direction_counts.get('up', 0)}, " f"neutral={direction_counts.get('neutral', 0)}, down={direction_counts.get('down', 0)}") # Clear signals using string comparison ai_bullish = (direction == 'up') ai_bearish = (direction == 'down') # Convert to float for scoring (0 or 1) ai_positive = ai_bullish.astype(float) elif regression_col in dataframe.columns: # REGRESSION MODE: Backward compatibility preds = dataframe[regression_col] non_zero = (preds != 0).sum() above_thresh = (preds > self.buy_pred_threshold.value).sum() logger.info(f"REGRESSION STATS: min={preds.min():.6f}, max={preds.max():.6f}, " f"mean={preds.mean():.6f}, non_zero={non_zero}/{len(preds)}, " f"above_thresh({self.buy_pred_threshold.value})={above_thresh}") ai_positive = (preds > self.buy_pred_threshold.value).astype(float) else: logger.warning("No prediction column found, using zeros for AI signal") ai_positive = pd.Series(0, index=dataframe.index).astype(float) # ===================================================== # LONG ENTRY - WEIGHTED SCORING SYSTEM # ===================================================== # Instead of AND logic (all must be True), use weighted scores # This allows partial confluence and better trade frequency control # Get flags once use_htf_ob_confluence = flags.get('htf_ob_confluence', True) # 2. HTF Order Block (25% weight) htf_ob_signal = np.zeros(len(dataframe)) if use_htf_ob_confluence: if '%-testing_bull_ob_4h' in dataframe.columns: htf_ob_signal = (dataframe['%-testing_bull_ob_4h'] > 0).astype(float).values elif '%-testing_bull_ob_1h' in dataframe.columns: htf_ob_signal = (dataframe['%-testing_bull_ob_1h'] > 0).astype(float).values if '%-ob_fib_bull_confluence' in dataframe.columns: htf_ob_signal = np.maximum(htf_ob_signal, (dataframe['%-ob_fib_bull_confluence'] > 0).astype(float).values) # 3. ADX/Regime (10% weight) - Trending market adx_signal = np.zeros(len(dataframe)) if '%-ker_10' in dataframe.columns: adx_signal = (dataframe['%-ker_10'] > 0.4).astype(float).values elif 'adx' in dataframe.columns: adx_signal = (dataframe['adx'] > self.buy_adx_threshold.value).astype(float).values # 4. Momentum Quality (20% weight) - CONTINUOUS score momentum_signal = np.zeros(len(dataframe)) if '%-momentum_quality' in dataframe.columns: # Use continuous value clipped to positive range [0, 1] momentum_signal = dataframe['%-momentum_quality'].clip(0, 1).values elif '%-momentum_confluence' in dataframe.columns: momentum_signal = (dataframe['%-momentum_confluence'] > 0.4).astype(float).values # 5. Money Pressure (15% weight) pressure_signal = np.zeros(len(dataframe)) if '%-money_pressure' in dataframe.columns: pressure_signal = (dataframe['%-money_pressure'] > 0).astype(float).values # 6. Volatility Expansion (15% weight) - Uses hyperopt params volatility_signal = np.zeros(len(dataframe)) if '%-volatility_expanding' in dataframe.columns: # Use pre-computed feature as base volatility_signal = dataframe['%-volatility_expanding'].values elif '%-atr_pct' in dataframe.columns: # Compute dynamically using hyperopt params atr_pct = dataframe['%-atr_pct'] atr_mean = atr_pct.rolling(100).mean() volatility_ratio = atr_pct / (atr_mean + 1e-10) volatility_signal = np.select( [volatility_ratio > self.volatility_high_threshold.value, volatility_ratio > self.volatility_mid_threshold.value], [1.0, 0.5], default=0.0 ) # 7. DISCOUNT ZONE BONUS (15% weight) - OPTION B: Golden Pocket # Price in Fib 0.382-0.618 zone = golden pocket (optimal buying area) # This rewards entries at "value" levels, improving R/R discount_signal = np.zeros(len(dataframe)) use_golden_pocket = flags.get('golden_pocket_entry', True) # Look for fib_position with any timeframe suffix (e.g., %-fib_position_5m) fib_pos_cols = [c for c in dataframe.columns if 'fib_position' in c and '%-' in c] if fib_pos_cols: # Use base timeframe (5m) fib_position, or first available fib_col = next((c for c in fib_pos_cols if '5m' in c or c == '%-fib_position'), fib_pos_cols[0]) fib_pos = dataframe[fib_col].values # OPTION B: Golden pocket (0.382-0.618) instead of wider 0.22-0.50 if use_golden_pocket: discount_signal = np.where( (fib_pos >= 0.382) & (fib_pos <= 0.618), 1.0, # Full bonus in golden pocket np.where((fib_pos >= 0.22) & (fib_pos < 0.382), 0.5, 0.0) # Partial outside ) else: discount_signal = np.where( (fib_pos >= 0.22) & (fib_pos <= 0.50), 1.0, np.where(fib_pos < 0.22, 0.5, 0.0) ) # Fallback: Use OB+Fib confluence indicator ob_fib_cols = [c for c in dataframe.columns if 'ob_fib_bull_confluence' in c] if discount_signal.sum() == 0 and ob_fib_cols: discount_signal = (dataframe[ob_fib_cols[0]] > 0).astype(float).values # Calculate total LONG score (sum of weighted signals) # Weights adjusted: 20+20+15+15+15+15 = 100% long_score = ( ai_positive.values * 0.20 + # AI: 20% (was 25%) htf_ob_signal * 0.20 + # HTF OB: 20% (same) momentum_signal * 0.15 + # Momentum: 15% (was 20%) pressure_signal * 0.15 + # Pressure: 15% (same) volatility_signal * 0.15 + # Volatility: 15% (was 20%) discount_signal * 0.15 # Golden Pocket: 15% (NEW) ) # ===================================================== # OPTION B: REQUIRED ENTRY FILTERS (Reduce SL Hits) # ===================================================== # Volume filter (must have volume - not scored, just required) has_volume = dataframe['volume'] > 0 # OPTION B.1: ATR Breakout Filter (volatility expanding) # Only enter when ATR > 20-period MA (market is moving) use_atr_breakout = flags.get('atr_breakout_filter', True) atr_breakout = pd.Series(True, index=dataframe.index) if use_atr_breakout and 'atr' in dataframe.columns: atr_ma = dataframe['atr'].rolling(20).mean() atr_breakout = dataframe['atr'] > atr_ma # OPTION B.2: Volume Confirmation (breakout needs volume) # Volume > 1.5x 20-period MA = strong interest use_volume_confirm = flags.get('volume_confirmation', True) volume_confirmed = pd.Series(True, index=dataframe.index) if use_volume_confirm: vol_ma = dataframe['volume'].rolling(20).mean() volume_confirmed = dataframe['volume'] > (vol_ma * 1.5) # Mean Reversion filter - avoid entering at overbought (for longs) # Uses hyperopt params for dynamic thresholds not_overbought = pd.Series(True, index=dataframe.index) if '%-rsi_normalized' in dataframe.columns: # Convert normalized RSI back to 0-100 scale and check against param rsi = dataframe['%-rsi_normalized'] * 50 + 50 not_overbought = rsi < self.mr_overbought.value elif '%-mean_reversion_zone' in dataframe.columns: not_overbought = dataframe['%-mean_reversion_zone'] >= 0 # LONG ENTRY: # Option A: AI positive + Score >= threshold (AI-confirmed entry) # Option B: Score >= 0.6 (high confluence entry without AI) # PLUS: Must pass ATR breakout OR volume confirmation (one of the B filters) ai_confirmed = (ai_positive > 0) & (long_score >= self.entry_score_threshold.value) high_confluence = long_score >= 0.60 # Strong signals even without AI # ITER 19: TIGHTEN - Need BOTH filters (evidence: Iter 17 vs 18 trade-off) # Iter 17: strict = 71% WR, Iter 18: OR logic = 42% WR # AND logic = balance between quality and volume entry_confirmed = atr_breakout & volume_confirmed dataframe.loc[ has_volume & not_overbought & entry_confirmed & (ai_confirmed | high_confluence), 'enter_long'] = 1 # Log score distribution for debugging if len(dataframe) > 0: logger.info(f"LONG Score stats: mean={long_score.mean():.3f}, max={long_score.max():.3f}, entries={(has_volume & not_overbought & entry_confirmed & (ai_confirmed | high_confluence)).sum()}") # ===================================================== # SHORT ENTRY - WEIGHTED SCORING SYSTEM # ===================================================== use_trend_filter = flags.get('trend_filter', True) # 1. AI Prediction (30% weight) - REQUIRED BASE # Use ai_bearish from classification mode, or compute from regression if use_classification and classification_col in dataframe.columns: ai_negative = (dataframe[classification_col] == 'down').astype(float) # STRING: 'down' elif regression_col in dataframe.columns: ai_negative = (dataframe[regression_col] < self.sell_pred_threshold.value).astype(float) else: ai_negative = pd.Series(0, index=dataframe.index).astype(float) # 2. HTF Order Block - Bear (25% weight) htf_ob_bear = np.zeros(len(dataframe)) if use_htf_ob_confluence: if '%-testing_bear_ob_4h' in dataframe.columns: htf_ob_bear = (dataframe['%-testing_bear_ob_4h'] > 0).astype(float).values elif '%-testing_bear_ob_1h' in dataframe.columns: htf_ob_bear = (dataframe['%-testing_bear_ob_1h'] > 0).astype(float).values if '%-ob_fib_bear_confluence' in dataframe.columns: htf_ob_bear = np.maximum(htf_ob_bear, (dataframe['%-ob_fib_bear_confluence'] > 0).astype(float).values) # 3. ADX/Regime (10% weight) adx_short = np.zeros(len(dataframe)) if '%-ker_10' in dataframe.columns: adx_short = (dataframe['%-ker_10'] > 0.4).astype(float).values elif 'adx' in dataframe.columns: adx_short = (dataframe['adx'] > self.buy_adx_threshold.value).astype(float).values # 4. Momentum Quality - Bearish (20% weight) - CONTINUOUS score momentum_short = np.zeros(len(dataframe)) if '%-momentum_quality' in dataframe.columns: # Use negative momentum (inverted and clipped to [0, 1]) momentum_short = (-dataframe['%-momentum_quality']).clip(0, 1).values elif '%-momentum_confluence' in dataframe.columns: momentum_short = (dataframe['%-momentum_confluence'] < 0.6).astype(float).values # 5. Money Pressure - Selling (15% weight) pressure_short = np.zeros(len(dataframe)) if '%-money_pressure' in dataframe.columns: pressure_short = (dataframe['%-money_pressure'] < 0).astype(float).values # 6. Volatility Expansion (15% weight) - Uses hyperopt params volatility_short = np.zeros(len(dataframe)) if '%-volatility_expanding' in dataframe.columns: volatility_short = dataframe['%-volatility_expanding'].values elif '%-atr_pct' in dataframe.columns: # Compute dynamically using hyperopt params atr_pct = dataframe['%-atr_pct'] atr_mean = atr_pct.rolling(100).mean() volatility_ratio = atr_pct / (atr_mean + 1e-10) volatility_short = np.select( [volatility_ratio > self.volatility_high_threshold.value, volatility_ratio > self.volatility_mid_threshold.value], [1.0, 0.5], default=0.0 ) # 7. PREMIUM ZONE BONUS (15% weight) - NEW # Price in Fib 50-78.6% zone from LOW = expensive area (premium) # This rewards SHORT entries at "overpriced" levels, improving R/R premium_signal = np.zeros(len(dataframe)) # Look for fib_position with any timeframe suffix fib_pos_cols = [c for c in dataframe.columns if 'fib_position' in c and '%-' in c] if fib_pos_cols: fib_col = next((c for c in fib_pos_cols if '5m' in c or c == '%-fib_position'), fib_pos_cols[0]) fib_pos = dataframe[fib_col].values # Premium zone for SHORT: 0.50-0.78 (near swing high = expensive) premium_signal = np.where( (fib_pos >= 0.50) & (fib_pos <= 0.78), 1.0, # Full bonus in premium zone np.where(fib_pos > 0.78, 0.5, 0.0) # Partial if very expensive ) # Fallback: Use OB+Fib confluence indicator ob_fib_cols = [c for c in dataframe.columns if 'ob_fib_bear_confluence' in c] if premium_signal.sum() == 0 and ob_fib_cols: premium_signal = (dataframe[ob_fib_cols[0]] > 0).astype(float).values # Calculate total SHORT score # Weights adjusted: 20+20+15+15+15+15 = 100% short_score = ( ai_negative.values * 0.20 + # AI: 20% (was 25%) htf_ob_bear * 0.20 + # HTF OB: 20% (same) momentum_short * 0.15 + # Momentum: 15% (was 20%) pressure_short * 0.15 + # Pressure: 15% (same) volatility_short * 0.15 + # Volatility: 15% (was 20%) premium_signal * 0.15 # Premium Zone: 15% (NEW) ) # ============================================== # STRICTER SHORT FILTER: Only short in bearish markets # ============================================== # 1. EMA 200 filter (price < EMA 200) ema_filter = pd.Series(True, index=dataframe.index) if use_trend_filter: if '%-dist_to_ema_200' in dataframe.columns: ema_filter = dataframe['%-dist_to_ema_200'] < 0 elif 'ema_200' in dataframe.columns: ema_filter = dataframe['close'] < dataframe['ema_200'] else: ema_200 = ta.EMA(dataframe['close'], timeperiod=200) ema_filter = dataframe['close'] < ema_200 # 2. Trend direction filter (must be BEARISH or at least NEUTRAL) trend_filter = pd.Series(True, index=dataframe.index) if 'trend_direction' in dataframe.columns: # Only allow shorts when NOT in BULLISH trend trend_filter = dataframe['trend_direction'] != 'BULLISH' # 3. Mean Reversion filter - avoid entering at oversold (for shorts) # Uses hyperopt params for dynamic thresholds not_oversold = pd.Series(True, index=dataframe.index) if '%-rsi_normalized' in dataframe.columns: # Convert normalized RSI back to 0-100 scale and check against param rsi = dataframe['%-rsi_normalized'] * 50 + 50 not_oversold = rsi > self.mr_oversold.value elif '%-mean_reversion_zone' in dataframe.columns: not_oversold = dataframe['%-mean_reversion_zone'] <= 0 # SHORT ENTRY: Use same threshold as longs (trend filter provides protection) short_threshold = self.entry_score_threshold.value # SHORT ENTRY: # Option A: AI negative + Score >= threshold (AI-confirmed entry) # Option B: Score >= 0.6 (high confluence entry without AI) ai_short_confirmed = (ai_negative > 0) & (short_score >= short_threshold) high_short_confluence = short_score >= 0.60 dataframe.loc[ has_volume & ema_filter & trend_filter & not_oversold & (ai_short_confirmed | high_short_confluence), 'enter_short'] = 1 # Log score distribution if len(dataframe) > 0: short_entries = ((short_score >= short_threshold) & (ai_negative > 0) & ema_filter & trend_filter & not_oversold).sum() logger.info(f"SHORT Score stats: mean={short_score.mean():.3f}, max={short_score.max():.3f}, threshold={short_threshold:.2f}, entries={short_entries}") else: logger.warning("No prediction columns found (classification or regression) - FreqAI not working!") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Exit signal using 5-Layer Confluence System: LONG EXIT when ANY of these conditions: 1. AI Prediction < sell_pred_threshold (bearish forecast) 2. Trend confluence < 0.5 (trend weakening) 3. Momentum confluence < 0.4 (momentum fading) 4. Pattern net score becomes negative (bearish patterns forming) 5. RSI overbought (> sell_rsi_threshold) 6. SMC: Price at Order Block resistance / FVG filled 7. Extreme Fear (panic in market) SHORT EXIT when ANY of these conditions: 1. AI Prediction > buy_pred_threshold (bullish forecast) 2. Trend confluence > 0.5 (trend reversing up) 3. Momentum confluence > 0.6 (momentum turning bullish) 4. Pattern net score becomes positive (bullish patterns forming) 5. RSI oversold (< buy_rsi_low) 6. SMC: Price at Order Block support / Bullish FVG """ if self.config['freqai']['enabled']: # Regression target column (predicts % price change) prediction_col = '&-price_change_pct' if prediction_col in dataframe.columns: # ===================================================== # LONG EXIT CONDITIONS # ===================================================== # AI prediction bearish exit_prediction = ( (dataframe[prediction_col] < self.sell_pred_threshold.value) & (dataframe['volume'] > 0) ) # Trend confluence weakening exit_trend = False if '%-trend_confluence' in dataframe.columns: exit_trend = dataframe['%-trend_confluence'] < 0.4 # Momentum fading exit_momentum = False if '%-momentum_confluence' in dataframe.columns: exit_momentum = dataframe['%-momentum_confluence'] < 0.3 # Money pressure turning negative (selling) exit_pressure = False if '%-money_pressure' in dataframe.columns: exit_pressure = dataframe['%-money_pressure'] < -0.3 # Bearish pattern forming exit_pattern = False if '%-pattern_net_score' in dataframe.columns: exit_pattern = dataframe['%-pattern_net_score'] < -1 # RSI overbought exit_rsi = False if 'rsi' in dataframe.columns: exit_rsi = dataframe['rsi'] > self.sell_rsi_threshold.value # SMC: At resistance (Order Block bear) or Bearish FVG exit_smc = False if '%-fvg_bear' in dataframe.columns: exit_smc = dataframe['%-fvg_bear'] == 1 elif '%-order_block_bear' in dataframe.columns: exit_smc = dataframe['%-order_block_bear'] == 1 # Extreme Fear (market panic - exit to safety) exit_fear = False if '%-is_extreme_fear' in dataframe.columns: exit_fear = dataframe['%-is_extreme_fear'] == 1 # Combine LONG exit conditions (THRESHOLD-based, need 3+ conditions) # Rationale: OR logic had only 30.4% win rate, exiting too early # Changed from: ANY condition triggers exit # Changed to: Need 3+ conditions to trigger exit (more conservative) exit_count = ( exit_prediction.astype(int) + (exit_trend if isinstance(exit_trend, pd.Series) else pd.Series(exit_trend, index=dataframe.index)).astype(int) + (exit_momentum if isinstance(exit_momentum, pd.Series) else pd.Series(exit_momentum, index=dataframe.index)).astype(int) + (exit_pressure if isinstance(exit_pressure, pd.Series) else pd.Series(exit_pressure, index=dataframe.index)).astype(int) + (exit_pattern if isinstance(exit_pattern, pd.Series) else pd.Series(exit_pattern, index=dataframe.index)).astype(int) + (exit_rsi if isinstance(exit_rsi, pd.Series) else pd.Series(exit_rsi, index=dataframe.index)).astype(int) ) dataframe.loc[exit_count >= 3, 'exit_long'] = 1 # ===================================================== # SHORT EXIT CONDITIONS (mirror of Long exit) # ===================================================== # AI prediction bullish short_exit_prediction = ( (dataframe[prediction_col] > self.buy_pred_threshold.value) & (dataframe['volume'] > 0) ) # Trend confluence turning bullish short_exit_trend = False if '%-trend_confluence' in dataframe.columns: short_exit_trend = dataframe['%-trend_confluence'] > 0.6 # Momentum turning bullish short_exit_momentum = False if '%-momentum_confluence' in dataframe.columns: short_exit_momentum = dataframe['%-momentum_confluence'] > 0.7 # Money pressure positive (buying coming in) short_exit_pressure = False if '%-money_pressure' in dataframe.columns: short_exit_pressure = dataframe['%-money_pressure'] > 0.3 # Bullish pattern forming short_exit_pattern = False if '%-pattern_net_score' in dataframe.columns: short_exit_pattern = dataframe['%-pattern_net_score'] > 1 # RSI oversold (potential bounce) short_exit_rsi = False if 'rsi' in dataframe.columns: short_exit_rsi = dataframe['rsi'] < self.buy_rsi_low.value # SMC: At support (Order Block bull) or Bullish FVG short_exit_smc = False if '%-fvg_bull' in dataframe.columns: short_exit_smc = dataframe['%-fvg_bull'] == 1 elif '%-order_block_bull' in dataframe.columns: short_exit_smc = dataframe['%-order_block_bull'] == 1 # Extreme Fear (market bottoming, shorts may get squeezed) short_exit_fear = False if '%-is_extreme_fear' in dataframe.columns: short_exit_fear = dataframe['%-is_extreme_fear'] == 1 # Combine SHORT exit conditions (ANY triggers exit) dataframe.loc[ short_exit_prediction | short_exit_trend | short_exit_momentum | short_exit_pressure | short_exit_pattern | short_exit_rsi | short_exit_smc | short_exit_fear, 'exit_short'] = 1 return dataframe