49. Multi-Strategy Super Scheduler¶

services/super-scheduler-center/
├── src/
│   ├── main.py
│   ├── registry/
│   │   ├── strategy_registry.py
│   │   ├── strategy_classifier.py
│   │   └── metadata_manager.py
│   ├── scheduler/
│   │   ├── strategy_scheduler.py
│   │   ├── priority_manager.py
│   │   ├── batch_optimizer.py
│   │   └── load_balancer.py
│   ├── executor/
│   │   ├── scheduler_executor.py
│   │   ├── strategy_launcher.py
│   │   ├── execution_monitor.py
│   │   └── failure_handler.py
│   ├── monitor/
│   │   ├── resource_monitor.py
│   │   ├── latency_monitor.py
│   │   ├── performance_analyzer.py
│   │   └── health_checker.py
│   ├── resource/
│   │   ├── resource_allocator.py
│   │   ├── capacity_planner.py
│   │   └── starvation_detector.py
│   ├── api/
│   │   ├── scheduler_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class StrategyRegistry:
    def __init__(self, database_client):
        self.database_client = database_client
        self.strategies = {}
        self.strategy_groups = {}
        self.metadata_cache = {}

    def register_strategy(self, strategy_id: str, config: Dict) -> bool:
        """Register a new strategy with complete metadata"""

        strategy_metadata = {
            "strategy_id": strategy_id,
            "name": config.get("name", strategy_id),
            "priority": config.get("priority", "medium"),
            "category": config.get("category", "general"),
            "frequency": config.get("frequency", "medium"),
            "resource_requirements": config.get("resource_requirements", {
                "cpu_cores": 1,
                "memory_mb": 512,
                "network_mbps": 10
            }),
            "execution_timeout": config.get("execution_timeout", 30),
            "max_retries": config.get("max_retries", 3),
            "dependencies": config.get("dependencies", []),
            "tags": config.get("tags", []),
            "status": "registered",
            "created_at": datetime.now().isoformat(),
            "last_updated": datetime.now().isoformat()
        }

        # Store in database
        self.database_client.insert("strategies", strategy_metadata)

        # Update cache
        self.strategies[strategy_id] = strategy_metadata

        # Add to appropriate group
        self.add_to_group(strategy_id, strategy_metadata)

        return True

    def add_to_group(self, strategy_id: str, metadata: Dict):
        """Add strategy to appropriate group based on category and frequency"""

        group_key = f"{metadata['category']}_{metadata['frequency']}"

        if group_key not in self.strategy_groups:
            self.strategy_groups[group_key] = []

        self.strategy_groups[group_key].append(strategy_id)

    def get_strategy(self, strategy_id: str) -> Dict:
        """Get strategy metadata"""

        if strategy_id in self.strategies:
            return self.strategies[strategy_id]

        # Query database if not in cache
        result = self.database_client.query(
            "SELECT * FROM strategies WHERE strategy_id = %s",
            (strategy_id,)
        )

        if result:
            strategy_data = result[0]
            self.strategies[strategy_id] = strategy_data
            return strategy_data

        return {}

    def get_strategies_by_priority(self, priority: str) -> List[Dict]:
        """Get all strategies with specific priority"""

        strategies = []
        for strategy_id, metadata in self.strategies.items():
            if metadata.get("priority") == priority:
                strategies.append(metadata)

        return strategies

    def get_strategies_by_category(self, category: str) -> List[Dict]:
        """Get all strategies in specific category"""

        strategies = []
        for strategy_id, metadata in self.strategies.items():
            if metadata.get("category") == category:
                strategies.append(metadata)

        return strategies

    def get_strategies_by_frequency(self, frequency: str) -> List[Dict]:
        """Get all strategies with specific frequency"""

        strategies = []
        for strategy_id, metadata in self.strategies.items():
            if metadata.get("frequency") == frequency:
                strategies.append(metadata)

        return strategies

    def update_strategy_status(self, strategy_id: str, status: str):
        """Update strategy status"""

        if strategy_id in self.strategies:
            self.strategies[strategy_id]["status"] = status
            self.strategies[strategy_id]["last_updated"] = datetime.now().isoformat()

            # Update database
            self.database_client.update(
                "strategies",
                {"status": status, "last_updated": datetime.now().isoformat()},
                {"strategy_id": strategy_id}
            )

    def get_active_strategies(self) -> List[Dict]:
        """Get all active strategies"""

        active_strategies = []
        for strategy_id, metadata in self.strategies.items():
            if metadata.get("status") in ["running", "scheduled", "queued"]:
                active_strategies.append(metadata)

        return active_strategies

    def get_strategy_groups(self) -> Dict:
        """Get all strategy groups"""

        return self.strategy_groups

