REAL-TIME RISK ENGINES: ARCHITECTURE & SCALING FOR TRADING SYSTEMS

class PreTradeRiskEngine:
    def __init__(self,
                 position_store: PositionStore,
                 limit_store: LimitStore,
                 circuit_breaker: CircuitBreaker):
        self.position_store = position_store
        self.limit_store = limit_store
        self.circuit_breaker = circuit_breaker
        self.checks = [
            PositionLimitCheck(position_store, limit_store),
            ExposureLimitCheck(position_store, limit_store),
            VelocityCheck(position_store, limit_store),
            CounterpartyLimitCheck(position_store, limit_store),
            ShortSaleCheck(position_store, limit_store),
            ClientLimitCheck(position_store, limit_store)
        ]

    async def evaluate_order(self, order: Order) -> RiskDecision:
        """Run all pre-trade risk checks in parallel."""

        # Start timer
        start_time = time.perf_counter()

        # Run all checks in parallel
        results = await asyncio.gather(
            *[check.evaluate(order) for check in self.checks],
            return_exceptions=True
        )

        # Collect results
        decision = RiskDecision(
            order_id=order.id,
            approved=True,
            reasons=[],
            latency_ns=(time.perf_counter() - start_time) * 1e9
        )

        for result in results:
            if isinstance(result, Exception):
                # Check failed with exception
                decision.approved = False
                decision.reasons.append(f"Check error: {str(result)}")

            elif not result.approved:
                # Check failed with reason
                decision.approved = False
                decision.reasons.extend(result.reasons)

        # Log decision
        await self.audit_log.log_decision(decision)

        # Update metrics
        self.metrics.record_evaluation(decision)

        return decision

class PositionLimitCheck:
    def __init__(self, position_store: PositionStore, limit_store: LimitStore):
        self.position_store = position_store
        self.limit_store = limit_store

    async def evaluate(self, order: Order) -> CheckResult:
        """Check if order would breach position limits."""

        # Get current position
        current_position = await self.position_store.get_position(
            account_id=order.account_id,
            symbol=order.symbol
        )

        # Calculate new position
        if order.side == 'BUY':
            new_position = current_position + order.quantity
        else:
            new_position = current_position - order.quantity

        # Get position limit
        limit = await self.limit_store.get_position_limit(
            account_id=order.account_id,
            symbol=order.symbol
        )

        # Check limit
        if abs(new_position) > limit.max_position:
            return CheckResult(
                approved=False,
                reasons=[
                    f"Position limit breach: {symbol} @ {abs(new_position)}/{limit.max_position}"
                ]
            )

        # Check long/short limits
        if new_position > 0 and new_position > limit.max_long:
            return CheckResult(
                approved=False,
                reasons=[
                    f"Long position limit breach: {new_position}/{limit.max_long}"
                ]
            )

        if new_position < 0 and abs(new_position) > limit.max_short:
            return CheckResult(
                approved=False,
                reasons=[
                    f"Short position limit breach: {abs(new_position)}/{limit.max_short}"
                ]
            )

        return CheckResult(approved=True)

class VelocityCheck:
    def __init__(self, position_store: PositionStore, limit_store: LimitStore):
        self.position_store = position_store
        self.limit_store = limit_store

    async def evaluate(self, order: Order) -> CheckResult:
        """Check velocity limits (order rate, value per time window)."""

        account_id = order.account_id
        now = time.time()

        # Get velocity limit
        limit = await self.limit_store.get_velocity_limit(account_id)

        # Get recent orders (sliding window)
        window_start = now - limit.window_seconds
        recent_orders = await self.position_store.get_orders_in_window(
            account_id=account_id,
            start_time=window_start
        )

        # Check order count
        if len(recent_orders) >= limit.max_orders:
            return CheckResult(
                approved=False,
                reasons=[
                    f"Velocity limit: {len(recent_orders)}/{limit.max_orders} orders in {limit.window_seconds}s"
                ]
            )

        # Check order value
        recent_value = sum(o.quantity * o.price for o in recent_orders)
        order_value = order.quantity * order.price

        if recent_value + order_value > limit.max_value:
            return CheckResult(
                approved=False,
                reasons=[
                    f"Value limit: ${(recent_value + order_value):,.2f}/${limit.max_value:,.2f}"
                ]
            )

        return CheckResult(approved=True)

