57. Real-Time Trade Log Streaming and Analytics System¶

services/trade-log-stream-center/
├── src/
│   ├── main.py
│   ├── collector/
│   │   ├── event_collector.py
│   │   ├── trade_event_collector.py
│   │   └── order_event_collector.py
│   ├── processor/
│   │   ├── event_cleaner.py
│   │   ├── event_enricher.py
│   │   ├── anomaly_marker.py
│   │   └── stream_processor.py
│   ├── storage/
│   │   ├── log_storage.py
│   │   ├── kafka_producer.py
│   │   └── timescaledb_writer.py
│   ├── analyzer/
│   │   ├── real_time_analyzer.py
│   │   ├── quality_metrics.py
│   │   ├── latency_analyzer.py
│   │   └── slippage_analyzer.py
│   ├── api/
│   │   ├── log_stream_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class EventCollector:
    def __init__(self, trade_event_collector, order_event_collector):
        self.trade_event_collector = trade_event_collector
        self.order_event_collector = order_event_collector
        self.event_stream = []
        self.collection_stats = {
            "total_events": 0,
            "trade_events": 0,
            "order_events": 0,
            "last_collection": None
        }

    async def start_collection(self):
        """Start collecting all trading events"""

        # Start trade event collection
        await self.start_trade_collection()

        # Start order event collection
        await self.start_order_collection()

        # Start system event collection
        await self.start_system_collection()

    async def start_trade_collection(self):
        """Start collecting trade execution events"""

        trade_collector = TradeEventCollector()
        await trade_collector.start_monitoring()

        async for trade_event in trade_collector.get_trade_events():
            await self.process_trade_event(trade_event)

    async def start_order_collection(self):
        """Start collecting order lifecycle events"""

        order_collector = OrderEventCollector()
        await order_collector.start_monitoring()

        async for order_event in order_collector.get_order_events():
            await self.process_order_event(order_event)

    async def start_system_collection(self):
        """Start collecting system-level events"""

        system_collector = SystemEventCollector()
        await system_collector.start_monitoring()

        async for system_event in system_collector.get_system_events():
            await self.process_system_event(system_event)

    async def process_trade_event(self, trade_event):
        """Process a trade execution event"""

        event_record = {
            "timestamp": datetime.now().isoformat(),
            "event_type": "trade_execution",
            "event_id": trade_event.get("trade_id"),
            "order_id": trade_event.get("order_id"),
            "symbol": trade_event.get("symbol"),
            "side": trade_event.get("side"),
            "quantity": trade_event.get("quantity"),
            "price": trade_event.get("price"),
            "amount": trade_event.get("amount"),
            "account_id": trade_event.get("account_id"),
            "strategy_id": trade_event.get("strategy_id"),
            "venue": trade_event.get("venue"),
            "counterparty": trade_event.get("counterparty"),
            "execution_time": trade_event.get("execution_time"),
            "raw_data": trade_event
        }

        # Add to event stream
        await self.add_to_event_stream(event_record)

        # Update statistics
        self.collection_stats["trade_events"] += 1
        self.collection_stats["total_events"] += 1

    async def process_order_event(self, order_event):
        """Process an order lifecycle event"""

        event_record = {
            "timestamp": datetime.now().isoformat(),
            "event_type": "order_lifecycle",
            "event_id": order_event.get("event_id"),
            "order_id": order_event.get("order_id"),
            "symbol": order_event.get("symbol"),
            "side": order_event.get("side"),
            "order_type": order_event.get("order_type"),
            "quantity": order_event.get("quantity"),
            "price": order_event.get("price"),
            "status": order_event.get("status"),
            "action": order_event.get("action"),  # new, modify, cancel, fill
            "account_id": order_event.get("account_id"),
            "strategy_id": order_event.get("strategy_id"),
            "venue": order_event.get("venue"),
            "raw_data": order_event
        }

        # Add to event stream
        await self.add_to_event_stream(event_record)

        # Update statistics
        self.collection_stats["order_events"] += 1
        self.collection_stats["total_events"] += 1

    async def process_system_event(self, system_event):
        """Process a system-level event"""

        event_record = {
            "timestamp": datetime.now().isoformat(),
            "event_type": "system_event",
            "event_id": system_event.get("event_id"),
            "event_category": system_event.get("category"),
            "severity": system_event.get("severity"),
            "message": system_event.get("message"),
            "component": system_event.get("component"),
            "raw_data": system_event
        }

        # Add to event stream
        await self.add_to_event_stream(event_record)

    async def add_to_event_stream(self, event_record):
        """Add event to processing stream"""

        # Add to in-memory stream
        self.event_stream.append(event_record)

        # Keep stream manageable
        if len(self.event_stream) > 10000:
            self.event_stream = self.event_stream[-10000:]

        # Update last collection time
        self.collection_stats["last_collection"] = datetime.now().isoformat()

        # Trigger stream processing
        await self.trigger_stream_processing(event_record)

    async def trigger_stream_processing(self, event_record):
        """Trigger stream processing pipeline"""

        # Send to stream processor
        await self.stream_processor.process_event(event_record)

    async def get_recent_events(self, minutes: int = 60) -> List[Dict]:
        """Get recent events from stream"""

        cutoff_time = datetime.now() - timedelta(minutes=minutes)

        recent_events = [
            event for event in self.event_stream
            if datetime.fromisoformat(event["timestamp"]) > cutoff_time
        ]

        return recent_events

    async def get_collection_statistics(self) -> Dict:
        """Get collection statistics"""

        return {
            **self.collection_stats,
            "stream_size": len(self.event_stream),
            "collection_active": True
        }