class StrategyScheduler:
    def __init__(self, strategy_registry, resource_monitor, priority_manager):
        self.strategy_registry = strategy_registry
        self.resource_monitor = resource_monitor
        self.priority_manager = priority_manager
        self.schedule_queue = []
        self.execution_history = {}

    def generate_schedule(self) -> Dict:
        """Generate optimal execution schedule"""

        # Get current system resources
        system_resources = self.resource_monitor.get_system_resources()

        # Get all active strategies
        active_strategies = self.strategy_registry.get_active_strategies()

        # Categorize strategies by priority and frequency
        high_freq_high_priority = []
        high_freq_medium_priority = []
        medium_freq_strategies = []
        low_freq_strategies = []

        for strategy in active_strategies:
            frequency = strategy.get("frequency", "medium")
            priority = strategy.get("priority", "medium")

            if frequency == "high" and priority == "high":
                high_freq_high_priority.append(strategy)
            elif frequency == "high":
                high_freq_medium_priority.append(strategy)
            elif frequency == "medium":
                medium_freq_strategies.append(strategy)
            else:
                low_freq_strategies.append(strategy)

        # Generate schedule based on system load
        schedule = self.create_adaptive_schedule(
            high_freq_high_priority,
            high_freq_medium_priority,
            medium_freq_strategies,
            low_freq_strategies,
            system_resources
        )

        return {
            "timestamp": datetime.now().isoformat(),
            "system_resources": system_resources,
            "schedule": schedule,
            "total_strategies": len(active_strategies),
            "scheduled_strategies": len(schedule["immediate"]) + len(schedule["batch"]) + len(schedule["idle"])
        }

    def create_adaptive_schedule(self, high_priority_high_freq, high_freq_medium_priority,
                               medium_freq, low_freq, system_resources):
        """Create adaptive schedule based on system resources"""

        cpu_usage = system_resources.get("cpu_percent", 0)
        memory_usage = system_resources.get("memory_percent", 0)

        schedule = {
            "immediate": [],  # Execute immediately (ultra-low latency)
            "batch": [],      # Batch execution
            "idle": [],       # Execute during idle time
            "deferred": []    # Defer execution
        }

        # High-frequency, high-priority strategies always get immediate execution
        schedule["immediate"].extend(high_priority_high_freq)

        # High-frequency, medium-priority strategies get immediate execution if resources allow
        if cpu_usage < 80 and memory_usage < 85:
            schedule["immediate"].extend(high_freq_medium_priority)
        else:
            # Limit immediate execution based on available resources
            available_slots = max(1, int((100 - cpu_usage) / 10))
            schedule["immediate"].extend(high_freq_medium_priority[:available_slots])
            schedule["batch"].extend(high_freq_medium_priority[available_slots:])

        # Medium-frequency strategies go to batch processing
        if cpu_usage < 90:
            schedule["batch"].extend(medium_freq)
        else:
            schedule["deferred"].extend(medium_freq)

        # Low-frequency strategies execute during idle time
        if cpu_usage < 70:
            schedule["idle"].extend(low_freq)
        else:
            schedule["deferred"].extend(low_freq)

        return schedule

    def optimize_batch_execution(self, strategies: List[Dict]) -> List[List[Dict]]:
        """Optimize batch execution by grouping similar strategies"""

        if not strategies:
            return []

        # Group strategies by category and resource requirements
        strategy_groups = {}

        for strategy in strategies:
            category = strategy.get("category", "general")
            resource_req = strategy.get("resource_requirements", {})
            cpu_cores = resource_req.get("cpu_cores", 1)

            group_key = f"{category}_{cpu_cores}"

            if group_key not in strategy_groups:
                strategy_groups[group_key] = []

            strategy_groups[group_key].append(strategy)

        # Create optimized batches
        optimized_batches = []

        for group_key, group_strategies in strategy_groups.items():
            # Split large groups into smaller batches
            batch_size = self.calculate_optimal_batch_size(group_strategies[0])

            for i in range(0, len(group_strategies), batch_size):
                batch = group_strategies[i:i + batch_size]
                optimized_batches.append(batch)

        return optimized_batches

    def calculate_optimal_batch_size(self, sample_strategy: Dict) -> int:
        """Calculate optimal batch size based on strategy characteristics"""

        resource_req = sample_strategy.get("resource_requirements", {})
        cpu_cores = resource_req.get("cpu_cores", 1)
        memory_mb = resource_req.get("memory_mb", 512)

        # Base batch size on resource requirements
        if cpu_cores >= 4:
            return 1  # High-resource strategies run individually
        elif cpu_cores >= 2:
            return 3  # Medium-resource strategies in small batches
        else:
            return 10  # Low-resource strategies in larger batches

    def get_execution_priority(self, strategy: Dict) -> float:
        """Calculate execution priority score"""

        base_priority = self.priority_manager.get_priority_score(strategy.get("priority", "medium"))
        frequency_multiplier = self.get_frequency_multiplier(strategy.get("frequency", "medium"))
        performance_bonus = self.get_performance_bonus(strategy.get("strategy_id"))

        return base_priority * frequency_multiplier * performance_bonus

    def get_frequency_multiplier(self, frequency: str) -> float:
        """Get frequency multiplier for priority calculation"""

        multipliers = {
            "high": 2.0,
            "medium": 1.0,
            "low": 0.5
        }

        return multipliers.get(frequency, 1.0)

    def get_performance_bonus(self, strategy_id: str) -> float:
        """Get performance bonus based on historical performance"""

        if strategy_id in self.execution_history:
            history = self.execution_history[strategy_id]
            success_rate = history.get("success_rate", 0.5)
            avg_latency = history.get("avg_latency", 1000)

            # Bonus for high success rate and low latency
            success_bonus = 1.0 + (success_rate - 0.5) * 0.5
            latency_bonus = 1.0 + max(0, (1000 - avg_latency) / 1000) * 0.3

            return success_bonus * latency_bonus

        return 1.0

