# 核心模块伪代码：空间编码器 + 时间编码器 + 融合头
class CarbonIntensityPredictor(nn.Module):
    def __init__(self, num_nodes, node_feat_dim, hist_len, pred_len):
        super().__init__()
        # 空间分支：图注意力网络捕捉电网拓扑传导
        self.spatial_encoder = GraphAttentionNet(
            in_dim=node_feat_dim,
            hid_dim=128,
            num_heads=4,
            num_layers=3,
            adj_matrix=grid_topology_adj  # 电网拓扑邻接矩阵
        )
        # 时间分支：TFT处理多尺度时序与静态协变量
        self.temporal_encoder = TemporalFusionTransformer(
            input_size=node_feat_dim + 128,  # 原始特征 + 空间嵌入
            hidden_size=256,
            lstm_layers=2,
            attention_heads=8,
            dropout=0.1,
            output_size=pred_len,
            static_covariates=['zone_type', 'renewable_capacity_ratio'],
            known_future_inputs=['weather_forecast', 'market_clearing'],
            observed_past_inputs=['historical_ci', 'load', 'thermal_output']
        )
        # 融合预测头：量化回归输出预测区间
        self.quantile_head = QuantileRegressionHead(
            in_dim=256, quantiles=[0.1, 0.5, 0.9]  # P10/P50/P90
        )

    def forward(self, x_hist, x_future, static_feat, adj):
        # x_hist: [B, N, L_hist, D], x_future: [B, N, L_future, D_known]
        spatial_emb = self.spatial_encoder(x_hist[:, :, -1, :], adj)  # 取最新时刻节点特征
        # 拼接空间嵌入到时序特征
        x_hist_enriched = torch.cat([x_hist, spatial_emb.unsqueeze(2).repeat(1,1,x_hist.size(2),1)], dim=-1)
        x_future_enriched = torch.cat([x_future, spatial_emb.unsqueeze(2).repeat(1,1,x_future.size(2),1)], dim=-1)
        temporal_out = self.temporal_encoder(x_hist_enriched, x_future_enriched, static_feat)
        return self.quantile_head(temporal_out)  # [B, N, L_pred, 3]

# PD-PPO 核心更新逻辑
class PDPPOAgent:
    def __init__(self, state_dim, action_dim, constraint_dim):
        self.actor = PolicyNet(state_dim, action_dim)
        self.critic = ValueNet(state_dim)
        self.cost_critic = ValueNet(state_dim)  # 约束价值网络
        self.lagrange_multipliers = nn.Parameter(torch.zeros(constraint_dim))
        
    def update(self, batch):
        states, actions, rewards, costs, next_states, dones = batch
        
        # 1. 计算优势函数 (GAE)
        advantages = self.compute_gae(rewards, states, next_states, dones)
        cost_advantages = self.compute_gae(costs, states, next_states, dones)
        
        # 2. 策略损失 (带约束的PPO-Clip)
        ratio = torch.exp(new_log_prob - old_log_prob)
        surr1 = ratio * advantages
        surr2 = torch.clamp(ratio, 1-clip_eps, 1+clip_eps) * advantages
        policy_loss = -torch.min(surr1, surr2).mean()
        
        # 3. 约束损失：拉格朗日对偶
        lagrangian = (self.lagrange_multipliers * cost_advantages).sum()
        total_loss = policy_loss + lagrangian + value_loss + cost_value_loss
        
        # 4. 对偶变量更新 (梯度上升)
        with torch.no_grad():
            constraint_violation = cost_advantages.mean(dim=0) - constraint_threshold
            self.lagrange_multipliers.data = torch.clamp(
                self.lagrange_multipliers + lr_dual * constraint_violation, min=0
            )
        
        return total_loss

┌─────────────────────────────────────────────────────────────┐
│                    应用接入层 (API Gateway)                    │
│  请求分类/画像标记 | SLA标签注入 | 熔断降级策略 | 观测埋点      │
└─────────────────────────────────────────────────────────────┘
                              ↓ gRPC/HTTP2
┌─────────────────────────────────────────────────────────────┐
│                  碳感知调度中控层 (Scheduler Core)             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐           │
│  │ 碳强度预测服务 │  │ 决策推理引擎  │  │ 状态感知模块  │           │
│  │ (ST-GNN+TFT) │→│ (PD-PPO)    │←│ (集群/队列/模型)│           │
│  └─────────────┘  └─────────────┘  └─────────────┘           │
│         ↑                ↑                ↑                    │
│         └────────────────┴────────────────┘                    │
│                    配置下发 / 模型热更新                          │
└─────────────────────────────────────────────────────────────┘
                              ↓ xDS/Envoy
