""" ========================================= CV6 Advanced Risk Guard Protects capital through multiple layers: 1. Emergency Kill Switch (manual + auto) 2. Daily / Weekly / Monthly loss limits 3. Max Drawdown protection 4. Sector exposure cap 5. Correlation risk guard 6. Circuit filter (limit-up / limit-down) 7. Volatility filter (VIX spike) 8. News risk filter 9. Market holiday check ========================================= """ from __future__ import annotations import threading from datetime import date, datetime, timedelta from typing import Dict, List, Optional, Tuple from loguru import logger # ── Risk Configuration ───────────────────────────────────────────────────────── class RiskConfig: """All thresholds are configurable at runtime.""" def __init__(self): # Kill switch self.kill_switch_active: bool = False self.kill_switch_reason: str = "" # Loss limits (₹) self.daily_loss_limit: float = 5_000.0 self.weekly_loss_limit: float = 15_000.0 self.monthly_loss_limit: float = 40_000.0 # Max drawdown from equity peak (%) self.max_drawdown_pct: float = 10.0 # Sector concentration cap (% of total deployed capital) self.max_sector_pct: float = 30.0 # Correlation — max positions in same sector before blocking self.max_same_sector: int = 3 # Volatility — block new entries when India VIX > threshold self.vix_block_threshold: float = 22.0 # News risk — block symbols with NEGATIVE news sentiment self.news_filter_active: bool = True # Circuit filter — skip stocks already at 5% circuit self.circuit_filter_pct: float = 4.5 # block if move >= this % # Market hour check self.enforce_market_hours: bool = True # PAPER mode bypasses this # NSE market holidays (ISO date strings) self.market_holidays: List[str] = [] # ── Loss Tracker ─────────────────────────────────────────────────────────────── class LossTracker: """Tracks realised P&L over rolling windows.""" def __init__(self): self._lock = threading.Lock() # List of (datetime_str, pnl_float) tuples self._trades: List[Tuple[str, float]] = [] def record(self, pnl: float, ts: Optional[str] = None) -> None: ts = ts or datetime.now().strftime("%Y-%m-%d %H:%M:%S") with self._lock: self._trades.append((ts, pnl)) def _total_since(self, since: datetime) -> float: with self._lock: return sum( p for ts, p in self._trades if datetime.strptime(ts[:19], "%Y-%m-%d %H:%M:%S") >= since ) @property def today_pnl(self) -> float: today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) return self._total_since(today) @property def week_pnl(self) -> float: monday = datetime.now() - timedelta(days=datetime.now().weekday()) monday = monday.replace(hour=0, minute=0, second=0, microsecond=0) return self._total_since(monday) @property def month_pnl(self) -> float: first = datetime.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0) return self._total_since(first) def summary(self) -> dict: return { "today_pnl": round(self.today_pnl, 2), "week_pnl": round(self.week_pnl, 2), "month_pnl": round(self.month_pnl, 2), } # ── Drawdown Tracker ─────────────────────────────────────────────────────────── class DrawdownTracker: def __init__(self): self._lock = threading.Lock() self._peak = 0.0 self._current = 0.0 def update(self, pnl_delta: float) -> float: """Add a P&L delta. Returns current drawdown %.""" with self._lock: self._current += pnl_delta if self._current > self._peak: self._peak = self._current if self._peak > 0: dd_pct = (self._peak - self._current) / self._peak * 100 else: dd_pct = 0.0 return round(dd_pct, 2) @property def drawdown_pct(self) -> float: with self._lock: if self._peak > 0: return round((self._peak - self._current) / self._peak * 100, 2) return 0.0 def reset_peak(self): with self._lock: self._peak = max(self._current, 0.0) # ── Risk Verdict ─────────────────────────────────────────────────────────────── class RiskVerdict: def __init__(self, allowed: bool, reason: str, layer: str = ""): self.allowed = allowed self.reason = reason self.layer = layer # which guard blocked def to_dict(self) -> dict: return {"allowed": self.allowed, "reason": self.reason, "layer": self.layer} def __bool__(self): return self.allowed @staticmethod def ok() -> "RiskVerdict": return RiskVerdict(True, "OK", "none") # ── Risk Guard ──────────────────────────────────────────────────────────────── class RiskGuard: """ Central risk gate. Call `check_entry()` before any new position. Call `on_trade_closed()` after each exit to update loss trackers. """ def __init__(self): self.config = RiskConfig() self.loss = LossTracker() self.drawdown = DrawdownTracker() self._lock = threading.Lock() # ── External API ────────────────────────────────────────────────────────── def check_entry( self, symbol: str, sector: str, ltp: float, change_pct: float, vix: float, news_sentiment: str, # POSITIVE | NEUTRAL | NEGATIVE open_positions: List[dict], # list of dicts with 'sector' key capital: float, deployed: float, mode: str = "PAPER", exchange: str = "NSE", # NSE / BSE / MCX — determines market hours ) -> RiskVerdict: """ Run all risk checks. Returns RiskVerdict(allowed=False, ...) on first failure. """ checks = [ self._check_kill_switch, lambda: self._check_loss_limits(), lambda: self._check_drawdown(), lambda: self._check_market_hours(mode, exchange), lambda: self._check_circuit(symbol, change_pct), lambda: self._check_volatility(vix), lambda: self._check_news(symbol, news_sentiment), lambda: self._check_sector_exposure(sector, open_positions, deployed), ] for check in checks: verdict = check() if callable(check) else check if not verdict.allowed: logger.warning(f"[RiskGuard] BLOCKED {symbol}: [{verdict.layer}] {verdict.reason}") return verdict return RiskVerdict.ok() def on_trade_closed(self, pnl: float, ts: Optional[str] = None) -> None: """Call after every trade closes.""" self.loss.record(pnl, ts) self.drawdown.update(pnl) def activate_kill_switch(self, reason: str = "Manual") -> None: with self._lock: self.config.kill_switch_active = True self.config.kill_switch_reason = reason logger.critical(f"[RiskGuard] ⛔ KILL SWITCH ACTIVATED: {reason}") def deactivate_kill_switch(self) -> None: with self._lock: self.config.kill_switch_active = False self.config.kill_switch_reason = "" logger.info("[RiskGuard] ✅ Kill switch deactivated") def update_config(self, updates: dict) -> None: """Bulk-update risk config at runtime.""" with self._lock: for key, value in updates.items(): if hasattr(self.config, key): setattr(self.config, key, value) logger.info(f"[RiskGuard] Config updated: {list(updates.keys())}") def get_status(self) -> dict: return { "kill_switch": self.config.kill_switch_active, "kill_reason": self.config.kill_switch_reason, "drawdown_pct": self.drawdown.drawdown_pct, "pnl": self.loss.summary(), "limits": { "daily": self.config.daily_loss_limit, "weekly": self.config.weekly_loss_limit, "monthly": self.config.monthly_loss_limit, "drawdown": self.config.max_drawdown_pct, "vix_block": self.config.vix_block_threshold, "sector_cap": self.config.max_sector_pct, }, } def get_config_dict(self) -> dict: c = self.config return { "kill_switch_active": c.kill_switch_active, "kill_switch_reason": c.kill_switch_reason, "daily_loss_limit": c.daily_loss_limit, "weekly_loss_limit": c.weekly_loss_limit, "monthly_loss_limit": c.monthly_loss_limit, "max_drawdown_pct": c.max_drawdown_pct, "max_sector_pct": c.max_sector_pct, "max_same_sector": c.max_same_sector, "vix_block_threshold": c.vix_block_threshold, "news_filter_active": c.news_filter_active, "circuit_filter_pct": c.circuit_filter_pct, "enforce_market_hours": c.enforce_market_hours, "market_holidays": c.market_holidays, } # ── Individual checks ────────────────────────────────────────────────────── def _check_kill_switch(self) -> RiskVerdict: if self.config.kill_switch_active: return RiskVerdict(False, f"Kill switch active: {self.config.kill_switch_reason}", "kill_switch") return RiskVerdict.ok() def _check_loss_limits(self) -> RiskVerdict: today = self.loss.today_pnl week = self.loss.week_pnl month = self.loss.month_pnl if today <= -abs(self.config.daily_loss_limit): return RiskVerdict(False, f"Daily loss limit hit: ₹{today:.0f}", "daily_limit") if week <= -abs(self.config.weekly_loss_limit): return RiskVerdict(False, f"Weekly loss limit hit: ₹{week:.0f}", "weekly_limit") if month <= -abs(self.config.monthly_loss_limit): return RiskVerdict(False, f"Monthly loss limit hit: ₹{month:.0f}", "monthly_limit") return RiskVerdict.ok() def _check_drawdown(self) -> RiskVerdict: dd = self.drawdown.drawdown_pct if dd >= self.config.max_drawdown_pct: return RiskVerdict(False, f"Max drawdown breached: {dd:.1f}%", "drawdown") return RiskVerdict.ok() def _check_market_hours(self, mode: str, exchange: str = "NSE") -> RiskVerdict: if mode == "PAPER" or not self.config.enforce_market_hours: return RiskVerdict.ok() if date.today().weekday() >= 5: return RiskVerdict(False, "Markets closed on weekends", "market_hours") today_str = date.today().isoformat() if today_str in self.config.market_holidays: return RiskVerdict(False, f"NSE holiday: {today_str}", "market_holiday") now = datetime.now().time() exch = exchange.upper() # MCX commodity futures: 09:00 – 23:30 IST if exch == "MCX": market_open = datetime.strptime("09:00", "%H:%M").time() market_close = datetime.strptime("23:30", "%H:%M").time() # NSE / BSE / NFO / BFO / CDS equities & derivatives: 09:15 – 15:30 else: market_open = datetime.strptime("09:15", "%H:%M").time() market_close = datetime.strptime("15:30", "%H:%M").time() if not (market_open <= now <= market_close): return RiskVerdict( False, f"Outside {exch} market hours: {now.strftime('%H:%M')} " f"(allowed {market_open.strftime('%H:%M')}–{market_close.strftime('%H:%M')})", "market_hours" ) return RiskVerdict.ok() def _check_circuit(self, symbol: str, change_pct: float) -> RiskVerdict: threshold = self.config.circuit_filter_pct if abs(change_pct) >= threshold: direction = "upper" if change_pct > 0 else "lower" return RiskVerdict( False, f"{symbol} near {direction} circuit ({change_pct:+.1f}%)", "circuit_filter" ) return RiskVerdict.ok() def _check_volatility(self, vix: float) -> RiskVerdict: if vix > 0 and vix >= self.config.vix_block_threshold: return RiskVerdict(False, f"VIX too high: {vix:.1f}", "volatility_filter") return RiskVerdict.ok() def _check_news(self, symbol: str, sentiment: str) -> RiskVerdict: if self.config.news_filter_active and sentiment == "NEGATIVE": return RiskVerdict(False, f"{symbol} has negative news sentiment", "news_filter") return RiskVerdict.ok() def _check_sector_exposure( self, sector: str, open_positions: List[dict], deployed: float, ) -> RiskVerdict: # Count open positions in same sector same_sector = [p for p in open_positions if p.get("sector") == sector] if len(same_sector) >= self.config.max_same_sector: return RiskVerdict( False, f"Too many open {sector} positions ({len(same_sector)})", "sector_exposure" ) return RiskVerdict.ok() # ── Singleton ───────────────────────────────────────────────────────────────── risk_guard = RiskGuard()