Model Training Scheduler (Auto-Rent on Price Drop)
Last updated
Was this helpful?
Was this helpful?
┌─────────────────────────────────────────────────────────────────┐
│ Training Scheduler │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Price Monitor │ │ Job Queue │ │ Executor │ │
│ │ (Every 5 min) │──│ (Redis/Memory) │──│ (Training) │ │
│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Clore.ai API │ │
│ │ /v1/marketplace /v1/create_order /v1/my_orders │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘# 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)# 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)
}# 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
}
}#!/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()# 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