# source: https://raw.githubusercontent.com/M1ch43lV/strategy-audit/d3572689679015b9eb76af06e757097f03574039/repos/OfficialGIGA_freqtrade-ml-strategy/ultimate_alpha_v16.py # directory_url: https://github.com/M1ch43lV/strategy-audit/blob/main/repos/OfficialGIGA_freqtrade-ml-strategy/ # User: M1ch43lV # Repository: strategy-audit # --------------------""" ultimate_alpha_v16.py — Ultimate Alpha Bot v16.0 Complete rewrite of the previous v14 script. Every component has a reason to exist and either works or isn't there. No theater. Architecture: 1. Features — computed by the shared features.py module (same code as trainer). 2. Model — one LightGBM classifier trained offline, loaded at startup. Outputs P(trade wins | primary signal fired). No torch, no ensemble. 3. Regime — BTC-driven risk_on / neutral / risk_off with hysteresis. 4. Scoring — additive fusion of model probability, regime adjustment, trend filter, and rank bonus. Bounded [0, 1]. No multiplicative explosion. 5. Entry — score must exceed a regime-dependent threshold AND pass trend + execution-cost gates. 6. Exit — ROI table + trailing stop + ATR-based custom stoploss + score-collapse exit. 7. Position sizing — Kelly fraction scaled by regime and portfolio heat. What's removed from v14: - Untrained 6-model torch ensemble (produced random noise). - Genetic weight optimizer (noise amplifier on tiny samples). - "True RL policy update" (was not RL). - Market memory with hash-state keys (keys never collided usefully). - Champion/challenger (never actually swapped). - 5 fake on-chain/DeFi/tokenomics/whale/microstructure scores. - "Quantum elite 20/10 master" naming. What's kept and fixed: - BTC macro regime detection (was solid — now has proper hysteresis). - merge_asof for 1d informative (correctly implemented). - Portfolio heat & correlation-aware stake sizing. - Partial take-profits. - Persistent trade journal. - CryptoQuant netflow (now ACTUALLY calls the API, reads from env var). Environment: CRYPTOQUANT_TOKEN (optional) — if missing, netflow signal defaults to 0. ALPHA_MODEL_DIR (optional) — defaults to user_data/models. """ from __future__ import annotations import json import logging import os import sys import time from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame, Series from freqtrade.persistence import Trade from freqtrade.strategy import ( DecimalParameter, IntParameter, IStrategy, ) # Make our sibling modules importable regardless of where Freqtrade launches us. _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) from features import engineer_features, get_feature_columns # noqa: E402 log = logging.getLogger(__name__) # ═══════════════════════════════════════════════════════════════════════ # Model loader — loaded once, shared across all pairs # ═══════════════════════════════════════════════════════════════════════ class _ModelBundle: """Lazy-loaded LightGBM model (single or ensemble) + metadata. Backward-compatible: handles both v16.0 single-model .pkl files AND v16.1 list-of-models ensemble .pkl files. """ def __init__(self): self.models: list = [] # List of 1-or-more LightGBM models self.feature_cols: list[str] = [] self.metrics: dict = {} self.loaded = False def load(self, model_dir: Path) -> bool: import joblib pkl = model_dir / "alpha_model.pkl" meta = model_dir / "alpha_model.meta.json" if not pkl.exists() or not meta.exists(): log.warning(f"Model files not found in {model_dir}. " "Strategy will run with model probability fixed at 0.5. " "Run train_model.py first.") return False try: loaded = joblib.load(pkl) # Accept either a single model or a list of models if isinstance(loaded, list): self.models = loaded else: self.models = [loaded] with open(meta) as f: payload = json.load(f) self.feature_cols = payload["feature_columns"] self.metrics = payload.get("metrics", {}) self.loaded = True log.info(f"Loaded {len(self.models)}-model ensemble from {pkl} — " f"AUC {self.metrics.get('walk_forward_auc_mean', 0):.4f}, " f"{len(self.feature_cols)} features") # Sanity: make sure our feature engineering matches the trained set current = get_feature_columns() if current != self.feature_cols: log.error("FEATURE MISMATCH between trained model and live features. " "Retrain or sync features.py. Falling back to 0.5.") self.loaded = False return False return True except Exception as e: log.error(f"Failed to load model: {e}") return False def predict_proba(self, features: np.ndarray) -> float: """Return averaged P(win) across ensemble for a single row.""" if not self.loaded or not self.models: return 0.5 try: X = features.reshape(1, -1) probas = [float(m.predict_proba(X)[0, 1]) for m in self.models] p = float(np.mean(probas)) if np.isnan(p) or p < 0 or p > 1: return 0.5 return p except Exception as e: log.debug(f"predict_proba failed: {e}") return 0.5 def predict_proba_batch(self, X: np.ndarray) -> np.ndarray: """Ensemble-averaged P(win) for a batch. Used in populate_indicators.""" if not self.loaded or not self.models: return np.full(len(X), 0.5, dtype=np.float32) try: probas = np.column_stack([m.predict_proba(X)[:, 1] for m in self.models]) return probas.mean(axis=1) except Exception as e: log.debug(f"predict_proba_batch failed: {e}") return np.full(len(X), 0.5, dtype=np.float32) # ═══════════════════════════════════════════════════════════════════════ # Strategy # ═══════════════════════════════════════════════════════════════════════ class Github_M1ch43lV_strategy_audit__ultimate_alpha_v16__20260916_180122(IStrategy): INTERFACE_VERSION = 3 timeframe = "4h" informative_timeframe = "1d" startup_candle_count: int = 250 # 250 × 4h = ~42 days warmup for 200 EMA + features process_only_new_candles = True can_short = False # ── ROI & stops ────────────────────────────────────────────────── # Aligned with train_model.py meta-label barriers (4h timeframe): # TARGET_PROFIT = 0.030 (take 3% wins) # STOP_LOSS = -0.020 (cut 2% losers) # HORIZON_BARS = 42 (~7 days holding max) # ROI minutes: one 4h candle = 240 min, one day = 1440 min, one week = 10080 min minimal_roi = { "0": 0.060, # First 4h: let winners run up to 6% "240": 0.040, # After 1d: take 4%+ (1 candle in) "1440": 0.030, # After 1d: take 3% (aligned with model target) "4320": 0.015, # After 3d: take 1.5% (partial scale-out) "10080": 0.0, # After 7d: close on any profit (matches horizon) } stoploss = -0.05 # Hard floor; custom_stoploss uses -2% baseline # ─ TRAILING STOP DISABLED (learned from 1h backtests) ───────────── # Trailing stops combined with the volatility profile of our targets # net-destroyed winners in both bull and bear markets. Pure ROI + hard # stoploss only. trailing_stop = False trailing_stop_positive = None trailing_stop_positive_offset = 0.0 trailing_only_offset_is_reached = False use_custom_stoploss = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False position_adjustment_enable = False order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False, "emergency_exit": "market", } # ── Hyperopt parameters (safe, meaningful ranges) ──────────────── entry_threshold_neutral = DecimalParameter(0.50, 0.72, default=0.58, decimals=3, space="buy") entry_threshold_risk_on_bonus = DecimalParameter(-0.10, 0.0, default=-0.05, decimals=3, space="buy") entry_threshold_risk_off_penalty = DecimalParameter(0.0, 0.15, default=0.08, decimals=3, space="buy") model_prob_weight = DecimalParameter(0.3, 0.7, default=0.50, decimals=2, space="buy") kelly_fraction = DecimalParameter(0.2, 0.6, default=0.35, decimals=2, space="buy") max_portfolio_heat = DecimalParameter(0.4, 0.8, default=0.60, decimals=2, space="buy") # ── Class-level state: RESET IN __init__ to avoid cross-instance bleed ── _model = _ModelBundle() _regime_state: dict[str, Any] = {} _btc_macro: dict[str, float] = {} _breadth: float = 0.5 _onchain_skew: float = 0.0 _pair_scores: dict[str, float] = {} _last_apex_log: dict[str, float] = {} _last_bot_loop_ts: float = 0.0 _last_skew_fetch_ts: float = 0.0 _trade_journal: list[dict] = [] # ═══════════════════════════════════════════════════════════════ # Lifecycle # ═══════════════════════════════════════════════════════════════ def __init__(self, config: dict) -> None: super().__init__(config) # Instance-level state (NOT shared between instances) self._apex_log_cache: dict[str, float] = {} self._last_regime_logged: str | None = None self._regime_smoothed: float = 0.0 self._current_regime: str = "neutral" self._current_vol: float = 1.0 self._pair_ranking: list[tuple[str, float]] = [] self._last_rank_ts: float = 0.0 def version(self) -> str: return "v16.0" def informative_pairs(self): # All whitelisted pairs on 1d for multi-timeframe context pairs = self.dp.current_whitelist() if self.dp else [] return [(p, self.informative_timeframe) for p in pairs] def bot_start(self, **kwargs) -> None: log.info("=" * 72) log.info("Ultimate Alpha v16.0 starting") log.info("=" * 72) # Locate model dir: env var > user_data/models user_data = Path(self.config.get("user_data_dir", "user_data")) model_dir = Path(os.environ.get("ALPHA_MODEL_DIR", user_data / "models")) self._model.load(model_dir) # Restore trade journal journal_path = user_data / "alpha_trade_journal.json" if journal_path.exists(): try: with open(journal_path) as f: self._trade_journal = json.load(f) log.info(f"Restored {len(self._trade_journal)} trades from journal") except Exception as e: log.warning(f"Failed to restore trade journal: {e}") log.info(f"Model loaded: {self._model.loaded}") log.info(f"CryptoQuant token set: {'CRYPTOQUANT_TOKEN' in os.environ}") def bot_loop_start(self, current_time, **kwargs) -> None: """ Runs once per candle processing cycle across all pairs. Updates market-wide state. Each step isolated — one failure won't stop the others. """ now_ts = time.time() # ── BTC macro context ───────────────────────────────────── try: self._btc_macro = self._compute_btc_macro() except Exception as e: log.warning(f"bot_loop_start | btc_macro failed: {e}") # ── Breadth ─────────────────────────────────────────────── try: self._breadth = self._compute_breadth() except Exception as e: log.warning(f"bot_loop_start | breadth failed: {e}") # ── Regime ──────────────────────────────────────────────── try: prev = self._current_regime self._current_regime = self._detect_regime() if prev != self._current_regime and self._current_regime != self._last_regime_logged: log.info(f"REGIME CHANGE: {prev} → {self._current_regime}") self._last_regime_logged = self._current_regime except Exception as e: log.warning(f"bot_loop_start | regime failed: {e}") # ── On-chain skew (every 10 minutes) ────────────────────── if now_ts - self._last_skew_fetch_ts > 600: try: self._onchain_skew = self._fetch_cryptoquant_skew() except Exception as e: log.debug(f"skew fetch failed (safe): {e}") self._last_skew_fetch_ts = now_ts # ── Pair ranking (every 3 candles) ──────────────────────── tf_s = self.timeframe_to_seconds(self.timeframe) if now_ts - self._last_rank_ts > tf_s * 3: try: self._pair_ranking = self._rank_pairs() except Exception as e: log.warning(f"bot_loop_start | ranking failed: {e}") self._last_rank_ts = now_ts self._last_bot_loop_ts = now_ts @staticmethod def timeframe_to_seconds(tf: str) -> int: return {"1m": 60, "5m": 300, "15m": 900, "1h": 3600, "4h": 14400, "1d": 86400}.get(tf, 3600) # ═══════════════════════════════════════════════════════════════ # Market-wide computations # ═══════════════════════════════════════════════════════════════ def _compute_btc_macro(self) -> dict[str, float]: btc = self.dp.get_pair_dataframe("BTC/USDT", self.timeframe) if btc is None or len(btc) < 200: return self._btc_macro or {"risk_state": 0.0, "ret24h": 0.0, "above_ema200": 0.5} close = btc["close"] ema200 = pd.Series(np.asarray(ta.EMA(close, timeperiod=200)), index=close.index) ret24 = float(close.pct_change(24).iloc[-1]) above = 1.0 if float(close.iloc[-1]) > float(ema200.iloc[-1]) else 0.0 atr_raw = ta.ATR(btc["high"], btc["low"], btc["close"], timeperiod=14) atr = pd.Series(np.asarray(atr_raw), index=close.index) atr_last = float(atr.iloc[-1]) btc_vol_pct = (atr_last / float(close.iloc[-1])) * 100 self._current_vol = btc_vol_pct risk_state = np.clip( (above * 2 - 1) * 0.4 + np.clip(ret24 * 8, -0.5, 0.5) * 0.6, -1.0, 1.0, ) return { "risk_state": float(risk_state), "ret24h": ret24, "above_ema200": above, "vol_pct": btc_vol_pct, } def _compute_breadth(self) -> float: wl = self.dp.current_whitelist() if self.dp else [] if not wl: return 0.5 above = 0 total = 0 for p in wl: try: df = self.dp.get_pair_dataframe(p, self.timeframe) if df is None or len(df) < 51: continue ema_raw = ta.EMA(df["close"], timeperiod=50) ema = pd.Series(np.asarray(ema_raw), index=df["close"].index) if float(df["close"].iloc[-1]) > float(ema.iloc[-1]): above += 1 total += 1 except Exception: continue return (above / total) if total > 0 else 0.5 def _detect_regime(self) -> str: """ Regime state machine with hysteresis. Once in a regime, requires a stronger signal in the opposite direction to flip. Prevents whipsaws. """ risk_state = float(self._btc_macro.get("risk_state", 0.0)) breadth_signal = (self._breadth - 0.5) * 2.0 # [-1, 1] vol = float(self._current_vol) vol_penalty = max(0.0, (vol - 2.5) / 2.5) # 0 below 2.5%, grows above raw = risk_state * 0.55 + breadth_signal * 0.35 - vol_penalty * 0.30 raw = float(np.clip(raw, -1.0, 1.0)) alpha = 0.25 self._regime_smoothed = alpha * raw + (1 - alpha) * self._regime_smoothed s = self._regime_smoothed current = self._current_regime if current == "risk_on": if s < -0.35: return "risk_off" if s < 0.20: return "neutral" return "risk_on" if current == "risk_off": if s > 0.35: return "risk_on" if s > -0.20: return "neutral" return "risk_off" # neutral if s > 0.30: return "risk_on" if s < -0.30: return "risk_off" return "neutral" def _fetch_cryptoquant_skew(self) -> float: """ Real BTC exchange netflow from CryptoQuant. Positive value = inflows = bearish (coins moving to exchanges to sell). We return a tanh-squashed score in [-1, 1] where positive = bullish (outflows). """ token = os.environ.get("CRYPTOQUANT_TOKEN") if not token: return 0.0 try: import requests resp = requests.get( "https://api.cryptoquant.com/v1/btc/exchange-flows/netflow", headers={"Authorization": f"Bearer {token}"}, timeout=10, ) if resp.status_code != 200: log.debug(f"CryptoQuant returned {resp.status_code}") return 0.0 data = resp.json().get("data", []) if not data: return 0.0 latest = float(data[-1].get("value", 0)) return float(-np.tanh(latest / 800.0)) except Exception as e: log.debug(f"CryptoQuant fetch failed (safe): {e}") return 0.0 def _rank_pairs(self) -> list[tuple[str, float]]: """Rank whitelist pairs by a cheap proxy score for allocation priority.""" wl = self.dp.current_whitelist() if self.dp else [] scored: list[tuple[str, float]] = [] for p in wl: try: df, _ = self.dp.get_analyzed_dataframe(p, self.timeframe) if df is None or len(df) < 30 or "final_score" not in df.columns: continue s = float(df["final_score"].iloc[-1]) scored.append((p, s)) except Exception: continue scored.sort(key=lambda x: x[1], reverse=True) return scored[:12] def _rank_multiplier(self, pair: str) -> float: """Score-based, not rank-based. Top pairs get up to 15% bonus.""" if not self._pair_ranking: return 1.0 pairs = [p for p, _ in self._pair_ranking] if pair not in pairs: return 1.0 idx = pairs.index(pair) # Linear interpolation: top = 1.15, bottom of top-12 = 0.95 frac = 1.0 - (idx / max(len(pairs) - 1, 1)) return float(0.95 + frac * 0.20) # ═══════════════════════════════════════════════════════════════ # Freqtrade hooks # ═══════════════════════════════════════════════════════════════ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] df = dataframe.copy() # Standardize index for merge_asof if "date" not in df.columns: df = df.reset_index().rename(columns={df.index.name or "index": "date"}) df["date"] = pd.to_datetime(df["date"], utc=True) # ── Core TA indicators ───────────────────────────────────── df["rsi"] = ta.RSI(df["close"], timeperiod=14) df["atr"] = ta.ATR(df["high"], df["low"], df["close"], timeperiod=14) df["atr_pct"] = (df["atr"] / df["close"].clip(lower=1e-10)) * 100 macd_val, macd_sig, macd_hist = ta.MACD(df["close"]) df["macd"] = macd_val df["macd_signal"] = macd_sig df["macd_hist"] = macd_hist df["ema_9"] = ta.EMA(df["close"], timeperiod=9) df["ema_21"] = ta.EMA(df["close"], timeperiod=21) df["ema_50"] = ta.EMA(df["close"], timeperiod=50) df["ema_200"] = ta.EMA(df["close"], timeperiod=200) df["adx"] = ta.ADX(df["high"], df["low"], df["close"], timeperiod=14) df["volume_ma"] = df["volume"].rolling(24).mean() df["volume_ratio"] = df["volume"] / df["volume_ma"].clip(lower=1e-10) # ── 1d informative merge ─────────────────────────────────── df["trend_1d"] = 0.0 df["close_1d"] = df["close"] try: inf = self.dp.get_pair_dataframe(pair, self.informative_timeframe) if inf is not None and len(inf) > 10: inf = inf.copy() if "date" not in inf.columns: inf = inf.reset_index() inf["date"] = pd.to_datetime(inf["date"], utc=True) inf["trend_1d"] = inf["close"].pct_change(5).fillna(0) inf["close_1d"] = inf["close"] merged = pd.merge_asof( df.sort_values("date"), inf[["date", "trend_1d", "close_1d"]].sort_values("date"), on="date", direction="backward", suffixes=("", "_inf"), ) df["trend_1d"] = merged["trend_1d_inf"].fillna(0).values if "trend_1d_inf" in merged.columns else merged["trend_1d"].fillna(0).values df["close_1d"] = merged["close_1d_inf"].ffill().values if "close_1d_inf" in merged.columns else merged["close_1d"].ffill().values except Exception as e: log.debug(f"{pair} | 1d merge skipped: {e}") # ── Feature engineering (shared with trainer) ────────────── btc = self.dp.get_pair_dataframe("BTC/USDT", self.timeframe) if pair != "BTC/USDT" else None # Need datetime index on btc for features module if btc is not None and not btc.empty: btc_fe = btc.copy() if "date" in btc_fe.columns: btc_fe = btc_fe.set_index(pd.to_datetime(btc_fe["date"], utc=True)) else: btc_fe = None # features.py expects datetime index df_fe = df.set_index("date") df_fe = engineer_features(df_fe, btc_df=btc_fe) df = df_fe.reset_index() # ── Model probability ────────────────────────────────────── df = self._apply_model(df, pair) # ── Final score ──────────────────────────────────────────── df = self._compute_final_score(df, pair) return df def _apply_model(self, df: DataFrame, pair: str) -> DataFrame: """Apply the LightGBM model to the latest rows. Only the last row matters for live decisions, but we fill the column for backtests.""" if not self._model.loaded or "feat_000" not in df.columns: df["model_prob"] = 0.5 return df feat_cols = self._model.feature_cols missing = [c for c in feat_cols if c not in df.columns] if missing: log.warning(f"{pair} | missing {len(missing)} features — skipping model") df["model_prob"] = 0.5 return df try: X = df[feat_cols].values.astype(np.float32) # LightGBM handles NaN natively; still, guard against inf X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0) proba = self._model.predict_proba_batch(X) proba = np.clip(proba, 0.01, 0.99) df["model_prob"] = proba except Exception as e: log.warning(f"{pair} | model inference failed: {e}") df["model_prob"] = 0.5 return df def _compute_final_score(self, df: DataFrame, pair: str) -> DataFrame: """ Additive score in [0, 1]. No multiplication, no explosion. Weights sum to 1.0 for the base, then small additive adjustments. """ if len(df) < 60: df["final_score"] = 0.5 return df # Model probability (primary signal) model_prob = df["model_prob"].fillna(0.5) mw = float(self.model_prob_weight.value) # Classic technical components rsi_norm = (df["rsi"].fillna(50) / 100).clip(0, 1) rsi_score = 1.0 - (rsi_norm - 0.5).abs() * 2 # peaks at RSI=50 rsi_score = rsi_score * 0.5 + rsi_norm * 0.5 # blend with trend direction macd_hist_norm = (df["macd_hist"].fillna(0) / df["close"].clip(lower=1e-10) * 100).clip(-2, 2) macd_score = (macd_hist_norm + 2) / 4 ema_stack_score = ( (df["close"] > df["ema_21"]).astype(float) * 0.3 + (df["ema_21"] > df["ema_50"]).astype(float) * 0.3 + (df["close"] > df["ema_200"]).astype(float) * 0.4 ) vol_score = df["volume_ratio"].fillna(1.0).clip(0.3, 3.0) vol_score = ((vol_score - 0.3) / 2.7).clip(0, 1) adx_score = (df["adx"].fillna(20) / 60).clip(0, 1) technical = ( rsi_score * 0.15 + macd_score * 0.25 + ema_stack_score * 0.35 + vol_score * 0.15 + adx_score * 0.10 ) base_score = model_prob * mw + technical * (1 - mw) # ── Additive adjustments ─────────────────────────────────── regime = self._current_regime regime_adj = {"risk_on": +0.04, "neutral": 0.0, "risk_off": -0.06}.get(regime, 0.0) skew_adj = float(self._onchain_skew) * 0.03 rank_adj = (self._rank_multiplier(pair) - 1.0) * 0.10 # Toxic volume spike without price movement = kill toxicity = np.where( (df["volume_ratio"].fillna(1.0) > 4.0) & (df["close"].pct_change(3).abs().fillna(0) < 0.002), -0.08, 0.0, ) final = (base_score + regime_adj + skew_adj + rank_adj + toxicity).clip(0.0, 1.0) df["final_score"] = final.fillna(0.5) # ── Throttled logging (once per pair per 5 min) ──────────── now = time.time() if now - self._apex_log_cache.get(pair, 0) > 300: last = df.iloc[-1] log.info( f"{pair:<12} | score={last['final_score']:.3f} " f"model_p={last['model_prob']:.3f} tech={technical.iloc[-1]:.3f} " f"regime={regime} rank_adj={rank_adj:+.3f} " f"skew={self._onchain_skew:+.3f}" ) self._apex_log_cache[pair] = now return df def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] df = dataframe.copy() df["enter_long"] = 0 if len(df) < 60 or "final_score" not in df.columns: return df regime = self._current_regime base = float(self.entry_threshold_neutral.value) threshold = base if regime == "risk_on": threshold = base + float(self.entry_threshold_risk_on_bonus.value) elif regime == "risk_off": threshold = base + float(self.entry_threshold_risk_off_penalty.value) threshold = float(np.clip(threshold, 0.45, 0.80)) # Volatility adjustment: raise threshold when vol is unusually high atr_roll = df["atr_pct"].rolling(48).mean() atr_cur = df["atr_pct"] vol_ratio = (atr_cur / atr_roll.clip(lower=1e-6)).fillna(1.0) threshold_series = threshold + (vol_ratio - 1.0).clip(-0.05, 0.10) * 0.5 threshold_series = threshold_series.clip(0.45, 0.82) # Gates — must mirror train_model.py generate_primary_signal so the # model is inferring on the same state distribution it trained on. score_ok = df["final_score"] > threshold_series # === Primary signal state (mirrors trainer) === macd_bull = df["macd"] > df["macd_signal"] above_200 = df["close"] > df["ema_200"] ema_stack = df["ema_21"] > df["ema_50"] rsi_ok = (df["rsi"] > 50) & (df["rsi"] < 75) # === Live-only safety checks === # 1d trend not deeply negative (avoid trading against higher-TF bear) trend_ok = (df["trend_1d"] > -0.08) | (df["ema_21"].pct_change(6).fillna(0) > 0.003) # Not overextended above fast EMA (don't chase parabolic moves) not_extended = df["close"] < df["ema_9"] * 1.04 # Regime gate: never enter in risk_off unless score is exceptional regime_ok = (regime != "risk_off") | (df["final_score"] > 0.75) entry = (score_ok & macd_bull & above_200 & ema_stack & rsi_ok & trend_ok & not_extended & regime_ok) df["enter_long"] = entry.astype(int) df["enter_tag"] = np.where(entry, f"v16_{regime}", "") # Log the last candle's decision once per 5 min last = df.iloc[-1] if bool(last["enter_long"]): log.info( f"{pair:<12} ENTRY | score={last['final_score']:.3f} " f"thresh={float(threshold_series.iloc[-1]):.3f} " f"model_p={last['model_prob']:.3f} regime={regime}" ) return df def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: df = dataframe.copy() df["exit_long"] = 0 if len(df) < 60 or "final_score" not in df.columns: return df # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # EXIT POLICY — The model was trained to predict ENTRIES, not exits. # Using model score as an exit signal is a category error that killed # winners before they reached their ROI target in v1 backtests. # # We rely on: # 1. ROI table (take profit at 4.5% / 2.5% / 1.5% / 0%) # 2. custom_stoploss (ATR-adaptive, -2% baseline) # 3. Trailing stop (locks in after +3%) # 4. This function: ONLY an extreme-volatility emergency exit # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Extreme volatility spike — true risk signal (3x normal ATR). # Rare event. When it fires, market structure has broken. atr_roll = df["atr_pct"].rolling(48).mean().clip(lower=1e-6) extreme_vol = df["atr_pct"] > atr_roll * 3.0 exit_signal = extreme_vol # Never exit on the same candle as entry if "enter_long" in df.columns: exit_signal = exit_signal & (df["enter_long"] != 1) df["exit_long"] = exit_signal.astype(int) return df # ═══════════════════════════════════════════════════════════════ # Custom stoploss — PURE hard stop, no profit-tightening # ═══════════════════════════════════════════════════════════════ def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs ) -> float: """ Returns a NEGATIVE ratio — a pure hard stop. IMPORTANT: Does NOT tighten with profit. The previous version did, which Freqtrade labeled 'trailing_stop_loss' and destroyed winners. This version only widens the stop for high-volatility pairs so normal noise doesn't stop us out — it never raises the stop above baseline. """ df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is None or len(df) < 20: return self.stoploss atr_pct = float(df["atr_pct"].iloc[-1]) / 100 # ratio # Base stop = -2.0% (training barrier). Widen up to -5% on high-vol pairs. # Never tighten with profit — that was the bug. atr_based = max(0.020, min(atr_pct * 2.0, 0.05)) return -atr_based # ═══════════════════════════════════════════════════════════════ # Partial exits via custom_exit # ═══════════════════════════════════════════════════════════════ def custom_exit(self, pair: str, trade: Trade, current_time, current_rate: float, current_profit: float, **kwargs): """ Bonus exits for exceptional profit. With TARGET_PROFIT=3%, anything running past 8% is a runner — we grab at 8%, and at 15%+ we're in rare-but-real multi-day-breakout territory. Order matters: check strongest condition first. """ if current_profit > 0.15: return "runner_take_15pct" if current_profit > 0.08: return "partial_take_8pct" # No special exit — let ROI / stoploss handle it return None # ═══════════════════════════════════════════════════════════════ # Position sizing — Kelly × regime × portfolio-heat # ═══════════════════════════════════════════════════════════════ def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: Optional[float], leverage: float, entry_tag: Optional[str], side: str, **kwargs ) -> float: try: if not isinstance(proposed_stake, (int, float)): proposed_stake = float(self.config.get("stake_amount", 100)) df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) score = float(df["final_score"].iloc[-1]) if df is not None and len(df) > 0 else 0.5 model_p = float(df["model_prob"].iloc[-1]) if df is not None and "model_prob" in df.columns else 0.5 # Kelly fraction: (p*b - q)/b, but we fractionalize to cap exposure. # Assume payoff ratio b ≈ 1 (our TP and SL are similar scale). # edge = p - (1 - p) = 2p - 1 edge = max(0.0, 2 * model_p - 1) kelly = edge * float(self.kelly_fraction.value) # fractional Kelly # Scale by score above neutral score_mult = 0.5 + max(0.0, score - 0.5) # range [0.5, 1.0] # Portfolio heat check heat = self._portfolio_heat() heat_cap = float(self.max_portfolio_heat.value) heat_mult = 1.0 if heat < heat_cap else max(0.3, (1.5 - heat / heat_cap)) # Regime scale regime_mult = {"risk_on": 1.15, "neutral": 1.0, "risk_off": 0.55}.get(self._current_regime, 1.0) # Final stake mult = (0.5 + kelly) * score_mult * heat_mult * regime_mult stake = proposed_stake * np.clip(mult, 0.3, 1.6) if min_stake is not None: stake = max(stake, float(min_stake) * 1.05) if max_stake is not None: stake = min(stake, float(max_stake)) log.info( f"STAKE {pair:<12} | proposed=${proposed_stake:.2f} " f"→ final=${stake:.2f} | score={score:.3f} model_p={model_p:.3f} " f"kelly={kelly:.3f} heat={heat:.1%} regime={self._current_regime}" ) return float(stake) except Exception as e: log.warning(f"custom_stake_amount fallback for {pair}: {e}") return float(proposed_stake) * 0.5 def _portfolio_heat(self) -> float: """Fraction of total balance currently deployed in open trades.""" try: open_trades = Trade.get_trades_proxy(is_open=True) total = float(self.wallets.get_total(self.config["stake_currency"]) or 0.0) if total <= 0: return 0.0 deployed = sum(float(t.stake_amount or 0.0) for t in open_trades) return deployed / total except Exception: return 0.0 # ═══════════════════════════════════════════════════════════════ # Trade journaling — persistent record for analysis # ═══════════════════════════════════════════════════════════════ 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: try: profit_ratio = float(trade.calc_profit_ratio(rate)) if rate else 0.0 entry = { "ts": current_time.isoformat() if hasattr(current_time, "isoformat") else str(current_time), "pair": pair, "entry_rate": float(trade.open_rate), "exit_rate": float(rate) if rate else None, "profit_ratio": profit_ratio, "exit_reason": exit_reason, "enter_tag": trade.enter_tag or "", "regime": self._current_regime, "trade_duration_mins": int((current_time - trade.open_date_utc).total_seconds() / 60) if trade.open_date_utc else 0, } self._trade_journal.append(entry) if len(self._trade_journal) > 2000: self._trade_journal = self._trade_journal[-2000:] self._persist_journal() except Exception as e: log.debug(f"journal write failed: {e}") return True def _persist_journal(self): try: user_data = Path(self.config.get("user_data_dir", "user_data")) user_data.mkdir(parents=True, exist_ok=True) path = user_data / "alpha_trade_journal.json" with open(path, "w") as f: json.dump(self._trade_journal, f, default=str) except Exception as e: log.debug(f"persist journal failed: {e}")