feat(performance): 批量执行合并报告实现（数据持久化 + 聚合算法 + API）

- PerformanceBatchExecution / PerformanceProjectReport 模型实现（表已建立）
- _aggregate_task_results 聚合算法: 求和/重算错误率与TPS/加权平均响应时间/
  取极值/加权近似分位数(p50/p90/p95/p99)/聚合标准差重算, 排除失败任务
- 批量执行完成后自动生成合并报告并持久化, 返回 report_id
- GET /api/performance/reports 按 project_id/batch_id 查询, ?latest=1 取最新
- ProjectReportResponse/ProjectReportListResponse/TaskSummaryItem schema,
  BatchRunResponse 增加 report_id 字段
Co-Authored-By: Claude <noreply@anthropic.com>

backend/app/routers/performance.py

@@ -41,6 +41,8 @@ from app.schemas.performance import (
     PerformanceProjectStatsResponse,
     BatchRunRequest,
     BatchRunResponse,
+    ProjectReportResponse,
+    ProjectReportListResponse,
 )
 from app.services.performance_service import PerformanceService
 from app.utils.curl_parser import parse_curl
@@ -527,6 +529,7 @@ async def batch_run_tasks(
         message=f"批量执行完成: {result['success_count']} 成功, {result['fail_count']} 失败",
         batch_id=result["batch_id"],
         task_count=result["total"],
+        report_id=result.get("report_id"),
     )
 
 
@@ -554,4 +557,52 @@ async def run_project_all(
         message=f"项目执行完成: {result['success_count']} 成功, {result['fail_count']} 失败",
         batch_id=result["batch_id"],
         task_count=result["total"],
-    )
+        report_id=result.get("report_id"),
\ No newline at end of file
+    )
+
+
+# ==================== 合并报告 ====================
+
+
+@router.get("/projects/{project_id}/report", response_model=ProjectReportResponse)
+async def get_project_report(
+    project_id: str,
+    service: PerformanceService = Depends(get_perf_service),
+):
+    """
+    获取项目最新合并报告
+
+    返回该项目最近一次批量执行生成的合并报告。
+    """
+    report = await service.get_project_report(project_id)
+    if not report:
+        raise HTTPException(status_code=404, detail="该项目暂无合并报告")
+    return report
+
+
+@router.get("/batch-reports/{batch_id}", response_model=ProjectReportResponse)
+async def get_batch_report(
+    batch_id: str,
+    service: PerformanceService = Depends(get_perf_service),
+):
+    """
+    按批次ID获取合并报告
+    """
+    report = await service.get_batch_report(batch_id)
+    if not report:
+        raise HTTPException(status_code=404, detail="合并报告不存在")
+    return report
+
+
+@router.get("/projects/{project_id}/reports", response_model=ProjectReportListResponse)
+async def get_project_reports(
+    project_id: str,
+    page: int = Query(1, ge=1, description="页码"),
+    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
+    service: PerformanceService = Depends(get_perf_service),
+):
+    """
+    获取项目历史合并报告列表
+    """
+    result = await service.list_project_reports(project_id, page, page_size)
+    return ProjectReportListResponse(**result)
\ No newline at end of file

backend/app/schemas/performance.py

@@ -460,5 +460,71 @@ class BatchRunResponse(BaseModel):
     message: str
     batch_id: str
     task_count: int