class EventCleaner:
    def __init__(self):
        self.cleaning_rules = self.load_cleaning_rules()
        self.field_mappings = self.load_field_mappings()

    async def clean_event(self, raw_event: Dict) -> Dict:
        """Clean and standardize event data"""

        # Extract basic fields
        cleaned_event = {
            "timestamp": self.standardize_timestamp(raw_event["timestamp"]),
            "event_type": self.standardize_event_type(raw_event["event_type"]),
            "event_id": self.clean_event_id(raw_event.get("event_id")),
            "order_id": self.clean_order_id(raw_event.get("order_id")),
            "symbol": self.standardize_symbol(raw_event.get("symbol")),
            "side": self.standardize_side(raw_event.get("side")),
            "quantity": self.clean_numeric_field(raw_event.get("quantity")),
            "price": self.clean_numeric_field(raw_event.get("price")),
            "amount": self.clean_numeric_field(raw_event.get("amount")),
            "account_id": self.clean_account_id(raw_event.get("account_id")),
            "strategy_id": self.clean_strategy_id(raw_event.get("strategy_id")),
            "venue": self.standardize_venue(raw_event.get("venue")),
            "status": self.standardize_status(raw_event.get("status")),
            "action": self.standardize_action(raw_event.get("action")),
            "order_type": self.standardize_order_type(raw_event.get("order_type")),
            "cleaned_at": datetime.now().isoformat()
        }

        # Apply cleaning rules
        cleaned_event = await self.apply_cleaning_rules(cleaned_event)

        # Validate cleaned event
        if await self.validate_cleaned_event(cleaned_event):
            return cleaned_event
        else:
            # Mark as invalid
            cleaned_event["valid"] = False
            cleaned_event["validation_errors"] = self.get_validation_errors(cleaned_event)
            return cleaned_event

    def standardize_timestamp(self, timestamp) -> str:
        """Standardize timestamp format"""

        if isinstance(timestamp, str):
            try:
                # Parse and standardize
                dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
                return dt.isoformat()
            except:
                return datetime.now().isoformat()
        else:
            return datetime.now().isoformat()

    def standardize_event_type(self, event_type: str) -> str:
        """Standardize event type"""

        event_type_mapping = {
            "trade": "trade_execution",
            "execution": "trade_execution",
            "fill": "trade_execution",
            "order": "order_lifecycle",
            "order_event": "order_lifecycle",
            "system": "system_event",
            "error": "system_event"
        }

        return event_type_mapping.get(event_type.lower(), event_type.lower())

    def clean_numeric_field(self, value) -> float:
        """Clean numeric fields"""

        if value is None:
            return 0.0

        try:
            return float(value)
        except (ValueError, TypeError):
            return 0.0

    def standardize_symbol(self, symbol: str) -> str:
        """Standardize symbol format"""

        if symbol:
            return symbol.upper().strip()
        else:
            return ""

    def standardize_side(self, side: str) -> str:
        """Standardize side field"""

        if side:
            side_lower = side.lower()
            if side_lower in ["buy", "b", "long"]:
                return "buy"
            elif side_lower in ["sell", "s", "short"]:
                return "sell"

        return ""

    async def apply_cleaning_rules(self, cleaned_event: Dict) -> Dict:
        """Apply specific cleaning rules"""

        # Remove negative quantities
        if cleaned_event["quantity"] < 0:
            cleaned_event["quantity"] = abs(cleaned_event["quantity"])

        # Remove negative prices
        if cleaned_event["price"] < 0:
            cleaned_event["price"] = 0.0

        # Ensure amount is calculated if missing
        if cleaned_event["amount"] == 0 and cleaned_event["quantity"] > 0 and cleaned_event["price"] > 0:
            cleaned_event["amount"] = cleaned_event["quantity"] * cleaned_event["price"]

        return cleaned_event

    async def validate_cleaned_event(self, cleaned_event: Dict) -> bool:
        """Validate cleaned event"""

        # Check required fields
        required_fields = ["timestamp", "event_type", "event_id"]
        for field in required_fields:
            if not cleaned_event.get(field):
                return False

        # Check numeric fields
        numeric_fields = ["quantity", "price", "amount"]
        for field in numeric_fields:
            if cleaned_event.get(field) is not None and cleaned_event[field] < 0:
                return False

        return True

    def get_validation_errors(self, cleaned_event: Dict) -> List[str]:
        """Get validation errors"""

        errors = []

        if not cleaned_event.get("timestamp"):
            errors.append("Missing timestamp")

        if not cleaned_event.get("event_type"):
            errors.append("Missing event type")

        if cleaned_event.get("quantity", 0) < 0:
            errors.append("Negative quantity")

        if cleaned_event.get("price", 0) < 0:
            errors.append("Negative price")

        return errors

    def load_cleaning_rules(self) -> Dict:
        """Load cleaning rules"""

        return {
            "remove_negative_values": True,
            "standardize_timestamps": True,
            "validate_required_fields": True,
            "auto_calculate_amount": True
        }

    def load_field_mappings(self) -> Dict:
        """Load field mappings"""

        return {
            "event_types": {
                "trade": "trade_execution",
                "execution": "trade_execution",
                "order": "order_lifecycle"
            },
            "sides": {
                "b": "buy",
                "s": "sell",
                "long": "buy",
                "short": "sell"
            }
        }