class SchedulerExecutor:
    def __init__(self, strategy_launcher, execution_monitor, failure_handler):
        self.strategy_launcher = strategy_launcher
        self.execution_monitor = execution_monitor
        self.failure_handler = failure_handler
        self.active_executions = {}
        self.execution_queue = asyncio.Queue()

    async def execute_schedule(self, schedule: Dict) -> Dict:
        """Execute the generated schedule"""

        execution_results = {
            "immediate": [],
            "batch": [],
            "idle": [],
            "failed": [],
            "total_executed": 0
        }

        # Execute immediate strategies
        immediate_strategies = schedule.get("immediate", [])
        for strategy in immediate_strategies:
            result = await self.execute_strategy_immediate(strategy)
            execution_results["immediate"].append(result)
            if result["status"] == "success":
                execution_results["total_executed"] += 1
            else:
                execution_results["failed"].append(result)

        # Execute batch strategies
        batch_strategies = schedule.get("batch", [])
        if batch_strategies:
            batch_results = await self.execute_strategies_batch(batch_strategies)
            execution_results["batch"].extend(batch_results["successful"])
            execution_results["failed"].extend(batch_results["failed"])
            execution_results["total_executed"] += batch_results["total_executed"]

        # Schedule idle strategies
        idle_strategies = schedule.get("idle", [])
        if idle_strategies:
            await self.schedule_idle_executions(idle_strategies)

        return execution_results

    async def execute_strategy_immediate(self, strategy: Dict) -> Dict:
        """Execute strategy immediately with ultra-low latency"""

        strategy_id = strategy["strategy_id"]
        start_time = time.time()

        try:
            # Launch strategy
            execution_id = await self.strategy_launcher.launch_strategy(strategy)

            # Monitor execution
            execution_result = await self.execution_monitor.monitor_execution(
                execution_id, strategy.get("execution_timeout", 30)
            )

            end_time = time.time()
            latency = (end_time - start_time) * 1000  # Convert to milliseconds

            result = {
                "strategy_id": strategy_id,
                "execution_id": execution_id,
                "status": "success" if execution_result["status"] == "completed" else "failed",
                "latency_ms": latency,
                "start_time": start_time,
                "end_time": end_time,
                "result": execution_result
            }

            # Update execution history
            self.update_execution_history(strategy_id, result)

            return result

        except Exception as e:
            # Handle execution failure
            await self.failure_handler.handle_execution_failure(strategy, str(e))

            return {
                "strategy_id": strategy_id,
                "status": "failed",
                "error": str(e),
                "start_time": start_time,
                "end_time": time.time()
            }

    async def execute_strategies_batch(self, strategies: List[Dict]) -> Dict:
        """Execute strategies in optimized batches"""

        # Optimize batch execution
        optimized_batches = self.optimize_batch_execution(strategies)

        batch_results = {
            "successful": [],
            "failed": [],
            "total_executed": 0
        }

        for batch in optimized_batches:
            # Execute batch concurrently
            batch_tasks = [self.execute_strategy_immediate(strategy) for strategy in batch]
            batch_results_list = await asyncio.gather(*batch_tasks, return_exceptions=True)

            for result in batch_results_list:
                if isinstance(result, Exception):
                    batch_results["failed"].append({
                        "error": str(result),
                        "status": "failed"
                    })
                elif result["status"] == "success":
                    batch_results["successful"].append(result)
                    batch_results["total_executed"] += 1
                else:
                    batch_results["failed"].append(result)

        return batch_results

    async def schedule_idle_executions(self, strategies: List[Dict]):
        """Schedule strategies for idle-time execution"""

        for strategy in strategies:
            await self.execution_queue.put({
                "strategy": strategy,
                "priority": "idle",
                "scheduled_time": time.time()
            })

    def update_execution_history(self, strategy_id: str, result: Dict):
        """Update execution history for performance tracking"""

        if strategy_id not in self.execution_history:
            self.execution_history[strategy_id] = {
                "total_executions": 0,
                "successful_executions": 0,
                "failed_executions": 0,
                "total_latency": 0,
                "latencies": []
            }

        history = self.execution_history[strategy_id]
        history["total_executions"] += 1

        if result["status"] == "success":
            history["successful_executions"] += 1
            history["total_latency"] += result["latency_ms"]
            history["latencies"].append(result["latency_ms"])

            # Keep only recent latencies for performance calculation
            if len(history["latencies"]) > 100:
                history["latencies"] = history["latencies"][-100:]
        else:
            history["failed_executions"] += 1

        # Calculate performance metrics
        history["success_rate"] = history["successful_executions"] / history["total_executions"]
        if history["latencies"]:
            history["avg_latency"] = sum(history["latencies"]) / len(history["latencies"])
            history["p95_latency"] = sorted(history["latencies"])[int(len(history["latencies"]) * 0.95)]
        else:
            history["avg_latency"] = 0
            history["p95_latency"] = 0