+    report_id: Optional[str] = Field(None, description="合并报告ID（批量执行完成后生成）")
+
+    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+
+
+# ==================== 合并报告 ====================
+
+class TaskSummaryItem(BaseModel):
+    """合并报告中的任务明细项"""
+    task_id: str = Field(..., description="任务ID")
+    task_name: str = Field("", description="任务名称")
+    status: str = Field("", description="任务状态")
+    total_requests: int = Field(0, description="请求数")
+    success_count: int = Field(0, description="成功数")
+    fail_count: int = Field(0, description="失败数")
+    avg_response_time: float = Field(0.0, description="平均响应时间(ms)")
+    actual_tps: float = Field(0.0, description="实际TPS")
+    error_rate: float = Field(0.0, description="错误率")
+
+    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+
+
+class ProjectReportSummary(BaseModel):
+    """合并报告聚合指标摘要"""
+    total_requests: int = 0
+    success_count: int = 0
+    fail_count: int = 0
+    error_rate: float = 0.0
+    actual_tps: float = 0.0
+    peak_tps: float = 0.0
+    avg_response_time: float = 0.0
+    min_response_time: float = 0.0
+    max_response_time: float = 0.0
+    p50_response_time: float = 0.0
+    p90_response_time: float = 0.0
+    p95_response_time: float = 0.0
+    p99_response_time: float = 0.0
+    std_dev: float = 0.0
+    apdex: float = 1.0
+    status_2xx: int = Field(0, alias="status2xx")
+    status_3xx: int = Field(0, alias="status3xx")
+    status_4xx: int = Field(0, alias="status4xx")
+    status_5xx: int = Field(0, alias="status5xx")
+    total_sent_bytes: int = 0
+    total_received_bytes: int = 0
+    duration_actual: float = 0.0
+
+    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+
+
+class ProjectReportResponse(BaseModel):
+    """合并报告完整响应"""
+    id: str
+    project_id: str
+    batch_id: Optional[str] = None
+    created_at: Optional[str] = None
+    summary: ProjectReportSummary
+    task_summaries: List[TaskSummaryItem] = []
+
+    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+
+
+class ProjectReportListResponse(BaseModel):
+    """合并报告列表响应"""
+    total: int
+    items: List[ProjectReportResponse]
 
     model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
\ No newline at end of file

backend/app/services/performance_service.py

@@ -24,6 +24,8 @@ from app.models.performance import (
     PerformanceTask,
     PerformanceSnapshot,
     PerformanceProject,
+    PerformanceBatchExecution,
+    PerformanceProjectReport,
 )
 from app.models.performance_output import PerformanceTaskOutput
 from app.models.api_preset import ApiPreset
@@ -597,6 +599,130 @@ class PerformanceService:
             "snapshots": [s.to_dict() for s in snapshots],
         }
 