class PositionManager:
    def __init__(self, redis_client: Redis, db: Database):
        self.redis = redis_client
        self.db = db

    async def update_position(self, trade: Trade):
        """Update position after trade execution."""

        key = f"position:{trade.account_id}:{trade.symbol}"

        # Get current position from Redis
        current = await self.redis.get(key)

        if current:
            position = json.loads(current)
        else:
            # Load from database on cache miss
            position = await self.db.get_position(trade.account_id, trade.symbol)

        # Calculate new position
        if trade.side == 'BUY':
            position['quantity'] += trade.quantity
            position['avg_price'] = (
                (position['quantity'] * position['avg_price'] +
                 trade.quantity * trade.price) /
                (position['quantity'] + trade.quantity)
            )
        else:
            position['quantity'] -= trade.quantity

        position['last_updated'] = time.time()

        # Update Redis (async write-through)
        await self.redis.set(key, json.dumps(position))

        # Persist to database (fire-and-forget)
        asyncio.create_task(self.db.save_position(trade.account_id, trade.symbol, position))

        # Publish position update
        await self.message_bus.publish(
            topic='position_updates',
            message=PositionUpdate(
                account_id=trade.account_id,
                symbol=trade.symbol,
                quantity=position['quantity'],
                avg_price=position['avg_price'],
                trade_id=trade.id
            )
        )

class PositionAggregator:
    def __init__(self, position_store: PositionStore):
        self.position_store = position_store

    async def get_aggregated_position(self, account_group: List[str], symbol: str) -> AggregatedPosition:
        """Aggregate positions across multiple accounts."""

        positions = await asyncio.gather(
            *[self.position_store.get_position(account_id, symbol) for account_id in account_group]
        )

        total_long = sum(p.quantity for p in positions if p.quantity > 0)
        total_short = sum(abs(p.quantity) for p in positions if p.quantity < 0)
        net_position = total_long - total_short

        return AggregatedPosition(
            symbol=symbol,
            total_long=total_long,
            total_short=total_short,
            net_position=net_position,
            account_count=len(positions)
        )

class CircuitBreaker:
    def __init__(self, redis_client: Redis, alert_manager: AlertManager):
        self.redis = redis_client
        self.alert_manager = alert_manager

    async def check_and_halt(self, breach: LimitBreach):
        """Check if breach requires trading halt."""

        # Determine halt scope
        if breach.breach_type == 'ACCOUNT_LIMIT':
            # Halt trading for this account
            await self.halt_account(breach.account_id, breach.reason)

        elif breach.breach_type == 'SYMBOL_LIMIT':
            # Halt trading for this symbol across all accounts
            await self.halt_symbol(breach.symbol, breach.reason)

        elif breach.breach_type == 'EXPOSURE_LIMIT':
            # Halt all trading
            await self.halt_all(breach.reason)

        # Alert
        await self.alert_manager.critical(
            f"Trading halt: {breach.breach_type} - {breach.reason}"
        )

    async def halt_account(self, account_id: str, reason: str):
        """Halt trading for specific account."""

        halt_key = f"halt:account:{account_id}"
        await self.redis.setex(
            halt_key,
            300,  # 5-minute halt
            json.dumps({
                'halted_at': time.time(),
                'reason': reason,
                'halted_by': 'risk_engine'
            })
        )

    async def halt_symbol(self, symbol: str, reason: str):
        """Halt trading for specific symbol."""

        halt_key = f"halt:symbol:{symbol}"
        await self.redis.setex(
            halt_key,
            300,  # 5-minute halt
            json.dumps({
                'halted_at': time.time(),
                'reason': reason,
                'halted_by': 'risk_engine'
            })
        )

    async def is_halted(self, account_id: str, symbol: str) -> bool:
        """Check if trading is halted."""

        # Check account halt
        account_halt = await self.redis.get(f"halt:account:{account_id}")
        if account_halt:
            return True

        # Check symbol halt
        symbol_halt = await self.redis.get(f"halt:symbol:{symbol}")
        if symbol_halt:
            return True

        return False

class RiskEngineCluster:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards
        self.shards = []

    def get_shard_for_account(self, account_id: str) -> int:
        """Determine which shard handles this account."""

        # Consistent hashing
        hash_value = hash(account_id)
        return hash_value % self.num_shards

    async def route_order(self, order: Order):
        """Route order to appropriate shard."""

        shard_id = self.get_shard_for_account(order.account_id)

        # Send to shard's message queue
        await self.message_bus.send(
            topic=f"risk_shard_{shard_id}",
            message=order
        )

class OptimizedRiskEngine:
    """Single-node optimizations for maximum throughput."""

    def __init__(self):
        # Use memory-mapped data structures
        self.positions = mmap.mmap(-1, 1024*1024*1024)  # 1GB shared memory

        # Lock-free data structures
        self.position_locks = [threading.Lock() for _ in range(256)]

        # Pre-allocated object pools
        self.order_pool = ObjectPool(Order, size=10000)
        self.result_pool = ObjectPool(RiskDecision, size=10000)

    def get_position_lock(self, account_id: str) -> threading.Lock:
        """Get lock for this account (striping)."""
        lock_index = hash(account_id) % len(self.position_locks)
        return self.position_locks[lock_index]