class EventEnricher:
    def __init__(self, latency_calculator, slippage_calculator):
        self.latency_calculator = latency_calculator
        self.slippage_calculator = slippage_calculator
        self.enrichment_cache = {}

    async def enrich_event(self, cleaned_event: Dict) -> Dict:
        """Enrich event with derived metrics"""

        enriched_event = cleaned_event.copy()

        # Calculate latency for trade executions
        if enriched_event["event_type"] == "trade_execution":
            latency = await self.calculate_execution_latency(enriched_event)
            enriched_event["execution_latency_ms"] = latency

        # Calculate slippage for trade executions
        if enriched_event["event_type"] == "trade_execution":
            slippage = await self.calculate_slippage(enriched_event)
            enriched_event["slippage_bps"] = slippage

        # Calculate fill rate for orders
        if enriched_event["event_type"] == "order_lifecycle":
            fill_rate = await self.calculate_fill_rate(enriched_event)
            enriched_event["fill_rate"] = fill_rate

        # Add market context
        market_context = await self.get_market_context(enriched_event)
        enriched_event["market_context"] = market_context

        # Add strategy context
        strategy_context = await self.get_strategy_context(enriched_event)
        enriched_event["strategy_context"] = strategy_context

        # Add venue context
        venue_context = await self.get_venue_context(enriched_event)
        enriched_event["venue_context"] = venue_context

        # Add enrichment timestamp
        enriched_event["enriched_at"] = datetime.now().isoformat()

        return enriched_event

    async def calculate_execution_latency(self, event: Dict) -> float:
        """Calculate execution latency in milliseconds"""

        order_id = event.get("order_id")
        if not order_id:
            return 0.0

        # Get order submission time
        order_submission_time = await self.get_order_submission_time(order_id)
        if not order_submission_time:
            return 0.0

        # Calculate latency
        execution_time = datetime.fromisoformat(event["timestamp"])
        submission_time = datetime.fromisoformat(order_submission_time)

        latency_ms = (execution_time - submission_time).total_seconds() * 1000

        return max(0.0, latency_ms)

    async def calculate_slippage(self, event: Dict) -> float:
        """Calculate slippage in basis points"""

        symbol = event.get("symbol")
        execution_price = event.get("price")

        if not symbol or not execution_price:
            return 0.0

        # Get expected price (mid price at order time)
        expected_price = await self.get_expected_price(symbol, event.get("order_id"))
        if not expected_price or expected_price == 0:
            return 0.0

        # Calculate slippage in basis points
        slippage_bps = ((execution_price - expected_price) / expected_price) * 10000

        return slippage_bps

    async def calculate_fill_rate(self, event: Dict) -> float:
        """Calculate fill rate for order"""

        order_id = event.get("order_id")
        if not order_id:
            return 0.0

        # Get order details
        order_details = await self.get_order_details(order_id)
        if not order_details:
            return 0.0

        original_quantity = order_details.get("original_quantity", 0)
        filled_quantity = order_details.get("filled_quantity", 0)

        if original_quantity == 0:
            return 0.0

        fill_rate = filled_quantity / original_quantity
        return min(1.0, max(0.0, fill_rate))

    async def get_market_context(self, event: Dict) -> Dict:
        """Get market context for event"""

        symbol = event.get("symbol")
        timestamp = event.get("timestamp")

        if not symbol or not timestamp:
            return {}

        # Get market data at event time
        market_data = await self.get_market_data_at_time(symbol, timestamp)

        return {
            "bid": market_data.get("bid", 0),
            "ask": market_data.get("ask", 0),
            "spread": market_data.get("spread", 0),
            "volume": market_data.get("volume", 0),
            "volatility": market_data.get("volatility", 0)
        }

    async def get_strategy_context(self, event: Dict) -> Dict:
        """Get strategy context for event"""

        strategy_id = event.get("strategy_id")
        if not strategy_id:
            return {}

        # Get strategy information
        strategy_info = await self.get_strategy_info(strategy_id)

        return {
            "strategy_name": strategy_info.get("name", ""),
            "strategy_type": strategy_info.get("type", ""),
            "risk_level": strategy_info.get("risk_level", ""),
            "target_latency": strategy_info.get("target_latency", 0)
        }

    async def get_venue_context(self, event: Dict) -> Dict:
        """Get venue context for event"""

        venue = event.get("venue")
        if not venue:
            return {}

        # Get venue information
        venue_info = await self.get_venue_info(venue)

        return {
            "venue_name": venue_info.get("name", ""),
            "venue_type": venue_info.get("type", ""),
            "average_latency": venue_info.get("average_latency", 0),
            "reliability": venue_info.get("reliability", 0)
        }

    async def get_order_submission_time(self, order_id: str) -> Optional[str]:
        """Get order submission time"""

        # This would query order database
        # For now, return placeholder
        return None

    async def get_expected_price(self, symbol: str, order_id: str) -> Optional[float]:
        """Get expected price at order time"""

        # This would query market data at order time
        # For now, return placeholder
        return None

    async def get_order_details(self, order_id: str) -> Optional[Dict]:
        """Get order details"""

        # This would query order database
        # For now, return placeholder
        return None

