商汤算电协同Agent与TPW新指标：从PUE到Tokens Per Watt的AIDC系统级算效革命深度解析

// TPW计算核心结构
package tpw

import (
	"math"
	"time"
)

type TPWMetrics struct {
	TotalTokens     int64           // 有效Token总产出
	PowerConsumed   float64         // 总耗电量（kWh）
	ElectricityCost float64         // 总电费（元）
	TimeWindow      time.Duration   // 统计时间窗口
}

// TPWByCost: 单位电力成本的Token产出
func (m *TPWMetrics) TPWByCost() float64 {
	if m.ElectricityCost == 0 {
		return 0
	}
	return float64(m.TotalTokens) / m.ElectricityCost
}

// TPWByEnergy: 单位电量的Token产出
func (m *TPWMetrics) TPWByEnergy() float64 {
	if m.PowerConsumed == 0 {
		return 0
	}
	return float64(m.TotalTokens) / m.PowerConsumed
}

// 时间序列TPW追踪
type TPWTimeSeries struct {
	Points []TPWPoint
}

type TPWPoint struct {
	Timestamp time.Time
	TPWValue  float64
	LoadLevel float64 // 负载水平 0-1
	PricePhase string // 电价时段: peak/flat/valley
}

// 计算加权TPW，按电价时段加权
func (ts *TPWTimeSeries) WeightedTPW() float64 {
	if len(ts.Points) == 0 {
		return 0
	}
	var totalWeighted float64
	var totalWeight float64
	for _, p := range ts.Points {
		weight := 1.0
		switch p.PricePhase {
		case "peak":
			weight = 3.0 // 峰值电价权重高
		case "flat":
			weight = 1.5
		case "valley":
			weight = 1.0
		}
		totalWeighted += p.TPWValue * weight
		totalWeight += weight
	}
	return totalWeighted / totalWeight
}

