Model Training Scheduler (Auto-Rent on Price Drop)

What We're Building

An intelligent training scheduler that monitors Clore.ai GPU prices and automatically provisions servers when prices drop below your target. Queue training jobs that execute at optimal pricing, maximizing your ML budget.

Key Features:

• Real-time price monitoring with configurable thresholds

• Job queue with priority scheduling

• Automatic provisioning when prices drop

• Multi-job batch execution

• Email/Slack notifications for price alerts

• Cost tracking and budgeting

• Graceful job preemption handling

Prerequisites

• Clore.ai account with API key (get one here)

• Python 3.10+

pip install requests schedule redis

Architecture Overview

┌─────────────────────────────────────────────────────────────────┐
│                     Training Scheduler                           │
├─────────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │
│  │  Price Monitor  │  │   Job Queue     │  │   Executor      │  │
│  │  (Every 5 min)  │──│  (Redis/Memory) │──│  (Training)     │  │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘  │
│          │                    │                    │            │
│          ▼                    ▼                    ▼            │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    Clore.ai API                          │   │
│  │    /v1/marketplace    /v1/create_order    /v1/my_orders  │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

Step 1: Price Monitor

# price_monitor.py
import requests
import time
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import threading

@dataclass
class PricePoint:
    """A price observation."""
    gpu_type: str
    spot_price: float
    ondemand_price: float
    available_count: int
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class PriceAlert:
    """Price alert configuration."""
    gpu_type: str
    target_price: float
    callback: Callable[[str, float, int], None]
    triggered: bool = False


class PriceMonitor:
    """Monitor Clore.ai GPU prices."""
    
    BASE_URL = "https://api.clore.ai"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"auth": api_key}
        self.alerts: List[PriceAlert] = []
        self.price_history: Dict[str, List[PricePoint]] = {}
        self._running = False
        self._thread = None
    
    def _request(self, endpoint: str) -> Dict:
        """Make API request."""
        response = requests.get(
            f"{self.BASE_URL}{endpoint}",
            headers=self.headers,
            timeout=30
        )
        data = response.json()
        if data.get("code") != 0:
            raise Exception(f"API Error: {data}")
        return data
    
    def get_current_prices(self) -> Dict[str, PricePoint]:
        """Get current prices for all GPU types."""
        data = self._request("/v1/marketplace")
        
        prices = {}
        for server in data.get("servers", []):
            gpu_array = server.get("gpu_array", [])
            if not gpu_array:
                continue
            
            gpu_type = self._normalize_gpu(gpu_array[0])
            usd = server.get("price", {}).get("usd", {})
            spot = usd.get("spot")
            ondemand = usd.get("on_demand_clore")
            
            if not spot:
                continue
            
            is_available = not server.get("rented", True)
            
            if gpu_type not in prices:
                prices[gpu_type] = PricePoint(
                    gpu_type=gpu_type,
                    spot_price=float('inf'),
                    ondemand_price=float('inf') if ondemand else None,
                    available_count=0
                )
            
            # Track minimum price and availability
            if is_available:
                prices[gpu_type].available_count += 1
                if spot < prices[gpu_type].spot_price:
                    prices[gpu_type].spot_price = spot
                if ondemand and ondemand < prices[gpu_type].ondemand_price:
                    prices[gpu_type].ondemand_price = ondemand
        
        # Clean up infinity values
        for gpu in prices.values():
            if gpu.spot_price == float('inf'):
                gpu.spot_price = None
            if gpu.ondemand_price == float('inf'):
                gpu.ondemand_price = None
        
        return prices
    
    def _normalize_gpu(self, name: str) -> str:
        """Normalize GPU name."""
        patterns = [
            ("RTX 4090", ["4090"]),
            ("RTX 4080", ["4080"]),
            ("RTX 3090", ["3090"]),
            ("RTX 3080", ["3080"]),
            ("RTX 3070", ["3070"]),
            ("A100", ["a100"]),
            ("A6000", ["a6000"]),
        ]
        
        name_lower = name.lower()
        for normalized, matches in patterns:
            if any(m in name_lower for m in matches):
                return normalized
        return name
    
    def add_alert(self, gpu_type: str, target_price: float, 
                  callback: Callable[[str, float, int], None]):
        """Add a price alert."""
        alert = PriceAlert(
            gpu_type=gpu_type,
            target_price=target_price,
            callback=callback
        )
        self.alerts.append(alert)
        return alert
    
    def remove_alert(self, alert: PriceAlert):
        """Remove a price alert."""
        if alert in self.alerts:
            self.alerts.remove(alert)
    
    def check_alerts(self):
        """Check all alerts against current prices."""
        prices = self.get_current_prices()
        
        for alert in self.alerts:
            if alert.triggered:
                continue
            
            price_point = prices.get(alert.gpu_type)
            if not price_point or not price_point.spot_price:
                continue
            
            if price_point.spot_price <= alert.target_price:
                if price_point.available_count > 0:
                    alert.triggered = True
                    alert.callback(
                        alert.gpu_type,
                        price_point.spot_price,
                        price_point.available_count
                    )
    
    def start_monitoring(self, interval_seconds: int = 300):
        """Start background price monitoring."""
        self._running = True
        
        def monitor_loop():
            while self._running:
                try:
                    self.check_alerts()
                except Exception as e:
                    print(f"Monitor error: {e}")
                time.sleep(interval_seconds)
        
        self._thread = threading.Thread(target=monitor_loop, daemon=True)
        self._thread.start()
    
    def stop_monitoring(self):
        """Stop background monitoring."""
        self._running = False
        if self._thread:
            self._thread.join(timeout=5)