class FastRiskEngine:
    def __init__(self):
        # Pre-load limits into memory
        self.limits = {}
        self.positions = {}  # Dict for O(1) lookup

        # Compile validation functions
        self.check_functions = self.compile_checks()

    def compile_checks(self):
        """JIT-compile hot paths."""

        @numba.jit(nopython=True)
        def check_position_limit(position: float, limit: float) -> bool:
            return abs(position) <= limit

        @numba.jit(nopython=True)
        def check_exposure(positions: np.ndarray, prices: np.ndarray, limit: float) -> bool:
            exposure = np.sum(positions * prices)
            return abs(exposure) <= limit

        return {
            'position_limit': check_position_limit,
            'exposure': check_exposure
        }

    async def evaluate_order_fast(self, order: Order) -> RiskDecision:
        """Optimized evaluation path (hot code)."""

        # Inline position lookup (no async overhead)
        key = (order.account_id, order.symbol)
        position = self.positions.get(key, 0.0)

        # Calculate new position
        new_position = position + order.signed_quantity

        # Inline limit check (compiled function)
        limit = self.limits.get(f"position:{order.account_id}:{order.symbol}")

        if not self.check_functions['position_limit'](new_position, limit):
            return RiskDecision(approved=False, reasons=["Position limit"])

        return RiskDecision(approved=True)

class BatchRiskEngine:
    async def evaluate_batch(self, orders: List[Order]) -> List[RiskDecision]:
        """Evaluate multiple orders in batch (vectorized)."""

        # Group orders by account
        orders_by_account = defaultdict(list)
        for order in orders:
            orders_by_account[order.account_id].append(order)

        # Process accounts in parallel
        tasks = [
            self.evaluate_account_batch(account_id, account_orders)
            for account_id, account_orders in orders_by_account.items()
        ]

        results = await asyncio.gather(*tasks)

        # Flatten results
        return [decision for account_results in results for decision in account_results]

    async def evaluate_account_batch(self, account_id: str, orders: List[Order]) -> List[RiskDecision]:
        """Evaluate all orders for an account (can share position lookups)."""

        # Load position once
        positions = await self.position_store.get_all_positions(account_id)

        # Evaluate each order
        decisions = []
        for order in orders:
            # Use shared position data (no repeated lookups)
            decision = await self.evaluate_with_positions(order, positions)
            decisions.append(decision)

        return decisions

class RiskEngineFailover:
    async def handle_engine_failure(self, failed_shard: int):
        """Handle failure of a risk engine shard."""

        # 1. Detect failure (heartbeat timeout)
        # 2. Mark shard as down
        await self.service_registry.mark_down(f"risk_shard_{failed_shard}")

        # 3. Route to backup shard
        backup_shard = self.get_backup_shard(failed_shard)

        # 4. Replay in-flight messages from backup
        await self.replay_messages(backup_shard, failed_shard)

        # 5. Alert operations
        await self.alert_manager.critical(
            f"Risk engine shard {failed_shard} failed, failing over to {backup_shard}"
        )

class PositionDataMonitor:
    async def detect_stale_positions(self):
        """Monitor for stale position data."""

        while True:
            # Check all accounts
            accounts = await self.position_store.get_all_accounts()

            for account_id in accounts:
                # Get last update time
                last_update = await self.position_store.get_last_update_time(account_id)

                # Check staleness threshold
                if time.time() - last_update > 60:  # 60 seconds
                    await self.alert_manager.warning(
                        f"Stale position data: {account_id} (last update {last_update})"
                    )

                    # Mark account for reconciliation
                    await self.reconciliation_queue.enqueue(account_id)

            await asyncio.sleep(10)  # Check every 10 seconds

class RiskEngineMetrics:
    def __init__(self):
        self.check_latency = Histogram(
            'risk_check_latency_microseconds',
            'Risk check latency',
            ['check_type', 'shard']
        )

        self.throughput = Gauge(
            'risk_checks_per_second',
            'Risk check throughput',
            ['shard']
        )

        self.reject_rate = Gauge(
            'risk_check_reject_rate',
            'Percentage of checks rejected',
            ['reason']
        )

        self.position_update_lag = Gauge(
            'position_update_lag_seconds',
            'Time between trade and position update',
            ['account']
        )

        self.circuit_breaker_trips = Counter(
            'circuit_breaker_trips_total',
            'Circuit breaker activations',
            ['scope', 'reason']
        )

class RiskEngineLoadTest:
    async def test_throughput(self, orders_per_second: int, duration_seconds: int):
        """Test risk engine throughput."""

        engine = PreTradeRiskEngine(...)

        # Generate test orders
        orders = self.generate_test_orders(count=orders_per_second * duration_seconds)

        # Start timer
        start_time = time.time()

        # Process orders
        tasks = [engine.evaluate_order(order) for order in orders]
        results = await asyncio.gather(*tasks)

        # Calculate metrics
        end_time = time.time()
        duration = end_time - start_time
        actual_throughput = len(orders) / duration

        # Check latency requirements
        latencies = [r.latency_ns for r in results]
        p99_latency = np.percentile(latencies, 99)

        assert p99_latency < 500_000, f"P99 latency {p99_latency}ns exceeds 500μs"
        assert actual_throughput >= orders_per_second, f"Throughput {actual_throughput} < target {orders_per_second}"