第八章：性能分析与调试

@torch.compile
def perception_model(image, lidar_points):
    # 图像特征提取
    image_features = self.image_backbone(image)

    # 使用 torch 操作替代 Python 控制流
    num_points = lidar_points.shape[0]
    max_points = 10000

    # 方案 1：使用 torch.where 避免条件分支
    indices = torch.arange(num_points, device=lidar_points.device)
    # 确定性采样（如每隔 N 个点采样）
    step = torch.maximum(torch.tensor(1), num_points // max_points)
    sampled_indices = indices[::step][:max_points]
    lidar_points = lidar_points[sampled_indices]

    # 方案 2：使用 padding/masking 处理变长输入
    # 将所有输入 pad 到固定大小，使用 mask 标记有效数据

    # 点云特征提取
    point_features = self.point_encoder(lidar_points)

    # 特征融合
    fused = torch.cat([image_features, point_features], dim=1)

    return self.detection_head(fused)

class RobotController:
    def __init__(self):
        self.perception_model = torch.compile(PerceptionNet())
        self.planning_model = torch.compile(PlanningNet())
        self.history_buffer = []

    def control_loop(self):
        while True:
            # 感知 - 使用 inference_mode 避免梯度
            with torch.inference_mode():
                sensor_data = self.get_sensor_data()
                perception_output = self.perception_model(sensor_data)

            # 历史记录 - 只保存必要的数据，detach 并 clone
            self.history_buffer.append({
                'timestamp': time.time(),
                'perception': perception_output.detach().clone().cpu(),  # 移到 CPU 节省显存
                'sensor': None  # 不保存原始传感器数据，或只保存摘要
            })

            # 规划
            if len(self.history_buffer) > 10:
                with torch.inference_mode():
                    # 将历史数据移回 GPU 进行推理
                    recent_history = torch.stack([
                        h['perception'].cuda() for h in self.history_buffer[-10:]
                    ])
                    plan = self.planning_model(recent_history)

            # 执行动作
            self.execute_action(plan.cpu().numpy())  # 转换为 numpy 避免保留张量

            # 积极清理
            if len(self.history_buffer) > 100:
                self.history_buffer = self.history_buffer[-50:]
                # 强制垃圾回收
                torch.cuda.empty_cache()

            # 定期监控
            if time.time() % 3600 == 0:  # 每小时
                print(f"Memory allocated: {torch.cuda.memory_allocated() / 1e9:.2f} GB")

import torch
import time
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
import pandas as pd

@dataclass
class BenchmarkResult:
    batch_size: int
    compile_mode: str
    cold_start_time: float
    warm_up_times: List[float]
    steady_state_times: List[float]
    peak_memory: float
    compile_time: float
    accuracy_loss: float

class CompileBenchmark:
    def __init__(self, model, test_data):
        self.model = model
        self.test_data = test_data
        self.results = []

    def run_benchmark(self):
        batch_sizes = [1, 4, 8, 16, 32]
        compile_modes = [
            None,  # Eager mode
            "default",
            "reduce-overhead",
            "max-autotune"
        ]

        for bs in batch_sizes:
            for mode in compile_modes:
                result = self.benchmark_single_config(bs, mode)
                self.results.append(result)

        return self.analyze_results()

    def benchmark_single_config(self, batch_size, compile_mode):
        # 准备模型
        model = self.model.clone()
        if compile_mode:
            model = torch.compile(model, mode=compile_mode)

        # 准备数据
        test_batch = self.prepare_batch(batch_size)

        # 1. 编译时间（首次运行）
        torch.cuda.synchronize()
        compile_start = time.time()
        with torch.inference_mode():
            _ = model(test_batch)
        torch.cuda.synchronize()
        compile_time = time.time() - compile_start

        # 2. 冷启动时间
        torch.cuda.empty_cache()
        torch.cuda.synchronize()
        cold_start = time.time()
        with torch.inference_mode():
            _ = model(test_batch)
        torch.cuda.synchronize()
        cold_start_time = time.time() - cold_start

        # 3. 预热阶段
        warm_up_times = []
        for _ in range(10):
            torch.cuda.synchronize()
            start = time.time()
            with torch.inference_mode():
                _ = model(test_batch)
            torch.cuda.synchronize()
            warm_up_times.append(time.time() - start)

        # 4. 稳态性能
        steady_state_times = []
        for _ in range(100):
            torch.cuda.synchronize()
            start = time.time()
            with torch.inference_mode():
                output = model(test_batch)
            torch.cuda.synchronize()
            steady_state_times.append(time.time() - start)

        # 5. 内存使用
        torch.cuda.reset_peak_memory_stats()
        with torch.inference_mode():
            _ = model(test_batch)
        peak_memory = torch.cuda.max_memory_allocated()

        # 6. 数值精度
        with torch.inference_mode():
            compiled_output = model(test_batch)
            eager_output = self.model(test_batch)
        accuracy_loss = self.compute_accuracy_loss(compiled_output, eager_output)

        return BenchmarkResult(
            batch_size=batch_size,
            compile_mode=compile_mode or "eager",
            cold_start_time=cold_start_time,
            warm_up_times=warm_up_times,
            steady_state_times=steady_state_times,
            peak_memory=peak_memory,
            compile_time=compile_time,
            accuracy_loss=accuracy_loss
        )

    def compute_accuracy_loss(self, output1, output2):
        # Cosine similarity
        cos_sim = torch.nn.functional.cosine_similarity(
            output1.flatten(), 
            output2.flatten(), 
            dim=0
        )
        return (1 - cos_sim).item()

    def analyze_results(self):
        df = pd.DataFrame([
            {
                'batch_size': r.batch_size,
                'mode': r.compile_mode,
                'p50_latency': np.percentile(r.steady_state_times, 50),
                'p95_latency': np.percentile(r.steady_state_times, 95),
                'p99_latency': np.percentile(r.steady_state_times, 99),
                'memory_mb': r.peak_memory / 1e6,
                'compile_time': r.compile_time,
                'accuracy_loss': r.accuracy_loss
            }
            for r in self.results
        ])

        # 计算加速比
        eager_times = df[df['mode'] == 'eager'].set_index('batch_size')['p50_latency']
        for mode in df['mode'].unique():
            if mode != 'eager':
                mode_times = df[df['mode'] == mode].set_index('batch_size')['p50_latency']
                speedup = eager_times / mode_times
                df.loc[df['mode'] == mode, 'speedup'] = speedup.values

        # 生成报告
        report = {
            'summary': df.groupby('mode').agg({
                'p50_latency': 'mean',
                'memory_mb': 'mean',
                'speedup': 'mean'
            }),
            'details': df,
            'recommendations': self.generate_recommendations(df)
        }

        return report

    def generate_recommendations(self, df):
        recommendations = []

        # 基于结果生成建议
        best_mode = df.groupby('mode')['speedup'].mean().idxmax()
        recommendations.append(f"推荐使用 {best_mode} 模式，平均加速 {df[df['mode']==best_mode]['speedup'].mean():.2f}x")

        # 检查数值精度
        if df['accuracy_loss'].max() > 1e-3:
            recommendations.append("警告：编译后数值精度损失较大，建议检查模型")

        # 内存使用建议
        memory_increase = (df[df['mode'] != 'eager']['memory_mb'].mean() - 
                          df[df['mode'] == 'eager']['memory_mb'].mean())
        if memory_increase > 100:
            recommendations.append(f"编译后内存增加 {memory_increase:.0f} MB，注意边缘设备部署")

        return recommendations

# 监控系统架构
"""
数据流：
Models → Metrics Collector → Time Series DB → Alerting System
                ↓                   ↓
           Stream Processor    Visualization Dashboard
                ↓
           Anomaly Detector → Auto Scaling/Degradation
"""

import asyncio
from datetime import datetime
from typing import Dict, Any
import aioredis
from prometheus_client import Counter, Histogram, Gauge
import numpy as np

class ProductionMonitor:
    def __init__(self):
        # Prometheus 指标
        self.latency_histogram = Histogram(
            'model_latency_seconds',
            'Model inference latency',
            ['model_name', 'model_version', 'device_id'],
            buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
        )

        self.throughput_gauge = Gauge(
            'model_throughput_fps',
            'Model throughput in FPS',
            ['model_name', 'device_id']
        )

        self.error_counter = Counter(
            'model_errors_total',
            'Total model errors',
            ['model_name', 'error_type']
        )

        self.gpu_memory_gauge = Gauge(
            'gpu_memory_usage_bytes',
            'GPU memory usage',
            ['device_id']
        )

        # Redis 用于实时流处理
        self.redis = None

        # 异常检测器
        self.anomaly_detectors = {}

    async def initialize(self):
        self.redis = await aioredis.create_redis_pool('redis://localhost')

    async def monitor_inference(self, model_name: str, model_version: str, 
                               device_id: str, func):
        """包装模型推理函数，添加监控"""

        async def monitored_func(*args, **kwargs):
            start_time = time.time()

            try:
                # 执行推理
                result = await func(*args, **kwargs)

                # 记录成功指标
                latency = time.time() - start_time
                self.latency_histogram.labels(
                    model_name=model_name,
                    model_version=model_version,
                    device_id=device_id
                ).observe(latency)

                # 实时流处理
                await self.stream_metrics({
                    'timestamp': datetime.utcnow().isoformat(),
                    'model': model_name,
                    'version': model_version,
                    'device': device_id,
                    'latency': latency,
                    'status': 'success'
                })

                # 异常检测
                if self.detect_anomaly(model_name, latency):
                    await self.handle_anomaly(model_name, latency)

                return result

            except Exception as e:
                # 记录错误
                self.error_counter.labels(
                    model_name=model_name,
                    error_type=type(e).__name__
                ).inc()

                # 触发降级
                await self.trigger_degradation(model_name, str(e))

                raise

        return monitored_func

    async def stream_metrics(self, metrics: Dict[str, Any]):
        """流式处理指标数据"""
        # 发送到 Redis Stream
        await self.redis.xadd(
            'metrics:stream',
            {'data': json.dumps(metrics)}
        )

        # 实时聚合（滑动窗口）
        window_key = f"window:{metrics['model']}:5min"
        await self.redis.zadd(
            window_key,
            metrics['timestamp'],
            metrics['latency']
        )

        # 清理旧数据
        cutoff = time.time() - 300  # 5 分钟窗口
        await self.redis.zremrangebyscore(window_key, 0, cutoff)

    def detect_anomaly(self, model_name: str, latency: float) -> bool:
        """基于统计的异常检测"""
        if model_name not in self.anomaly_detectors:
            self.anomaly_detectors[model_name] = OnlineAnomalyDetector()

        detector = self.anomaly_detectors[model_name]
        return detector.is_anomaly(latency)

    async def handle_anomaly(self, model_name: str, latency: float):
        """处理检测到的异常"""
        alert = {
            'severity': 'warning',
            'model': model_name,
            'message': f'Latency anomaly detected: {latency:.3f}s',
            'timestamp': datetime.utcnow().isoformat()
        }

        # 发送告警
        await self.send_alert(alert)

        # 自动降级逻辑
        if latency > 0.5:  # 严重延迟
            await self.trigger_degradation(model_name, f"High latency: {latency}")

    async def trigger_degradation(self, model_name: str, reason: str):
        """触发模型降级"""
        # 切换到备用模型或降低精度模式
        degradation_config = {
            'model': model_name,
            'action': 'switch_to_lite_model',
            'reason': reason,
            'timestamp': datetime.utcnow().isoformat()
        }

        await self.redis.publish('degradation:trigger', json.dumps(degradation_config))

        # 记录降级事件
        self.error_counter.labels(
            model_name=model_name,
            error_type='degradation'
        ).inc()

class OnlineAnomalyDetector:
    """在线异常检测器（基于 EWMA）"""

    def __init__(self, alpha=0.1, threshold=3):
        self.alpha = alpha
        self.threshold = threshold
        self.mean = None
        self.var = None

    def is_anomaly(self, value: float) -> bool:
        if self.mean is None:
            self.mean = value
            self.var = 0
            return False

        # 更新 EWMA
        self.mean = self.alpha * value + (1 - self.alpha) * self.mean
        self.var = self.alpha * (value - self.mean) ** 2 + (1 - self.alpha) * self.var

        # 检测异常
        std = np.sqrt(self.var)
        z_score = abs(value - self.mean) / (std + 1e-8)

        return z_score > self.threshold

# 使用示例
monitor = ProductionMonitor()

# 包装模型
@monitor.monitor_inference("perception_model", "v2.1", "cuda:0")
async def run_perception(image, lidar):
    # 实际推理代码
    return model(image, lidar)

# Grafana Dashboard 配置
dashboard_config = {
    "dashboard": {
        "title": "自动驾驶感知系统监控",
        "panels": [
            {
                "title": "模型延迟分布",
                "type": "heatmap",
                "targets": [{
                    "expr": "rate(model_latency_seconds_bucket[5m])"
                }]
            },
            {
                "title": "错误率",
                "type": "graph",
                "targets": [{
                    "expr": "rate(model_errors_total[1m])"
                }]
            },
            {
                "title": "GPU 内存使用",
                "type": "graph",
                "targets": [{
                    "expr": "gpu_memory_usage_bytes / 1e9"
                }]
            }
        ]
    }
}

import torch
from functools import lru_cache
import math

class CompiledModelCache:
    def __init__(self, model, cache_size=10):
        self.base_model = model
        self.cache = {}
        self.cache_size = cache_size
        self.hit_count = 0
        self.miss_count = 0

    def get_bucket_size(self, size, bucket_width=32):
        """将输入尺寸映射到桶"""
        # 向上取整到最近的 bucket_width 的倍数
        return math.ceil(size / bucket_width) * bucket_width

    def get_compiled_model(self, input_shape):
        """获取对应输入形状的编译模型"""
        # 方案 1：尺寸分桶
        bucket_shape = tuple(
            self.get_bucket_size(dim) if i > 0 else dim
            for i, dim in enumerate(input_shape)
        )

        cache_key = bucket_shape

        if cache_key in self.cache:
            self.hit_count += 1
            return self.cache[cache_key], bucket_shape

        self.miss_count += 1

        # 方案 2：使用动态形状
        if len(self.cache) >= self.cache_size:
            # 使用动态形状编译，支持范围内的所有尺寸
            compiled_model = torch.compile(
                self.base_model,
                dynamic=True,
                options={
                    "shape_padding": True,  # 启用形状填充
                    "assume_static_by_default": False
                }
            )
            # 清空缓存，只保留动态版本
            self.cache.clear()
            self.cache["dynamic"] = compiled_model
            return compiled_model, input_shape

        # 方案 3：为特定桶编译
        compiled_model = torch.compile(
            self.base_model,
            options={
                "max_autotune": False,  # 快速编译
                "epilogue_fusion": True,
                "aggressive_fusion": True
            }
        )

        self.cache[cache_key] = compiled_model
        return compiled_model, bucket_shape

    def forward(self, input_tensor):
        """智能前向传播"""
        input_shape = tuple(input_tensor.shape)
        compiled_model, target_shape = self.get_compiled_model(input_shape)

        # 如果需要，pad 输入到目标形状
        if input_shape != target_shape:
            padded_input = self.pad_to_shape(input_tensor, target_shape)
            output = compiled_model(padded_input)
            # 裁剪输出回原始尺寸
            output = self.crop_to_shape(output, input_shape)
        else:
            output = compiled_model(input_tensor)

        return output

    def pad_to_shape(self, tensor, target_shape):
        """将张量 pad 到目标形状"""
        padding = []
        for i in range(len(tensor.shape) - 1, -1, -1):
            diff = target_shape[i] - tensor.shape[i]
            padding.extend([0, diff])

        if any(p > 0 for p in padding):
            tensor = torch.nn.functional.pad(tensor, padding)

        return tensor

    def crop_to_shape(self, tensor, target_shape):
        """裁剪张量到目标形状"""
        slices = tuple(slice(0, dim) for dim in target_shape)
        return tensor[slices]

    def get_stats(self):
        """获取缓存统计"""
        total = self.hit_count + self.miss_count
        hit_rate = self.hit_count / total if total > 0 else 0

        return {
            "hit_rate": hit_rate,
            "miss_rate": 1 - hit_rate,
            "cache_size": len(self.cache),
            "total_requests": total
        }

# 高级策略：预编译常见尺寸
class PrecompiledModelCache(CompiledModelCache):
    def __init__(self, model, common_sizes):
        super().__init__(model)
        self.precompile(common_sizes)

    def precompile(self, sizes):
        """预编译常见尺寸"""
        print(f"预编译 {len(sizes)} 个常见尺寸...")

        for size in sizes:
            dummy_input = torch.randn(size)
            cache_key = tuple(size)

            compiled_model = torch.compile(
                self.base_model,
                options={"max_autotune": True}  # 预编译时可以用更激进的优化
            )

            # 触发编译
            with torch.no_grad():
                _ = compiled_model(dummy_input)

            self.cache[cache_key] = compiled_model

        print(f"预编译完成，缓存大小: {len(self.cache)}")

# 使用示例
model = YourModel()

# 分析历史数据找出常见尺寸
common_sizes = [
    (1, 3, 128, 128),
    (1, 3, 160, 160),
    (1, 3, 192, 192),
]

cached_model = PrecompiledModelCache(model, common_sizes)

# 生产使用
for input_data in data_stream:
    output = cached_model.forward(input_data)

    # 定期打印统计
    if step % 1000 == 0:
        stats = cached_model.get_stats()
        print(f"Cache hit rate: {stats['hit_rate']:.2%}")

import torch
import git
import json
from pathlib import Path
import subprocess
from scipy import stats
import pandas as pd

class PerformanceRegressionTest:
    def __init__(self, repo_path, model_path, test_data_path):
        self.repo = git.Repo(repo_path)
        self.model_path = model_path
        self.test_data_path = test_data_path
        self.baseline_db = Path("baselines.json")

    def run_regression_test(self, commit_hash=None):
        """运行性能回归测试"""
        # 1. 获取当前和基线性能
        current_perf = self.measure_performance(commit_hash)
        baseline_perf = self.get_baseline()

        # 2. 统计分析
        regression_report = self.analyze_regression(baseline_perf, current_perf)

        # 3. 生成报告
        report = self.generate_report(regression_report)

        # 4. CI/CD 决策
        return self.make_decision(regression_report)

    def measure_performance(self, commit_hash=None):
        """测量指定 commit 的性能"""
        if commit_hash:
            self.repo.git.checkout(commit_hash)

        # 加载模型
        model = torch.load(self.model_path)
        model = torch.compile(model)
        model.eval()

        # 加载测试数据
        test_data = torch.load(self.test_data_path)

        metrics = {
            'latencies': [],
            'memory_usage': [],
            'accuracy': [],
            'compile_time': None
        }

        # 测量编译时间
        start = time.time()
        with torch.no_grad():
            _ = model(test_data[0])
        metrics['compile_time'] = time.time() - start

        # 性能测试（多次运行获得统计数据）
        for i in range(100):
            # 预热
            if i < 10:
                with torch.no_grad():
                    _ = model(test_data[i % len(test_data)])
                continue

            batch = test_data[i % len(test_data)]

            # 测量延迟
            torch.cuda.synchronize()
            start = time.time()
            with torch.no_grad():
                output = model(batch)
            torch.cuda.synchronize()
            latency = time.time() - start
            metrics['latencies'].append(latency)

            # 测量内存
            metrics['memory_usage'].append(
                torch.cuda.memory_allocated() / 1e6  # MB
            )

            # 测量准确性（如果有 ground truth）
            if hasattr(batch, 'labels'):
                acc = self.compute_accuracy(output, batch.labels)
                metrics['accuracy'].append(acc)

        return metrics

    def analyze_regression(self, baseline, current):
        """分析性能回归"""
        analysis = {}

        # 1. 延迟分析
        baseline_latencies = np.array(baseline['latencies'])
        current_latencies = np.array(current['latencies'])

        # T 检验
        t_stat, p_value = stats.ttest_ind(
            baseline_latencies, 
            current_latencies,
            equal_var=False  # Welch's t-test
        )

        # 效应量（Cohen's d）
        pooled_std = np.sqrt(
            (np.std(baseline_latencies)**2 + np.std(current_latencies)**2) / 2
        )
        cohens_d = (np.mean(current_latencies) - np.mean(baseline_latencies)) / pooled_std

        analysis['latency'] = {
            'baseline_mean': np.mean(baseline_latencies),
            'current_mean': np.mean(current_latencies),
            'change_percent': (np.mean(current_latencies) - np.mean(baseline_latencies)) / np.mean(baseline_latencies) * 100,
            'p_value': p_value,
            'cohens_d': cohens_d,
            'significant': p_value < 0.05 and abs(cohens_d) > 0.2,  # 统计和实际显著性
            'regression': p_value < 0.05 and cohens_d > 0.2  # 性能退化
        }

        # 2. 内存分析
        analysis['memory'] = {
            'baseline_mean': np.mean(baseline['memory_usage']),
            'current_mean': np.mean(current['memory_usage']),
            'change_percent': (np.mean(current['memory_usage']) - np.mean(baseline['memory_usage'])) / np.mean(baseline['memory_usage']) * 100
        }

        # 3. 编译时间分析
        analysis['compile_time'] = {
            'baseline': baseline['compile_time'],
            'current': current['compile_time'],
            'change_percent': (current['compile_time'] - baseline['compile_time']) / baseline['compile_time'] * 100
        }

        # 4. 尾部延迟分析（P95, P99）
        analysis['tail_latency'] = {
            'p95_baseline': np.percentile(baseline_latencies, 95),
            'p95_current': np.percentile(current_latencies, 95),
            'p99_baseline': np.percentile(baseline_latencies, 99),
            'p99_current': np.percentile(current_latencies, 99),
        }

        return analysis

    def generate_report(self, analysis):
        """生成 Markdown 报告"""
        report = f"""
# 性能回归测试报告

## 摘要

- **延迟变化**: {analysis['latency']['change_percent']:.2f}%
- **内存变化**: {analysis['memory']['change_percent']:.2f}%
- **统计显著性**: p={analysis['latency']['p_value']:.4f}
- **结论**: {'⚠️ 检测到性能回归' if analysis['latency']['regression'] else '✅ 性能正常'}

## 详细分析

### 推理延迟
| 指标 | 基线 | 当前 | 变化 |

| 指标 | 基线 | 当前 | 变化 |
|------|------|------|------|
| 平均值 | {analysis['latency']['baseline_mean']*1000:.2f}ms | {analysis['latency']['current_mean']*1000:.2f}ms | {analysis['latency']['change_percent']:+.2f}% |
| P95 | {analysis['tail_latency']['p95_baseline']*1000:.2f}ms | {analysis['tail_latency']['p95_current']*1000:.2f}ms | - |
| P99 | {analysis['tail_latency']['p99_baseline']*1000:.2f}ms | {analysis['tail_latency']['p99_current']*1000:.2f}ms | - |

### 资源使用
| 指标 | 基线 | 当前 | 变化 |

| 指标 | 基线 | 当前 | 变化 |
|------|------|------|------|
| 内存 | {analysis['memory']['baseline_mean']:.1f}MB | {analysis['memory']['current_mean']:.1f}MB | {analysis['memory']['change_percent']:+.2f}% |
| 编译时间 | {analysis['compile_time']['baseline']:.2f}s | {analysis['compile_time']['current']:.2f}s | {analysis['compile_time']['change_percent']:+.2f}% |

### 统计分析

- **p-value**: {analysis['latency']['p_value']:.6f}
- **Cohen's d**: {analysis['latency']['cohens_d']:.3f}
- **解释**: {self.interpret_cohens_d(analysis['latency']['cohens_d'])}
"""

        # 保存报告
        with open("performance_report.md", "w") as f:
            f.write(report)

        return report

    def interpret_cohens_d(self, d):
        """解释 Cohen's d 效应量"""
        d = abs(d)
        if d < 0.2:
            return "可忽略的差异"
        elif d < 0.5:
            return "小效应"
        elif d < 0.8:
            return "中等效应"
        else:
            return "大效应"

    def make_decision(self, analysis):
        """CI/CD 决策"""
        # 定义阈值
        LATENCY_THRESHOLD = 5  # 5% 性能退化阈值
        MEMORY_THRESHOLD = 10  # 10% 内存增加阈值

        failures = []
        warnings = []

        # 检查性能回归
        if analysis['latency']['regression']:
            if analysis['latency']['change_percent'] > LATENCY_THRESHOLD:
                failures.append(f"性能回归: 延迟增加 {analysis['latency']['change_percent']:.1f}%")
            else:
                warnings.append(f"轻微性能回归: 延迟增加 {analysis['latency']['change_percent']:.1f}%")

        # 检查内存使用
        if analysis['memory']['change_percent'] > MEMORY_THRESHOLD:
            warnings.append(f"内存使用增加: {analysis['memory']['change_percent']:.1f}%")

        # 生成 CI 输出
        if failures:
            print("❌ 性能测试失败")
            for f in failures:
                print(f"  - {f}")
            return False
        elif warnings:
            print("⚠️ 性能测试通过（有警告）")
            for w in warnings:
                print(f"  - {w}")
            return True
        else:
            print("✅ 性能测试通过")
            return True

    def update_baseline(self, metrics):
        """更新性能基线"""
        baselines = {}
        if self.baseline_db.exists():
            with open(self.baseline_db) as f:
                baselines = json.load(f)

        baselines[self.repo.head.commit.hexsha] = {
            'timestamp': datetime.now().isoformat(),
            'metrics': metrics
        }

        with open(self.baseline_db, 'w') as f:
            json.dump(baselines, f, indent=2)

# GitHub Actions 集成
"""
name: Performance Regression Test

on:
  pull_request:
    branches: [ main ]

jobs:
  performance-test:
    runs-on: [self-hosted, gpu]

    steps:

    - uses: actions/checkout@v2
      with:
        fetch-depth: 0

    - name: Run Performance Test
      run: |
        python perf_regression_test.py --commit ${{ github.sha }}

    - name: Upload Report
      uses: actions/upload-artifact@v2
      with:
        name: performance-report
        path: performance_report.md

    - name: Comment PR
      uses: actions/github-script@v6
      with:
        script: |
          const fs = require('fs');
          const report = fs.readFileSync('performance_report.md', 'utf8');
          github.rest.issues.createComment({
            issue_number: context.issue.number,
            owner: context.repo.owner,
            repo: context.repo.repo,
            body: report
          });
"""