shaibu01/Titan-Engine
0
1"""2TITAN ADAPTIVE ENGINE (v42.0 - REGIME-LEARNING + SIGNAL DECAY MONITOR)3TYPE: Dynamic Signal Normalizer, Stop-Loss Optimizer & IC-Weighted Threshold Engine4MATH: Online Z-Score, Volatility Scaling, Regime-Based Parity,5 Bayesian Regime Parameter Bank, Spearman IC Decay Detection6 7UPGRADES OVER v41.0:8 1. RegimeParameterBank — per-regime SL/TP parameter learning via EMA:9 param_new = (1-α)*param_old + α*observed_value (α = 0.1)10 Replaces fixed if/else multipliers with statistics-driven adaptive multipliers.11 12 2. SignalDecayMonitor — rolling Spearman IC tracking:13 IC(signal, t) = Spearman(signal_t, return_{t+5})14 If IC < 0.02 for 20+ consecutive bars → signal downweighted.15 If IC drops > 1σ below historical mean → early-decay warning.16 Returns weight ∈ [0.1, 1.0].17 18 3. get_thresholds now applies IC-based signal weights to dynamic thresholds.19 20PRESERVED:21 - FastMath / C++ OFFLOAD / Python fallback22 - AdaptiveEngine.scale_risk (now delegates to RegimeParameterBank)23 - AdaptiveEngine.update / _calculate_regime_modifier24 - Mutex (threading.Lock) around all shared state25 - Z-score normalisation, Bollinger-logic thresholds, safety clamps26"""27 28import numpy as np29from collections import deque30from scipy import stats as scipy_stats # for Spearman correlation31import logging32import threading33 34logger = logging.getLogger("TITAN")35 36# --- C++ CORE INJECTION ---37try:38 import titan_engine39 HAS_CPP_CORE = True40except ImportError:41 HAS_CPP_CORE = False42 43 44# ==============================================================================45# FAST MATH (preserved)46# ==============================================================================47class FastMath:48 @staticmethod49 def get_stats(array_data):50 """Ultra-fast Mean and Standard Deviation computation."""51 # RAM OPTIMIZATION: Zero-copy cast from deque to contiguous C-array52 np_arr = np.fromiter(array_data, dtype=np.float64)53 54 # C++ OFFLOAD55 if HAS_CPP_CORE and hasattr(titan_engine, 'calc_stats'):56 try:57 mean, std = titan_engine.calc_stats(np_arr)58 return float(mean), float(std)59 except Exception:60 pass61 62 # PYTHON FALLBACK63 if len(np_arr) == 0:64 return 0.0, 0.065 return float(np.mean(np_arr)), float(np.std(np_arr))66 67 68fast_math = FastMath()69 70 71# ==============================================================================72# 1. REGIME PARAMETER BANK (NEW)73# ==============================================================================74class RegimeParameterBank:75 """76 Online Bayesian parameter learning per market regime.77 78 For each regime {0=BULL, 1=CHOP, 2=BEAR, 3=CRASH} we maintain:79 sl_mult — stop-loss distance multiplier (learned from SL-hit rate)80 tp_mult — take-profit distance multiplier (learned from avg bars-to-target)81 n_obs — number of completed trade observations in this regime82 83 MATH — Exponential Moving Average update rule:84 param_new = (1 - α) * param_old + α * observed_value85 α = 0.10 (slow learning rate → stability over reactivity)86 87 SL multiplier update:88 sl_triggered == True → tighten: observed_sl_mult = sl_mult * 0.989 sl_triggered == False → widen slightly: observed_sl_mult = sl_mult * 1.0590 91 TP multiplier update:92 If trade was profitable (final_pnl_pct > 0):93 bars_factor = clip(bars_held / 10.0, 0.5, 2.0)94 observed_tp_mult = tp_mult * bars_factor95 Else: leave tp_mult unchanged (SL drove the exit, not TP)96 97 Replaces hard-coded if/else regime multipliers in AdaptiveEngine.scale_risk.98 """99 100 def __init__(self):101 self.lock = threading.Lock()102 103 # Default initialisation: conservative values informed by regime intuition104 # BULL (0): normal SL, let profits run (1.5x TP)105 # CHOP (1): tight SL, tight TP (mean-reversion)106 # BEAR (2): normal SL, wide TP for shorts (1.8x)107 # CRASH (3): very tight SL, quick TP (avoid whipsaws)108 self.regime_params = {109 0: {"sl_mult": 1.00, "tp_mult": 1.50, "n_obs": 0}, # BULL110 1: {"sl_mult": 0.80, "tp_mult": 0.80, "n_obs": 0}, # CHOP111 2: {"sl_mult": 1.00, "tp_mult": 1.80, "n_obs": 0}, # BEAR112 3: {"sl_mult": 0.60, "tp_mult": 0.70, "n_obs": 0}, # CRASH113 }114 self.alpha = 0.10 # Learning rate — controls speed of adaptation115 116 def update_from_trade(self, regime, sl_triggered, bars_held, final_pnl_pct):117 """118 Update per-regime SL/TP multipliers from a completed trade outcome.119 120 Args:121 regime (int) : 0=BULL, 1=CHOP, 2=BEAR, 3=CRASH122 sl_triggered (bool) : True if the stop-loss (not TP) caused the exit123 bars_held (int) : number of bars from entry to exit124 final_pnl_pct (float): signed P&L as a fraction (e.g. -0.003 or +0.005)125 """126 regime = int(regime)127 if regime not in self.regime_params:128 regime = 1 # Default to CHOP if unknown129 130 with self.lock:131 p = self.regime_params[regime]132 α = self.alpha133 old_sl = p["sl_mult"]134 old_tp = p["tp_mult"]135 136 # --- SL multiplier update ---137 if sl_triggered:138 # SL was hit → current SL was too tight; loosen it slightly139 observed_sl = old_sl * 1.05140 else:141 # TP or manual exit → SL held; reward with a slight tighten142 observed_sl = old_sl * 0.98143 144 new_sl = (1.0 - α) * old_sl + α * observed_sl145 new_sl = float(max(0.30, min(3.00, new_sl))) # hard clamps146 147 # --- TP multiplier update (only if trade was profitable) ---148 if final_pnl_pct > 0 and bars_held > 0:149 bars_factor = float(np.clip(bars_held / 10.0, 0.5, 2.0))150 observed_tp = old_tp * bars_factor151 new_tp = (1.0 - α) * old_tp + α * observed_tp152 new_tp = float(max(0.30, min(5.00, new_tp)))153 else:154 new_tp = old_tp # No TP update on losing trades155 156 self.regime_params[regime]["sl_mult"] = new_sl157 self.regime_params[regime]["tp_mult"] = new_tp158 self.regime_params[regime]["n_obs"] += 1159 160 logger.debug(161 f"[RegimeBank] R={regime} | SL: {old_sl:.3f}→{new_sl:.3f} | "162 f"TP: {old_tp:.3f}→{new_tp:.3f} | n_obs={p['n_obs']}")163 164 def get_params(self, regime):165 """166 Return current (sl_mult, tp_mult) for the given regime.167 Falls back to CHOP defaults if regime is unknown.168 """169 regime = int(regime)170 with self.lock:171 p = self.regime_params.get(regime, self.regime_params[1])172 return float(p["sl_mult"]), float(p["tp_mult"])173 174 def get_all_stats(self):175 """Snapshot of all regime parameters (for monitoring / dashboards)."""176 with self.lock:177 return {r: dict(v) for r, v in self.regime_params.items()}178 179 180# ==============================================================================181# 2. SIGNAL DECAY MONITOR (NEW)182# ==============================================================================183class SignalDecayMonitor:184 """185 Detects when a predictive signal is losing its forward return predictiveness.186 187 MATH — Information Coefficient (IC):188 IC(t) = Spearman(signal_{t-5:t}, return_{t:t+5})189 (rolling window: last `window` IC observations)190 191 Decay conditions (checked in get_signal_weight):192 1. HARD DECAY: IC < 0.02 for >= 20 consecutive bars193 → weight = 0.10 (signal nearly disabled)194 2. SOFT DECAY: current_IC < (IC_mean - 1.0 * IC_std)195 → weight = 0.50 (signal half-weighted)196 3. HEALTHY: otherwise197 → weight = 1.00198 199 Weight is clamped to [0.10, 1.00].200 201 This is the institutional method for managing signal decay in live production202 (factor IC monitoring is standard at systematic hedge funds such as AQR, D.E. Shaw).203 """204 205 IC_WINDOW_DEFAULT = 60 # bars of IC history to retain206 HARD_DECAY_IC = 0.02 # IC below this is considered non-predictive207 HARD_DECAY_BARS = 20 # consecutive bars below HARD_DECAY_IC for full decay208 BUFFER_SIZE = 200 # (signal, fwd_return) pairs to retain per signal209 210 def __init__(self):211 self.lock = threading.Lock()212 # {signal_name: deque of IC floats}213 self.ic_history = {}214 # {signal_name: deque of (signal_value, fwd_return_5bar) tuples}215 self.signal_buffer = {}216 # {signal_name: int — consecutive bars with IC < HARD_DECAY_IC}217 self._low_ic_streak = {}218 219 def _ensure_signal(self, signal_name):220 """Initialise data structures for a new signal (call inside lock)."""221 if signal_name not in self.ic_history:222 self.ic_history[signal_name] = deque(maxlen=self.IC_WINDOW_DEFAULT)223 self.signal_buffer[signal_name] = deque(maxlen=self.BUFFER_SIZE)224 self._low_ic_streak[signal_name] = 0225 226 def record_signal(self, signal_name, signal_value, forward_return_5bar):227 """228 Record a (signal_value, 5-bar forward return) pair and recompute rolling IC.229 230 Call this AFTER the 5-bar forward return has been realised (i.e., 5 bars231 after the signal was generated). The caller is responsible for the 5-bar lag.232 233 Args:234 signal_name (str) : unique name for the signal235 signal_value (float): the signal at time t236 forward_return_5bar (float): realised return over [t, t+5]237 """238 with self.lock:239 self._ensure_signal(signal_name)240 self.signal_buffer[signal_name].append(241 (float(signal_value), float(forward_return_5bar)))242 243 # Recompute IC if we have at least 10 paired observations244 buf = self.signal_buffer[signal_name]245 if len(buf) >= 10:246 arr = np.array(buf, dtype=np.float64)247 sigs = arr[:, 0]248 rets = arr[:, 1]249 try:250 rho, _ = scipy_stats.spearmanr(sigs, rets)251 ic_val = float(rho) if not np.isnan(rho) else 0.0252 except Exception:253 ic_val = 0.0254 self.ic_history[signal_name].append(ic_val)255 256 # Update low-IC streak257 if ic_val < self.HARD_DECAY_IC:258 self._low_ic_streak[signal_name] += 1259 else:260 self._low_ic_streak[signal_name] = 0261 262 def get_signal_ic(self, signal_name, window=60):263 """264 Compute rolling mean Spearman IC for a signal over the last N observations.265 266 Returns mean IC (float). Returns 0.0 if insufficient history.267 """268 with self.lock:269 self._ensure_signal(signal_name)270 hist = list(self.ic_history[signal_name])271 272 if len(hist) < 5:273 return 0.0274 275 # Use the last `window` IC values276 recent = hist[-window:]277 return float(np.mean(recent))278 279 def get_signal_weight(self, signal_name):280 """281 Returns a weight multiplier ∈ [0.10, 1.00] based on IC decay status.282 283 Logic:284 1. If low-IC streak ≥ HARD_DECAY_BARS → weight = 0.10 (HARD DECAY)285 2. If current_IC < IC_mean - 1.0*IC_std → weight = 0.50 (SOFT DECAY)286 3. Otherwise → weight = 1.00 (HEALTHY)287 """288 with self.lock:289 self._ensure_signal(signal_name)290 hist = list(self.ic_history[signal_name])291 streak = self._low_ic_streak[signal_name]292 293 if len(hist) == 0:294 return 1.0 # No history yet: assume healthy, do not penalise295 296 # 1. Hard decay: consecutive low-IC bars297 if streak >= self.HARD_DECAY_BARS:298 logger.warning(f"[SignalDecay] HARD DECAY: {signal_name} | "299 f"IC < {self.HARD_DECAY_IC} for {streak} bars. Weight=0.10")300 return 0.10301 302 # 2. Soft decay: current IC dropped > 1σ below historical mean303 ic_arr = np.array(hist, dtype=np.float64)304 ic_mean = float(np.mean(ic_arr))305 ic_std = float(np.std(ic_arr)) + 1e-9306 current_ic = hist[-1] if hist else 0.0307 308 if current_ic < (ic_mean - 1.0 * ic_std):309 logger.info(f"[SignalDecay] SOFT DECAY: {signal_name} | "310 f"IC={current_ic:.4f} < mean-1σ ({ic_mean-ic_std:.4f}). Weight=0.50")311 return 0.50312 313 return 1.00314 315 def get_all_ic_summary(self):316 """Summary dict of all tracked signals (for monitoring)."""317 with self.lock:318 names = list(self.ic_history.keys())319 summary = {}320 for name in names:321 summary[name] = {322 "mean_ic": self.get_signal_ic(name),323 "weight": self.get_signal_weight(name),324 "n_ic_obs": len(self.ic_history.get(name, [])),325 }326 return summary327 328 329# ==============================================================================330# 3. ADAPTIVE ENGINE (upgraded)331# ==============================================================================332class AdaptiveEngine:333 def __init__(self, lookback_window=100):334 logger.info("🧬 ADAPTIVE RISK v42.0: ONLINE [REGIME LEARNING + SIGNAL DECAY MONITOR]")335 self.lookback = lookback_window336 self.history = {}337 self.volatility_state = {}338 self.lock = threading.Lock() # CRITICAL: Prevents Multi-Core Race Conditions339 340 # Upgraded sub-modules341 self.regime_bank = RegimeParameterBank()342 self.decay_monitor = SignalDecayMonitor()343 344 # --------------------------------------------------------------------------345 # 1. STOP-LOSS / TAKE-PROFIT SCALER (now delegates to RegimeParameterBank)346 # --------------------------------------------------------------------------347 def scale_risk(self, volatility, regime_state, trade_direction="BUY"):348 """349 MASTER-MASTER INTEGRATION:350 Dynamically adjusts Stop-Loss and Take-Profit distances.351 352 Accepts regime_state as integer (0-3) or legacy string ("BULL", "CHOP", "BEAR", "CRASH").353 Multipliers come from RegimeParameterBank (learned from trade outcomes) rather354 than fixed constants. Falls back to initial defaults until sufficient data355 accumulates (n_obs tracked per regime in RegimeParameterBank).356 357 MATH:358 base_sl = max(0.015, volatility * 2.5)359 base_tp = max(0.025, volatility * 4.0)360 dynamic_sl = base_sl * sl_mult * direction_adjust361 dynamic_tp = base_tp * tp_mult * direction_adjust362 """363 volatility = float(volatility)364 regime_str = str(regime_state).upper()365 direction = str(trade_direction).upper()366 367 # --- Regime integer mapping (supports both legacy strings and new ints) ---368 if regime_str in ("0", "BULL", "TRENDING"):369 regime_int = 0370 elif regime_str in ("1", "CHOP", "STATIC"):371 regime_int = 1372 elif regime_str in ("2", "BEAR"):373 regime_int = 2374 elif regime_str in ("3", "CRASH"):375 regime_int = 3376 else:377 regime_int = 1 # Default to CHOP378 379 # Base distances derived from asset ATR / GARCH volatility380 base_sl = max(0.015, volatility * 2.5)381 base_tp = max(0.025, volatility * 4.0)382 383 # Learned regime multipliers384 sl_mult, tp_mult = self.regime_bank.get_params(regime_int)385 386 # Direction adjustment (same directional logic as legacy v41.0)387 if regime_int == 0: # BULL388 if direction == "BUY":389 dir_sl = 1.0; dir_tp = 1.0 # Trend is your friend390 else:391 dir_sl = 0.85; dir_tp = 0.75 # Counter-trend short: quick profit392 elif regime_int == 1: # CHOP: both sides get same treatment393 dir_sl = 1.0; dir_tp = 1.0394 elif regime_int == 2: # BEAR395 if direction == "SELL":396 dir_sl = 1.0; dir_tp = 1.0 # Trend-following short397 else:398 dir_sl = 0.80; dir_tp = 0.75 # Knife-catching long: very tight399 else: # CRASH400 dir_sl = 0.90; dir_tp = 0.85 # Both sides: protect capital first401 402 dynamic_sl = base_sl * sl_mult * dir_sl403 dynamic_tp = base_tp * tp_mult * dir_tp404 405 # Safety clamps (ensure no fatal parameters)406 dynamic_sl = max(0.010, min(dynamic_sl, 0.08))407 dynamic_tp = max(0.015, min(dynamic_tp, 0.20))408 409 return float(round(dynamic_sl, 4)), float(round(dynamic_tp, 4))410 411 # --------------------------------------------------------------------------412 # 2. SIGNAL NORMALIZATION & Z-SCORE THRESHOLDS (preserved + IC weights)413 # --------------------------------------------------------------------------414 def update(self, symbol, current_score, current_volatility=0.01):415 """Ingests new data safely across multiple CPU threads."""416 with self.lock:417 if symbol not in self.history:418 self.history[symbol] = deque(maxlen=self.lookback)419 self.volatility_state[symbol] = deque(maxlen=self.lookback)420 421 self.history[symbol].append(float(current_score))422 self.volatility_state[symbol].append(float(current_volatility))423 424 def _calculate_regime_modifier(self, symbol):425 """Determines if thresholds should be loosened or tightened based on volatility trend."""426 with self.lock:427 if symbol not in self.volatility_state or len(self.volatility_state[symbol]) < 10:428 return 1.0429 vol_list = list(self.volatility_state[symbol])430 431 recent_vol = sum(vol_list[-5:]) / 5.0432 hist_vol = sum(vol_list) / len(vol_list)433 434 if hist_vol == 0:435 return 1.0436 437 # Clamp between 0.8 (Aggressive) and 1.5 (Defensive)438 return float(max(0.8, min(1.5, recent_vol / hist_vol)))439 440 def get_thresholds(self, symbol, signal_name=None):441 """442 Returns dynamic triggers for BOTH Long and Short entries based on Z-Score distribution.443 444 UPGRADE (v42.0): Applies IC-based signal weight from SignalDecayMonitor.445 446 MATH:447 z_target = 1.65 * k_factor * ic_weight448 long_entry = clamp(μ + z_target * σ, 0.2, 0.9)449 short_entry = clamp(μ - z_target * σ, -0.9, -0.2)450 451 ic_weight ∈ [0.10, 1.00]:452 1.00 → full confidence in signal453 0.10 → signal nearly decayed; thresholds pushed far out of reach454 (effective filtering of decayed signals at entry)455 456 Args:457 symbol (str): asset identifier for history lookup458 signal_name (str, optional): name for IC weight lookup.459 If None, IC weight = 1.0 (no decay adjustment).460 """461 with self.lock:462 if symbol not in self.history or len(self.history[symbol]) < 20:463 return 0.65, -0.65464 data_copy = list(self.history[symbol])465 466 # 1. C++ Accelerated Stats467 mu, sigma = fast_math.get_stats(data_copy)468 sigma += 1e-9 # Prevent div/0469 470 # 2. Regime Modifier (Noise Filter based on recent vs historical vol)471 k_factor = self._calculate_regime_modifier(symbol)472 473 # 3. IC-Based Signal Weight (NEW)474 if signal_name is not None:475 ic_weight = self.decay_monitor.get_signal_weight(signal_name)476 else:477 ic_weight = 1.0478 479 # 4. Dynamic Thresholds (Bollinger Logic on Signals, IC-adjusted)480 # ic_weight < 1 → shrink the σ contribution → thresholds move toward μ481 # This makes entry harder (higher bar) when signal is decaying.482 target_sigma = 1.65 * k_factor * ic_weight483 484 long_entry = mu + (target_sigma * sigma)485 short_entry = mu - (target_sigma * sigma)486 487 # 5. Institutional Safety Clamps488 long_entry = max(0.2, min(0.9, long_entry))489 short_entry = max(-0.9, min(-0.2, short_entry))490 491 if signal_name and ic_weight < 1.0:492 logger.info(f"[AdaptiveEngine] {symbol}/{signal_name}: "493 f"IC weight={ic_weight:.2f} → thresholds adjusted "494 f"(long={long_entry:.4f}, short={short_entry:.4f})")495 496 return float(round(long_entry, 4)), float(round(short_entry, 4))497 498 # --------------------------------------------------------------------------499 # 3. CONVENIENCE WRAPPERS FOR EXTERNAL CALLERS500 # --------------------------------------------------------------------------501 def record_trade_outcome(self, regime, sl_triggered, bars_held, final_pnl_pct):502 """503 Feed completed trade results into the RegimeParameterBank.504 Call this from the execution core when a position is closed.505 """506 self.regime_bank.update_from_trade(507 regime=int(regime),508 sl_triggered=bool(sl_triggered),509 bars_held=int(bars_held),510 final_pnl_pct=float(final_pnl_pct),511 )512 513 def record_signal_outcome(self, signal_name, signal_value, forward_return_5bar):514 """515 Feed (signal, 5-bar forward return) pair into the SignalDecayMonitor.516 Call this 5 bars after the signal was generated.517 """518 self.decay_monitor.record_signal(signal_name, signal_value, forward_return_5bar)519 520 def get_regime_stats(self):521 """Return a snapshot of all regime parameter banks (for monitoring)."""522 return self.regime_bank.get_all_stats()523 524 def get_signal_decay_report(self):525 """Return IC summary for all tracked signals (for monitoring)."""526 return self.decay_monitor.get_all_ic_summary()527