英伟达AI Compute Partnership：从"卖铲人"到"收租人"的算力金融化革命

// DSX AI Factory - Multi-Tenant GPU Scheduler
// Core scheduling engine for NVIDIA's AI factory architecture

package scheduler

import (
	"container/heap"
	"context"
	"fmt"
	"sync"
	"time"
)

// GPU资源单元
type GPUUnit struct {
	ID         string
	Model      string     // GB200, GB300, H100, etc.
	MemoryMB   int64
	ComputeCap float64    // TFLOPS
	Status     string     // free, allocated, reserved, maintenance
	PodID      string     // 所属DGX SuperPod
	NodeID     string
}

// 租户定义
type Tenant struct {
	ID             string
	Name           string
	Priority       int           // 优先级（0-100，越高越优先）
	GuaranteedGPU  int           // 保证GPU数
	MaxGPU         int           // 最大GPU数（弹性上限）
	BurstEnabled   bool          // 是否允许突发弹性使用
	ContractLevel  string        // premium, standard, basic
	RevenueShare   float64       // 收入分成比例
}

// 任务定义
type Task struct {
	ID             string
	TenantID       string
	Type           string        // training, inference, finetune
	GPUCount       int
	MinGPU         int
	MaxGPU         int
	Duration       time.Duration
	Priority       int
	SubmittedAt    time.Time
	Deadline       time.Time
	Preemptible    bool          // 是否可被抢占
	State          string        // queued, running, preempted, completed, failed
}

// 资源分配记录
type Allocation struct {
	TaskID    string
	GPUIDs    []string
	StartTime time.Time
	EndTime   time.Time
	Cost      float64     // 累计费用
}

// MultiTenantScheduler 多租户GPU调度器
type MultiTenantScheduler struct {
	mu          sync.RWMutex
	gpuPool     map[string]*GPUUnit
	tenants     map[string]*Tenant
	taskQueue   PriorityQueue
	running     map[string]*Allocation
	metrics     *SchedulerMetrics
	
	// 调度策略参数
	config SchedulerConfig
}

type SchedulerConfig struct {
	OvercommitRatio   float64       // 超分比（默认1.2）
	PreemptionGrace   time.Duration // 抢占宽限期
	MaxQueueDepth     int
	AccountingPeriod  time.Duration // 计费周期
}

type SchedulerMetrics struct {
	TotalGPUHours      float64
	UtilizationRate    float64
	PreemptionCount    int64
	AvgQueueWaitTime   time.Duration
	TenantUtilization  map[string]float64
}

// PriorityQueue 优先级队列（堆）
type PriorityQueue []*Task

func (pq PriorityQueue) Len() int { return len(pq) }

func (pq PriorityQueue) Less(i, j int) bool {
	// 优先级高的先调度；同优先级下提交时间早的先调度
	if pq[i].Priority != pq[j].Priority {
		return pq[i].Priority > pq[j].Priority
	}
	return pq[i].SubmittedAt.Before(pq[j].SubmittedAt)
}

func (pq PriorityQueue) Swap(i, j int) {
	pq[i], pq[j] = pq[j], pq[i]
}

func (pq *PriorityQueue) Push(x interface{}) {
	*pq = append(*pq, x.(*Task))
}

func (pq *PriorityQueue) Pop() interface{} {
	old := *pq
	n := len(old)
	item := old[n-1]
	old[n-1] = nil
	*pq = old[0 : n-1]
	return item
}

// NewScheduler 创建新的多租户调度器
func NewScheduler(config SchedulerConfig) *MultiTenantScheduler {
	return &MultiTenantScheduler{
		gpuPool:   make(map[string]*GPUUnit),
		tenants:   make(map[string]*Tenant),
		taskQueue: make(PriorityQueue, 0),
		running:   make(map[string]*Allocation),
		metrics: &SchedulerMetrics{
			TenantUtilization: make(map[string]float64),
		},
		config: config,
	}
	heap.Init(&s.taskQueue)
}

// RegisterTenant 注册租户
func (s *MultiTenantScheduler) RegisterTenant(tenant *Tenant) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	if _, exists := s.tenants[tenant.ID]; exists {
		return fmt.Errorf("tenant %s already registered", tenant.ID)
	}
	
	s.tenants[tenant.ID] = tenant
	s.metrics.TenantUtilization[tenant.ID] = 0.0
	return nil
}

// AddGPUNode 添加GPU节点
func (s *MultiTenantScheduler) AddGPUNode(podID, nodeID string, 
	gpus []GPUUnit) {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	for i := range gpus {
		gpus[i].PodID = podID
		gpus[i].NodeID = nodeID
		gpus[i].Status = "free"
		s.gpuPool[gpus[i].ID] = &gpus[i]
	}
}

