# source: https://raw.githubusercontent.com/acsvalentinacs/HOPE-Minibot/65e927a2984b68dfce48fafe50b7a35813bd1102/hope_ml/HopeStrategyV10.py # -*- coding: utf-8 -*- """ HopeStrategy V10.1 — ML-Enhanced Scalping ========================================== V9 entry conditions + Guard pipeline + ML ensemble filter. V10.1 fixes (2026-02-07): - FIX #2: ML Gate relaxed via hope_ml_signal.py changes - FIX #4: AI Learner feedback — better error logging, correct entry_day format - FIX #5: Spam log throttle — deduplicate repeated block messages per pair V9 Entry: RSI<55 + EMA7>EMA16 + Volume>=0.5x avg V10 Entry: V9 -> SessionLimits -> AntiChase -> AILearner -> ML(1/3 vote) -> ENTER Guards pipeline (cheap first, heavy last): 1. SessionLimits — circuit breaker (3 consecutive losses -> 30min pause) 2. AntiChase — block pump chasing (+1.5% in 3min) 3. AI Learner — learned pair/hour blacklist 4. ML Gate — ensemble prediction (1/3 vote + confidence sizing) + News Guard EXIT -> feedback -> GuardStats + AI Learner + SessionLimits Author: HOPE AI System Date: 2026-02-07 """ import logging from datetime import datetime, timezone from typing import Dict, Optional import numpy as np import pandas as pd from freqtrade.persistence import Trade from freqtrade.strategy import IStrategy, merge_informative_pair from pandas import DataFrame logger = logging.getLogger(__name__) # Try to import ML modules (graceful fallback if not deployed) ML_AVAILABLE = False try: import sys sys.path.insert(0, '/freqtrade/user_data/strategies/hope_ml') from hope_ml_signal import MLSignalGenerator ML_AVAILABLE = True logger.info("[V10] ML Signal Generator loaded successfully") except ImportError as e: logger.warning(f"[V10] ML not available, running in V9 mode: {e}") # Try to import Guards (graceful fallback) GUARDS_AVAILABLE = False try: sys.path.insert(0, '/freqtrade/user_data/strategies') from guards.session_limits import SessionLimits from guards.anti_chase import AntiChaseFilter from guards.ai_learner import AITradeLearner from guards.stats_collector import GuardStats GUARDS_AVAILABLE = True logger.info("[V10] Guards loaded successfully") except ImportError as e: logger.warning(f"[V10] Guards not available: {e}") # Spam throttle: log ML blocks at most once per this many seconds per pair ML_BLOCK_LOG_THROTTLE = 300 # 5 minutes class Github_acsvalentinacs_HOPE_Minibot__HopeStrategyV10__20260218_115300(IStrategy): """ HOPE Strategy V10 — ML-Enhanced Scalping with Guard Pipeline. """ INTERFACE_VERSION = 3 timeframe = '5m' can_short = False minimal_roi = { "0": 0.015, # 1.5% "30": 0.010, # 1.0% after 30 min "60": 0.005, # 0.5% after 1h "120": 0.003, # 0.3% after 2h } stoploss = -0.02 trailing_stop = True trailing_stop_positive = 0.005 trailing_stop_positive_offset = 0.01 trailing_only_offset_is_reached = True startup_candle_count = 200 process_only_new_candles = True # V10 config use_ml = True ml_use_dynamic_sizing = True ml_fallback_to_v9 = True # Counters _ml_entries = 0 _v9_entries = 0 _ml_blocks = 0 _guard_blocks = 0 def __init__(self, config: dict) -> None: super().__init__(config) self.ml_signal = None if self.use_ml and ML_AVAILABLE: try: self.ml_signal = MLSignalGenerator(config) logger.info("[V10] ML Signal Generator initialized") except Exception as e: logger.error(f"[V10] Failed to init ML: {e}") if self.ml_signal is None and not self.ml_fallback_to_v9: logger.error("[V10] ML not available and fallback disabled — FAIL-CLOSED") # Initialize Guards (each in try/except — no single point of failure) self.session_limits = None self.anti_chase = None self.ai_learner = None self.guard_stats = None if GUARDS_AVAILABLE: try: self.session_limits = SessionLimits() logger.info("[V10] Guard: SessionLimits initialized") except Exception as e: logger.error(f"[V10] SessionLimits init failed: {e}") try: self.anti_chase = AntiChaseFilter() logger.info("[V10] Guard: AntiChaseFilter initialized") except Exception as e: logger.error(f"[V10] AntiChaseFilter init failed: {e}") try: self.ai_learner = AITradeLearner() logger.info("[V10] Guard: AITradeLearner initialized") except Exception as e: logger.error(f"[V10] AITradeLearner init failed: {e}") try: self.guard_stats = GuardStats() logger.info("[V10] Guard: StatsCollector initialized") except Exception as e: logger.error(f"[V10] GuardStats init failed: {e}") # FIX #5: Spam log throttle — {pair: last_block_timestamp} self._last_ml_block_log = {} self._last_guard_block_log = {} # ML decision cache: (pair, decision_dict) — set in custom_stake_amount, read in confirm_trade_entry self._ml_decision_cache = (None, None) def informative_pairs(self): return [("BTC/USDT", self.timeframe)] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Calculate indicators for V9 rules + ML features.""" # === BTC INFORMATIVE DATA === if metadata.get('pair') != 'BTC/USDT': btc_df = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe=self.timeframe) if btc_df is not None and not btc_df.empty: btc_delta = btc_df['close'].diff() btc_gain = btc_delta.where(btc_delta > 0, 0.0).rolling(14).mean() btc_loss = (-btc_delta.where(btc_delta < 0, 0.0)).rolling(14).mean() btc_rs = btc_gain / btc_loss.replace(0, np.nan) btc_df['btc_rsi'] = 100 - (100 / (1 + btc_rs)) btc_df['btc_return_1'] = btc_df['close'].pct_change(1) * 100 btc_df['btc_return_12'] = btc_df['close'].pct_change(12) * 100 btc_df['btc_volume_sma'] = btc_df['volume'].rolling(20).mean() btc_df['btc_volume_ratio'] = btc_df['volume'] / btc_df['btc_volume_sma'].replace(0, np.nan) btc_tr = pd.concat([ btc_df['high'] - btc_df['low'], (btc_df['high'] - btc_df['close'].shift()).abs(), (btc_df['low'] - btc_df['close'].shift()).abs(), ], axis=1).max(axis=1) btc_df['btc_atr_pct'] = (btc_tr.rolling(14).mean() / btc_df['close']) * 100 btc_plus_dm = btc_df['high'].diff() btc_minus_dm = -btc_df['low'].diff() btc_plus_dm = btc_plus_dm.where((btc_plus_dm > btc_minus_dm) & (btc_plus_dm > 0), 0) btc_minus_dm = btc_minus_dm.where((btc_minus_dm > btc_plus_dm) & (btc_minus_dm > 0), 0) btc_atr14 = btc_tr.rolling(14).mean() btc_plus_di = 100 * (btc_plus_dm.rolling(14).mean() / btc_atr14.replace(0, np.nan)) btc_minus_di = 100 * (btc_minus_dm.rolling(14).mean() / btc_atr14.replace(0, np.nan)) btc_dx = 100 * (btc_plus_di - btc_minus_di).abs() / (btc_plus_di + btc_minus_di).replace(0, np.nan) btc_df['btc_adx'] = btc_dx.rolling(14).mean() btc_df['btc_correlation'] = dataframe['close'].rolling(20).corr(btc_df['close']) for col in ['btc_rsi', 'btc_return_1', 'btc_return_12', 'btc_volume_ratio', 'btc_atr_pct', 'btc_adx', 'btc_correlation']: if col in btc_df.columns: dataframe[col] = btc_df[col].values[:len(dataframe)] if len(btc_df) >= len(dataframe) else 0.0 # === V9 INDICATORS === delta = dataframe['close'].diff() gain = delta.where(delta > 0, 0.0).rolling(14).mean() loss = (-delta.where(delta < 0, 0.0)).rolling(14).mean() rs = gain / loss.replace(0, np.nan) dataframe['rsi'] = 100 - (100 / (1 + rs)) gain7 = delta.where(delta > 0, 0.0).rolling(7).mean() loss7 = (-delta.where(delta < 0, 0.0)).rolling(7).mean() rs7 = gain7 / loss7.replace(0, np.nan) dataframe['rsi_7'] = 100 - (100 / (1 + rs7)) dataframe['ema_7'] = dataframe['close'].ewm(span=7, adjust=False).mean() dataframe['ema_16'] = dataframe['close'].ewm(span=16, adjust=False).mean() dataframe['ema_50'] = dataframe['close'].ewm(span=50, adjust=False).mean() dataframe['ema_200'] = dataframe['close'].ewm(span=200, adjust=False).mean() dataframe['sma_20'] = dataframe['close'].rolling(20).mean() dataframe['ema7_dist'] = ((dataframe['close'] - dataframe['ema_7']) / dataframe['ema_7']) * 100 dataframe['ema16_dist'] = ((dataframe['close'] - dataframe['ema_16']) / dataframe['ema_16']) * 100 dataframe['ema50_dist'] = ((dataframe['close'] - dataframe['ema_50']) / dataframe['ema_50']) * 100 dataframe['ema_cross'] = (dataframe['ema_7'] > dataframe['ema_16']).astype(int) dataframe['volume_sma'] = dataframe['volume'].rolling(20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'].replace(0, np.nan) if 'taker_buy_base' in dataframe.columns: dataframe['buy_pressure'] = dataframe['taker_buy_base'] / dataframe['volume'].replace(0, np.nan) else: dataframe['buy_pressure'] = 0.5 tr = pd.concat([ dataframe['high'] - dataframe['low'], (dataframe['high'] - dataframe['close'].shift()).abs(), (dataframe['low'] - dataframe['close'].shift()).abs(), ], axis=1).max(axis=1) dataframe['atr'] = tr.rolling(14).mean() dataframe['atr_pct'] = (dataframe['atr'] / dataframe['close']) * 100 bb_std = dataframe['close'].rolling(20).std() dataframe['bb_upper'] = dataframe['sma_20'] + 2 * bb_std dataframe['bb_lower'] = dataframe['sma_20'] - 2 * bb_std dataframe['bb_width'] = ((dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['sma_20']) * 100 dataframe['bb_position'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) ema12 = dataframe['close'].ewm(span=12, adjust=False).mean() ema26 = dataframe['close'].ewm(span=26, adjust=False).mean() dataframe['macd'] = ema12 - ema26 dataframe['macd_signal'] = dataframe['macd'].ewm(span=9, adjust=False).mean() dataframe['macd_hist'] = dataframe['macd'] - dataframe['macd_signal'] plus_dm = dataframe['high'].diff() minus_dm = -dataframe['low'].diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0) atr14 = tr.rolling(14).mean() dataframe['plus_di'] = 100 * (plus_dm.rolling(14).mean() / atr14.replace(0, np.nan)) dataframe['minus_di'] = 100 * (minus_dm.rolling(14).mean() / atr14.replace(0, np.nan)) dx = 100 * (dataframe['plus_di'] - dataframe['minus_di']).abs() / (dataframe['plus_di'] + dataframe['minus_di']).replace(0, np.nan) dataframe['adx'] = dx.rolling(14).mean() dataframe['return_1'] = dataframe['close'].pct_change(1) * 100 dataframe['return_3'] = dataframe['close'].pct_change(3) * 100 dataframe['return_6'] = dataframe['close'].pct_change(6) * 100 dataframe['return_12'] = dataframe['close'].pct_change(12) * 100 dataframe['volatility_12'] = dataframe['return_1'].rolling(12).std() dataframe['volatility_48'] = dataframe['return_1'].rolling(48).std() dataframe['spread_pct'] = ((dataframe['high'] - dataframe['low']) / dataframe['close']) * 100 body = (dataframe['close'] - dataframe['open']).abs() wick = dataframe['high'] - dataframe['low'] dataframe['body_ratio'] = body / wick.replace(0, np.nan) dataframe['momentum_score'] = ( (dataframe['rsi_7'] - 50) / 50 * 0.3 + dataframe['ema7_dist'].clip(-5, 5) / 5 * 0.3 + (dataframe['macd_hist'] / dataframe['close'] * 1000).clip(-5, 5) / 5 * 0.2 + (dataframe['volume_ratio'] - 1).clip(-2, 2) / 2 * 0.2 ) if 'date' in dataframe.columns: dataframe['hour'] = pd.to_datetime(dataframe['date']).dt.hour dataframe['day_of_week'] = pd.to_datetime(dataframe['date']).dt.dayofweek else: dataframe['hour'] = 0 dataframe['day_of_week'] = 0 dataframe['is_weekend'] = (dataframe['day_of_week'] >= 5).astype(int) dataframe['session_asia'] = dataframe['hour'].isin(range(0, 8)).astype(int) dataframe['session_europe'] = dataframe['hour'].isin(range(8, 16)).astype(int) dataframe['session_us'] = dataframe['hour'].isin(range(13, 22)).astype(int) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """V9 entry conditions. ML filtering in confirm_trade_entry.""" dataframe.loc[ ( (dataframe['rsi'] < 55) & (dataframe['ema_7'] > dataframe['ema_16']) & (dataframe['volume_ratio'] >= 0.5) & (dataframe['volume'] > 0) ), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe['rsi'] > 81), 'exit_long' ] = 1 return dataframe def _should_log_block(self, cache_dict: dict, pair: str) -> bool: """Check if we should log this block (throttle to once per 5 min per pair).""" import time now = time.time() last = cache_dict.get(pair, 0) if now - last >= ML_BLOCK_LOG_THROTTLE: cache_dict[pair] = now return True return False def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs) -> bool: """ V10 Guard Pipeline (cheap -> heavy): 1. SessionLimits — circuit breaker 2. AntiChase — block pump chasing 3. AI Learner — learned pair/hour filter 4. ML Gate — ensemble vote + confidence sizing + News Guard """ now = datetime.now(timezone.utc) # GUARD 1: SessionLimits if self.session_limits: try: closed_trades = Trade.get_trades_proxy(is_open=False) can, reason = self.session_limits.can_trade(closed_trades) if not can: self._guard_blocks += 1 if self.guard_stats: self.guard_stats.record_block("session_limits", pair, reason) if self._should_log_block(self._last_guard_block_log, f"sl_{pair}"): logger.info(f"[V10] {pair} BLOCKED by SessionLimits: {reason}") return False if self.guard_stats: self.guard_stats.record_pass("session_limits", pair) except Exception as e: logger.error(f"[V10] SessionLimits check failed: {e}") # GUARD 2: AntiChase if self.anti_chase: try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is not None and not dataframe.empty: is_chasing, chase_reason = self.anti_chase.is_chasing(dataframe, -1, pair) if is_chasing: self._guard_blocks += 1 if self.guard_stats: self.guard_stats.record_block("anti_chase", pair, chase_reason) logger.info(f"[V10] {pair} BLOCKED by AntiChase: {chase_reason}") return False if self.guard_stats: self.guard_stats.record_pass("anti_chase", pair) except Exception as e: logger.error(f"[V10] AntiChase check failed: {e}") # GUARD 3: AI Learner if self.ai_learner: try: can_trade, ai_reason, ai_conf = self.ai_learner.should_trade( pair, now.hour, now.strftime('%A') ) if not can_trade: self._guard_blocks += 1 if self.guard_stats: self.guard_stats.record_block("ai_learner", pair, ai_reason) if self._should_log_block(self._last_guard_block_log, f"ai_{pair}"): logger.info(f"[V10] {pair} BLOCKED by AI Learner: {ai_reason}") return False if self.guard_stats: self.guard_stats.record_pass("ai_learner", pair) except Exception as e: logger.error(f"[V10] AI Learner check failed: {e}") # GUARD 4: ML Gate — use cached decision from custom_stake_amount (runs first) if self.ml_signal and self.use_ml: try: cached_pair, decision = self._ml_decision_cache if cached_pair != pair or decision is None: # Cache miss — compute fresh (shouldn't happen in normal flow) dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is not None and not dataframe.empty: decision = self.ml_signal.should_enter(pair, dataframe) # Clear cache after use self._ml_decision_cache = (None, None) if decision and not decision['enter']: self._ml_blocks += 1 if self.guard_stats: self.guard_stats.record_block("ml_gate", pair, decision['reason']) if self._should_log_block(self._last_ml_block_log, pair): logger.info(f"[V10] {pair} BLOCKED by ML: {decision['reason']} " f"(conf={decision['confidence']:.3f})") return False if decision and decision['enter']: self._ml_entries += 1 if self.guard_stats: self.guard_stats.record_pass("ml_gate", pair) logger.info(f"[V10] {pair} ML APPROVED: conf={decision['confidence']:.3f}, " f"size={decision['size_multiplier']:.2f}x") except Exception as e: logger.error(f"[V10] ML check failed: {e}") self._ml_decision_cache = (None, None) if not self.ml_fallback_to_v9: return False self._v9_entries += 1 return True 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: """ Dynamic stake sizing based on ML confidence. NOTE: Freqtrade calls this BEFORE confirm_trade_entry. So we pre-compute ML decision here and cache it for confirm_trade_entry. """ multiplier = 1.0 if self.ml_signal and self.use_ml and self.ml_use_dynamic_sizing: try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is not None and not dataframe.empty: decision = self.ml_signal.should_enter(pair, dataframe) # Cache for confirm_trade_entry (avoid double ML computation) self._ml_decision_cache = (pair, decision) if decision['enter']: multiplier = decision.get('size_multiplier', 1.0) # If ML blocks, multiplier stays 1.0 (irrelevant — trade will be blocked) except Exception as e: logger.error(f"[V10] ML sizing failed: {e}") self._ml_decision_cache = (None, None) adjusted_stake = proposed_stake * multiplier adjusted_stake = max(adjusted_stake, min_stake or 0) adjusted_stake = min(adjusted_stake, max_stake) if multiplier != 1.0: logger.info(f"[V10] {pair} stake: ${proposed_stake:.2f} x {multiplier:.2f} = ${adjusted_stake:.2f}") return adjusted_stake def confirm_trade_exit(self, pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, **kwargs) -> bool: """ Feedback loop: record trade result to all learning systems. FIX #4: Better error logging + correct entry_day format for AI Learner. """ profit_ratio = trade.calc_profit_ratio(rate) hold_minutes = (current_time - trade.open_date_utc).total_seconds() / 60 entry_hour = trade.open_date_utc.hour entry_day = trade.open_date_utc.strftime('%A') logger.info(f"[V10] {pair} EXIT: {exit_reason}, " f"profit={profit_ratio*100:.2f}%, hold={hold_minutes:.0f}min") # FEEDBACK: GuardStats if self.guard_stats: try: self.guard_stats.record_outcome( pair, profit_ratio, hold_minutes, exit_reason, entry_hour ) logger.info(f"[V10] GuardStats recorded: {pair} {profit_ratio*100:+.2f}%") except Exception as e: logger.error(f"[V10] GuardStats feedback FAILED for {pair}: {e}") # FEEDBACK: AI Learner if self.ai_learner: try: self.ai_learner.record_trade( pair=pair, profit_ratio=profit_ratio, hold_minutes=hold_minutes, exit_reason=exit_reason, entry_hour=entry_hour, entry_day=entry_day, ) logger.info(f"[V10] AI Learner recorded: {pair} {profit_ratio*100:+.2f}%") except Exception as e: logger.error(f"[V10] AI Learner feedback FAILED for {pair}: {e}") # FEEDBACK: SessionLimits if self.session_limits: try: self.session_limits.record_trade(profit_ratio * 100) except Exception as e: logger.error(f"[V10] SessionLimits feedback FAILED: {e}") return True def bot_loop_start(self, current_time: datetime = None, **kwargs) -> None: """Hourly status log.""" if current_time and current_time.minute == 0 and current_time.second < 60: logger.info(f"[V10 HEARTBEAT] ML entries={self._ml_entries}, " f"V9 entries={self._v9_entries}, " f"ML blocks={self._ml_blocks}, " f"Guard blocks={self._guard_blocks}, " f"ML active={'YES' if self.ml_signal else 'NO'}, " f"Guards active={'YES' if GUARDS_AVAILABLE else 'NO'}") if self.ml_signal: logger.info(f"[V10 ML STATS] {self.ml_signal.get_stats()}") if self.guard_stats: logger.info(f"[V10 GUARD STATS] {self.guard_stats.get_report()}")