"""
八级数据穿透体系 - 数据采集与聚合层
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import numpy as np
from enum import Enum

class MetricType(Enum):
    POWER = "power"          # 功率(W)
    TEMPERATURE = "temp"     # 温度(°C)
    UTILIZATION = "util"     # 利用率(%)
    LATENCY = "latency"      # 延迟(ms)
    THROUGHPUT = "throughput" # 吞吐量(tokens/s)

@dataclass
class MetricPoint:
    timestamp: datetime
    metric_type: MetricType
    value: float
    unit: str

@dataclass
class JobMetrics:
    """Job级 - 最细粒度"""
    job_id: str
    model_name: str
    gpu_count: int
    power_consumption: float  # W
    token_throughput: float   # tokens/s
    gpu_util: float
    def tpw_instant(self) -> float:
        """瞬时TPW：每秒Token产出 / 功率(W)"""
        return self.token_throughput / max(self.power_consumption, 1)

@dataclass
class PodMetrics:
    """Pod级 - 一组同任务Job"""
    pod_id: str
    jobs: List[JobMetrics] = field(default_factory=list)
    @property
    def total_power(self) -> float:
        return sum(j.power_consumption for j in self.jobs)
    @property
    def total_throughput(self) -> float:
        return sum(j.token_throughput for j in self.jobs)

@dataclass
class NodeMetrics:
    """Node级 - 一台物理服务器"""
    node_id: str
    gpu_model: str
    gpu_count: int
    pods: List[PodMetrics] = field(default_factory=list)
    node_power: float  # 包含CPU/内存/网络等非GPU功耗
    @property
    def total_power(self) -> float:
        return self.node_power + sum(p.total_power for p in self.pods)

@dataclass
class ServerMetrics:
    """Server级 - 机架服务器组"""
    server_id: str
    nodes: List[NodeMetrics] = field(default_factory=list)

@dataclass
class CabinetMetrics:
    """机柜级"""
    cabinet_id: str
    servers: List[ServerMetrics] = field(default_factory=list)

@dataclass
class RoomMetrics:
    """包间级"""
    room_id: str
    cabinets: List[CabinetMetrics] = field(default_factory=list)
    room_cooling_power: float  # 该包间制冷功耗

@dataclass
class ClusterMetrics:
    """集群级"""
    cluster_id: str
    rooms: List[RoomMetrics] = field(default_factory=list)

@dataclass
class CampusMetrics:
    """园区/楼栋级 - 最粗粒度"""
    campus_id: str
    clusters: List[ClusterMetrics] = field(default_factory=list)
    total_incoming_power: float  # 市电总入口
    solar_generation: float      # 光伏发电量
    battery_soc: float           # 储能SOC (0-1)

"""
五大算法决策链实现
"""
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
import joblib

class ElectricityPricePredictor:
    """决策链1: 负荷→电价 - 电价预测模型"""
    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=300, 
            max_depth=6, 
            learning_rate=0.05
        )
        self.scaler = StandardScaler()
    
    def _extract_features(self, historical_load: np.ndarray, 
                          hour: int, day_of_week: int, 
                          is_holiday: bool) -> np.ndarray:
        """特征工程：从历史负荷和时段信息提取电价预测特征"""
        features = []
        # 过去24小时平均负荷
        features.append(np.mean(historical_load[-24:]))
        # 过去1小时负荷变化率
        features.append(historical_load[-1] - historical_load[-2] 
                       if len(historical_load) >= 2 else 0)
        # 过去7天同期负荷均值
        features.append(np.mean(historical_load[-168::24]) 
                       if len(historical_load) >= 168 else 0)
        # 时间特征
        features.append(np.sin(2 * np.pi * hour / 24))
        features.append(np.cos(2 * np.pi * hour / 24))
        features.append(day_of_week)
        features.append(1 if is_holiday else 0)
        # 气温特征（影响空调负荷和电价）
        features.append(0)  # placeholder for temperature
        return np.array(features).reshape(1, -1)
    
    def predict_price(self, load_series: np.ndarray, 
                      hour: int, dow: int, holiday: bool) -> float:
        """预测下一个时段的电价（元/kWh）"""
        features = self._extract_features(load_series, hour, dow, holiday)
        features_scaled = self.scaler.transform(features)
        return self.model.predict(features_scaled)[0]


class BatteryOptimizer:
    """决策链2: 电价→储能 - 储能充放电策略优化"""
    def __init__(self, capacity_kwh: float, max_power_kw: float, 
                 efficiency: float = 0.92):
        self.capacity = capacity_kwh
        self.max_power = max_power_kw
        self.efficiency = efficiency  # 充放电效率
    
    def optimize_schedule(self, price_forecast: np.ndarray, 
                          current_soc: float, 
                          hours_ahead: int = 24) -> np.ndarray:
        """
        基于电价预测优化储能充放电计划
        使用动态规划求解最优策略
        """
        n = hours_ahead
        # DP表: dp[t][soc_level] = 最小成本
        soc_levels = 20  # 离散化SOC等级
        dp = np.full((n + 1, soc_levels), float('inf'))
        action = np.zeros((n, soc_levels), dtype=int)
        # -1=放电, 0=闲置, 1=充电
        
        initial_soc_idx = int(current_soc * (soc_levels - 1))
        dp[0][initial_soc_idx] = 0
        
        for t in range(n):
            for s in range(soc_levels):
                if dp[t][s] == float('inf'):
                    continue
                current_energy = s / (soc_levels - 1) * self.capacity
                
                for a in [-1, 0, 1]:  # 放电/闲置/充电
                    if a == -1:  # 放电
                        discharge = self.max_power * 0.5  # 30分钟放电
                        new_energy = max(0, current_energy - discharge / self.efficiency)
                        cost = -discharge * price_forecast[t]  # 收益
                    elif a == 1:  # 充电
                        charge = self.max_power * 0.5
                        new_energy = min(self.capacity, current_energy + charge * self.efficiency)
                        cost = charge * price_forecast[t]
                    else:  # 闲置
                        new_energy = current_energy
                        cost = 0
                    
                    new_s = int(new_energy / self.capacity * (soc_levels - 1))
                    new_s = min(soc_levels - 1, max(0, new_s))
                    
                    if dp[t][s] + cost < dp[t + 1][new_s]:
                        dp[t + 1][new_s] = dp[t][s] + cost
                        action[t][s] = a
        
        # 回溯最优路径
        best_soc = np.argmin(dp[n])
        schedule = np.zeros(n, dtype=int)
        for t in range(n - 1, -1, -1):
            schedule[t] = action[t][best_soc]
            # 回溯前一步状态
            prev_energy = best_soc / (soc_levels - 1) * self.capacity
            if schedule[t] == -1:
                prev_energy += self.max_power * 0.5 / self.efficiency
            elif schedule[t] == 1:
                prev_energy -= self.max_power * 0.5 * self.efficiency
            prev_energy = max(0, min(self.capacity, prev_energy))
            best_soc = int(prev_energy / self.capacity * (soc_levels - 1))
            best_soc = min(soc_levels - 1, max(0, best_soc))
        
        return schedule


class TPWOptimizer:
    """
    决策链5: 电价→Token - TPW导向的算力调度优化
    核心逻辑：在电价低时执行高Token产出任务，电价高时执行高价值低Token任务
    """
    def __init__(self, token_value_fn=None):
        self.token_value_fn = token_value_fn or (lambda t: 1.0)
    
    def optimize_job_scheduling(self, jobs: List[Dict], 
                                price_forecast: np.ndarray,
                                gpu_capacity: int) -> Dict:
        """
        基于电价和TPW优化任务调度
        jobs: [{'id': str, 'tokens_per_hour': float, 'importance': float, 'deadline': int}]
        """
        n_jobs = len(jobs)
        n_slots = len(price_forecast)
        
        # 计算每个任务在每个时间段的TPW贡献
        # TPW = tokens_per_hour / (gpu_power * price)
        gpu_power = 700  # W, 单卡平均功耗
        
        # 贪心调度：按TPW排序分配低电价时段
        job_scores = []
        for j, job in enumerate(jobs):
            tph = job['tokens_per_hour']
            for t in range(min(n_slots, job.get('deadline', n_slots))):
                tpw = tph / (gpu_power * price_forecast[t] / 1000)
                job_scores.append((tpw * job['importance'], j, t))
        
        job_scores.sort(reverse=True)
        
        schedule = {j: [] for j in range(n_jobs)}
        slot_usage = np.zeros(n_slots)
        
        for score, j, t in job_scores:
            if slot_usage[t] < gpu_capacity:
                schedule[j].append(t)
                slot_usage[t] += 1
        
        return schedule

// 负载预测系统的核心实现
package prediction

import (
	"math"
	"time"
)

type LoadPredictor struct {
	history      []LoadPoint
	seasonalPeriod int // 季节性周期（小时）
}

type LoadPoint struct {
	Timestamp time.Time
	LoadKW    float64
	TokenRate float64 // tokens/s
	TPW       float64
}

// 基于三重指数平滑的负载预测
func (lp *LoadPredictor) TripleExponentialSmoothing(
	alpha, beta, gamma float64, 
	horizon int,
) []float64 {
	n := len(lp.history)
	if n == 0 {
		return nil
	}
	
	// 提取负载序列
	loads := make([]float64, n)
	for i, p := range lp.history {
		loads[i] = p.LoadKW
	}
	
	// 初始化
	seasonLen := lp.seasonalPeriod
	seasonal := make([]float64, seasonLen)
	for i := 0; i < seasonLen; i++ {
		seasonal[i] = loads[i] / lp.initialLevel()
	}
	
	level := lp.initialLevel()
	trend := lp.initialTrend()
	
	predictions := make([]float64, horizon)
	
	for i := 0; i < n+horizon; i++ {
		if i < n {
			// 更新
			lastLevel := level
			seasonIdx := i % seasonLen
			level = alpha*(loads[i]/seasonal[seasonIdx]) + (1-alpha)*(level+trend)
			trend = beta*(level-lastLevel) + (1-beta)*trend
			seasonal[seasonIdx] = gamma*(loads[i]/level) + (1-gamma)*seasonal[seasonIdx]
		} else {
			// 预测
			idx := i - n
			seasonIdx := i % seasonLen
			predictions[idx] = (level + float64(idx+1)*trend) * seasonal[seasonIdx]
		}
	}
	
	return predictions
}

func (lp *LoadPredictor) initialLevel() float64 {
	// 取第一个季节性周期的均值
	n := len(lp.history)
	if n < lp.seasonalPeriod {
		return 0
	}
	var sum float64
	for i := 0; i < lp.seasonalPeriod; i++ {
		sum += lp.history[i].LoadKW
	}
	return sum / float64(lp.seasonalPeriod)
}

func (lp *LoadPredictor) initialTrend() float64 {
	n := len(lp.history)
	if n < 2*lp.seasonalPeriod {
		return 0
	}
	var sum1, sum2 float64
	for i := 0; i < lp.seasonalPeriod; i++ {
		sum1 += lp.history[i].LoadKW
		sum2 += lp.history[i+lp.seasonalPeriod].LoadKW
	}
	return (sum2 - sum1) / float64(lp.seasonalPeriod*lp.seasonalPeriod)
}

"""
TPW驱动的算力调度决策引擎
"""
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import heapq
from datetime import datetime, timedelta

class TaskPriority(Enum):
    CRITICAL = 0   # 实时推理，必须立即执行
    HIGH = 1       # 在线服务，可容忍秒级延迟
    MEDIUM = 2     # 离线批处理，可容忍小时级延迟
    LOW = 3        # 训练任务，可灵活调度

@dataclass
class Task:
    task_id: str
    priority: TaskPriority
    gpu_required: int
    estimated_tokens: int       # 预估Token产出
    estimated_duration_min: int # 预估执行时长（分钟）
    deadline: Optional[datetime] = None
    hourly_value: float = 1.0   # 每小时产出价值
    
    @property
    def gpu_hours(self) -> float:
        return self.gpu_required * self.estimated_duration_min / 60.0

class TPWAwareScheduler:
    """
    TPW感知调度器 - 核心决策引擎
    """
    def __init__(self, 
                 gpu_power_watt: float = 700,
                 idle_power_ratio: float = 0.3):
        self.gpu_power = gpu_power_watt
        self.idle_power = gpu_power_watt * idle_power_ratio
        self.task_queue: List[Task] = []
    
    def calculate_tpw(self, task: Task, 
                      electricity_price: float) -> float:
        """
        计算任务在给定电价下的TPW
        TPW = tokens / (power * duration * price)
        """
        power_kw = task.gpu_required * self.gpu_power / 1000.0
        duration_h = task.estimated_duration_min / 60.0
        energy_cost = power_kw * duration_h * electricity_price
        if energy_cost <= 0:
            return float('inf')
        return task.estimated_tokens / energy_cost
    
    def schedule_tasks(self, 
                       tasks: List[Task],
                       price_forecast: List[float],
                       available_gpus: int,
                       time_slot_min: int = 30) -> List[Tuple[str, int, int]]:
        """
        调度任务到最优时间段
        返回: [(task_id, start_slot, end_slot)]
        """
        n_slots = len(price_forecast)
        slot_gpu_usage = [0] * n_slots
        schedule = []
        
        # 按优先级排序，同优先级按TPW排序
        sorted_tasks = sorted(tasks, key=lambda t: (
            t.priority.value,
            -self.calculate_tpw(t, min(price_forecast))
        ))
        
        for task in sorted_tasks:
            slots_needed = max(1, task.estimated_duration_min // time_slot_min)
            best_tpw = -1
            best_start = -1
            
            for start in range(n_slots - slots_needed + 1):
                # 检查GPU容量
                if max(slot_gpu_usage[start:start + slots_needed]) + task.gpu_required > available_gpus:
                    continue
                
                # 检查deadline
                if task.deadline and start * time_slot_min > task.deadline.hour * 60:
                    continue
                
                # 计算该时间段平均TPW
                avg_price = sum(price_forecast[start:start + slots_needed]) / slots_needed
                tpw = self.calculate_tpw(task, avg_price)
                
                if tpw > best_tpw:
                    best_tpw = tpw
                    best_start = start
            
            if best_start >= 0:
                for s in range(best_start, best_start + slots_needed):
                    slot_gpu_usage[s] += task.gpu_required
                schedule.append((
                    task.task_id, 
                    best_start, 
                    best_start + slots_needed
                ))
        
        return schedule
    
    def estimate_tpw_improvement(self, 
                                 before_schedule: List[Tuple[str, int, int]],
                                 after_schedule: List[Tuple[str, int, int]],
                                 price_forecast: List[float]) -> float:
        """评估TPW优化效果"""
        def total_cost(schedule, tasks_map):
            cost = 0.0
            total_tokens = 0
            for task_id, start, end in schedule:
                task = tasks_map[task_id]
                duration_h = (end - start) * 0.5  # 半小时slot
                power_kw = task.gpu_required * self.gpu_power / 1000.0
                avg_price = sum(price_forecast[start:end]) / (end - start)
                cost += power_kw * duration_h * avg_price
                total_tokens += task.estimated_tokens
            return total_tokens / max(cost, 0.001)
        
        tasks_map = {t.task_id: t for t in self.task_queue}
        before_tpw = total_cost(before_schedule, tasks_map)
        after_tpw = total_cost(after_schedule, tasks_map)
        
        return (after_tpw - before_tpw) / before_tpw * 100  # 百分比