┌─────────────────────────────────────────────────────────────┐
│                    模型服务层 (Model Serving Fleet)            │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐             │
│  │ Tiny-1.5B│ │ Small-7B│ │ Med-70B │ │Large-1.8T│  多区域部署 │
│  │  K8s HPA │ │  K8s HPA│ │  K8s HPA│ │  K8s HPA │  碳感知扩缩容│
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘             │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                    基础设施层 (Carbon-Aware Infra)             │
│  绿电直供数据中心 | 碳感知K8s调度器 | GPU功耗遥测 | 电力市场接口   │
└─────────────────────────────────────────────────────────────┘

// pkg/scheduler/carbonaware/score.go
func (p *CarbonAwarePlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    // 1. 获取节点实时碳强度 (来自Node Annotation/MetricServer)
    ci := getNodeCarbonIntensity(nodeName)  // gCO2/kWh
    
    // 2. 获取Pod请求的模型规模与质量要求
    modelSize := pod.Annotations["model-size"]  // tiny/small/medium/large
    qosClass := pod.Annotations["qos-class"]    // high/medium/low
    
    // 3. 计算碳效能得分 (0-100)
    // 得分 = 基础分(50) - 碳惩罚 + 质量匹配奖励 - 延迟惩罚
    baseScore := int64(50)
    carbonPenalty := int64(ci / 10)  // 每100 gCO2/kWh 扣10分
    qualityBonus := qualityMatchScore(modelSize, qosClass)  // 0~30
    latencyPenalty := latencyPenaltyScore(nodeName, pod)    // 0~20
    
    finalScore := baseScore - carbonPenalty + qualityBonus - latencyPenalty
    return clamp(finalScore, 0, 100), nil
}

# 伪代码：算子碳强度在线校准器
class OperatorCarbonProfiler:
    def __init__(self, hw_type: str):
        self.hw_type = hw_type
        self.surrogate_models = self._load_surrogate_models()  # 预训练代理模型
        self.pmu_collector = PMUCollector(hw_type)           # eBPF/Perf/DCGM采集
        self.calibration_buffer = RingBuffer(capacity=10000) # 存储(特征, 实测功耗)
        
    def profile_step(self, op_name: str, input_shape: Tuple, dtype: str):
        # 1. 采集PMU特征向量
        pmu_feats = self.pmu_collector.collect(op_name, input_shape, dtype)
        # 2. 代理模型预测
        pred_power = self.surrogate_models[op_name].predict(pmu_feats)
        # 3. 实测功耗 (通过IPMI/Redfish/DCGM获取整机功耗, 扣除静态功耗后归因)
        actual_power = self._attribute_power(op_name)
        # 4. 计算碳强度
        ci_grid = CarbonIntensityService.get_realtime_ci(self.zone_id)
        ci_op = (actual_power * ci_grid) / self._calc_throughput(op_name, input_shape)
        # 5. 缓存用于增量学习
        self.calibration_buffer.push((pmu_feats, actual_power))
        return ci_op
    
    def incremental_update(self):
        if len(self.calibration_buffer) > 500:
            X, y = self.calibration_buffer.sample_all()
            self.surrogate_models.partial_fit(X, y)  # 在线更新
            self.calibration_buffer.clear()

graph LR
    A[在线推理流量] --> B(数据采集层<br/>请求画像/碳强度/硬件遥测/决策日志)
    B --> C{特征工程与标注层<br/>自动化标注: 碳排放/质量/延迟}
    C --> D[模型训练与评估层<br/>碳预测模型/决策策略/功耗代理模型]
    D --> E[影子模式验证<br/>Shadow Mode: 仅记录决策不执行]
    E --> F[金丝雀发布<br/>Canary: 5%->100% 流量]
    F --> A
    D -.-> G[模型注册中心<br/>版本管理/回滚/元数据]
    G --> D

# Shadow Validation Config
shadow_validation:
  enabled: true
  traffic_mirror_ratio: 1.0  # 100% 镜像流量
  candidate_policies:
    - name: "ppo_v2.1_carbon_aware"
      model_version: "ci_pred_v3.2"
      decision_model: "pdppo_actor_v15"
    - name: "baseline_round_robin"
  metrics:
    - carbon_per_request: {threshold: -5%, direction: "lower"}  # 相对基线降5%
    - p99_latency: {threshold: +10%, direction: "lower"}       # 不劣化10%
    - quality_score: {threshold: -0.05, direction: "higher"}   # 不劣化0.05分
  duration: "7d"
  auto_promote: true
  rollback_on_alert: true