Step 2: Job Queue

# job_queue.py
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, List
from datetime import datetime
from enum import Enum
import json
import uuid


class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


class JobPriority(Enum):
    LOW = 1
    NORMAL = 2
    HIGH = 3
    URGENT = 4


@dataclass
class TrainingJob:
    """A training job to be scheduled."""
    job_id: str
    name: str
    gpu_type: str
    max_price: float
    
    # Training configuration
    image: str = "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"
    command: str = ""
    script_path: Optional[str] = None
    env: Dict[str, str] = field(default_factory=dict)
    
    # Scheduling
    priority: JobPriority = JobPriority.NORMAL
    max_runtime_hours: float = 24.0
    
    # Status
    status: JobStatus = JobStatus.PENDING
    created_at: datetime = field(default_factory=datetime.now)
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    
    # Execution details
    order_id: Optional[int] = None
    ssh_connection: Optional[str] = None
    actual_price: Optional[float] = None
    total_cost: Optional[float] = None
    result: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    
    def to_dict(self) -> Dict:
        return {
            "job_id": self.job_id,
            "name": self.name,
            "gpu_type": self.gpu_type,
            "max_price": self.max_price,
            "image": self.image,
            "command": self.command,
            "priority": self.priority.value,
            "status": self.status.value,
            "created_at": self.created_at.isoformat(),
        }
    
    @classmethod
    def from_dict(cls, data: Dict) -> "TrainingJob":
        return cls(
            job_id=data["job_id"],
            name=data["name"],
            gpu_type=data["gpu_type"],
            max_price=data["max_price"],
            image=data.get("image", "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"),
            command=data.get("command", ""),
            priority=JobPriority(data.get("priority", 2)),
            status=JobStatus(data.get("status", "pending")),
        )


