Building a Compliance Framework for Algo Trading

SEBI’s 2024 circulars mandate strict compliance for algo traders: registration, testing, kill switches, audit trails. Non-compliance brings penalties up to ₹25 crore.

This guide helps you build a complete compliance framework from scratch, ensuring you trade legally and sleep peacefully.

Compliance Architecture

Core Components

1. Pre-Trade Controls: Risk checks before order placement
2. Order Audit Trail: Complete order lifecycle logging
3. Kill Switch: Emergency stop mechanism
4. Position Limits: Automated position size controls
5. Reporting: Daily compliance reports for internal audit

import logging
from datetime import datetime
import json

class ComplianceFramework:
    """
    Complete compliance framework for algo trading
    """
    
    def __init__(self, config: dict):
        self.config = config
        self.setup_logging()
        self.kill_switch_active = False
        
    def setup_logging(self):
        """Setup audit trail logging"""
        logging.basicConfig(
            filename=f"audit_{datetime.now().strftime('%Y%m%d')}.log",
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    
    def pre_trade_checks(self, order: dict) -> dict:
        """Pre-trade risk checks"""
        checks = {
            'kill_switch': self.check_kill_switch(),
            'position_limit': self.check_position_limit(order),
            'order_value': self.check_order_value(order),
            'price_collar': self.check_price_collar(order),
            'daily_loss': self.check_daily_loss_limit()
        }
        
        # Log all checks
        self.logger.info(f"Pre-trade checks for {order}: {checks}")
        
        # Block order if any check fails
        passed = all(checks.values())
        
        return {
            'passed': passed,
            'checks': checks,
            'reason': None if passed else self.get_failure_reason(checks)
        }
    
    def check_kill_switch(self) -> bool:
        """Check if kill switch is active"""
        return not self.kill_switch_active
    
    def check_position_limit(self, order: dict) -> bool:
        """Check position limits"""
        current_position = self.get_current_position(order['symbol'])
        new_position = current_position + (order['quantity'] if order['side'] == 'buy' else -order['quantity'])
        
        max_position = self.config['position_limits'].get(order['symbol'], 1000)
        
        return abs(new_position) <= max_position
    
    def check_order_value(self, order: dict) -> bool:
        """Check single order value limit"""
        order_value = order['quantity'] * order.get('price', 0)
        max_order_value = self.config['max_order_value']
        
        return order_value <= max_order_value
    
    def audit_order(self, order: dict, status: str, reason: str = None):
        """Log order to audit trail"""
        audit_entry = {
            'timestamp': datetime.now().isoformat(),
            'order_id': order.get('order_id'),
            'symbol': order['symbol'],
            'side': order['side'],
            'quantity': order['quantity'],
            'price': order.get('price'),
            'order_type': order['order_type'],
            'status': status,
            'reason': reason
        }
        
        self.logger.info(f"AUDIT: {json.dumps(audit_entry)}")
    
    def activate_kill_switch(self, reason: str):
        """Activate emergency kill switch"""
        self.kill_switch_active = True
        self.logger.critical(f"KILL SWITCH ACTIVATED: {reason}")
        
        # Cancel all pending orders
        self.cancel_all_orders()
        
        # Close all positions
        self.emergency_position_closure()
    
    def generate_daily_report(self) -> dict:
        """Generate compliance report"""
        return {
            'date': datetime.now().strftime('%Y-%m-%d'),
            'total_orders': self.count_orders(),
            'rejected_orders': self.count_rejected_orders(),
            'kill_switch_activations': self.count_kill_switch_events(),
            'compliance_violations': self.check_violations()
        }

# Usage
config = {
    'position_limits': {'NIFTY': 1000, 'BANKNIFTY': 500},
    'max_order_value': 1000000,
    'max_daily_loss': 50000
}

framework = ComplianceFramework(config)

Building compliance isn’t optional—it’s survival. Regulators watching closely. Implement these controls today, trade confidently tomorrow.