Risk Management Frameworks for Algorithmic Trading

Risk management isn’t optional—it’s what separates surviving traders from bankrupt ones. A single unmanaged risk event can wipe out months of profits. This guide implements production-grade risk frameworks for Indian algorithmic trading.

Core Risk Management Principles

The Three Pillars

1. Position Limits: Never risk too much on single trade
2. Stop Losses: Cut losses before they spiral
3. Portfolio Limits: Control aggregate exposure

import pandas as pd
import numpy as np
from typing import Dict, List
from datetime import datetime

class RiskManager:
    """
    Comprehensive risk management system
    """
    
    def __init__(self, account_size: float, config: Dict):
        self.account_size = account_size
        self.config = config
        self.positions = {}
        self.daily_pnl = 0
        self.peak_value = account_size
        
    def check_position_limit(self, symbol: str, quantity: int, price: float) -> Dict:
        """Check if trade violates position limits"""
        position_value = quantity * price
        position_pct = position_value / self.account_size
        
        max_position_pct = self.config.get('max_position_pct', 0.10)  # 10% default
        
        if position_pct > max_position_pct:
            return {
                'allowed': False,
                'reason': f'Position size {position_pct:.1%} exceeds limit {max_position_pct:.1%}',
                'max_quantity': int(self.account_size * max_position_pct / price)
            }
        
        return {'allowed': True, 'reason': 'Within position limits'}
    
    def check_daily_loss_limit(self) -> Dict:
        """Check if daily loss limit reached"""
        max_daily_loss = self.config.get('max_daily_loss', 0.02) * self.account_size
        
        if self.daily_pnl < -max_daily_loss:
            return {
                'trading_allowed': False,
                'reason': f'Daily loss limit reached: ₹{abs(self.daily_pnl):,.0f}'
            }
        
        return {'trading_allowed': True}
    
    def check_drawdown_limit(self, current_value: float) -> Dict:
        """Check if drawdown limit reached"""
        if current_value > self.peak_value:
            self.peak_value = current_value
        
        drawdown = (self.peak_value - current_value) / self.peak_value
        max_drawdown = self.config.get('max_drawdown', 0.15)  # 15% default
        
        if drawdown > max_drawdown:
            return {
                'trading_allowed': False,
                'reason': f'Drawdown {drawdown:.1%} exceeds limit {max_drawdown:.1%}'
            }
        
        return {'trading_allowed': True, 'current_drawdown': drawdown}
    
    def calculate_position_size(
        self,
        signal_strength: float,
        volatility: float,
        account_value: float
    ) -> int:
        """
        Kelly Criterion-based position sizing
        
        Position Size = (Edge / Volatility) * Capital
        """
        # Risk per trade
        risk_per_trade = self.config.get('risk_per_trade', 0.02)  # 2% default
        
        # Adjust for signal strength (0-1 scale)
        adjusted_risk = risk_per_trade * signal_strength
        
        # Kelly fraction (fractional Kelly for safety)
        kelly_fraction = 0.25  # Use 25% of full Kelly
        
        # Position size
        position_size = (adjusted_risk / volatility) * account_value * kelly_fraction
        
        return max(1, int(position_size))

# Example usage
risk_config = {
    'max_position_pct': 0.10,      # 10% per position
    'max_daily_loss': 0.02,         # 2% daily loss limit
    'max_drawdown': 0.15,           # 15% max drawdown
    'risk_per_trade': 0.01          # 1% risk per trade
}

risk_mgr = RiskManager(account_size=1000000, config=risk_config)

# Check position limit
check = risk_mgr.check_position_limit('RELIANCE', 100, 2500)
print(f"Position Check: {check}")

# Calculate position size
position_size = risk_mgr.calculate_position_size(
    signal_strength=0.8,
    volatility=0.02,
    account_value=1000000
)
print(f"Recommended Position Size: {position_size}")