""" Risk Manager — Circuit breakers, depeg detection, and position limits ====================================================================== Monitors portfolio health and enforces risk constraints before any rebalancing action is executed on-chain. """ import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Tuple import numpy as np from config.constants import RiskConfig, PortfolioConfig logger = logging.getLogger("risk_manager") class RiskLevel(Enum): """Risk assessment levels.""" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class CircuitBreakerState(Enum): """Circuit breaker states.""" CLOSED = "closed" # normal operation OPEN = "open" # halted — no trades HALF_OPEN = "half_open" # testing if conditions improved @dataclass class RiskAssessment: """Complete risk assessment for a proposed allocation.""" timestamp: float overall_risk: RiskLevel risk_score: float # 0 (safe) to 1 (extreme) # Component scores depeg_risk: float = 0.0 volatility_risk: float = 0.0 concentration_risk: float = 0.0 smart_contract_risk: float = 0.0 liquidity_risk: float = 0.0 drawdown_risk: float = 0.0 # Flags circuit_breaker_triggered: bool = False emergency_exit_recommended: bool = False rebalance_approved: bool = True # Adjustments adjusted_weights: Optional[np.ndarray] = None warnings: List[str] = field(default_factory=list) def to_dict(self) -> Dict: return { "timestamp": self.timestamp, "overall_risk": self.overall_risk.value, "risk_score": round(self.risk_score, 4), "depeg_risk": round(self.depeg_risk, 4), "volatility_risk": round(self.volatility_risk, 4), "concentration_risk": round(self.concentration_risk, 4), "smart_contract_risk": round(self.smart_contract_risk, 4), "liquidity_risk": round(self.liquidity_risk, 4), "drawdown_risk": round(self.drawdown_risk, 4), "circuit_breaker": self.circuit_breaker_triggered, "emergency_exit": self.emergency_exit_recommended, "approved": self.rebalance_approved, "warnings": self.warnings, } class RiskManager: """ Multi-layer risk management system. Layers: 1. Depeg Detection — monitors USDY/USD and mETH/ETH pegs 2. Volatility Guard — scales down risky assets in high-vol regimes 3. Concentration Limits — enforces max/min position sizes 4. Smart Contract Risk — weights by protocol audit status 5. Drawdown Protection — circuit breaker on excessive losses 6. Liquidity Check — ensures sufficient DEX liquidity for rebalance size """ def __init__( self, risk_config: Optional[RiskConfig] = None, portfolio_config: Optional[PortfolioConfig] = None, ): self.risk_cfg = risk_config or RiskConfig() self.portfolio_cfg = portfolio_config or PortfolioConfig() # Circuit breaker state self.circuit_breaker = CircuitBreakerState.CLOSED self.circuit_breaker_triggered_at: Optional[float] = None # Tracking self.portfolio_value_history: List[Tuple[float, float]] = [] # (timestamp, value) self.peak_portfolio_value: float = 0.0 self.risk_assessment_history: List[RiskAssessment] = [] # Depeg tracking (rolling window) self.usdy_peg_history: List[float] = [] self.meth_peg_history: List[float] = [] def assess_risk( self, proposed_weights: np.ndarray, current_weights: np.ndarray, snapshot, # YieldSnapshot portfolio_value: float, ) -> RiskAssessment: """ Comprehensive risk assessment of a proposed allocation. Args: proposed_weights: Target weights [USDY, mETH, MI4] current_weights: Current weights [USDY, mETH, MI4] snapshot: Current market data (YieldSnapshot) portfolio_value: Current portfolio value in USD Returns: RiskAssessment with approval/rejection and adjustments """ assessment = RiskAssessment( timestamp=time.time(), overall_risk=RiskLevel.LOW, risk_score=0.0, ) # ─── Layer 1: Depeg Detection ─── self._check_depeg(snapshot, proposed_weights, assessment) # ─── Layer 2: Volatility Guard ─── self._check_volatility(snapshot, proposed_weights, assessment) # ─── Layer 3: Concentration Limits ─── self._check_concentration(proposed_weights, assessment) # ─── Layer 4: Smart Contract Risk ─── self._check_smart_contract_risk(proposed_weights, assessment) # ─── Layer 5: Drawdown Protection ─── self._check_drawdown(portfolio_value, assessment) # ─── Layer 6: Circuit Breaker ─── self._check_circuit_breaker(assessment) # ─── Compute Overall Score ─── risk_components = [ assessment.depeg_risk * 0.25, assessment.volatility_risk * 0.20, assessment.concentration_risk * 0.15, assessment.smart_contract_risk * 0.15, assessment.drawdown_risk * 0.15, assessment.liquidity_risk * 0.10, ] assessment.risk_score = sum(risk_components) # Determine risk level if assessment.risk_score < 0.2: assessment.overall_risk = RiskLevel.LOW elif assessment.risk_score < 0.4: assessment.overall_risk = RiskLevel.MEDIUM elif assessment.risk_score < 0.7: assessment.overall_risk = RiskLevel.HIGH else: assessment.overall_risk = RiskLevel.CRITICAL # Apply risk-adjusted weights if needed if assessment.adjusted_weights is None: assessment.adjusted_weights = proposed_weights.copy() # Block rebalance if critical if assessment.overall_risk == RiskLevel.CRITICAL: assessment.rebalance_approved = False assessment.warnings.append("CRITICAL risk level — rebalance blocked") if assessment.circuit_breaker_triggered: assessment.rebalance_approved = False # Log self.risk_assessment_history.append(assessment) logger.info( f"Risk Assessment: {assessment.overall_risk.value} " f"(score={assessment.risk_score:.3f}) " f"approved={assessment.rebalance_approved} " f"warnings={len(assessment.warnings)}" ) return assessment def _check_depeg(self, snapshot, weights: np.ndarray, assessment: RiskAssessment): """Check USDY and mETH peg health.""" # USDY peg check usdy_deviation = abs(1.0 - snapshot.usdy_peg) self.usdy_peg_history.append(snapshot.usdy_peg) if len(self.usdy_peg_history) > 1000: self.usdy_peg_history = self.usdy_peg_history[-500:] if usdy_deviation > self.risk_cfg.usdy_depeg_threshold: assessment.depeg_risk += 0.5 assessment.warnings.append( f"⚠️ USDY depeg detected: ${snapshot.usdy_peg:.4f} " f"(deviation: {usdy_deviation*100:.2f}%)" ) # Reduce USDY exposure if assessment.adjusted_weights is None: assessment.adjusted_weights = weights.copy() assessment.adjusted_weights[0] = min(weights[0], 0.20) # mETH peg check meth_deviation = abs(1.0 - snapshot.meth_peg) self.meth_peg_history.append(snapshot.meth_peg) if len(self.meth_peg_history) > 1000: self.meth_peg_history = self.meth_peg_history[-500:] if meth_deviation > self.risk_cfg.meth_depeg_threshold: assessment.depeg_risk += 0.5 assessment.warnings.append( f"⚠️ mETH depeg detected: ratio={snapshot.meth_peg:.4f} " f"(deviation: {meth_deviation*100:.2f}%)" ) if assessment.adjusted_weights is None: assessment.adjusted_weights = weights.copy() assessment.adjusted_weights[1] = min(weights[1], 0.15) # Severe depeg — emergency exit if usdy_deviation > 0.03 or meth_deviation > 0.05: assessment.emergency_exit_recommended = True assessment.warnings.append("🚨 SEVERE DEPEG — emergency exit to USDC recommended") def _check_volatility(self, snapshot, weights: np.ndarray, assessment: RiskAssessment): """Scale down risky assets during high-volatility regimes.""" eth_vol = snapshot.eth_volatility_30d if eth_vol > 0.8: assessment.volatility_risk = min(eth_vol / 1.5, 1.0) assessment.warnings.append( f"⚠️ High ETH volatility: {eth_vol:.2f} (annualized)" ) # Reduce mETH exposure, increase USDY (safe haven) if assessment.adjusted_weights is None: assessment.adjusted_weights = weights.copy() vol_penalty = min((eth_vol - 0.8) / 0.5, 0.5) assessment.adjusted_weights[1] *= (1 - vol_penalty) assessment.adjusted_weights[0] += weights[1] * vol_penalty assessment.adjusted_weights /= assessment.adjusted_weights.sum() elif eth_vol > 0.6: assessment.volatility_risk = 0.3 def _check_concentration(self, weights: np.ndarray, assessment: RiskAssessment): """Enforce position size limits.""" max_w = self.portfolio_cfg.max_single_asset_weight min_w = self.portfolio_cfg.min_single_asset_weight asset_names = ["USDY", "mETH", "MI4"] if assessment.adjusted_weights is None: assessment.adjusted_weights = weights.copy() for i, (w, name) in enumerate(zip(weights, asset_names)): if w > max_w: assessment.concentration_risk += 0.3 assessment.warnings.append( f"⚠️ {name} weight {w:.2f} exceeds max {max_w:.2f}" ) assessment.adjusted_weights[i] = max_w elif w < min_w: assessment.concentration_risk += 0.1 assessment.adjusted_weights[i] = min_w # Renormalize assessment.adjusted_weights /= assessment.adjusted_weights.sum() def _check_smart_contract_risk(self, weights: np.ndarray, assessment: RiskAssessment): """Weight-adjusted protocol risk score.""" protocol_risks = self.risk_cfg.protocol_risk_scores asset_protocols = ["USDY", "mETH", "MI4"] weighted_risk = sum( weights[i] * protocol_risks.get(asset_protocols[i], 0.5) for i in range(len(weights)) ) assessment.smart_contract_risk = weighted_risk def _check_drawdown(self, current_value: float, assessment: RiskAssessment): """Check portfolio drawdown against limits.""" self.portfolio_value_history.append((time.time(), current_value)) self.peak_portfolio_value = max(self.peak_portfolio_value, current_value) if self.peak_portfolio_value > 0: drawdown = (self.peak_portfolio_value - current_value) / self.peak_portfolio_value else: drawdown = 0.0 assessment.drawdown_risk = min(drawdown / self.portfolio_cfg.max_drawdown_pct, 1.0) if drawdown > self.portfolio_cfg.max_drawdown_pct: assessment.warnings.append( f"🚨 Drawdown {drawdown*100:.2f}% exceeds max {self.portfolio_cfg.max_drawdown_pct*100:.0f}%" ) assessment.circuit_breaker_triggered = True def _check_circuit_breaker(self, assessment: RiskAssessment): """Manage circuit breaker state machine.""" if assessment.circuit_breaker_triggered: self.circuit_breaker = CircuitBreakerState.OPEN self.circuit_breaker_triggered_at = time.time() assessment.rebalance_approved = False logger.warning("🔴 Circuit breaker OPEN — all rebalancing halted") return if self.circuit_breaker == CircuitBreakerState.OPEN: # Check if cooldown has passed elapsed_hours = (time.time() - (self.circuit_breaker_triggered_at or 0)) / 3600 if elapsed_hours > self.risk_cfg.circuit_breaker_cooldown_hours: self.circuit_breaker = CircuitBreakerState.HALF_OPEN logger.info("🟡 Circuit breaker HALF-OPEN — testing conditions") else: assessment.rebalance_approved = False assessment.circuit_breaker_triggered = True if self.circuit_breaker == CircuitBreakerState.HALF_OPEN: # Allow only conservative rebalances if assessment.risk_score < 0.3: self.circuit_breaker = CircuitBreakerState.CLOSED logger.info("🟢 Circuit breaker CLOSED — normal operations resumed") else: assessment.rebalance_approved = False def get_emergency_exit_weights(self) -> np.ndarray: """Return safe-haven weights for emergency exit.""" # 90% USDY (T-bills), 5% mETH, 5% MI4 return np.array([0.90, 0.05, 0.05]) def get_risk_summary(self) -> Dict: """Return summary of current risk state.""" recent = self.risk_assessment_history[-1] if self.risk_assessment_history else None return { "circuit_breaker": self.circuit_breaker.value, "peak_value": self.peak_portfolio_value, "current_drawdown": ( (self.peak_portfolio_value - self.portfolio_value_history[-1][1]) / self.peak_portfolio_value if self.portfolio_value_history and self.peak_portfolio_value > 0 else 0.0 ), "latest_risk_score": recent.risk_score if recent else 0.0, "latest_risk_level": recent.overall_risk.value if recent else "unknown", "total_assessments": len(self.risk_assessment_history), "total_warnings": sum(len(a.warnings) for a in self.risk_assessment_history), } # ─────────────────────── Correlation Risk ─────────────────────────── class CorrelationMonitor: """ Monitors rolling correlations between portfolio assets. High correlation = reduced diversification benefit. """ def __init__(self, window: int = 168): # 1 week of hourly data self.window = window self.returns = {"usdy": [], "meth": [], "mi4": []} def update(self, usdy_return: float, meth_return: float, mi4_return: float): """Add new return observations.""" self.returns["usdy"].append(usdy_return) self.returns["meth"].append(meth_return) self.returns["mi4"].append(mi4_return) # Trim to window for key in self.returns: if len(self.returns[key]) > self.window: self.returns[key] = self.returns[key][-self.window:] def get_correlation_matrix(self) -> np.ndarray: """Compute correlation matrix from recent returns.""" if len(self.returns["usdy"]) < 10: return np.eye(3) data = np.array([ self.returns["usdy"], self.returns["meth"], self.returns["mi4"], ]) return np.corrcoef(data) def get_diversification_ratio(self, weights: np.ndarray) -> float: """ Diversification ratio: weighted avg vol / portfolio vol. Higher = better diversification (>1 means diversification benefit). """ if len(self.returns["usdy"]) < 10: return 1.0 vols = np.array([ np.std(self.returns["usdy"]), np.std(self.returns["meth"]), np.std(self.returns["mi4"]), ]) corr = self.get_correlation_matrix() cov = np.outer(vols, vols) * corr weighted_vol = np.dot(weights, vols) portfolio_vol = np.sqrt(weights @ cov @ weights) if portfolio_vol == 0: return 1.0 return weighted_vol / portfolio_vol