class AnomalyMarker:
    def __init__(self):
        self.anomaly_rules = self.load_anomaly_rules()
        self.anomaly_history = []

    async def mark_anomalies(self, enriched_event: Dict) -> Dict:
        """Mark anomalies in enriched event"""

        marked_event = enriched_event.copy()
        anomalies = []

        # Check latency anomalies
        latency_anomaly = await self.check_latency_anomaly(marked_event)
        if latency_anomaly:
            anomalies.append(latency_anomaly)

        # Check slippage anomalies
        slippage_anomaly = await self.check_slippage_anomaly(marked_event)
        if slippage_anomaly:
            anomalies.append(slippage_anomaly)

        # Check size anomalies
        size_anomaly = await self.check_size_anomaly(marked_event)
        if size_anomaly:
            anomalies.append(size_anomaly)

        # Check price anomalies
        price_anomaly = await self.check_price_anomaly(marked_event)
        if price_anomaly:
            anomalies.append(price_anomaly)

        # Check timing anomalies
        timing_anomaly = await self.check_timing_anomaly(marked_event)
        if timing_anomaly:
            anomalies.append(timing_anomaly)

        # Add anomaly information to event
        if anomalies:
            marked_event["anomalies"] = anomalies
            marked_event["has_anomalies"] = True
            marked_event["anomaly_count"] = len(anomalies)
            marked_event["anomaly_severity"] = self.calculate_anomaly_severity(anomalies)
        else:
            marked_event["has_anomalies"] = False
            marked_event["anomaly_count"] = 0
            marked_event["anomaly_severity"] = "none"

        # Add marking timestamp
        marked_event["anomaly_marked_at"] = datetime.now().isoformat()

        return marked_event

    async def check_latency_anomaly(self, event: Dict) -> Optional[Dict]:
        """Check for latency anomalies"""

        latency = event.get("execution_latency_ms", 0)
        threshold = self.anomaly_rules["latency_threshold_ms"]

        if latency > threshold:
            return {
                "type": "high_latency",
                "severity": "medium",
                "description": f"Execution latency {latency}ms exceeds threshold {threshold}ms",
                "value": latency,
                "threshold": threshold
            }

        return None

    async def check_slippage_anomaly(self, event: Dict) -> Optional[Dict]:
        """Check for slippage anomalies"""

        slippage = abs(event.get("slippage_bps", 0))
        threshold = self.anomaly_rules["slippage_threshold_bps"]

        if slippage > threshold:
            return {
                "type": "high_slippage",
                "severity": "high",
                "description": f"Slippage {slippage}bps exceeds threshold {threshold}bps",
                "value": slippage,
                "threshold": threshold
            }

        return None

    async def check_size_anomaly(self, event: Dict) -> Optional[Dict]:
        """Check for size anomalies"""

        quantity = event.get("quantity", 0)
        symbol = event.get("symbol", "")

        if symbol:
            avg_size = await self.get_average_trade_size(symbol)
            threshold = self.anomaly_rules["size_threshold_multiplier"]

            if avg_size > 0 and quantity > avg_size * threshold:
                return {
                    "type": "large_size",
                    "severity": "medium",
                    "description": f"Trade size {quantity} exceeds {threshold}x average ({avg_size})",
                    "value": quantity,
                    "threshold": avg_size * threshold
                }

        return None

    async def check_price_anomaly(self, event: Dict) -> Optional[Dict]:
        """Check for price anomalies"""

        price = event.get("price", 0)
        symbol = event.get("symbol", "")

        if symbol and price > 0:
            market_context = event.get("market_context", {})
            bid = market_context.get("bid", 0)
            ask = market_context.get("ask", 0)

            if bid > 0 and ask > 0:
                spread = ask - bid
                mid_price = (bid + ask) / 2
                threshold = self.anomaly_rules["price_deviation_threshold"]

                price_deviation = abs(price - mid_price) / mid_price

                if price_deviation > threshold:
                    return {
                        "type": "price_deviation",
                        "severity": "high",
                        "description": f"Price deviation {price_deviation:.2%} exceeds threshold {threshold:.2%}",
                        "value": price_deviation,
                        "threshold": threshold
                    }

        return None

    async def check_timing_anomaly(self, event: Dict) -> Optional[Dict]:
        """Check for timing anomalies"""

        timestamp = datetime.fromisoformat(event["timestamp"])

        # Check for after-hours trading
        if self.is_after_hours(timestamp):
            return {
                "type": "after_hours_trading",
                "severity": "low",
                "description": "Trading activity detected outside normal hours",
                "value": timestamp.strftime("%H:%M"),
                "threshold": "09:00-17:00"
            }

        # Check for weekend trading
        if self.is_weekend(timestamp):
            return {
                "type": "weekend_trading",
                "severity": "low",
                "description": "Trading activity detected on weekend",
                "value": timestamp.strftime("%A"),
                "threshold": "Monday-Friday"
            }

        return None

    def calculate_anomaly_severity(self, anomalies: List[Dict]) -> str:
        """Calculate overall anomaly severity"""

        if not anomalies:
            return "none"

        severities = [anomaly["severity"] for anomaly in anomalies]

        if "high" in severities:
            return "high"
        elif "medium" in severities:
            return "medium"
        else:
            return "low"

    def is_after_hours(self, timestamp: datetime) -> bool:
        """Check if timestamp is after hours"""

        hour = timestamp.hour
        return hour < 9 or hour > 17

    def is_weekend(self, timestamp: datetime) -> bool:
        """Check if timestamp is on weekend"""

        return timestamp.weekday() >= 5

    async def get_average_trade_size(self, symbol: str) -> float:
        """Get average trade size for symbol"""

        # This would query historical data
        # For now, return placeholder
        return 1000.0

    def load_anomaly_rules(self) -> Dict:
        """Load anomaly detection rules"""

        return {
            "latency_threshold_ms": 1000,  # 1 second
            "slippage_threshold_bps": 50,  # 50 basis points
            "size_threshold_multiplier": 10,  # 10x average
            "price_deviation_threshold": 0.01,  # 1%
            "after_hours_start": "17:00",
            "after_hours_end": "09:00"
        }