+    # ==================== 合并报告查询 ====================
+
+    @staticmethod
+    def _report_to_dict(report: PerformanceProjectReport) -> Dict[str, Any]:
+        """将合并报告 ORM 对象转为 API 响应字典"""
+        task_summaries = []
+        try:
+            task_summaries = json.loads(report.task_summaries or "[]")
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+        return {
+            "id": report.id,
+            "project_id": report.project_id,
+            "batch_id": report.batch_id,
+            "created_at": report.created_at.isoformat() if report.created_at else None,
+            "summary": {
+                "total_requests": report.total_requests,
+                "success_count": report.success_count,
+                "fail_count": report.fail_count,
+                "error_rate": report.error_rate,
+                "actual_tps": report.actual_tps,
+                "avg_response_time": report.avg_response_time,
+                "min_response_time": report.min_response_time,
+                "max_response_time": report.max_response_time,
+                "p50_response_time": report.p50_response_time,
+                "p90_response_time": report.p90_response_time,
+                "p95_response_time": report.p95_response_time,
+                "p99_response_time": report.p99_response_time,
+                "std_dev": report.std_dev,
+                "apdex": report.apdex,
+                "status_2xx": report.status_2xx,
+                "status_3xx": report.status_3xx,
+                "status_4xx": report.status_4xx,
+                "status_5xx": report.status_5xx,
+                "total_sent_bytes": report.total_sent_bytes,
+                "total_received_bytes": report.total_received_bytes,
+                "peak_tps": report.peak_tps,
+                "duration_actual": report.duration_actual,
+            },
+            "task_summaries": task_summaries,
+        }
+
+    async def get_project_report(self, project_id: str) -> Optional[Dict[str, Any]]:
+        """
+        获取项目最新合并报告
+
+        Args:
+            project_id: 项目ID
+
+        Returns:
+            dict: 合并报告数据，无则返回 None
+        """
+        result = await self.db.execute(
+            select(PerformanceProjectReport)
+            .where(PerformanceProjectReport.project_id == project_id)
+            .order_by(PerformanceProjectReport.created_at.desc())
+            .limit(1)
+        )
+        report = result.scalar_one_or_none()
+        if not report:
+            return None
+        return self._report_to_dict(report)
+
+    async def get_batch_report(self, batch_id: str) -> Optional[Dict[str, Any]]:
+        """
+        按批次ID获取合并报告
+
+        Args:
+            batch_id: 批量执行ID
+
+        Returns:
+            dict: 合并报告数据，无则返回 None
+        """
+        result = await self.db.execute(
+            select(PerformanceProjectReport).where(
+                PerformanceProjectReport.batch_id == batch_id
+            )
+        )
+        report = result.scalar_one_or_none()
+        if not report:
+            return None
+        return self._report_to_dict(report)
+
+    async def list_project_reports(
+        self, project_id: str, page: int = 1, page_size: int = 20
+    ) -> Dict[str, Any]:
+        """
+        获取项目历史合并报告列表
+
+        Args:
+            project_id: 项目ID
+            page: 页码（从 1 开始）
+            page_size: 每页数量
+
+        Returns:
+            dict: {total, items}
+        """
+        base_query = select(PerformanceProjectReport).where(
+            PerformanceProjectReport.project_id == project_id
+        )
+
+        # 总数
+        count_result = await self.db.execute(
+            select(func.count()).select_from(PerformanceProjectReport).where(
+                PerformanceProjectReport.project_id == project_id
+            )
+        )
+        total = count_result.scalar() or 0
+
+        # 分页列表
+        result = await self.db.execute(
+            base_query
+            .order_by(PerformanceProjectReport.created_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+        )
+        reports = list(result.scalars().all())
+
+        return {
+            "total": total,
+            "items": [self._report_to_dict(r) for r in reports],
+        }
+
     # ==================== 内部方法 ====================
 
     @staticmethod
@@ -995,6 +1121,161 @@ class PerformanceService:
 
     # ==================== 批量执行 ====================
 
+    @staticmethod
+    def _weighted_percentile(
+        percentiles: List[float], counts: List[int], p: float
+    ) -> float:
+        """
+        加权近似分位数合并算法
+
+        各任务的 P 分位数独立计算，合并时无法精确还原全局分位数，
+        采用加权近似法：按分位数值排序，累计请求数到达目标位置时取值。
+
+        Args:
+            percentiles: 各任务的指定分位数列表
+            counts: 各任务请求数列表
+            p: 目标分位数 (50/90/95/99)
+
+        Returns:
+            float: 加权近似分位数
+        """
+        if not percentiles:
+            return 0.0
+        # 过滤请求数为 0 的任务（无效权重）
+        pairs = [(v, c) for v, c in zip(percentiles, counts) if c > 0]
+        if not pairs:
+            return 0.0
+        pairs.sort(key=lambda x: x[0])
+        total = sum(c for _, c in pairs)
+        if total <= 0:
+            return 0.0
+        target = total * p / 100.0
+        cumulative = 0
+        for val, cnt in pairs:
+            cumulative += cnt
+            if cumulative >= target:
+                return val
+        return pairs[-1][0]
+
+    async def _aggregate_task_results(self, task_ids: List[str]) -> Dict[str, Any]:
+        """
+        聚合多个任务的执行结果为合并报告指标
+
+        聚合算法（参考 PRD 3.3）：
+        - 求和：total_requests / success_count / fail_count / 状态码 / 字节数 / duration_actual
+        - 重算：error_rate = fail / total，actual_tps = total_requests / duration_actual
+        - 加权平均：avg_response_time / apdex（权重为各任务请求数）
+        - 取极值：min_response_time / max_response_time / peak_tps
+        - 加权近似：p50 / p90 / p95 / p99（_weighted_percentile）
+        - 标准差：基于聚合均值和方差重算（√(Σ((σ_i² + μ_i²) × n_i) / N - 总体均值²)）
+
+        Args:
+            task_ids: 任务ID列表
+
+        Returns:
+            dict: 合并报告聚合指标 + 任务明细
+        """
+        if not task_ids:
+            return {}
+
+        result = await self.db.execute(
+            select(PerformanceTask).where(PerformanceTask.id.in_(task_ids))
+        )
+        tasks = list(result.scalars().all())
+        if not tasks:
+            return {}
+
+        # ---- 基础求和 ----
+        total_requests = sum(t.total_requests for t in tasks)
+        success_count = sum(t.success_count for t in tasks)
+        fail_count = sum(t.fail_count for t in tasks)
+        status_2xx = sum(t.status_2xx for t in tasks)
+        status_3xx = sum(t.status_3xx for t in tasks)
+        status_4xx = sum(t.status_4xx for t in tasks)
+        status_5xx = sum(t.status_5xx for t in tasks)
+        total_sent_bytes = sum(t.total_sent_bytes for t in tasks)
+        total_received_bytes = sum(t.total_received_bytes for t in tasks)
+        duration_actual = sum(t.duration_actual or 0 for t in tasks)
+
+        # ---- 重算比率 ----
+        error_rate = (fail_count / total_requests) if total_requests > 0 else 0.0
+        actual_tps = (total_requests / duration_actual) if duration_actual > 0 else 0.0
+
+        # ---- 加权平均 ----
+        avg_response_time = (
+            sum((t.avg_response_time or 0) * t.total_requests for t in tasks) / total_requests
+            if total_requests > 0 else 0.0
+        )
+        apdex = (
+            sum((t.apdex if t.apdex is not None else 1.0) * t.total_requests for t in tasks) / total_requests
+            if total_requests > 0 else 0.0
+        )
+
+        # ---- 极值 ----
+        min_response_time = min((t.min_response_time or 0) for t in tasks)
+        max_response_time = max((t.max_response_time or 0) for t in tasks)
+        peak_tps = max((t.peak_tps or 0) for t in tasks)
+
+        # ---- 加权近似分位数 ----
+        counts = [t.total_requests for t in tasks]
+        p50 = self._weighted_percentile([t.p50_response_time or 0 for t in tasks], counts, 50)
+        p90 = self._weighted_percentile([t.p90_response_time or 0 for t in tasks], counts, 90)
+        p95 = self._weighted_percentile([t.p95_response_time or 0 for t in tasks], counts, 95)
+        p99 = self._weighted_percentile([t.p99_response_time or 0 for t in tasks], counts, 99)
+
+        # ---- 标准差（合并方差公式） ----
+        # σ_total² = Σ((σ_i² + μ_i²) × n_i) / N - μ_total²
+        if total_requests > 0:
+            pooled_e2 = sum(
+                ((t.std_dev or 0) ** 2 + (t.avg_response_time or 0) ** 2) * t.total_requests
+                for t in tasks
+            ) / total_requests
+            std_dev = max(0.0, (pooled_e2 - avg_response_time ** 2) ** 0.5)
+        else:
+            std_dev = 0.0
+
+        # ---- 任务明细 ----
+        task_summaries = []
+        for t in tasks:
+            t_error_rate = (t.fail_count / t.total_requests) if t.total_requests > 0 else 0.0
+            task_summaries.append({
+                "task_id": t.id,
+                "task_name": t.name,
+                "status": t.status,
+                "total_requests": t.total_requests,
+                "success_count": t.success_count,
+                "fail_count": t.fail_count,
+                "avg_response_time": t.avg_response_time,
+                "actual_tps": t.actual_tps,
+                "error_rate": t_error_rate,
+            })
+
+        return {
+            "total_requests": total_requests,
+            "success_count": success_count,
+            "fail_count": fail_count,
+            "error_rate": round(error_rate, 6),
+            "actual_tps": round(actual_tps, 2),
+            "avg_response_time": round(avg_response_time, 2),
+            "min_response_time": round(min_response_time, 2),
+            "max_response_time": round(max_response_time, 2),
+            "p50_response_time": round(p50, 2),
+            "p90_response_time": round(p90, 2),
+            "p95_response_time": round(p95, 2),
+            "p99_response_time": round(p99, 2),
+            "std_dev": round(std_dev, 2),
+            "apdex": round(apdex, 4),
+            "status_2xx": status_2xx,
+            "status_3xx": status_3xx,
+            "status_4xx": status_4xx,
+            "status_5xx": status_5xx,
+            "total_sent_bytes": total_sent_bytes,
+            "total_received_bytes": total_received_bytes,
+            "peak_tps": round(peak_tps, 2),
+            "duration_actual": round(duration_actual, 2),
+            "task_summaries": task_summaries,
+        }
+
     async def batch_run_tasks(
         self,
         task_ids: List[str],
@@ -1020,6 +1301,21 @@ class PerformanceService:
 
         ws_group = "perf_batch"
 
+        # 持久化批量执行记录（running）
+        batch_record = PerformanceBatchExecution(
+            id=batch_id,
+            project_id=project_id,
+            task_ids=json.dumps(task_ids, ensure_ascii=False),
+            status="running",
+            total_count=total,
+            success_count=0,
+            fail_count=0,
+            started_at=datetime.utcnow(),
+            created_at=datetime.utcnow(),
+        )
+        self.db.add(batch_record)
+        await self.db.commit()
+
         # 广播批量执行开始
         await manager.broadcast(ws_group, {
             "type": "perf_batch_progress",
@@ -1107,6 +1403,61 @@ class PerformanceService:
             "fail_count": fail_count,
         })
 
+        # ---- 生成合并报告 ----
+        report_id = None
+        try:
+            if project_id:
+                aggregated = await self._aggregate_task_results(task_ids)
+                if aggregated:
+                    report_id = generate_id("rpt")
+                    report = PerformanceProjectReport(
+                        id=report_id,
+                        project_id=project_id,
+                        batch_id=batch_id,
+                        task_ids=json.dumps(task_ids, ensure_ascii=False),
+                        total_requests=aggregated.get("total_requests", 0),
+                        success_count=aggregated.get("success_count", 0),
+                        fail_count=aggregated.get("fail_count", 0),
+                        error_rate=aggregated.get("error_rate", 0.0),
+                        actual_tps=aggregated.get("actual_tps", 0.0),
+                        avg_response_time=aggregated.get("avg_response_time", 0.0),
+                        min_response_time=aggregated.get("min_response_time", 0.0),
+                        max_response_time=aggregated.get("max_response_time", 0.0),
+                        p50_response_time=aggregated.get("p50_response_time", 0.0),
+                        p90_response_time=aggregated.get("p90_response_time", 0.0),
+                        p95_response_time=aggregated.get("p95_response_time", 0.0),
+                        p99_response_time=aggregated.get("p99_response_time", 0.0),
+                        std_dev=aggregated.get("std_dev", 0.0),
+                        apdex=aggregated.get("apdex", 1.0),
+                        status_2xx=aggregated.get("status_2xx", 0),
+                        status_3xx=aggregated.get("status_3xx", 0),
+                        status_4xx=aggregated.get("status_4xx", 0),
+                        status_5xx=aggregated.get("status_5xx", 0),
+                        total_sent_bytes=aggregated.get("total_sent_bytes", 0),
+                        total_received_bytes=aggregated.get("total_received_bytes", 0),
+                        peak_tps=aggregated.get("peak_tps", 0.0),
+                        duration_actual=aggregated.get("duration_actual", 0.0),
+                        task_summaries=json.dumps(
+                            aggregated.get("task_summaries", []), ensure_ascii=False
+                        ),
+                        created_at=datetime.utcnow(),
+                    )
+                    self.db.add(report)
+        except Exception as e:
+            logger.error(f"生成合并报告失败: {e}")
+            report_id = None
+
+        # 更新批量执行记录状态
+        try:
+            batch_record.status = final_status
+            batch_record.success_count = success_count
+            batch_record.fail_count = fail_count
+            batch_record.completed_at = datetime.utcnow()
+            await self.db.commit()
+        except Exception as e:
+            logger.error(f"更新批量执行记录失败: {e}")
+            await self.db.rollback()
+
         return {
             "batch_id": batch_id,
             "total": total,
@@ -1114,6 +1465,7 @@ class PerformanceService:
             "fail_count": fail_count,
             "status": final_status,
             "results": results,
+            "report_id": report_id,
         }
 
     async def run_project_all(