Autonomous trading agents can execute hundreds of orders per hour without fatigue or emotional interference. That is a substantial edge over human traders. But it is also how an agent can blow up an entire portfolio before any human notices. Risk management is not optional — it is the difference between an agent that compounds wealth over time and one that zeros out in a single bad session.
This guide covers the key risk metrics every agent must track, how to implement position limits and stop-losses via the Purple Flea Trading API, and a full Python RiskManager class you can drop into any agent architecture.
The Core Risk Metrics
Every trading agent should continuously monitor five metrics. If any breach a threshold, the agent should reduce exposure or halt trading.
1. Value at Risk (VaR)
VaR estimates the maximum expected loss over a given time period at a given confidence level. A 95% 1-day VaR of $500 means: there is a 5% probability the portfolio loses more than $500 in a single day.
For agent portfolios, use the historical simulation method: take the last N days of P&L, sort them, and read off the 5th percentile. It requires no distributional assumptions and captures fat tails naturally.
2. Maximum Drawdown
Max drawdown is the peak-to-trough decline in portfolio value over a given period. It is the single most intuitive measure of downside risk. An agent that has a maximum drawdown of 40% is dangerous regardless of its Sharpe ratio — it means at some point it lost nearly half its capital.
Track both the current drawdown (how far below the most recent high are we now?) and the historical maximum drawdown. Set a circuit breaker at a threshold — e.g., halt trading if current drawdown exceeds 15%.
3. Sharpe Ratio
The Sharpe ratio normalizes returns by volatility: (mean return - risk-free rate) / standard deviation of returns. A Sharpe above 1.0 is acceptable. Above 2.0 is excellent. Below 0.5 means you are taking too much risk for your return.
Measure it on a rolling 30-day window. A declining Sharpe ratio is an early warning sign that market conditions have shifted against your strategy.
4. Position Concentration
No single position should represent more than a fixed percentage of total capital. Common limits: 5% per position for aggressive agents, 2% for conservative ones. High concentration is a single-point-of-failure risk — one bad trade wipes a meaningful fraction of capital.
5. Correlation Exposure
In a multi-position portfolio, if all positions are highly correlated, a market-wide shock hits everything simultaneously. Monitor the average pairwise correlation of open positions. If correlation exceeds 0.7 across the board, the portfolio is effectively one large leveraged bet.
| Metric | Healthy Range | Warning | Halt Trading |
|---|---|---|---|
| Daily VaR (95%) | < 2% of capital | 2–5% | > 5% |
| Current Drawdown | < 5% | 5–15% | > 15% |
| Rolling Sharpe (30d) | > 1.0 | 0.3–1.0 | < 0.3 |
| Max Position Size | < 5% per trade | 5–10% | > 10% |
| Portfolio Correlation | < 0.5 avg | 0.5–0.7 | > 0.7 |
| Daily P&L Loss | < 3% of capital | 3–6% | > 6% |
Position Limits: Never Bet More Than X%
The rule is simple: before any trade, compute the position size as a percentage of total capital. If it exceeds your limit, reduce the size. Never override this — not for "high-conviction" trades, not under any circumstances.
The Kelly Criterion gives the mathematically optimal fraction to risk per trade given a known edge and win rate. Full Kelly is aggressive; most practitioners use half-Kelly or quarter-Kelly for safety.
def kelly_fraction(win_rate: float, win_loss_ratio: float) -> float: """ Kelly Criterion: f* = (p * b - q) / b p = win_rate, q = 1 - p, b = win/loss ratio Returns fraction of capital to risk. """ q = 1 - win_rate f_star = (win_rate * win_loss_ratio - q) / win_loss_ratio return max(0.0, f_star) # never negative def safe_position_size( capital: float, win_rate: float, win_loss_ratio: float, kelly_fraction_cap: float = 0.25, # use quarter-Kelly hard_cap_pct: float = 0.05, # never exceed 5% of capital ) -> float: """Return the safe dollar position size for a trade.""" kelly = kelly_fraction(win_rate, win_loss_ratio) fractional_kelly = kelly * kelly_fraction_cap capped = min(fractional_kelly, hard_cap_pct) return capital * capped # Example: 55% win rate, 1.5:1 win/loss ratio, $10,000 capital size = safe_position_size( capital=10_000, win_rate=0.55, win_loss_ratio=1.5, ) print(f"Safe position size: ${size:.2f}") # ~$125
Stop-Loss Automation via Purple Flea Trading API
The Purple Flea Trading API supports server-side stop-loss orders. Set them at the time of opening a position — do not rely on your agent polling to close positions manually. A polling loop that stalls during a network partition is a risk event waiting to happen.
import requests TRADING_BASE = "https://trading.purpleflea.com" def open_position_with_stop( api_key: str, symbol: str, side: str, # "buy" or "sell" size_usd: float, entry_price: float, stop_loss_pct: float = 0.02, # 2% stop-loss take_profit_pct: float = 0.04, # 4% take-profit (2:1 R/R) ) -> dict: """Open a leveraged position with server-side stop-loss.""" if side == "buy": stop_price = entry_price * (1 - stop_loss_pct) tp_price = entry_price * (1 + take_profit_pct) else: stop_price = entry_price * (1 + stop_loss_pct) tp_price = entry_price * (1 - take_profit_pct) payload = { "symbol": symbol, "side": side, "size_usd": size_usd, "order_type": "market", "stop_loss": round(stop_price, 4), "take_profit": round(tp_price, 4), } r = requests.post( f"{TRADING_BASE}/api/v1/orders", json=payload, headers={"Authorization": f"Bearer {api_key}"}, timeout=10, ) r.raise_for_status() order = r.json() print(f"Opened {side} {symbol}: stop={stop_price:.4f} tp={tp_price:.4f}") return order
Always set stop-losses server-side at order creation time. Never depend on your agent loop to monitor and close positions — network latency, crashes, or model timeouts can cause catastrophic unprotected drawdowns.
Correlation Monitoring Across Positions
Correlation monitoring prevents the hidden risk of a portfolio that looks diversified but moves as one. If you hold BTC, ETH, and SOL longs simultaneously, you effectively have a single leveraged crypto long — all three are highly correlated in market stress events.
Compute rolling correlations on 5-minute price returns. When average pairwise correlation exceeds your threshold, reduce position sizes across the board or exit the most correlated positions.
import numpy as np from collections import deque from typing import Dict, List class CorrelationMonitor: def __init__(self, window: int = 100, threshold: float = 0.7): self.window = window self.threshold = threshold self.returns: Dict[str, deque] = {} def update(self, symbol: str, price: float): """Feed in a new price tick for a symbol.""" if symbol not in self.returns: self.returns[symbol] = deque(maxlen=self.window + 1) self.returns[symbol].append(price) def avg_pairwise_correlation(self, symbols: List[str]) -> float: """Compute average pairwise Pearson correlation of open positions.""" if len(symbols) < 2: return 0.0 series = [] for sym in symbols: prices = np.array(self.returns.get(sym, [])) if len(prices) < 2: continue rets = np.diff(prices) / prices[:-1] series.append(rets) if len(series) < 2: return 0.0 # Align lengths min_len = min(len(s) for s in series) mat = np.vstack([s[-min_len:] for s in series]) corr_matrix = np.corrcoef(mat) # Average upper-triangle (exclude diagonal) n = corr_matrix.shape[0] total, count = 0.0, 0 for i in range(n): for j in range(i + 1, n): total += corr_matrix[i, j] count += 1 return total / count if count > 0 else 0.0 def is_overexposed(self, open_symbols: List[str]) -> bool: avg_corr = self.avg_pairwise_correlation(open_symbols) return avg_corr > self.threshold
Circuit Breakers: Pause Trading on Daily Loss Threshold
A circuit breaker is the most important safety mechanism in any trading agent. The rule: if cumulative losses in the current trading day exceed a threshold (e.g., 6% of capital), halt all trading for the remainder of the day. No exceptions.
Circuit breakers exist because drawdowns tend to be mean-reverting in human traders but can accelerate in agents. A strategy that loses 6% in one day is operating outside the regime it was designed for. Continuing to trade in an unrecognised regime is not rational risk-taking — it is compounding a bad situation.
from datetime import date class CircuitBreaker: def __init__(self, daily_loss_limit_pct: float = 0.06): self.daily_loss_limit_pct = daily_loss_limit_pct self._tripped = False self._trip_date = None self._start_equity = None self._day = None def reset_if_new_day(self, current_equity: float): today = date.today() if self._day != today: self._day = today self._tripped = False self._start_equity = current_equity def check(self, current_equity: float) -> bool: """Returns True if trading is allowed, False if halted.""" self.reset_if_new_day(current_equity) if self._tripped: return False if self._start_equity is None or self._start_equity == 0: return True daily_loss_pct = (self._start_equity - current_equity) / self._start_equity if daily_loss_pct >= self.daily_loss_limit_pct: self._tripped = True self._trip_date = date.today() print( f"[CIRCUIT BREAKER] Daily loss {daily_loss_pct:.1%} exceeded " f"{self.daily_loss_limit_pct:.1%} limit. Trading halted for today." ) return False return True @property def is_tripped(self) -> bool: return self._tripped
Full Python RiskManager Class
The following class integrates all five risk controls into a single object you can use in any agent architecture. Call pre_trade_check() before any order. Call update() after each tick or position change.
import numpy as np from dataclasses import dataclass, field from typing import Dict, List, Optional from datetime import date from collections import deque @dataclass class RiskConfig: # Position limits max_position_pct: float = 0.05 # max 5% per trade kelly_fraction: float = 0.25 # quarter-Kelly # Drawdown limits max_drawdown_pct: float = 0.15 # halt at 15% drawdown daily_loss_limit_pct: float = 0.06 # halt at 6% daily loss # VaR var_confidence: float = 0.95 # 95% confidence var_limit_pct: float = 0.05 # halt if VaR > 5% capital # Sharpe sharpe_window_days: int = 30 min_sharpe: float = 0.3 # halt if Sharpe < 0.3 # Correlation correlation_window: int = 100 max_avg_correlation: float = 0.7 class RiskManager: def __init__(self, capital: float, config: Optional[RiskConfig] = None): self.capital = capital self.config = config or RiskConfig() # State self.peak_equity = capital self.daily_start = capital self.daily_date = date.today() self.pnl_history: deque = deque(maxlen=self.config.sharpe_window_days) self.price_history: Dict[str, deque] = {} self.halted = False self.halt_reason: Optional[str] = None def update_equity(self, equity: float): """Call this whenever portfolio equity changes.""" today = date.today() if today != self.daily_date: # New day: record yesterday's P&L and reset daily_pnl_pct = (equity - self.daily_start) / self.daily_start self.pnl_history.append(daily_pnl_pct) self.daily_start = equity self.daily_date = today self.halted = False self.halt_reason = None self.capital = equity self.peak_equity = max(self.peak_equity, equity) self._check_limits() def _current_drawdown(self) -> float: if self.peak_equity == 0: return 0.0 return (self.peak_equity - self.capital) / self.peak_equity def _daily_loss(self) -> float: if self.daily_start == 0: return 0.0 loss = (self.daily_start - self.capital) / self.daily_start return max(0.0, loss) def _rolling_sharpe(self) -> Optional[float]: if len(self.pnl_history) < 5: return None rets = np.array(self.pnl_history) std = np.std(rets) if std == 0: return None return np.mean(rets) / std * np.sqrt(252) # annualised def _historical_var(self) -> Optional[float]: """95% 1-day VaR as a fraction of capital.""" if len(self.pnl_history) < 10: return None rets = np.array(self.pnl_history) var_pct = 1.0 - self.config.var_confidence return abs(np.percentile(rets, var_pct * 100)) def _check_limits(self): if self.halted: return dd = self._current_drawdown() if dd >= self.config.max_drawdown_pct: self._halt(f"Max drawdown {dd:.1%} exceeded {self.config.max_drawdown_pct:.1%}") return dl = self._daily_loss() if dl >= self.config.daily_loss_limit_pct: self._halt(f"Daily loss {dl:.1%} exceeded {self.config.daily_loss_limit_pct:.1%}") return sharpe = self._rolling_sharpe() if sharpe is not None and sharpe < self.config.min_sharpe: self._halt(f"Rolling Sharpe {sharpe:.2f} below minimum {self.config.min_sharpe:.2f}") return var = self._historical_var() if var is not None and var >= self.config.var_limit_pct: self._halt(f"1-day VaR {var:.1%} exceeded limit {self.config.var_limit_pct:.1%}") def _halt(self, reason: str): self.halted = True self.halt_reason = reason print(f"[RISK HALT] {reason}") def pre_trade_check( self, trade_size_usd: float, open_symbols: Optional[List[str]] = None, ) -> bool: """ Returns True if trade is allowed, False if blocked. Call this before every order submission. """ if self.halted: print(f"[RISK] Trade blocked — system halted: {self.halt_reason}") return False # Check position size limit position_pct = trade_size_usd / self.capital if self.capital > 0 else 1.0 if position_pct > self.config.max_position_pct: print( f"[RISK] Trade blocked — size {position_pct:.1%} exceeds " f"limit {self.config.max_position_pct:.1%}" ) return False return True def status(self) -> dict: return { "capital": self.capital, "halted": self.halted, "halt_reason": self.halt_reason, "current_drawdown": f"{self._current_drawdown():.2%}", "daily_loss": f"{self._daily_loss():.2%}", "rolling_sharpe": self._rolling_sharpe(), "var_95": self._historical_var(), }
Putting It Together: A Risk-Aware Agent Loop
The following sketch shows how a complete risk-aware trading agent integrates the RiskManager into its main loop. The risk manager is checked at every decision point.
import time, requests def run_trading_agent(api_key: str, starting_capital: float): risk = RiskManager(capital=starting_capital) while True: # 1. Fetch current equity equity_resp = requests.get( "https://trading.purpleflea.com/api/v1/account", headers={"Authorization": f"Bearer {api_key}"}, ).json() equity = equity_resp["equity_usd"] risk.update_equity(equity) # 2. Log risk status every loop status = risk.status() print(status) # 3. Skip trading if halted if risk.halted: print("Risk system halted — sleeping 1 hour") time.sleep(3600) continue # 4. Run your signal generation logic signal = get_trading_signal() # your function if signal is None: time.sleep(60) continue # 5. Compute safe position size trade_size = safe_position_size( capital=equity, win_rate=signal["win_rate"], win_loss_ratio=signal["win_loss_ratio"], ) # 6. Pre-trade risk check if not risk.pre_trade_check(trade_size_usd=trade_size): time.sleep(60) continue # 7. Execute with server-side stop-loss open_position_with_stop( api_key=api_key, symbol=signal["symbol"], side=signal["side"], size_usd=trade_size, entry_price=signal["price"], ) time.sleep(30)
Risk management is only effective if the agent enforces it unconditionally. Never add code paths that bypass the pre-trade check or override the circuit breaker. A single bypass is enough to lose everything in a bad market event.
Next Steps
Implement the RiskManager class in your agent and connect it to the Purple Flea Trading API. Set conservative thresholds to start — you can loosen them once you have real performance data. Track the status() output continuously and alert on any approaching thresholds before they trip the circuit breaker.
- Start trading: trading.purpleflea.com
- Full Trading API docs: purpleflea.com/docs
- Claim free $1 to start: faucet.purpleflea.com
- Multi-agent strategies: AI Agent Hedge Fund guide