@router.get("/log_stream/recent")
async def get_recent_logs(minutes: int = 60):
    """Get recent log entries"""

    logs = await event_collector.get_recent_events(minutes)

    return {
        "logs": logs,
        "total_count": len(logs),
        "time_period_minutes": minutes,
        "timestamp": datetime.now().isoformat()
    }

@router.get("/log_stream/statistics")
async def get_log_statistics():
    """Get log stream statistics"""

    stats = await event_collector.get_collection_statistics()

    return {
        "collection_stats": stats,
        "processing_stats": await stream_processor.get_processing_stats(),
        "storage_stats": await log_storage.get_storage_stats()
    }

@router.get("/log_stream/quality_metrics")
async def get_quality_metrics(hours: int = 24):
    """Get execution quality metrics"""

    metrics = await real_time_analyzer.get_quality_metrics(hours)

    return {
        "time_period_hours": hours,
        "metrics": metrics,
        "timestamp": datetime.now().isoformat()
    }

@router.get("/log_stream/anomalies")
async def get_anomalies(hours: int = 24):
    """Get recent anomalies"""

    anomalies = await anomaly_marker.get_recent_anomalies(hours)

    return {
        "anomalies": anomalies,
        "total_count": len(anomalies),
        "severity_breakdown": {
            "high": len([a for a in anomalies if a["severity"] == "high"]),
            "medium": len([a for a in anomalies if a["severity"] == "medium"]),
            "low": len([a for a in anomalies if a["severity"] == "low"])
        },
        "time_period_hours": hours
    }

