Risk Engine Design Patterns for Financial AI Systems
Why Risk Engines Are Non-Negotiable
I have seen AI trading systems generate brilliant signals and then blow up spectacularly because nobody built a proper risk engine. The signal is only half the system. The risk engine is what keeps you in the game when your model encounters market conditions it has never seen before.
Here are the design patterns I use for every financial AI system I build.
Pattern 1: The Layered Gate Architecture
Every trade signal passes through multiple independent risk gates before execution. Each gate can approve, modify, or reject the trade. This defense-in-depth approach means no single failure mode can cause catastrophic losses.
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class TradeSignal:
symbol: str
direction: str # 'long' or 'short'
size: float
price: float
strategy_id: str
@dataclass
class RiskDecision:
approved: bool
modified_size: float | None = None
reason: str = ""
class RiskGate(ABC):
@abstractmethod
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
pass
class RiskEngine:
def __init__(self):
self.gates: list[RiskGate] = []
def add_gate(self, gate: RiskGate):
self.gates.append(gate)
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
current_size = signal.size
for gate in self.gates:
decision = gate.evaluate(
TradeSignal(**{**signal.__dict__, 'size': current_size}),
portfolio
)
if not decision.approved:
return decision
if decision.modified_size is not None:
current_size = decision.modified_size
return RiskDecision(approved=True, modified_size=current_size)Pattern 2: Position Limits
Position limits prevent any single position from dominating your portfolio. I implement both absolute limits (maximum dollar exposure) and relative limits (maximum percentage of portfolio).
class PositionLimitGate(RiskGate):
def __init__(self, max_position_pct: float = 0.05, max_total_exposure: float = 1.0):
self.max_position_pct = max_position_pct
self.max_total_exposure = max_total_exposure
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
portfolio_value = portfolio['total_value']
position_value = signal.size * signal.price
position_pct = position_value / portfolio_value
if position_pct > self.max_position_pct:
adjusted_size = (self.max_position_pct * portfolio_value) / signal.price
return RiskDecision(
approved=True,
modified_size=adjusted_size,
reason=f"Position capped at {self.max_position_pct:.0%}"
)
total_exposure = portfolio['total_exposure'] + position_value
if total_exposure / portfolio_value > self.max_total_exposure:
return RiskDecision(
approved=False,
reason="Total exposure limit exceeded"
)
return RiskDecision(approved=True)Pattern 3: Drawdown Circuit Breaker
This is the most important safety mechanism. When the portfolio drawdown exceeds a threshold, the circuit breaker halts all new positions and optionally reduces existing ones.
class DrawdownCircuitBreaker(RiskGate):
def __init__(self, max_drawdown: float = 0.10, cooldown_hours: int = 24):
self.max_drawdown = max_drawdown
self.cooldown_hours = cooldown_hours
self.tripped_at = None
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
current_drawdown = portfolio['current_drawdown']
if current_drawdown >= self.max_drawdown:
self.tripped_at = datetime.utcnow()
return RiskDecision(
approved=False,
reason=f"Circuit breaker: drawdown {current_drawdown:.1%} exceeds {self.max_drawdown:.0%}"
)
if self.tripped_at:
hours_since = (datetime.utcnow() - self.tripped_at).total_seconds() / 3600
if hours_since < self.cooldown_hours:
return RiskDecision(
approved=False,
reason=f"Circuit breaker cooldown: {self.cooldown_hours - hours_since:.0f}h remaining"
)
self.tripped_at = None
return RiskDecision(approved=True)Pattern 4: Correlation Guard
AI models often generate correlated signals across related instruments. Without a correlation guard, you end up with a portfolio that looks diversified but actually has concentrated directional exposure.
class CorrelationGuard(RiskGate):
def __init__(self, max_correlated_exposure: float = 0.15):
self.max_correlated_exposure = max_correlated_exposure
self.correlation_matrix = {}
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
correlated_exposure = 0
for position in portfolio['positions']:
corr = self.get_correlation(signal.symbol, position['symbol'])
if corr > 0.7:
correlated_exposure += position['value'] / portfolio['total_value']
if correlated_exposure > self.max_correlated_exposure:
return RiskDecision(
approved=False,
reason=f"Correlated exposure {correlated_exposure:.0%} exceeds limit"
)
return RiskDecision(approved=True)Pattern 5: Volatility Scaling
Position sizes should shrink when volatility expands. This is not just prudent risk management; it also improves risk-adjusted returns because you take smaller positions when the market is most uncertain.
class VolatilityScaler(RiskGate):
def __init__(self, target_vol: float = 0.15):
self.target_vol = target_vol
def evaluate(self, signal: TradeSignal, portfolio: dict) -> RiskDecision:
current_vol = portfolio.get('realized_vol_20d', 0.15)
scale_factor = self.target_vol / max(current_vol, 0.01)
scale_factor = min(scale_factor, 2.0) # Cap upward scaling
adjusted_size = signal.size * scale_factor
return RiskDecision(
approved=True,
modified_size=adjusted_size,
reason=f"Vol scaled by {scale_factor:.2f}x"
)Wiring It All Together
engine = RiskEngine()
engine.add_gate(DrawdownCircuitBreaker(max_drawdown=0.10))
engine.add_gate(PositionLimitGate(max_position_pct=0.05))
engine.add_gate(CorrelationGuard(max_correlated_exposure=0.15))
engine.add_gate(VolatilityScaler(target_vol=0.15))
# Every signal goes through all gates
decision = engine.evaluate(signal, portfolio_state)
if decision.approved:
execute_trade(signal, size=decision.modified_size)
else:
log_rejection(signal, decision.reason)The Golden Rule
Build your risk engine before your signal generator. If that sounds backwards, consider this: a mediocre signal with excellent risk management will survive long enough to improve. A brilliant signal with no risk management will eventually find the one market condition that destroys it. The risk engine is what keeps the lights on.