import psutil
import asyncio

class ResourceMonitor:
    def __init__(self):
        self.resource_history = []
        self.alert_thresholds = {
            "cpu_percent": 90,
            "memory_percent": 85,
            "disk_percent": 80,
            "network_bandwidth": 1000  # Mbps
        }

    def get_system_resources(self) -> Dict:
        """Get current system resource utilization"""

        # CPU usage
        cpu_percent = psutil.cpu_percent(interval=0.1)
        cpu_count = psutil.cpu_count()

        # Memory usage
        memory = psutil.virtual_memory()
        memory_percent = memory.percent
        memory_available = memory.available / (1024 ** 3)  # GB

        # Disk usage
        disk = psutil.disk_usage('/')
        disk_percent = disk.percent
        disk_free = disk.free / (1024 ** 3)  # GB

        # Network I/O
        network = psutil.net_io_counters()
        network_bytes_sent = network.bytes_sent
        network_bytes_recv = network.bytes_recv

        # Process count
        process_count = len(psutil.pids())

        resources = {
            "timestamp": datetime.now().isoformat(),
            "cpu_percent": cpu_percent,
            "cpu_count": cpu_count,
            "memory_percent": memory_percent,
            "memory_available_gb": memory_available,
            "disk_percent": disk_percent,
            "disk_free_gb": disk_free,
            "network_bytes_sent": network_bytes_sent,
            "network_bytes_recv": network_bytes_recv,
            "process_count": process_count,
            "load_average": self.get_load_average()
        }

        # Store in history
        self.resource_history.append(resources)

        # Keep only recent history
        if len(self.resource_history) > 1000:
            self.resource_history = self.resource_history[-1000:]

        return resources

    def get_load_average(self) -> List[float]:
        """Get system load average"""

        try:
            return psutil.getloadavg()
        except:
            return [0.0, 0.0, 0.0]

    def check_resource_alerts(self) -> List[Dict]:
        """Check for resource utilization alerts"""

        current_resources = self.get_system_resources()
        alerts = []

        # CPU alert
        if current_resources["cpu_percent"] > self.alert_thresholds["cpu_percent"]:
            alerts.append({
                "type": "cpu_high",
                "severity": "warning",
                "message": f"CPU usage is {current_resources['cpu_percent']:.1f}%",
                "value": current_resources["cpu_percent"],
                "threshold": self.alert_thresholds["cpu_percent"]
            })

        # Memory alert
        if current_resources["memory_percent"] > self.alert_thresholds["memory_percent"]:
            alerts.append({
                "type": "memory_high",
                "severity": "warning",
                "message": f"Memory usage is {current_resources['memory_percent']:.1f}%",
                "value": current_resources["memory_percent"],
                "threshold": self.alert_thresholds["memory_percent"]
            })

        # Disk alert
        if current_resources["disk_percent"] > self.alert_thresholds["disk_percent"]:
            alerts.append({
                "type": "disk_high",
                "severity": "warning",
                "message": f"Disk usage is {current_resources['disk_percent']:.1f}%",
                "value": current_resources["disk_percent"],
                "threshold": self.alert_thresholds["disk_percent"]
            })

        return alerts

    def get_resource_trends(self, minutes: int = 60) -> Dict:
        """Get resource utilization trends"""

        if not self.resource_history:
            return {}

        # Filter recent history
        cutoff_time = datetime.now() - timedelta(minutes=minutes)
        recent_history = [
            h for h in self.resource_history 
            if datetime.fromisoformat(h["timestamp"]) > cutoff_time
        ]

        if not recent_history:
            return {}

        # Calculate trends
        cpu_values = [h["cpu_percent"] for h in recent_history]
        memory_values = [h["memory_percent"] for h in recent_history]

        return {
            "cpu_trend": {
                "current": cpu_values[-1],
                "average": sum(cpu_values) / len(cpu_values),
                "max": max(cpu_values),
                "min": min(cpu_values)
            },
            "memory_trend": {
                "current": memory_values[-1],
                "average": sum(memory_values) / len(memory_values),
                "max": max(memory_values),
                "min": min(memory_values)
            },
            "data_points": len(recent_history)
        }

