# source: https://raw.githubusercontent.com/FlyerStud/apex-bot-2026-syntax-fix/c3a6b9aa94d716d20266a40f65b8b5d653086ede/user_data/strategies/ApexBot2026.py """ Github_FlyerStud_apex_bot_2026_syntax_fix__ApexBot2026__20260413_181726 — Freqtrade Strategy EV-Maximized, Regime-Adaptive, Multi-Strategy Hybrid Bot HOW IMPORTS WORK ---------------- Freqtrade loads this file from: /user_data/strategies/Github_FlyerStud_apex_bot_2026_syntax_fix__ApexBot2026__20260413_181726.py The custom modules live at: /modules/ sys.path is extended at startup (below) to make `import modules.xxx` work regardless of the working directory Freqtrade uses. """ # ─── PATH FIX — must be first, before any project imports ───────────────────── import sys from pathlib import Path # Navigate from user_data/strategies/ → project root _STRATEGY_DIR = Path(__file__).resolve().parent # .../user_data/strategies _PROJECT_ROOT = _STRATEGY_DIR.parent.parent # .../apex-bot-2026 if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) # ────────────────────────────────────────────────────────────────────────────── import logging import numpy as np import pandas as pd import pandas_ta as ta from typing import Optional, Dict from freqtrade.strategy import (IStrategy, DecimalParameter, IntParameter, BooleanParameter) from freqtrade.persistence import Trade from pandas import DataFrame # Project modules (importable now that _PROJECT_ROOT is on sys.path) from modules.regime_detector import ( RegimeDetector, RegimeResult as DetectorRegimeResult, RegimeType, ScalingParams as RegimeScalingParams, ) from modules.signal_engine import SignalEngine from modules.ev_calculator import EVCalculator, EVInput, SignalBundle as EVBundle, RegimeResult as EVRegime from modules.risk_manager import RiskAgent, ProposedTrade, ScalingParams from modules.ml_pipeline import MLPipeline logger = logging.getLogger(__name__) class Github_FlyerStud_apex_bot_2026_syntax_fix__ApexBot2026__20260413_181726(IStrategy): """ APEX-BOT 2026 — Hybrid Multi-Strategy Trading System. Architecture: Module 1 RegimeDetector → BULL_STRONG / BULL_WEAK / SIDEWAYS / BEAR / HIGH_VOL Module 2 SignalEngine → 5 strategies scored -1 to +1 Module 3 MLPipeline → LightGBM + XGBoost direction + probability Module 4 EVCalculator → Expected Value decision + sizing Module 5 RiskAgent → Stoploss / trailing / circuit breakers """ # ── Freqtrade metadata ──────────────────────────────────────────────────── INTERFACE_VERSION = 3 timeframe = "1h" startup_candle_count: int = 500 # ML training needs more data can_short = False # spot default; set True for futures config # ── ROI / stoploss (2x LEVERAGE OPTIMIZED) ──── minimal_roi = {"0": 0.50, "720": 0.20, "1440": 0.05, "2880": 0} stoploss = -0.025 trailing_stop = False trailing_stop_positive = None process_only_new_candles = True # ── Order types (required for FreqTrade 2026.3) ─────────────────────────── order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False, "emergency_exit": "market", "force_exit": "market", "force_entry": "market" } order_time_in_force = { "entry": "gtc", "exit": "gtc", "stoploss": "gtc" } # ── Additional FreqTrade 2026.3 requirements ────────────────────────────── use_custom_stoploss = False position_stacking = False # ── Hyperopt-tunable parameters (✅ OPTIMIZED) ──────── min_ev_pct = DecimalParameter(0.0005, 0.003, default=0.0006, space="buy", optimize=True) disable_ml_req = BooleanParameter(default=False, space="buy", optimize=True) min_ml_conf = DecimalParameter(0.25, 0.50, default=0.261, space="buy", optimize=True) # ✅ 0.28 → 0.261 min_signal_score = DecimalParameter(0.25, 0.50, default=0.265, space="buy", optimize=True) # ✅ 0.28 → 0.265 stoploss_atr_mult= DecimalParameter(1.0, 3.0, default=2.359, space="sell", optimize=False) rr_target = DecimalParameter(1.5, 4.0, default=2.186, space="sell", optimize=False) # ── Internal agents (initialised once) ─────────────────────────────────── _regime_detector : Optional[RegimeDetector] = None _signal_engine : Optional[SignalEngine] = None _ev_calc : Optional[EVCalculator] = None _risk_agent : Optional[RiskAgent] = None _ml_pipe : Optional[MLPipeline] = None # ── Daily P&L tracking for loss limits ──────────────────────────────────── _daily_pnl_cache = {} # Format: {date_str: pnl_value} _last_update_day = None # Track date changes def __init__(self, config: dict) -> None: super().__init__(config) cfg_path = str(_PROJECT_ROOT / "apex_config.yml") runmode = str(config.get("runmode", "")).lower() ml_config: Dict[str, float | int] = {} # Backtesting and dry-run need a looser WFO gate so startup can train a model. if "backtest" in runmode or "dry_run" in runmode: ml_config.update({ "wfo_min_windows": 1, "wfo_min_dir_acc": 0.0, "wfo_max_dir_std": 1.0, "allow_bootstrap_no_wfo": True, "bootstrap_min_rows": 500, }) self._regime_detector = RegimeDetector(cfg_path) self._signal_engine = SignalEngine() self._ev_calc = EVCalculator() self._risk_agent = RiskAgent(cfg_path) self._ml_pipe = MLPipeline( model_dir = str(_PROJECT_ROOT / "models" / "ml"), log_dir = str(_PROJECT_ROOT / "logs" / "ml"), config = ml_config, ) logger.info("Github_FlyerStud_apex_bot_2026_syntax_fix__ApexBot2026__20260413_181726 initialised — all agents ready") # ── Informative timeframes ──────────────────────────────────────────────── def informative_pairs(self): """Declare the extra timeframes we consume in populate_indicators.""" pairs = self.dp.current_whitelist() return [(p, tf) for p in pairs for tf in ("4h", "1d")] # ── Indicators ──────────────────────────────────────────────────────────── def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Compute all technical indicators and run the full agent pipeline. Results are stored as columns on the dataframe so Freqtrade can persist them across candles. """ pair = metadata["pair"] # ── Check daily P&L limit (improvement #1) ──────────────────────────── # Cache daily P&L to avoid excessive Trade queries # Note: This works in dry_run and live modes; in backtest, trades are simulated from datetime import datetime, timezone try: current_date = datetime.now(timezone.utc).date() if self._last_update_day != current_date: # New day: recalculate P&L from trades self._last_update_day = current_date daily_trades = [t for t in Trade.get_trades_proxy() if t.close_date and t.close_date.date() == current_date] daily_pnl = sum(t.close_profit * t.stake_amount for t in daily_trades) if daily_trades else 0.0 self._daily_pnl_cache[str(current_date)] = daily_pnl if daily_pnl < -0.02: # -2% daily loss threshold logger.warning(f"⚠️ Daily P&L limit hit: {daily_pnl:.2%} < -2%. Blocking new entries.") except Exception as e: # Backtest mode or missing Trade data - skip silently pass # Initialize pipeline result columns with defaults if 'ev_verdict' not in dataframe.columns: dataframe['ev_verdict'] = "APPROVE_STANDARD" if 'ev_score' not in dataframe.columns: dataframe['ev_score'] = 0.002 if 'ml_confidence' not in dataframe.columns: dataframe['ml_confidence'] = 0.5 if 'composite_score' not in dataframe.columns: dataframe['composite_score'] = 0.5 # Default moderate signal if 'signal_quality' not in dataframe.columns: dataframe['signal_quality'] = "MODERATE" if 'trade_dir' not in dataframe.columns: dataframe['trade_dir'] = "long" if 'is_hv' not in dataframe.columns: dataframe['is_hv'] = 0.0 if 'disable_ml_req' not in dataframe.columns: dataframe['disable_ml_req'] = float(self.disable_ml_req.value) # ── Base indicators needed by all agents ───────────────────────────── dataframe["ema_20"] = ta.ema(dataframe["close"], length=20) dataframe["ema_50"] = ta.ema(dataframe["close"], length=50) dataframe["ema_200"] = ta.ema(dataframe["close"], length=200) adx = ta.adx(dataframe["high"], dataframe["low"], dataframe["close"], length=14) dataframe["adx_14"] = adx["ADX_14"] dataframe["adx_plus_di"] = adx["DMP_14"] dataframe["adx_minus_di"]= adx["DMN_14"] dataframe["rsi_14"] = ta.rsi(dataframe["close"], length=14) dataframe["atr_14"] = ta.atr(dataframe["high"], dataframe["low"], dataframe["close"], length=14) # ATR ratio: current volatility vs 20-bar average — spikes signal stop-loss risk dataframe["atr_ratio"] = dataframe["atr_14"] / dataframe["atr_14"].rolling(20).mean().replace(0, float("nan")) macd = ta.macd(dataframe["close"]) dataframe["macd"] = macd["MACD_12_26_9"] dataframe["macd_signal"] = macd["MACDs_12_26_9"] dataframe["macd_hist"] = macd["MACDh_12_26_9"] bb = ta.bbands(dataframe["close"], length=20, std=2.0) dataframe["bb_upper"] = bb["BBU_20_2.0"] dataframe["bb_middle"] = bb["BBM_20_2.0"] dataframe["bb_lower"] = bb["BBL_20_2.0"] dataframe["volume_ma_20"] = dataframe["volume"].rolling(20).mean() # Backtests evaluate the whole dataframe, while the full pipeline only updates # the last candle. Provide a conservative vectorized regime baseline so # historical rows are not left with NaN regime values. bull_strong_mask = ( (dataframe["ema_20"] > dataframe["ema_50"]) & (dataframe["close"] > dataframe["ema_200"]) & (dataframe["adx_14"] >= 25) ) bull_weak_mask = ( (dataframe["ema_20"] > dataframe["ema_50"]) & (dataframe["close"] > dataframe["ema_200"]) ) bear_strong_mask = ( (dataframe["ema_20"] < dataframe["ema_50"]) & (dataframe["close"] < dataframe["ema_200"]) & (dataframe["adx_14"] >= 25) ) bear_weak_mask = ( (dataframe["ema_20"] < dataframe["ema_50"]) & (dataframe["close"] < dataframe["ema_200"]) ) regime_proxy = np.select( [bull_strong_mask, bull_weak_mask, bear_strong_mask, bear_weak_mask], ["BULL_STRONG", "BULL_WEAK", "BEAR_STRONG", "BEAR_WEAK"], default="SIDEWAYS", ) regime_conf_proxy = np.select( [bull_strong_mask | bear_strong_mask, bull_weak_mask | bear_weak_mask], [0.60, 0.45], default=0.25, ) transition_risk_proxy = np.where( dataframe["adx_14"].fillna(0) < 18, 0.70, 0.35, ) if "regime" not in dataframe.columns: dataframe["regime"] = regime_proxy else: dataframe["regime"] = dataframe["regime"].fillna(pd.Series(regime_proxy, index=dataframe.index)) if "regime_confidence" not in dataframe.columns: dataframe["regime_confidence"] = regime_conf_proxy else: dataframe["regime_confidence"] = dataframe["regime_confidence"].fillna(pd.Series(regime_conf_proxy, index=dataframe.index)) if "transition_risk" not in dataframe.columns: dataframe["transition_risk"] = transition_risk_proxy else: dataframe["transition_risk"] = dataframe["transition_risk"].fillna(pd.Series(transition_risk_proxy, index=dataframe.index)) if "regime_encoded" not in dataframe.columns: dataframe["regime_encoded"] = pd.Series(regime_proxy, index=dataframe.index).map( {"BULL_STRONG": 2, "BULL_WEAK": 1, "SIDEWAYS": 0, "BEAR_WEAK": -1, "BEAR_STRONG": -2} ).fillna(0) else: dataframe["regime_encoded"] = dataframe["regime_encoded"].fillna( pd.Series(regime_proxy, index=dataframe.index).map( {"BULL_STRONG": 2, "BULL_WEAK": 1, "SIDEWAYS": 0, "BEAR_WEAK": -1, "BEAR_STRONG": -2} ).fillna(0) ) # ── Multi-timeframe data from informative pairs ─────────────────────── df_4h = self._get_informative(pair, "4h", dataframe) df_1d = self._get_informative(pair, "1d", dataframe) # ── Run agent pipeline on the LAST candle only (avoid recompute) ───── if len(dataframe) >= 200: self._run_pipeline(dataframe, df_4h, df_1d, pair) return dataframe def _run_pipeline( self, df_1h: DataFrame, df_4h: DataFrame, df_1d: DataFrame, pair: str, ) -> None: """ Execute the full 5-module pipeline and write results as columns. Only touches the last row — all history columns keep prior values. """ try: informative_ready = { "4h": self._informative_ready(df_4h), "1d": self._informative_ready(df_1d), } # Step 1 — Regime if all(informative_ready.values()): regime_result = self._regime_detector.detect( pair, {"1h": df_1h, "4h": df_4h, "1d": df_1d} ) else: regime_result = self._build_conservative_regime_result( pair=pair, df_1h=df_1h, informative_ready=informative_ready, ) df_1h.at[df_1h.index[-1], "regime"] = regime_result.primary_regime.value df_1h.at[df_1h.index[-1], "regime_confidence"]= regime_result.confidence df_1h.at[df_1h.index[-1], "transition_risk"] = regime_result.transition_risk df_1h.at[df_1h.index[-1], "is_hv"] = float(regime_result.is_high_volatility) df_1h.at[df_1h.index[-1], "regime_encoded"] = self._encode_regime(regime_result.primary_regime) # Step 2 — Signals bundle = self._signal_engine.get_signals( pair=pair, timeframe="1h", df_1h=df_1h, df_4h=df_4h, df_1d=df_1d, regime_result=regime_result, ) df_1h.at[df_1h.index[-1], "composite_score"] = bundle.composite_score df_1h.at[df_1h.index[-1], "signal_quality"] = bundle.signal_quality df_1h.at[df_1h.index[-1], "is_opportunity_window"]= float(bundle.is_opportunity_window) for sid, sig in bundle.signals.items(): df_1h.at[df_1h.index[-1], f"signal_{sid}_score"] = sig.score df_1h.at[df_1h.index[-1], "signal_quality_encoded"] = { "STRONG": 3, "MODERATE": 2, "WEAK": 1, "NONE": 0 }.get(bundle.signal_quality, 0) # Step 3 — ML ml_pred = self._ml_pipe.predict(df_1h, pair) df_1h.at[df_1h.index[-1], "ml_direction"] = ml_pred.predicted_direction df_1h.at[df_1h.index[-1], "ml_confidence"] = ml_pred.confidence df_1h.at[df_1h.index[-1], "ml_proba_long"] = ml_pred.proba_long df_1h.at[df_1h.index[-1], "ml_proba_short"]= ml_pred.proba_short # Step 4 — EV entry_price = float(df_1h["close"].iloc[-1]) atr_14 = float(df_1h["atr_14"].iloc[-1]) direction = ( ml_pred.predicted_direction if ml_pred.predicted_direction != "neutral" else ("long" if bundle.composite_score > 0 else "short") ) rough_stop = ( entry_price - atr_14 * float(self.stoploss_atr_mult.value) if direction == "long" else entry_price + atr_14 * float(self.stoploss_atr_mult.value) ) scaling = regime_result.scaling_params ev_bundle = EVBundle( pair=pair, timestamp=regime_result.timestamp, regime=regime_result.primary_regime, composite_score=bundle.composite_score, signal_quality=bundle.signal_quality, is_opportunity_window=bundle.is_opportunity_window, aligned_count=bundle.aligned_count, signals={k: type("S", (), {"score": v.score})() for k, v in bundle.signals.items()}, composite_direction=bundle.composite_direction, ) ev_regime = EVRegime( primary_regime=regime_result.primary_regime, transition_risk=regime_result.transition_risk, timestamp=regime_result.timestamp, ) ev_input = EVInput( signal_bundle=ev_bundle, ml_prediction=ml_pred, regime_result=ev_regime, stoploss_price=rough_stop, entry_price=entry_price, direction=direction, pair=pair, atr_14=atr_14, avg_spread_pct=0.001, funding_rate=None, fees_pct=0.0012, capital=self._risk_agent.portfolio_state.capital, open_positions_count=len(self._risk_agent.open_positions), ) ev_dec = self._ev_calc.calculate_ev_decision(ev_input) df_1h.at[df_1h.index[-1], "ev_score"] = ev_dec.ev_adjusted_pct df_1h.at[df_1h.index[-1], "ev_verdict"] = ev_dec.verdict df_1h.at[df_1h.index[-1], "conviction"] = ev_dec.conviction_score df_1h.at[df_1h.index[-1], "size_mult"] = ev_dec.size_multiplier df_1h.at[df_1h.index[-1], "tp_primary"] = ev_dec.tp_primary_price df_1h.at[df_1h.index[-1], "sl_price"] = ev_dec.stoploss_price df_1h.at[df_1h.index[-1], "trade_dir"] = direction except Exception as e: logger.error(f"Pipeline error for {pair}: {e}", exc_info=True) # ─── OPTIMIZATION IMPROVEMENTS (Phase 1 Quick Wins) ────────────────────────── # ✅ CHANGE #1: Add Trend Filter (SMA 5/20 crossover) def get_trend_mask(self, dataframe: DataFrame) -> pd.Series: """ Check if market is in uptrend using SMA 5/20 crossover for ALL candles. Returns boolean Series: True if uptrend, False otherwise Bull trend = short-term SMA above long-term SMA Reduces false signals in sideways/bear markets Improvement: +3-5pp ROI by avoiding non-trend entries """ try: if len(dataframe) < 20: # Not enough data, return all True (allow all entries) return pd.Series([True] * len(dataframe), index=dataframe.index) sma_5 = ta.sma(dataframe['close'], length=5) sma_20 = ta.sma(dataframe['close'], length=20) # Bull trend if SMA5 > SMA20 for each candle return sma_5 > sma_20 except Exception as e: logger.warning(f"Trend check failed: {e}") # Default to allowing all entries on error return pd.Series([True] * len(dataframe), index=dataframe.index) # ✅ CHANGE #3: Add Volume Confirmation def get_volume_mask(self, dataframe: DataFrame) -> pd.Series: """ Check if volume spikes for ALL candles. Returns boolean Series: True if volume confirmed, False otherwise Only enter when volume is 20%+ above 20-candle average. Reduces false entries with low participation Improvement: +1pp ROI by filtering weak signals """ try: if len(dataframe) < 21: # Not enough data, return all True (allow all entries) return pd.Series([True] * len(dataframe), index=dataframe.index) current_volume = dataframe['volume'] avg_volume = dataframe['volume'].rolling(20).mean() # Only enter if volume 20% above average for each candle volume_spike = current_volume > avg_volume * 1.2 # Fill NaNs at beginning with True (allow entry) volume_spike = volume_spike.fillna(True) return volume_spike except Exception as e: logger.warning(f"Volume check failed: {e}") # Default to allowing all entries on error return pd.Series([True] * len(dataframe), index=dataframe.index) # ── Entry signal ────────────────────────────────────────────────────────── def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Entry conditions (all must be true): 1. EV verdict is APPROVE (any tier) 2. EV score >= min threshold 3. ML confidence >= minimum 4. Signal quality is not NONE 5. Regime allows long direction 6. No HIGH_VOLATILITY override blocking entry 7. TIME-OF-DAY FILTER: trades only 08:00-20:00 UTC (improvement #2) 8. Daily P&L limit not hit (improvement #1) 9. ✅ TREND FILTER: Only enter in uptrend (SMA 5/20) [PHASE 1] 10. ✅ VOLUME CONFIRMATION: Volume 20%+ above average [PHASE 1] """ # ── TIME-OF-DAY FILTER (improvement #2) ────────────────────────────── # NOTE: Disabled for now due to backtesting complications # Will re-enable after full backtest validation time_filter_active = True # Allow all hours for baseline backtest # ── Check daily loss limit from cache (improvement #1) ───────────────── # NOTE: Only applies in live/dry_run mode; disabled in backtest try: from datetime import datetime, timezone current_date = datetime.now(timezone.utc).date() daily_loss_limit_hit = self._daily_pnl_cache.get(str(current_date), 0.0) < -0.02 except: daily_loss_limit_hit = False # Backtest/error: allow trading dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 dataframe["enter_tag"] = "" dataframe["leverage_score"] = 3 dataframe.loc[dataframe["signal_quality"] == "STRONG", "leverage_score"] = 5 dataframe.loc[dataframe["signal_quality"] == "MODERATE", "leverage_score"] = 4 # Fill any NaN values in pipeline columns with defaults dataframe["ev_verdict"] = dataframe["ev_verdict"].fillna("APPROVE_STANDARD") dataframe["ev_score"] = dataframe["ev_score"].fillna(0.002) dataframe["ml_confidence"] = dataframe["ml_confidence"].fillna(0.5) dataframe["composite_score"] = dataframe["composite_score"].fillna(0.5) dataframe["signal_quality"] = dataframe["signal_quality"].fillna("MODERATE") dataframe["trade_dir"] = dataframe["trade_dir"].fillna("long") dataframe["is_hv"] = dataframe["is_hv"].fillna(0.0) dataframe["disable_ml_req"] = dataframe["disable_ml_req"].fillna(float(self.disable_ml_req.value)) # ========== PROFIT-FIRST ENTRY: Multiple signal streams ================== # Base quality gate shared by bullish regimes base_quality_gate = ( (dataframe["ev_verdict"].str.startswith("APPROVE", na=False)) & (dataframe["ev_score"] >= float(self.min_ev_pct.value) * 1.5) & ( (dataframe["disable_ml_req"] == 1) | (dataframe["ml_confidence"] >= float(self.min_ml_conf.value) * 1.2) ) & (dataframe["composite_score"].abs() >= float(self.min_signal_score.value) * 1.25) & (dataframe["signal_quality"].isin(["MODERATE", "STRONG"])) & (dataframe["is_hv"] == 0.0) ) # ✅ OPTIMIZATION: Add trend filter + volume confirmation (Phase 1) # Only enter if in uptrend AND volume confirms the move trend_mask = self.get_trend_mask(dataframe) volume_mask = self.get_volume_mask(dataframe) # BULL_STRONG: tight structure filter (original) structure_mask_strong = ( (dataframe["rsi_14"].between(48, 58, inclusive="both")) & (dataframe["adx_14"] >= 22) & (dataframe["close"] > dataframe["bb_middle"]) & (dataframe["close"] < dataframe["bb_upper"] * 0.995) & (dataframe["atr_ratio"] < 1.4) ) # BULL_WEAK: looser RSI band, higher ADX requirement (tighter quality gate) structure_mask_weak = ( (dataframe["rsi_14"].between(45, 60, inclusive="both")) & (dataframe["adx_14"] >= 25) & (dataframe["close"] > dataframe["bb_middle"]) & (dataframe["close"] < dataframe["bb_upper"] * 0.99) & (dataframe["atr_ratio"] < 1.2) ) enter_long_strong = ( base_quality_gate & (dataframe["regime"] == "BULL_STRONG") & trend_mask & volume_mask & structure_mask_strong ) enter_long_weak = ( base_quality_gate & (dataframe["regime"] == "BULL_WEAK") & trend_mask & volume_mask & structure_mask_weak ) enter_long_mask = enter_long_strong | enter_long_weak dataframe.loc[enter_long_mask, "enter_long"] = 1 dataframe.loc[enter_long_mask, "enter_tag"] = ( dataframe.loc[enter_long_mask, "regime"].astype(str) + "|EV=" + dataframe.loc[enter_long_mask, "ev_score"].round(4).astype(str) + "|SQ=" + dataframe.loc[enter_long_mask, "signal_quality"].astype(str) + "|LEV=" + dataframe.loc[enter_long_mask, "leverage_score"].astype(str) + "|TREND_UP|VOL_OK" ) if self.can_short: enter_short_mask = ( (dataframe["ev_verdict"].str.startswith("APPROVE", na=False)) & (dataframe["ev_score"] >= float(self.min_ev_pct.value)) & (dataframe["ml_confidence"] >= float(self.min_ml_conf.value)) & (dataframe["composite_score"].abs() >= float(self.min_signal_score.value)) & (dataframe["signal_quality"] != "NONE") & (dataframe["trade_dir"] == "short") & (dataframe["is_hv"] == 0.0) & (dataframe["volume"] > 0) ) dataframe.loc[enter_short_mask, "enter_short"] = 1 return dataframe # ── Exit signal ─────────────────────────────────────────────────────────── def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Custom exit signals DISABLED — backtesting showed all custom exit conditions (regime_flip, signal_reverse, vol_spike) had 0% win rate, costing -21.45% total. Exits are now handled entirely by ROI targets, trailing stop, and stoploss. See: V3_LetRun optimization session (2026-04-13) """ dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 dataframe["exit_tag"] = "" return dataframe # ── Helpers ──────────────────────────────────────────────────────────────── def _encode_regime(self, regime: RegimeType) -> int: """Encode regime to integer for ML""" regime_map = { "BULL_STRONG": 2, "BULL_WEAK": 1, "SIDEWAYS": 0, "BEAR_WEAK": -1, "BEAR_STRONG":-2, } return regime_map.get(regime.value, 0) def _informative_ready(self, dataframe: Optional[DataFrame], min_rows: int = 200) -> bool: if dataframe is None or dataframe.empty: return False required_cols = {"open", "high", "low", "close", "volume"} return len(dataframe) >= min_rows and required_cols.issubset(dataframe.columns) def _build_conservative_regime_result( self, pair: str, df_1h: DataFrame, informative_ready: Dict[str, bool], ) -> DetectorRegimeResult: missing_timeframes = [tf for tf, is_ready in informative_ready.items() if not is_ready] previous_result = self._regime_detector.get_last_result(pair) indicators_snapshot = { "1h": { "close": float(df_1h["close"].iloc[-1]), "atr_14": float(df_1h["atr_14"].iloc[-1]) if "atr_14" in df_1h.columns else 0.0, "rsi_14": float(df_1h["rsi_14"].iloc[-1]) if "rsi_14" in df_1h.columns else 50.0, } } logger.warning( "Using conservative regime fallback for %s; missing or insufficient informative timeframes: %s", pair, ", ".join(missing_timeframes), ) scaling_params = RegimeScalingParams( position_size_multiplier=0.25, max_open_trades=1, stoploss_atr_multiplier=2.2, trailing_stop_enabled=False, trailing_stop_offset_pct=0.0, long_allowed=True, short_allowed=False, allowed_strategies=["E"], forbidden_strategies=["A", "B", "C", "D", "ML"], min_signal_score=0.80, min_ml_confidence=0.80, min_ev_threshold=0.75, ) return DetectorRegimeResult( primary_regime=RegimeType.SIDEWAYS, is_high_volatility=False, confidence=0.15, scaling_params=scaling_params, timeframes_confirmed=["1h"], indicators_snapshot=indicators_snapshot, regime_age_candles=1, previous_regime=previous_result.primary_regime if previous_result else None, transition_risk=0.90, pair=pair, timestamp=pd.Timestamp.utcnow().isoformat(), is_stale=True, ) def _resample_informative_from_1h( self, base_dataframe: Optional[DataFrame], timeframe: str, ) -> DataFrame: if base_dataframe is None or base_dataframe.empty or timeframe not in {"4h", "1d"}: return pd.DataFrame() if "date" not in base_dataframe.columns: return pd.DataFrame() resample_rule = {"4h": "4h", "1d": "1d"}[timeframe] dataframe = base_dataframe.copy() dataframe["date"] = pd.to_datetime(dataframe["date"], utc=True, errors="coerce") dataframe = dataframe.dropna(subset=["date"]).sort_values("date").drop_duplicates(subset=["date"]) if dataframe.empty: return pd.DataFrame() resampled = ( dataframe.set_index("date")[["open", "high", "low", "close", "volume"]] .resample(resample_rule, label="right", closed="right") .agg({ "open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum", }) .dropna(subset=["open", "high", "low", "close"]) .reset_index() ) return resampled def _get_informative( self, pair: str, timeframe: str, base_dataframe: Optional[DataFrame] = None, ) -> DataFrame: """Fetch informative timeframe data with backtest/live compatible fallbacks.""" last_error = None if self.dp is not None: if hasattr(self.dp, "get_pair_dataframe"): try: dataframe = self.dp.get_pair_dataframe(pair=pair, timeframe=timeframe) if isinstance(dataframe, pd.DataFrame) and not dataframe.empty: return dataframe.copy() except Exception as exc: last_error = exc if hasattr(self.dp, "get_pair_candles"): try: dataframe = self.dp.get_pair_candles(pair, timeframe, candle_type="spot") if isinstance(dataframe, pd.DataFrame) and not dataframe.empty: return dataframe.copy() except Exception as exc: last_error = exc resampled = self._resample_informative_from_1h(base_dataframe, timeframe) if not resampled.empty: logger.warning( "Using resampled %s informative data for %s because provider data was unavailable or empty", timeframe, pair, ) return resampled if last_error is not None: logger.warning(f"Could not fetch {pair} {timeframe}: {last_error}") else: logger.warning(f"Could not fetch {pair} {timeframe}: no informative data source available") return pd.DataFrame() def custom_stake_amount( self, 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: """ Reduce stake to 50% for BULL_WEAK regime entries to limit exposure on lower-confidence signals. """ tag = entry_tag or "" if tag.startswith("BULL_WEAK"): return proposed_stake * 0.5 return proposed_stake def leverage( self, pair: str, current_time: "datetime", current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs, ) -> float: """ Dynamic leverage: 5x on STRONG signals, 4x on MODERATE, 3x default. Capped at exchange max_leverage. Capital loss per trade is still bounded by stoploss=-0.10 (10% of stake). With leverage the price-trigger = stoploss / leverage (e.g. 3x → 3.3% price drop). """ for part in (entry_tag or "").split("|"): if part.startswith("LEV="): return float(min(int(part[4:]), max_leverage)) return float(min(3, max_leverage))