Development of Risk Management System
Risk management is not a collection of tools, but an architecture that permeates the entire trading system. One unmanaged risk can destroy months of profitable trading. A risk management system determines when to trade, how much to trade, and when to stop completely.
Risk Hierarchy
Market risk — losses from adverse price movement. Managed through position sizing, stop-loss, diversification.
Execution risk — deviation of execution price from expected (slippage). Especially critical in high-frequency trading and large orders.
Liquidity risk — inability to close position at reasonable price. Relevant for positions in illiquid assets.
Counterparty risk — exchange risk (hack, bankruptcy, regulatory freeze). Solved through diversification across multiple exchanges and withdrawal of most funds from exchanges.
Operational risk — technical failures, connection loss, code errors. Solved through reliable infrastructure, circuit breakers and monitoring.
Concentration risk — excessive dependence on single asset or strategy.
Risk Control Levels
Trade level — control of each individual trade:
class TradeRiskCheck:
def __init__(self, max_position_size_pct=0.10, max_risk_per_trade_pct=0.02):
self.max_position_pct = max_position_size_pct
self.max_risk_pct = max_risk_per_trade_pct
def validate(self, trade, portfolio_value, current_positions):
# Check 1: maximum position size
position_value = trade.qty * trade.price
if position_value / portfolio_value > self.max_position_pct:
return False, "Position size exceeds limit"
# Check 2: risk per trade
trade_risk = abs(trade.price - trade.stop_loss) * trade.qty
if trade_risk / portfolio_value > self.max_risk_pct:
return False, "Risk per trade exceeds limit"
# Check 3: correlation with current portfolio
portfolio_corr = self.calculate_portfolio_correlation(
trade.symbol, current_positions
)
if portfolio_corr > 0.8:
return False, "Too correlated with existing positions"
return True, "OK"
Portfolio level — control of aggregate portfolio:
- Maximum number of open positions (e.g., 10)
- Maximum aggregate exposure (e.g., 80% of capital)
- Maximum Net Delta (sum of all position deltas)
- Sector concentration: no more than 30% in one sector (DeFi, Layer-1, etc.)
Session level — daily/weekly limits:
- Maximum daily loss: 3% of deposit → automatic trading halt
- Maximum weekly drawdown: 8% → requires manual intervention
- Maximum trades per day: 20 → protection against overtrading
Portfolio Risk Monitor Implementation
from dataclasses import dataclass
from typing import List, Dict
import numpy as np
@dataclass
class Position:
symbol: str
qty: float
avg_price: float
stop_loss: float
current_price: float
@property
def unrealized_pnl(self):
return (self.current_price - self.avg_price) * self.qty
@property
def position_value(self):
return self.current_price * self.qty
@property
def risk_amount(self):
return abs(self.current_price - self.stop_loss) * self.qty
class PortfolioRiskMonitor:
def __init__(self, initial_capital: float, config: dict):
self.initial_capital = initial_capital
self.peak_capital = initial_capital
self.config = config
self.positions: List[Position] = []
self.daily_pnl = 0
self.session_start_capital = initial_capital
def update_capital(self, current_capital: float):
self.peak_capital = max(self.peak_capital, current_capital)
self.daily_pnl = current_capital - self.session_start_capital
def get_current_drawdown(self, current_capital: float) -> float:
return (self.peak_capital - current_capital) / self.peak_capital
def check_circuit_breakers(self, current_capital: float) -> dict:
alerts = {}
# Drawdown check
dd = self.get_current_drawdown(current_capital)
if dd > self.config['max_drawdown']:
alerts['max_drawdown'] = f"CRITICAL: Drawdown {dd:.1%} exceeded limit"
# Daily loss check
daily_loss_pct = -self.daily_pnl / self.session_start_capital
if daily_loss_pct > self.config['max_daily_loss']:
alerts['daily_loss'] = f"HALT: Daily loss {daily_loss_pct:.1%} exceeded"
# Total risk exposure
total_risk = sum(p.risk_amount for p in self.positions)
risk_pct = total_risk / current_capital
if risk_pct > self.config['max_portfolio_risk']:
alerts['portfolio_risk'] = f"WARNING: Total risk {risk_pct:.1%}"
return alerts
def get_correlation_matrix(self, price_data: Dict[str, list]) -> np.ndarray:
symbols = list(price_data.keys())
returns = {s: np.diff(np.log(price_data[s])) for s in symbols}
return np.corrcoef([returns[s] for s in symbols])
Stress Testing and Scenario Analysis
Historical scenarios: how would the portfolio perform under:
- March 2020 crash (BTC -60% per week)
- May 2021 crash (BTC -50% per month)
- Terra/LUNA collapse (>99% in 48h)
- FTX bankruptcy (market panic -25%)
Hypothetical scenarios:
- All assets correlate 0.9 (as in crisis)
- Liquidity disappears (bid-ask spread ×10)
- Simultaneous stop-loss triggering
def stress_test_portfolio(positions, scenarios):
results = {}
for scenario_name, price_shocks in scenarios.items():
total_pnl = 0
for position in positions:
shock = price_shocks.get(position.symbol, price_shocks.get('DEFAULT', 0))
pnl = position.qty * position.current_price * shock
total_pnl += pnl
results[scenario_name] = total_pnl
return results
VAR (Value at Risk) Integration
Historical VAR: with current portfolio, at 95% probability, maximum one-day loss will not exceed X USD. Calculated based on portfolio historical returns over past 252 days.
Real-time Monitoring
Dashboard (Grafana + Prometheus):
- Current P&L and drawdown
- All open positions with risk of each
- Aggregate portfolio risk
- Circuit breaker status
- Daily loss progress bar
Alerts (Telegram Bot with priority levels):
- Yellow: drawdown > 50% of limit
- Orange: drawdown > 75% of limit
- Red: circuit breaker triggered, trading halted
Audit log: every risk system decision (order block, circuit breaker trigger) is logged with timestamp and reason.
Integration with Trading Strategies
Risk Management System is an independent component embedded before execution of any order. All trading bots and strategies send orders through a single risk gateway:
Trading Strategy → Risk Gateway → Exchange API
Risk Gateway checks all limits, updates internal state, logs. On block, returns error with reason.
We develop a full risk management system: trade-level checks, portfolio monitoring, circuit breakers, stress testing module, realtime dashboard and Telegram alerts.