Fresh history cut from cynarAI/evoo antigravity-handoff/project-timeline (Voice First + project timeline). Brand docs/titles as evOo; keep @rakazo/* packages as tech debt. Excludes node_modules, caches, secrets, voice blobs.
41 lines
1.7 KiB
Python
41 lines
1.7 KiB
Python
"""
|
|
Apex Trading Risk Guard
|
|
Hard enforcement of risk constraints defined in BLUEPRINT.md.
|
|
"""
|
|
|
|
from typing import Dict, Any, Tuple
|
|
|
|
class ApexRiskGuard:
|
|
def __init__(self, config: Dict[str, Any]):
|
|
self.config = config.get("risk_limits", {})
|
|
self.max_drawdown = self.config.get("max_drawdown_percent", 1.5)
|
|
self.max_trade_risk = self.config.get("max_single_trade_risk_percent", 0.5)
|
|
self.max_positions = self.config.get("max_open_positions", 4)
|
|
self.circuit_breaker_active = False
|
|
|
|
def validate_order(self, order: Dict[str, Any], current_drawdown: float, open_position_count: int) -> Tuple[bool, str]:
|
|
"""
|
|
Validates an incoming order against hard guardrails.
|
|
Returns (is_approved: bool, reason: str).
|
|
"""
|
|
if self.circuit_breaker_active:
|
|
return False, "REJECTED: Circuit breaker is active."
|
|
|
|
if current_drawdown >= self.max_drawdown:
|
|
self.circuit_breaker_active = True
|
|
return False, f"REJECTED & CIRCUIT BREAKER TRIGGERED: Drawdown {current_drawdown}% >= limit {self.max_drawdown}%"
|
|
|
|
if open_position_count >= self.max_positions:
|
|
return False, f"REJECTED: Maximum concurrent positions ({self.max_positions}) reached."
|
|
|
|
order_risk_pct = order.get("risk_percent", 0.0)
|
|
if order_risk_pct > self.max_trade_risk:
|
|
return False, f"REJECTED: Order risk {order_risk_pct}% exceeds limit {self.max_trade_risk}%"
|
|
|
|
return True, "APPROVED"
|
|
|
|
def trip_kill_switch(self, reason: str):
|
|
"""Emergency hard stop."""
|
|
self.circuit_breaker_active = True
|
|
print(f"[APEX EMERGENCY KILL SWITCH] {reason}")
|