// pkg/scheduler/multitenant/carbon_preemption.go
func (p *MultiTenantCarbonPlugin) PostFilter(ctx context.Context, state *framework.CycleState, pod *v1.Pod, m framework.NodeToStatusMap) (*framework.PostFilterResult, *framework.Status) {
    tenantID := getTenantID(pod)
    carbonBudget := CarbonBudgetStore.GetRemaining(tenantID) // 剩余碳预算 gCO2
    estimatedCarbon := EstimateRequestCarbon(pod)             // 预估本次请求碳排放
    
    // 1. 预算充足，正常调度
    if carbonBudget >= estimatedCarbon {
        return nil, nil
    }
    
    // 2. 预算不足，尝试抢占低优先级租户的 Pod
    victims := p.findPreemptionVictims(ctx, pod, tenantID, estimatedCarbon - carbonBudget)
    if len(victims) > 0 {
        return &framework.PostFilterResult{NominatedPods: victims}, nil
    }
    
    // 3. 无法抢占，触发降级/排队/拒绝策略
    return p.handleBudgetExhaustion(pod, tenantID)
}

func (p *MultiTenantCarbonPlugin) findPreemptionVictims(ctx, pod, tenantID, carbonDeficit) []*v1.Pod {
    // 选取碳效能最低、优先级最低、可抢占的 Pod
    // 碳效能 = 请求质量价值 / 碳排放
    // 优先抢占: 低价值/高碳 的推理任务 (如: 批量离线Embedding生成)
    return selectVictimsByCarbonEfficiency(carbonDeficit)
}

# ChaosMesh Experiment: Carbon Prediction Failure
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: carbon-predictor-outage
spec:
  action: pod-failure
  mode: one
  duration: "10m"
  selector:
    namespaces: ["carbon-system"]
    labelSelectors:
      app: "ci-predictor"
  scheduler:
    cron: "@every 2h"  # 定期演练
---
# 验证指标 (自动化断言)
assertions:
  - metric: "scheduler_decision_latency_p99"
    operator: "lt"
    value: "200ms"  # 降级策略决策延迟不超200ms
  - metric: "carbon_emission_rate"
    operator: "lt"
    value: "baseline * 1.2"  # 碳排放不超基线1.2倍
  - metric: "request_success_rate"
    operator: "gt"
    value: "0.999"

ci_predictor:
  model:
    arch: "STGNN_TFT_Hybrid"
    version: "v3.2.1"
    update_strategy: "incremental_15min"
    prediction_horizons: ["15m", "1h", "4h", "24h"]
    quantiles: [0.1, 0.5, 0.9]
  data_sources:
    scada: {endpoint: "grpc://scada.grid.internal:50051", refresh: "4s"}
    weather: {provider: "ecmwf", resolution: "3km", refresh: "1h"}
    market: {provider: "power_exchange", refresh: "15m"}
  serving:
    replicas: 6
    resources: {cpu: "8", memory: "32Gi", gpu: "0"}
    autoscaling: {min: 3, max: 20, target_cpu: 60%}
  monitoring:
    mape_alert_threshold: 0.15
    latency_p99_threshold_ms: 50

pdppo_agent:
  network:
    hidden_dims: [512, 256, 128]
    activation: "gelu"
    lstm_layers: 1
  training:
    lr_actor: 3e-4
    lr_critic: 1e-3
    lr_dual: 1e-2
    gamma: 0.99
    gae_lambda: 0.95
    clip_eps: 0.2
    entropy_coef: 0.01
    max_grad_norm: 0.5
    ppo_epochs: 10
    batch_size: 4096
  constraints:
    latency_sla_ms: 1500
    quality_threshold: 7.5
    carbon_budget_daily_kg: 2000
  hysteresis:
    switch_threshold_pct: 0.15
    hold_duration_min: 30

# 根据电网实时状态动态调整碳权重 alpha
def calculate_dynamic_alpha(ci_current, ci_forecast_4h, grid_stress_level):
    """
    ci_current: 当前碳强度
    ci_forecast_4h: 4小时预测均值
    grid_stress_level: 0=宽松, 1=紧张, 2=严重 (来自电力调度指令)
    """
    base_alpha = 1.0
    # 1. 碳强度绝对值驱动：高碳时段加大权重
    if ci_current > 600: base_alpha *= 1.5
    elif ci_current > 400: base_alpha *= 1.2
    # 2. 趋势驱动：未来4小时将转低碳，当前可适当放宽（利用时间弹性）
    if ci_forecast_4h < ci_current * 0.7: base_alpha *= 0.8
    # 3. 电网应急响应：最高优先级
    if grid_stress_level == 2: base_alpha *= 3.0  # 强制低碳
    elif grid_stress_level == 1: base_alpha *= 1.5
    return min(max(base_alpha, 0.5), 5.0)  # 限幅 [0.5, 5.0]