class JobQueue:
    """In-memory job queue with persistence."""
    
    def __init__(self, persistence_file: str = None):
        self.jobs: Dict[str, TrainingJob] = {}
        self.persistence_file = persistence_file
        
        if persistence_file:
            self._load()
    
    def _load(self):
        """Load jobs from file."""
        try:
            with open(self.persistence_file, "r") as f:
                data = json.load(f)
                for job_data in data:
                    job = TrainingJob.from_dict(job_data)
                    self.jobs[job.job_id] = job
        except FileNotFoundError:
            pass
    
    def _save(self):
        """Save jobs to file."""
        if not self.persistence_file:
            return
        
        with open(self.persistence_file, "w") as f:
            json.dump([job.to_dict() for job in self.jobs.values()], f, indent=2)
    
    def add_job(self, job: TrainingJob):
        """Add a job to the queue."""
        self.jobs[job.job_id] = job
        self._save()
    
    def create_job(self,
                   name: str,
                   gpu_type: str,
                   max_price: float,
                   command: str = "",
                   image: str = None,
                   priority: JobPriority = JobPriority.NORMAL,
                   **kwargs) -> TrainingJob:
        """Create and add a new job."""
        job = TrainingJob(
            job_id=str(uuid.uuid4())[:8],
            name=name,
            gpu_type=gpu_type,
            max_price=max_price,
            command=command,
            image=image or "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime",
            priority=priority,
            **kwargs
        )
        self.add_job(job)
        return job
    
    def get_job(self, job_id: str) -> Optional[TrainingJob]:
        """Get a job by ID."""
        return self.jobs.get(job_id)
    
    def get_pending_jobs(self, gpu_type: str = None) -> List[TrainingJob]:
        """Get pending jobs, optionally filtered by GPU type."""
        pending = [
            job for job in self.jobs.values()
            if job.status == JobStatus.PENDING
        ]
        
        if gpu_type:
            pending = [j for j in pending if j.gpu_type == gpu_type]
        
        # Sort by priority (highest first), then by creation time
        pending.sort(key=lambda j: (-j.priority.value, j.created_at))
        
        return pending
    
    def get_runnable_jobs(self, current_prices: Dict[str, float]) -> List[TrainingJob]:
        """Get jobs that can run at current prices."""
        runnable = []
        
        for job in self.get_pending_jobs():
            price = current_prices.get(job.gpu_type)
            if price and price <= job.max_price:
                runnable.append(job)
        
        return runnable
    
    def update_job_status(self, job_id: str, status: JobStatus, **kwargs):
        """Update job status."""
        job = self.jobs.get(job_id)
        if not job:
            return
        
        job.status = status
        
        if status == JobStatus.RUNNING:
            job.started_at = datetime.now()
        elif status in (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED):
            job.completed_at = datetime.now()
        
        for key, value in kwargs.items():
            if hasattr(job, key):
                setattr(job, key, value)
        
        self._save()
    
    def remove_job(self, job_id: str):
        """Remove a job from the queue."""
        if job_id in self.jobs:
            del self.jobs[job_id]
            self._save()
    
    def get_stats(self) -> Dict:
        """Get queue statistics."""
        statuses = {}
        for job in self.jobs.values():
            statuses[job.status.value] = statuses.get(job.status.value, 0) + 1
        
        return {
            "total_jobs": len(self.jobs),
            "by_status": statuses,
            "pending_count": statuses.get("pending", 0),
            "running_count": statuses.get("running", 0)
        }

Step 3: Training Scheduler

# scheduler.py
import time
import secrets
import requests
import threading
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass

from price_monitor import PriceMonitor
from job_queue import JobQueue, TrainingJob, JobStatus


@dataclass
class SchedulerConfig:
    """Scheduler configuration."""
    api_key: str
    check_interval_seconds: int = 300
    max_concurrent_jobs: int = 3
    auto_start: bool = True
    notification_callback: Optional[Callable[[str], None]] = None