// SubmitTask 提交任务
func (s *MultiTenantScheduler) SubmitTask(task *Task) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	// 验证租户存在
	if _, exists := s.tenants[task.TenantID]; !exists {
		return fmt.Errorf("tenant %s not found", task.TenantID)
	}
	
	// 验证租户资源上限
	tenant := s.tenants[task.TenantID]
	if task.GPUCount > tenant.MaxGPU {
		return fmt.Errorf("task requires %d GPUs but tenant max is %d",
			task.GPUCount, tenant.MaxGPU)
	}
	
	// 检查队列深度
	if s.taskQueue.Len() >= s.config.MaxQueueDepth {
		return fmt.Errorf("task queue full (max %d)", s.config.MaxQueueDepth)
	}
	
	task.State = "queued"
	task.SubmittedAt = time.Now()
	heap.Push(&s.taskQueue, task)
	
	return nil
}

// Schedule 执行调度决策
func (s *MultiTenantScheduler) Schedule(ctx context.Context) {
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.scheduleCycle()
		}
	}
}

func (s *MultiTenantScheduler) scheduleCycle() {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	if s.taskQueue.Len() == 0 {
		return
	}
	
	// 计算可用GPU
	availableGPUs := s.countAvailableGPUs()
	
	// 评估是否需要抢占
	topTask := s.taskQueue[0]
	requiredGPU := topTask.MinGPU
	
	if availableGPUs < requiredGPU {
		s.attemptPreemption(topTask, requiredGPU-availableGPUs)
	}
	
	// 重新计算可用GPU
	availableGPUs = s.countAvailableGPUs()
	
	// 调度队列
	var remaining []*Task
	for s.taskQueue.Len() > 0 && availableGPUs > 0 {
		task := heap.Pop(&s.taskQueue).(*Task)
		
		allocGPU := min(task.MaxGPU, availableGPUs)
		if allocGPU < task.MinGPU {
			// 资源不足，放回队列或重新调度
			if task.Preemptible {
				remaining = append(remaining, task)
				continue
			}
		}
		
		// 分配GPU
		allocatedGPUs := s.allocateGPUs(task, allocGPU)
		
		allocation := &Allocation{
			TaskID:    task.ID,
			GPUIDs:    allocatedGPUs,
			StartTime: time.Now(),
		}
		s.running[task.ID] = allocation
		
		for _, gpuID := range allocatedGPUs {
			s.gpuPool[gpuID].Status = "allocated"
		}
		
		availableGPUs -= len(allocatedGPUs)
		task.State = "running"
	}
	
	// 放回未被调度的任务
	for _, task := range remaining {
		heap.Push(&s.taskQueue, task)
	}
	
	// 更新指标
	s.updateMetrics()
}

func (s *MultiTenantScheduler) countAvailableGPUs() int {
	count := 0
	for _, gpu := range s.gpuPool {
		if gpu.Status == "free" {
			count++
		}
	}
	// 应用超分比
	return int(float64(count) * s.config.OvercommitRatio)
}

func (s *MultiTenantScheduler) attemptPreemption(task *Task, needGPU int) {
	// 找出可抢占的低优先级任务
	var preemptable []*Allocation
	
	for _, alloc := range s.running {
		if allocTask := s.findTaskByID(alloc.TaskID); allocTask != nil {
			if allocTask.Preemptible && allocTask.Priority < task.Priority {
				preemptable = append(preemptable, alloc)
			}
		}
	}
	
	// 按优先级排序，释放GPU
	releasedGPU := 0
	for _, alloc := range preemptable {
		if releasedGPU >= needGPU {
			break
		}
		
		// 释放GPU
		for _, gpuID := range alloc.GPUIDs {
			s.gpuPool[gpuID].Status = "free"
		}
		releasedGPU += len(alloc.GPUIDs)
		
		// 被抢占任务放回队列前端
		if preemptedTask := s.findTaskByID(alloc.TaskID); preemptedTask != nil {
			preemptedTask.State = "preempted"
			heap.Push(&s.taskQueue, preemptedTask)
		}
		
		delete(s.running, alloc.TaskID)
		s.metrics.PreemptionCount++
	}
}

func (s *MultiTenantScheduler) allocateGPUs(task *Task, count int) []string {
	var allocated []string
	for _, gpu := range s.gpuPool {
		if gpu.Status == "free" {
			allocated = append(allocated, gpu.ID)
			if len(allocated) >= count {
				break
			}
		}
	}
	return allocated
}

func (s *MultiTenantScheduler) findTaskByID(id string) *Task {
	for _, task := range s.taskQueue {
		if task.ID == id {
			return task
		}
	}
	return nil
}