@router.get("/log_stream/latency_analysis")
async def get_latency_analysis(hours: int = 24):
    """Get latency analysis"""

    analysis = await real_time_analyzer.get_latency_analysis(hours)

    return {
        "time_period_hours": hours,
        "analysis": analysis,
        "histogram": await real_time_analyzer.get_latency_histogram(hours)
    }

@router.get("/log_stream/slippage_analysis")
async def get_slippage_analysis(hours: int = 24):
    """Get slippage analysis"""

    analysis = await real_time_analyzer.get_slippage_analysis(hours)

    return {
        "time_period_hours": hours,
        "analysis": analysis,
        "distribution": await real_time_analyzer.get_slippage_distribution(hours)
    }

const TradeLogStreamView: React.FC = () => {
  const [recentLogs, setRecentLogs] = useState<LogEntry[]>([]);
  const [qualityMetrics, setQualityMetrics] = useState<QualityMetrics | null>(null);
  const [anomalies, setAnomalies] = useState<Anomaly[]>([]);
  const [latencyAnalysis, setLatencyAnalysis] = useState<LatencyAnalysis | null>(null);

  return (
    <div className="trade-log-stream-dashboard">
      {/* Real-time Log Stream */}
      <LogStreamPanel 
        logs={recentLogs}
        onLogSelect={handleLogSelect}
      />

      {/* Quality Metrics */}
      <QualityMetricsPanel 
        metrics={qualityMetrics}
      />

      {/* Latency Analysis */}
      <LatencyAnalysisPanel 
        analysis={latencyAnalysis}
      />

      {/* Slippage Analysis */}
      <SlippageAnalysisPanel 
        analysis={qualityMetrics?.slippage}
      />

      {/* Anomaly Detection */}
      <AnomalyDetectionPanel 
        anomalies={anomalies}
        onAnomalySelect={handleAnomalySelect}
      />

      {/* Execution Quality Trends */}
      <ExecutionQualityTrendsPanel 
        metrics={qualityMetrics}
      />

      {/* Stream Statistics */}
      <StreamStatisticsPanel 
        logs={recentLogs}
      />

      {/* Performance Monitoring */}
      <PerformanceMonitoringPanel 
        metrics={qualityMetrics}
      />
    </div>
  );
};