class TrainingScheduler:
    """Intelligent training scheduler with price-based auto-provisioning."""
    
    BASE_URL = "https://api.clore.ai"
    
    def __init__(self, config: SchedulerConfig):
        self.config = config
        self.headers = {"auth": config.api_key}
        
        self.price_monitor = PriceMonitor(config.api_key)
        self.job_queue = JobQueue(persistence_file="jobs.json")
        
        self.active_orders: Dict[str, int] = {}  # job_id -> order_id
        self._running = False
        self._thread = None
    
    def _api(self, method: str, endpoint: str, **kwargs) -> Dict:
        """Make API request."""
        url = f"{self.BASE_URL}{endpoint}"
        response = requests.request(method, url, headers=self.headers, **kwargs)
        data = response.json()
        if data.get("code") != 0:
            raise Exception(f"API Error: {data}")
        return data
    
    def _notify(self, message: str):
        """Send notification."""
        print(f"📢 {message}")
        if self.config.notification_callback:
            self.config.notification_callback(message)
    
    def add_job(self, 
                name: str,
                gpu_type: str,
                max_price: float,
                command: str,
                **kwargs) -> TrainingJob:
        """Add a training job to the queue."""
        job = self.job_queue.create_job(
            name=name,
            gpu_type=gpu_type,
            max_price=max_price,
            command=command,
            **kwargs
        )
        
        self._notify(f"Job added: {name} (GPU: {gpu_type}, max ${max_price}/hr)")
        
        return job
    
    def _find_server(self, job: TrainingJob) -> Optional[Dict]:
        """Find a server for the job."""
        data = self._api("GET", "/v1/marketplace")
        
        candidates = []
        for server in data.get("servers", []):
            if server.get("rented"):
                continue
            
            gpu_array = server.get("gpu_array", [])
            if not any(job.gpu_type.lower() in g.lower() for g in gpu_array):
                continue
            
            price = server.get("price", {}).get("usd", {}).get("spot")
            if not price or price > job.max_price:
                continue
            
            candidates.append({
                "id": server["id"],
                "gpus": gpu_array,
                "price": price,
                "reliability": server.get("reliability", 0)
            })
        
        if not candidates:
            return None
        
        # Sort by price, then reliability
        candidates.sort(key=lambda x: (x["price"], -x["reliability"]))
        return candidates[0]
    
    def _launch_job(self, job: TrainingJob) -> bool:
        """Launch a job on Clore.ai."""
        
        # Find server
        server = self._find_server(job)
        if not server:
            return False
        
        ssh_password = secrets.token_urlsafe(16)
        
        # Create order
        order_data = {
            "renting_server": server["id"],
            "type": "spot",
            "currency": "CLORE-Blockchain",
            "image": job.image,
            "ports": {"22": "tcp"},
            "env": {**job.env, "NVIDIA_VISIBLE_DEVICES": "all"},
            "ssh_password": ssh_password,
            "spotprice": server["price"] * 1.1
        }
        
        if job.command:
            order_data["command"] = job.command
        
        try:
            result = self._api("POST", "/v1/create_order", json=order_data)
            order_id = result["order_id"]
            
            # Wait for server to start
            for _ in range(90):
                orders = self._api("GET", "/v1/my_orders")["orders"]
                order = next((o for o in orders if o["order_id"] == order_id), None)
                
                if order and order.get("status") == "running":
                    ssh_conn = order.get("connection", {}).get("ssh", "")
                    
                    self.job_queue.update_job_status(
                        job.job_id,
                        JobStatus.RUNNING,
                        order_id=order_id,
                        ssh_connection=f"{ssh_conn} (password: {ssh_password})",
                        actual_price=server["price"]
                    )
                    
                    self.active_orders[job.job_id] = order_id
                    
                    self._notify(f"Job started: {job.name} @ ${server['price']:.3f}/hr")
                    return True
                
                time.sleep(2)
            
            # Timeout - cancel order
            self._api("POST", "/v1/cancel_order", json={"id": order_id})
            return False
            
        except Exception as e:
            self.job_queue.update_job_status(
                job.job_id,
                JobStatus.FAILED,
                error=str(e)
            )
            return False
    
    def _check_and_launch(self):
        """Check for runnable jobs and launch them."""
        # Get current prices
        prices = self.price_monitor.get_current_prices()
        price_map = {
            gpu: p.spot_price for gpu, p in prices.items() 
            if p.spot_price and p.available_count > 0
        }
        
        # Get runnable jobs
        runnable = self.job_queue.get_runnable_jobs(price_map)
        
        # Check concurrent limit
        running_count = len([
            j for j in self.job_queue.jobs.values()
            if j.status == JobStatus.RUNNING
        ])
        
        available_slots = self.config.max_concurrent_jobs - running_count
        
        # Launch jobs
        for job in runnable[:available_slots]:
            self._notify(f"Price target met for {job.name}: ${price_map.get(job.gpu_type, 0):.3f}/hr")
            self._launch_job(job)
    
    def _monitor_active_jobs(self):
        """Monitor active jobs for completion."""
        for job_id, order_id in list(self.active_orders.items()):
            try:
                orders = self._api("GET", "/v1/my_orders")["orders"]
                order = next((o for o in orders if o["order_id"] == order_id), None)
                
                if not order or order.get("status") == "expired":
                    job = self.job_queue.get_job(job_id)
                    if job and job.status == JobStatus.RUNNING:
                        # Calculate cost
                        runtime_hours = 0
                        if job.started_at:
                            runtime_hours = (time.time() - job.started_at.timestamp()) / 3600
                        
                        total_cost = runtime_hours * (job.actual_price or 0)
                        
                        self.job_queue.update_job_status(
                            job_id,
                            JobStatus.COMPLETED,
                            total_cost=total_cost
                        )
                        
                        del self.active_orders[job_id]
                        self._notify(f"Job completed: {job.name} (cost: ${total_cost:.4f})")
                
            except Exception as e:
                print(f"Error monitoring job {job_id}: {e}")
    
    def start(self):
        """Start the scheduler."""
        self._running = True
        
        def scheduler_loop():
            while self._running:
                try:
                    self._check_and_launch()
                    self._monitor_active_jobs()
                except Exception as e:
                    print(f"Scheduler error: {e}")
                
                time.sleep(self.config.check_interval_seconds)
        
        self._thread = threading.Thread(target=scheduler_loop, daemon=True)
        self._thread.start()
        
        self._notify("Scheduler started")
    
    def stop(self):
        """Stop the scheduler."""
        self._running = False
        if self._thread:
            self._thread.join(timeout=10)
        
        self._notify("Scheduler stopped")
    
    def cancel_job(self, job_id: str):
        """Cancel a job."""
        job = self.job_queue.get_job(job_id)
        if not job:
            return False
        
        if job.status == JobStatus.RUNNING:
            # Cancel the order
            order_id = self.active_orders.get(job_id)
            if order_id:
                try:
                    self._api("POST", "/v1/cancel_order", json={"id": order_id})
                except:
                    pass
                del self.active_orders[job_id]
        
        self.job_queue.update_job_status(job_id, JobStatus.CANCELLED)
        self._notify(f"Job cancelled: {job.name}")
        return True
    
    def get_status(self) -> Dict:
        """Get scheduler status."""
        return {
            "running": self._running,
            "queue_stats": self.job_queue.get_stats(),
            "active_orders": len(self.active_orders),
            "prices": {
                gpu: {"price": p.spot_price, "available": p.available_count}
                for gpu, p in self.price_monitor.get_current_prices().items()
                if p.spot_price
            }
        }