func (s *MultiTenantScheduler) updateMetrics() {
	totalGPU := len(s.gpuPool)
	usedGPU := totalGPU - s.countAvailableGPUs()
	s.metrics.UtilizationRate = float64(usedGPU) / float64(totalGPU)
	
	// 更新各租户利用率
	for tenantID := range s.tenants {
		tenantUsage := 0
		for _, alloc := range s.running {
			task := s.findTaskByID(alloc.TaskID)
			if task != nil && task.TenantID == tenantID {
				tenantUsage += len(alloc.GPUIDs)
			}
		}
		if tenantMax := s.tenants[tenantID].MaxGPU; tenantMax > 0 {
			s.metrics.TenantUtilization[tenantID] = 
				float64(tenantUsage) / float64(tenantMax)
		}
	}
}

// GetMetrics 获取调度器指标
func (s *MultiTenantScheduler) GetMetrics() *SchedulerMetrics {
	s.mu.RLock()
	defer s.mu.RUnlock()
	
	metrics := *s.metrics
	return &metrics
}

// CalculateRevenue 计算收入分成
func (s *MultiTenantScheduler) CalculateRevenue(tenantID string, 
	gpuHours float64, ratePerHour float64) (float64, float64) {
	s.mu.RLock()
	tenant, exists := s.tenants[tenantID]
	s.mu.RUnlock()
	
	if !exists {
		return 0, 0
	}
	
	grossRevenue := gpuHours * ratePerHour
	nvidiaShare := grossRevenue * tenant.RevenueShare
	tenantShare := grossRevenue - nvidiaShare
	
	return nvidiaShare, tenantShare
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

"""
NVIDIA Compute Buyback & Revenue Share Model
Analysis of the CoreWeave precedent and AI Compute Partnership
"""

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import math


@dataclass
class ComputePartnership:
    """算力合作计划参数"""
    partner_name: str
    gpu_count: int
    gpu_model: str
    contract_years: int
    total_capex: float           # 总投资（亿美元）
    buyback_commitment: float     # 兜底额（亿美元）
    revenue_share_rate: float     # 收入分成比例
    initial_price_per_hour: float # 初始租赁单价（$/GPU-hour）
    
    def compute_metrics(self) -> Dict:
        """计算合作计划关键指标"""
        total_hours = self.gpu_count * 24 * 365 * self.contract_years
        total_revenue = total_hours * self.initial_price_per_hour / 1e8
        
        return {
            "total_gpu_hours_e9": round(total_hours / 1e9, 2),
            "estimated_total_revenue_b": round(total_revenue, 2),
            "nvidia_share_b": round(total_revenue * self.revenue_share_rate, 2),
            "partner_share_b": round(total_revenue * (1 - self.revenue_share_rate), 2),
            "buyback_ratio": round(self.buyback_commitment / total_revenue, 3),
            "break_even_utilization": round(
                self.buyback_commitment / total_revenue * 100, 1
            ),
        }


# CoreWeave协议分析
coreweave = ComputePartnership(
    partner_name="CoreWeave",
    gpu_count=500000,            # 假设~50万卡等效规模
    gpu_model="H100/B200混合",
    contract_years=7,            # 2025-2032
    total_capex=100,             # 100亿美元总投资
    buyback_commitment=63,       # 63亿美元兜底
    revenue_share_rate=0.15,     # 15%收入分成
    initial_price_per_hour=2.35, # $2.35/GPU-hour（2026年H100市场价）
)


# 新版AI Compute Partnership分析
class AIComputePartnershipAnalyzer:
    """
    AI Compute Partnership Program 分析器
    计算收入分成模式和信用支持的经济效益
    """
    
    def __init__(self, partnership: ComputePartnership):
        self.partnership = partnership
        self.scenarios: Dict[str, Dict] = {}
    
    def project_revenue(self, utilization_scenarios: List[float]) -> Dict[str, Dict]:
        """
        预测不同利用率下的收入分成
        
        Args:
            utilization_scenarios: 利用率场景列表（0.0-1.0）
        
        Returns:
            Dict: 各场景下的收入预测
        """
        p = self.partnership
        total_gpu_hours = p.gpu_count * 24 * 365 * p.contract_years
        
        results = {}
        for util in utilization_scenarios:
            utilized_hours = total_gpu_hours * util
            gross_revenue = utilized_hours * p.initial_price_per_hour
            
            # 英伟达分成收入
            nvidia_share = gross_revenue * p.revenue_share_rate
            
            # 云服务商收入
            partner_revenue = gross_revenue - nvidia_share
            
            # 若低于兜底线，英伟达需补差
            if nvidia_share < p.buyback_commitment * 1e8:
                shortfall = p.buyback_commitment * 1e8 - nvidia_share
                nvidia_share = p.buyback_commitment * 1e8
            else:
                shortfall = 0
            
            results[f"util_{util:.0%}"] = {
                "utilization": f"{util:.0%}",
                "gpu_hours": round(utilized_hours / 1e9, 2),
                "gross_revenue_b": round(gross_revenue / 1e8, 2),
                "nvidia_share_b": round(nvidia_share / 1e8, 2),
                "partner_revenue_b": round(partner_revenue / 1e8, 2),
                "buyback_shortfall_b": round(shortfall / 1e8, 2),
                "nvidia_effective_rate": f"{p.revenue_share_rate:.0%}" if shortfall == 0 
                    else f"{nvidia_share / gross_revenue:.1%}",
            }
        
        return results
    
    def compare_with_traditional_model(self, utilization: float) -> Dict:
        """
        比较新模式与传统硬件销售模式的收益
        
        Args:
            utilization: 预期利用率
        
        Returns:
            Dict: 两种模式的收益对比
        """
        p = self.partnership
        
        # 传统模式：一次性芯片销售
        traditional_revenue = p.total_capex  # 全额一次性收入
        
        # 新模式：硬件销售 + 持续性收入
        total_gpu_hours = p.gpu_count * 24 * 365 * p.contract_years
        utilized_hours = total_gpu_hours * utilization
        gross_revenue = utilized_hours * p.initial_price_per_hour
        
        # 假设硬件售价占总CapEx的60%
        hardware_revenue = p.total_capex * 0.6
        recurring_revenue = gross_revenue * p.revenue_share_rate
        new_model_revenue = hardware_revenue + recurring_revenue / 1e8
        
        return {
            "traditional_one_time_b": traditional_revenue,
            "new_model_hardware_b": round(hardware_revenue, 1),
            "new_model_recurring_b": round(recurring_revenue / 1e8, 2),
            "new_model_total_b": round(new_model_revenue, 2),
            "revenue_improvement": f"+{((new_model_revenue / traditional_revenue) - 1) * 100:.1f}%",
            "recurring_ratio": f"{recurring_revenue / (new_model_revenue * 1e8) * 100:.1f}%",
        }
    
    def risk_analysis(self) -> Dict:
        """
        风险分析：评估英伟达的潜在负债
        """
        p = self.partnership
        
        # 最差情景：利用率=0，全额兜底
        worst_case_loss = p.buyback_commitment
        
        # 基准情景：利用率=60%
        base_revenue = (p.gpu_count * 24 * 365 * p.contract_years * 0.6 
                        * p.initial_price_per_hour * p.revenue_share_rate / 1e8)
        base_loss = max(0, p.buyback_commitment - base_revenue)
        
        # 乐观情景：利用率=90%
        opt_revenue = (p.gpu_count * 24 * 365 * p.contract_years * 0.9 
                       * p.initial_price_per_hour * p.revenue_share_rate / 1e8)
        opt_loss = max(0, p.buyback_commitment - opt_revenue)
        
        return {
            "worst_case_loss_b": round(worst_case_loss, 1),
            "base_case_loss_b": round(base_loss, 2),
            "optimistic_case_loss_b": round(opt_loss, 2),
            "max_downside_ratio": f"{worst_case_loss / p.total_capex:.1%}",
            "break_even_util": p.compute_metrics()["break_even_utilization"],
        }


# 执行分析
analyzer = AIComputePartnershipAnalyzer(coreweave)

print("=" * 60)
print("CoreWeave 合作计划分析（英伟达AI Compute Partnership先驱）")
print("=" * 60)

metrics = coreweave.compute_metrics()
print(f"\n基础参数:")
print(f"  GPU规模:       {coreweave.gpu_count:,} 卡等效")
print(f"  合同期限:      {coreweave.contract_years} 年")
print(f"  总收入分成率:  {coreweave.revenue_share_rate:.0%}")
print(f"  兜底承诺:      ${coreweave.buyback_commitment}B")
print(f"\n关键指标:")
for k, v in metrics.items():
    print(f"  {k}: {v}")

print(f"\n不同利用率下的收入预测:")
scenarios = analyzer.project_revenue([0.3, 0.5, 0.7, 0.9])
for util, data in scenarios.items():
    print(f"  {data['utilization']}: "
          f"总营收${data['gross_revenue_b']}B → "
          f"NVIDIA分成${data['nvidia_share_b']}B")

print(f"\n与传统硬件销售模式对比 (60%利用率):")
comparison = analyzer.compare_with_traditional_model(0.6)
for k, v in comparison.items():
    print(f"  {k}: {v}")

print(f"\n风险分析:")
risk = analyzer.risk_analysis()
for k, v in risk.items():
    print(f"  {k}: {v}")