""" risk_engine.py — Adaptive risk management with consecutive-loss scaling, volatility-percentile-aware position sizing, and Kelly-influenced allocation. Key fixes vs prior version: - Consecutive loss counter drives a risk scale table (never compounds losses) - ATR stop multiplier is adaptive: widens in high-volatility to avoid noise stops - Position size caps at a hard notional limit regardless of risk fraction - Regime confidence feeds directly into risk fraction (low confidence = smaller size) - Separate max_drawdown_guard: if equity has drawn down >N% from peak, halt sizing """ from typing import Dict, Any, List import numpy as np from config import ( MAX_RISK_PER_TRADE, HIGH_VOL_THRESHOLD, LOW_VOL_THRESHOLD, REDUCED_RISK_FACTOR, ATR_STOP_MULT, RR_RATIO, DEFAULT_ACCOUNT_EQUITY, CONSEC_LOSS_RISK_SCALE, ) _MAX_NOTIONAL_FRACTION = 0.30 # never put more than 30% of equity in one trade _MAX_DRAWDOWN_HALT = 0.15 # halt new positions if equity is down 15% from peak _ADAPTIVE_STOP_MULT_HIGH = 3.0 # wider stop when vol ratio > HIGH_VOL_THRESHOLD _ADAPTIVE_STOP_MULT_LOW = 2.0 # tighter stop when vol is compressed def adaptive_stop_multiplier(vol_ratio: float, compressed: bool) -> float: """ Widen ATR stop in high volatility to avoid noise-out. Use tighter stop when entering from a compressed base (cleaner structure). """ if vol_ratio > HIGH_VOL_THRESHOLD: return _ADAPTIVE_STOP_MULT_HIGH if compressed: return _ADAPTIVE_STOP_MULT_LOW return ATR_STOP_MULT def consecutive_loss_scale(consec_losses: int) -> float: """ Step-down risk table — each loss reduces risk fraction. Prevents geometric compounding of losses during drawdown streaks. Table is defined in config.CONSEC_LOSS_RISK_SCALE. """ idx = min(consec_losses, len(CONSEC_LOSS_RISK_SCALE) - 1) return CONSEC_LOSS_RISK_SCALE[idx] def compute_dynamic_risk_fraction( vol_ratio: float, regime_score: float, volume_score: float, regime_confidence: float, consec_losses: int = 0, equity_drawdown_pct: float = 0.0, base_risk: float = MAX_RISK_PER_TRADE, ) -> float: """ Multi-factor risk fraction with hard halt on drawdown breach. Priority order (each multiplies, not adds): 1. Drawdown guard (hard gate) 2. Consecutive loss scale 3. Volatility regime adjustment 4. Regime score quality 5. Confidence floor """ # Hard halt: equity drawn down too far from peak if equity_drawdown_pct >= _MAX_DRAWDOWN_HALT: return 0.0 risk = base_risk # Consecutive loss scaling risk *= consecutive_loss_scale(consec_losses) # Volatility adjustment if vol_ratio > HIGH_VOL_THRESHOLD: risk *= REDUCED_RISK_FACTOR elif vol_ratio > HIGH_VOL_THRESHOLD * 0.75: risk *= 0.70 elif vol_ratio < LOW_VOL_THRESHOLD: risk *= 0.80 # also reduce in extreme low vol (thin market) # Regime quality if regime_score < 0.25: risk *= REDUCED_RISK_FACTOR elif regime_score < 0.45: risk *= 0.65 elif regime_score < 0.60: risk *= 0.85 # Confidence gate: confidence below threshold scales linearly to zero if regime_confidence < 0.30: risk *= 0.25 elif regime_confidence < 0.55: risk *= regime_confidence # proportional scaling return float(np.clip(risk, 0.001, base_risk)) def compute_position_size( account_equity: float, entry_price: float, stop_distance: float, risk_fraction: float, ) -> float: if stop_distance <= 0 or entry_price <= 0 or account_equity <= 0: return 0.0 dollar_risk = account_equity * risk_fraction units = dollar_risk / stop_distance notional = units * entry_price # Hard cap: never exceed _MAX_NOTIONAL_FRACTION of equity in one trade max_notional = account_equity * _MAX_NOTIONAL_FRACTION return float(min(notional, max_notional)) def evaluate_risk( close: float, atr: float, atr_pct: float, regime_score: float, vol_ratio: float, volume_score: float = 0.5, regime_confidence: float = 0.5, vol_compressed: bool = False, consec_losses: int = 0, equity_drawdown_pct: float = 0.0, account_equity: float = DEFAULT_ACCOUNT_EQUITY, rr_ratio: float = RR_RATIO, ) -> Dict[str, Any]: stop_mult = adaptive_stop_multiplier(vol_ratio, vol_compressed) stop_distance = atr * stop_mult risk_fraction = compute_dynamic_risk_fraction( vol_ratio=vol_ratio, regime_score=regime_score, volume_score=volume_score, regime_confidence=regime_confidence, consec_losses=consec_losses, equity_drawdown_pct=equity_drawdown_pct, base_risk=MAX_RISK_PER_TRADE, ) position_notional = compute_position_size( account_equity=account_equity, entry_price=close, stop_distance=stop_distance, risk_fraction=risk_fraction, ) dollar_at_risk = account_equity * risk_fraction reward_distance = stop_distance * rr_ratio leverage_implied = position_notional / account_equity if account_equity > 0 else 0.0 # Risk quality: composite readiness score quality = 1.0 if vol_ratio > HIGH_VOL_THRESHOLD: quality -= 0.25 if regime_score < 0.40: quality -= 0.20 if regime_confidence < 0.55: quality -= 0.15 if consec_losses >= 2: quality -= 0.15 risk_quality = float(np.clip(quality, 0.0, 1.0)) halted = equity_drawdown_pct >= _MAX_DRAWDOWN_HALT return { "entry_price": close, "atr": round(atr, 8), "atr_pct": round(atr_pct * 100, 3), "stop_mult": round(stop_mult, 2), "stop_distance": round(stop_distance, 8), "stop_long": round(close - stop_distance, 8), "stop_short": round(close + stop_distance, 8), "target_long": round(close + reward_distance, 8), "target_short": round(close - reward_distance, 8), "reward_distance": round(reward_distance, 8), "rr_ratio": rr_ratio, "risk_fraction": round(risk_fraction * 100, 4), "dollar_at_risk": round(dollar_at_risk, 2), "position_notional": round(position_notional, 2), "leverage_implied": round(leverage_implied, 3), "vol_ratio": round(vol_ratio, 3), "regime_score": round(regime_score, 4), "regime_confidence": round(regime_confidence, 4), "consec_losses": consec_losses, "equity_drawdown_pct": round(equity_drawdown_pct * 100, 2), "risk_quality": round(risk_quality, 3), "sizing_halted": halted, }