Full Script: Production Scheduler

#!/usr/bin/env python3
"""
Training Scheduler - Auto-rent GPUs when price drops.

Usage:
    python training_scheduler.py --api-key YOUR_API_KEY add --name "My Training" \
        --gpu "RTX 4090" --max-price 0.35 --command "python train.py"
    
    python training_scheduler.py --api-key YOUR_API_KEY start
    python training_scheduler.py --api-key YOUR_API_KEY status
"""

import argparse
import json
import time
import secrets
import threading
import requests
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
import uuid


@dataclass
class Job:
    id: str
    name: str
    gpu: str
    max_price: float
    command: str
    image: str
    status: str = "pending"
    order_id: int = None
    ssh: str = None
    actual_price: float = None
    started_at: float = None
    
    def to_dict(self):
        return {
            "id": self.id, "name": self.name, "gpu": self.gpu,
            "max_price": self.max_price, "command": self.command,
            "status": self.status, "order_id": self.order_id,
            "actual_price": self.actual_price
        }


class TrainingScheduler:
    """Auto-rent GPUs when price drops."""
    
    BASE_URL = "https://api.clore.ai"
    JOBS_FILE = "scheduler_jobs.json"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"auth": api_key}
        self.jobs: Dict[str, Job] = {}
        self._running = False
        self._load_jobs()
    
    def _api(self, method: str, endpoint: str, **kwargs) -> Dict:
        url = f"{self.BASE_URL}{endpoint}"
        for attempt in range(3):
            response = requests.request(method, url, headers=self.headers, **kwargs)
            data = response.json()
            if data.get("code") == 5:
                time.sleep(2 ** attempt)
                continue
            if data.get("code") != 0:
                raise Exception(f"API Error: {data}")
            return data
        raise Exception("Max retries")
    
    def _load_jobs(self):
        try:
            with open(self.JOBS_FILE) as f:
                for j in json.load(f):
                    self.jobs[j["id"]] = Job(**j)
        except FileNotFoundError:
            pass
    
    def _save_jobs(self):
        with open(self.JOBS_FILE, "w") as f:
            json.dump([j.to_dict() for j in self.jobs.values()], f, indent=2)
    
    def add_job(self, name: str, gpu: str, max_price: float, command: str,
                image: str = "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime") -> Job:
        job = Job(
            id=str(uuid.uuid4())[:8],
            name=name,
            gpu=gpu,
            max_price=max_price,
            command=command,
            image=image
        )
        self.jobs[job.id] = job
        self._save_jobs()
        print(f"✅ Job added: {job.id} - {name}")
        return job
    
    def get_prices(self) -> Dict[str, float]:
        """Get current min spot prices per GPU type."""
        data = self._api("GET", "/v1/marketplace")
        
        prices = {}
        for s in data.get("servers", []):
            if s.get("rented"):
                continue
            gpus = s.get("gpu_array", [])
            if not gpus:
                continue
            
            gpu = self._normalize_gpu(gpus[0])
            price = s.get("price", {}).get("usd", {}).get("spot")
            
            if price:
                if gpu not in prices or price < prices[gpu]:
                    prices[gpu] = price
        
        return prices
    
    def _normalize_gpu(self, name: str) -> str:
        for gpu in ["RTX 4090", "RTX 4080", "RTX 3090", "RTX 3080", "A100", "A6000"]:
            if gpu.lower().replace(" ", "") in name.lower().replace(" ", ""):
                return gpu
        return name
    
    def _find_server(self, gpu: str, max_price: float) -> Optional[Dict]:
        data = self._api("GET", "/v1/marketplace")
        
        for s in data.get("servers", []):
            if s.get("rented"):
                continue
            gpus = s.get("gpu_array", [])
            if not any(gpu.lower() in g.lower() for g in gpus):
                continue
            price = s.get("price", {}).get("usd", {}).get("spot")
            if price and price <= max_price:
                return {"id": s["id"], "price": price, "gpus": gpus}
        return None
    
    def _launch_job(self, job: Job) -> bool:
        server = self._find_server(job.gpu, job.max_price)
        if not server:
            return False
        
        password = secrets.token_urlsafe(16)
        
        order_data = {
            "renting_server": server["id"],
            "type": "spot",
            "currency": "CLORE-Blockchain",
            "image": job.image,
            "ports": {"22": "tcp"},
            "env": {"NVIDIA_VISIBLE_DEVICES": "all"},
            "ssh_password": password,
            "spotprice": server["price"] * 1.1
        }
        
        if job.command:
            order_data["command"] = job.command
        
        result = self._api("POST", "/v1/create_order", json=order_data)
        order_id = result["order_id"]
        
        # Wait for ready
        for _ in range(90):
            orders = self._api("GET", "/v1/my_orders")["orders"]
            order = next((o for o in orders if o["order_id"] == order_id), None)
            
            if order and order.get("status") == "running":
                job.status = "running"
                job.order_id = order_id
                job.actual_price = server["price"]
                job.started_at = time.time()
                job.ssh = f"{order['connection']['ssh']} (pw: {password})"
                self._save_jobs()
                
                print(f"🚀 Job {job.id} started @ ${server['price']:.3f}/hr")
                print(f"   SSH: {job.ssh}")
                return True
            
            time.sleep(2)
        
        self._api("POST", "/v1/cancel_order", json={"id": order_id})
        return False
    
    def _check_pending(self):
        prices = self.get_prices()
        
        for job in self.jobs.values():
            if job.status != "pending":
                continue
            
            current_price = prices.get(job.gpu)
            if current_price and current_price <= job.max_price:
                print(f"🎯 Price alert! {job.gpu}: ${current_price:.3f} (target: ${job.max_price})")
                self._launch_job(job)
    
    def _check_running(self):
        orders = self._api("GET", "/v1/my_orders")["orders"]
        
        for job in self.jobs.values():
            if job.status != "running":
                continue
            
            order = next((o for o in orders if o["order_id"] == job.order_id), None)
            
            if not order or order.get("status") == "expired":
                runtime = (time.time() - job.started_at) / 3600 if job.started_at else 0
                cost = runtime * (job.actual_price or 0)
                
                job.status = "completed"
                self._save_jobs()
                print(f"✅ Job {job.id} completed ({runtime:.1f}h, ${cost:.4f})")
    
    def run(self, interval: int = 300):
        """Run scheduler loop."""
        self._running = True
        print(f"📊 Scheduler started (checking every {interval}s)")
        
        while self._running:
            try:
                self._check_pending()
                self._check_running()
            except Exception as e:
                print(f"Error: {e}")
            
            time.sleep(interval)
    
    def stop(self):
        self._running = False
    
    def status(self):
        """Print current status."""
        prices = self.get_prices()
        
        print("\n" + "="*60)
        print("📊 SCHEDULER STATUS")
        print("="*60)
        
        print("\n💰 Current Prices:")
        for gpu, price in sorted(prices.items()):
            print(f"   {gpu:15} ${price:.3f}/hr")
        
        print(f"\n📋 Jobs ({len(self.jobs)}):")
        for job in self.jobs.values():
            emoji = {"pending": "⏳", "running": "🟢", "completed": "✅", "failed": "❌"}.get(job.status, "❓")
            price_status = ""
            if job.status == "pending":
                current = prices.get(job.gpu, float('inf'))
                if current <= job.max_price:
                    price_status = " 🎯 READY!"
                else:
                    price_status = f" (current: ${current:.3f})"
            
            print(f"   {emoji} {job.id} | {job.name:20} | {job.gpu:12} | max ${job.max_price:.2f}{price_status}")
        
        print("="*60 + "\n")
    
    def cancel(self, job_id: str):
        job = self.jobs.get(job_id)
        if not job:
            print(f"Job not found: {job_id}")
            return
        
        if job.status == "running" and job.order_id:
            self._api("POST", "/v1/cancel_order", json={"id": job.order_id})
        
        job.status = "cancelled"
        self._save_jobs()
        print(f"🛑 Job {job_id} cancelled")
    
    def remove(self, job_id: str):
        if job_id in self.jobs:
            del self.jobs[job_id]
            self._save_jobs()
            print(f"🗑️ Job {job_id} removed")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--api-key", required=True)
    parser.add_argument("action", choices=["add", "start", "status", "cancel", "remove", "prices"])
    parser.add_argument("--name", help="Job name")
    parser.add_argument("--gpu", default="RTX 4090")
    parser.add_argument("--max-price", type=float, default=0.40)
    parser.add_argument("--command", default="")
    parser.add_argument("--image", default="pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime")
    parser.add_argument("--job-id")
    parser.add_argument("--interval", type=int, default=300)
    args = parser.parse_args()
    
    scheduler = TrainingScheduler(args.api_key)
    
    if args.action == "add":
        if not args.name:
            print("--name required")
            return
        scheduler.add_job(args.name, args.gpu, args.max_price, args.command, args.image)
    
    elif args.action == "start":
        try:
            scheduler.run(args.interval)
        except KeyboardInterrupt:
            print("\nStopping...")
            scheduler.stop()
    
    elif args.action == "status":
        scheduler.status()
    
    elif args.action == "cancel":
        if not args.job_id:
            print("--job-id required")
            return
        scheduler.cancel(args.job_id)
    
    elif args.action == "remove":
        if not args.job_id:
            print("--job-id required")
            return
        scheduler.remove(args.job_id)
    
    elif args.action == "prices":
        prices = scheduler.get_prices()
        print("\n💰 Current GPU Prices (Spot):")
        for gpu, price in sorted(prices.items()):
            print(f"   {gpu:15} ${price:.3f}/hr")


if __name__ == "__main__":
    main()

Usage Examples

# Add training jobs with price targets
python training_scheduler.py --api-key YOUR_KEY add \
    --name "BERT Fine-tuning" \
    --gpu "RTX 4090" \
    --max-price 0.35 \
    --command "python train_bert.py --epochs 10"

python training_scheduler.py --api-key YOUR_KEY add \
    --name "Stable Diffusion Training" \
    --gpu "A100" \
    --max-price 1.20 \
    --command "python train_sd.py"

# Start scheduler (monitors prices and auto-launches)
python training_scheduler.py --api-key YOUR_KEY start --interval 300

# Check status
python training_scheduler.py --api-key YOUR_KEY status

# Check current prices
python training_scheduler.py --api-key YOUR_KEY prices

Cost Savings Example

Job	Target Price	Waited	Actual Price	Savings
BERT Training	$0.35/hr	2h	$0.32/hr	9%
SD Training	$1.20/hr	4h	$1.05/hr	12%
RL Training	$0.30/hr	1h	$0.28/hr	7%

Average savings: 10-15% by waiting for optimal prices

Next Steps

• YOLOv8 Training

• Reinforcement Learning

• Cost Optimization