@router.post("/scheduler/register")
async def register_strategy(strategy_config: StrategyConfig):
    """Register a new strategy"""
    result = strategy_registry.register_strategy(
        strategy_config.strategy_id, strategy_config.dict()
    )
    return {"status": "success", "strategy_id": strategy_config.strategy_id}

@router.get("/scheduler/schedule")
async def get_current_schedule():
    """Get current execution schedule"""
    schedule = strategy_scheduler.generate_schedule()
    return schedule

@router.post("/scheduler/execute")
async def execute_schedule():
    """Execute current schedule"""
    schedule = strategy_scheduler.generate_schedule()
    result = await scheduler_executor.execute_schedule(schedule)
    return result

@router.get("/scheduler/status")
async def get_scheduler_status():
    """Get scheduler status and statistics"""
    active_strategies = strategy_registry.get_active_strategies()
    system_resources = resource_monitor.get_system_resources()

    return {
        "active_strategies": len(active_strategies),
        "system_resources": system_resources,
        "scheduler_status": "running"
    }

@router.get("/scheduler/performance")
async def get_performance_metrics():
    """Get performance metrics for all strategies"""
    return scheduler_executor.get_performance_metrics()

@router.get("/scheduler/resources")
async def get_resource_utilization():
    """Get detailed resource utilization"""
    resources = resource_monitor.get_system_resources()
    trends = resource_monitor.get_resource_trends()
    alerts = resource_monitor.check_resource_alerts()

    return {
        "current": resources,
        "trends": trends,
        "alerts": alerts
    }

const SuperSchedulerView: React.FC = () => {
  const [schedulerStatus, setSchedulerStatus] = useState<SchedulerStatus | null>(null);
  const [resourceUtilization, setResourceUtilization] = useState<ResourceData | null>(null);
  const [activeStrategies, setActiveStrategies] = useState<Strategy[]>([]);
  const [performanceMetrics, setPerformanceMetrics] = useState<PerformanceData | null>(null);

  return (
    <div className="super-scheduler-dashboard">
      {/* System Overview */}
      <SystemOverviewPanel 
        schedulerStatus={schedulerStatus}
        resourceUtilization={resourceUtilization}
      />

      {/* Strategy Management */}
      <StrategyManagementPanel 
        activeStrategies={activeStrategies}
        onRegisterStrategy={registerStrategy}
        onUpdateSchedule={updateSchedule}
      />

      {/* Resource Monitoring */}
      <ResourceMonitoringPanel 
        resourceUtilization={resourceUtilization}
        alerts={resourceUtilization?.alerts}
      />

      {/* Performance Metrics */}
      <PerformanceMetricsPanel 
        performanceMetrics={performanceMetrics}
      />

      {/* Execution Queue */}
      <ExecutionQueuePanel 
        activeStrategies={activeStrategies}
      />

      {/* Latency Heatmap */}
      <LatencyHeatmapPanel 
        performanceMetrics={performanceMetrics}
      />
    </div>
  );
};