wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

refactor(api): Phase 17 metrics.py Router 層違規修復

Browse Source 
移除 Router 層直接 DB 存取，遵循 leWOOOgo 積木化原則：
- 新增 IMetricsRepository Protocol (interfaces.py)
- 新增 MetricsDBRepository 封裝 DB 查詢
- 新增 MetricsService 封裝業務邏輯
- Router 層只做 HTTP 轉發

架構: Router → Service → Repository → PostgreSQL

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-26 10:01:57 +08:00
parent 58b4004a18
commit e7f361db50
5 changed files with 664 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
apps/api/src/api/v1/metrics.py
View File

@@ -1,25 +1,32 @@
"""
Metrics API - 黃金指標端點 (Gold Metrics Endpoint)
===================================================
Phase 17 重構: Router 層只做 HTTP 轉發，業務邏輯移至 Service 層
統帥鐵律: 禁止假數據！所有指標必須來自 SignOz 真實血脈
Endpoints:
- GET /metrics/gold - 獲取 Gold Metrics (RPS, Error Rate, P99, AI Success)
- GET /metrics/health - Metrics 子系統健康檢查
Data Sources:
- SignOz ClickHouse: RPS, Error Rate, P99 Latency
- SQLite AuditLog: AI Success Rate (executed / total proposals)
架構層次:
- Router (此檔案) → MetricsService → MetricsRepository → PostgreSQL
- Router (此檔案) → MetricsService → SignOzClient → ClickHouse
版本: v2.0
更新: 2026-03-26 (台北時區)
更新者: Claude Code (Phase 17 技術債修復 - Router 層違規抽取)
原版: v1.0 by Claude Code
"""
from datetime import UTC, datetime, timedelta
from datetime import datetime
from typing import Any
from fastapi import APIRouter
from pydantic import BaseModel
from src.core.logging import get_logger
from src.db.base import get_db_context
from src.services.signoz_client import get_signoz_client
from src.services.metrics_service import get_metrics_service
logger = get_logger("awoooi.metrics")
router = APIRouter()
@@ -29,14 +36,17 @@ router = APIRouter()
# Response Models
# =============================================================================
class TrendData(BaseModel):
"""Sparkline 趨勢數據"""
values: list[float]
direction: str # up, down, stable
class GoldMetricItem(BaseModel):
"""單一黃金指標"""
label: str
value: float | str
unit: str | None = None
@@ -46,96 +56,18 @@ class GoldMetricItem(BaseModel):
class GoldMetricsResponse(BaseModel):
"""Gold Metrics API Response"""
timestamp: datetime
service_name: str
metrics: list[GoldMetricItem]
raw_data: dict[str, Any] | None = None
# =============================================================================
# AI Success Rate Calculator
# =============================================================================
async def calculate_ai_success_rate(hours: int = 24) -> tuple[float, list[float]]:
"""
計算 AI 提案成功執行率
統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假
Args:
hours: 統計時間範圍 (小時)
Returns:
(success_rate_percent, trend_values)
"""
try:
async with get_db_context() as session:
from sqlalchemy import text
# 時間範圍
cutoff = datetime.now(UTC) - timedelta(hours=hours)
cutoff_str = cutoff.isoformat()
# Query: 統計 executed vs total (approved + executed + execution_failed)
query = text("""
SELECT
COUNT(CASE WHEN status = 'executed' THEN 1 END) as executed_count,
COUNT(*) as total_count
FROM approval_records
WHERE created_at >= :cutoff
AND status IN ('approved', 'executed', 'execution_failed')
""")
result = await session.execute(query, {"cutoff": cutoff_str})
row = result.fetchone()
if row and row.total_count > 0:
executed = row.executed_count or 0
total = row.total_count
success_rate = (executed / total) * 100
else:
success_rate = 0.0
# Trend: 過去 10 個時間點的成功率 (每小時一點)
trend_query = text("""
SELECT
strftime('%Y-%m-%d %H:00:00', created_at) as hour_bucket,
COUNT(CASE WHEN status = 'executed' THEN 1 END) * 100.0 /
NULLIF(COUNT(*), 0) as hourly_rate
FROM approval_records
WHERE created_at >= :cutoff
AND status IN ('approved', 'executed', 'execution_failed')
GROUP BY hour_bucket
ORDER BY hour_bucket DESC
LIMIT 10
""")
trend_result = await session.execute(trend_query, {"cutoff": cutoff_str})
trend_rows = trend_result.fetchall()
if trend_rows:
trend_values = [float(r.hourly_rate or 0) for r in reversed(trend_rows)]
else:
trend_values = [0.0] * 10
logger.info(
"ai_success_rate_calculated",
success_rate=success_rate,
hours=hours,
)
return success_rate, trend_values
except Exception as e:
logger.warning("ai_success_rate_error", error=str(e))
# 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據
return 0.0, [0.0] * 10
# =============================================================================
# Endpoints
# =============================================================================
@router.get("/metrics/gold", response_model=GoldMetricsResponse)
async def get_gold_metrics(
service_name: str = "awoooi-api",
@@ -153,102 +85,35 @@ async def get_gold_metrics(
GoldMetricsResponse with RPS, Error Rate, P99, AI Success
"""
logger.info(
"gold_metrics_fetch",
"gold_metrics_request",
service=service_name,
window_minutes=time_window_minutes,
)
metrics_list: list[GoldMetricItem] = []
raw_data: dict[str, Any] = {}
# =========================================================================
# 1. SignOz Gold Metrics (RPS, Error Rate, P99)
# =========================================================================
try:
signoz = get_signoz_client()
gold = await signoz.get_gold_metrics(
service_name=service_name,
time_window_minutes=time_window_minutes,
)
# RPS
rps_status = "healthy" if gold.rps < 1000 else ("warning" if gold.rps < 5000 else "critical")
rps_trend = [gold.rps * (0.9 + i * 0.02) for i in range(10)] # 模擬趨勢
metrics_list.append(GoldMetricItem(
label="RPS",
value=round(gold.rps, 1),
unit="req/s",
trend=rps_trend,
status=rps_status,
))
# Error Rate
error_status = "healthy" if gold.error_rate < 1 else ("warning" if gold.error_rate < 5 else "critical")
error_trend = [gold.error_rate * (0.95 + i * 0.01) for i in range(10)]
metrics_list.append(GoldMetricItem(
label="Error Rate",
value=round(gold.error_rate, 2),
unit="%",
trend=error_trend,
status=error_status,
))
# P99 Latency
p99_status = "healthy" if gold.p99_latency_ms < 200 else ("warning" if gold.p99_latency_ms < 500 else "critical")
p99_trend = [gold.p99_latency_ms * (0.95 + i * 0.01) for i in range(10)]
metrics_list.append(GoldMetricItem(
label="P99 Latency",
value=round(gold.p99_latency_ms, 0),
unit="ms",
trend=p99_trend,
status=p99_status,
))
raw_data["signoz"] = {
"rps": gold.rps,
"error_rate": gold.error_rate,
"p99_latency_ms": gold.p99_latency_ms,
"total_requests": gold.total_requests,
"error_count": gold.error_count,
}
except Exception as e:
logger.warning("signoz_metrics_error", error=str(e))
# 統帥鐵律: SignOz 斷線時顯示 0，非假數據
metrics_list.extend([
GoldMetricItem(label="RPS", value=0, unit="req/s", trend=[0]*10, status="critical"),
GoldMetricItem(label="Error Rate", value=0, unit="%", trend=[0]*10, status="critical"),
GoldMetricItem(label="P99 Latency", value=0, unit="ms", trend=[0]*10, status="critical"),
])
raw_data["signoz_error"] = str(e)
# =========================================================================
# 2. AI Success Rate (from AuditLog)
# =========================================================================
ai_success, ai_trend = await calculate_ai_success_rate(hours=24)
ai_status = "healthy" if ai_success >= 90 else ("warning" if ai_success >= 70 else "critical")
metrics_list.append(GoldMetricItem(
label="AI Success",
value=round(ai_success, 1),
unit="%",
trend=ai_trend,
status=ai_status,
))
raw_data["ai_success"] = {
"rate": ai_success,
"hours": 24,
}
# =========================================================================
# Response
# =========================================================================
return GoldMetricsResponse(
timestamp=datetime.now(UTC),
# 透過 Service 取得數據
service = get_metrics_service()
result = await service.get_gold_metrics(
service_name=service_name,
time_window_minutes=time_window_minutes,
)
# 轉換為 Response Model
metrics_list = [
GoldMetricItem(
label=m.label,
value=m.value,
unit=m.unit,
trend=m.trend,
status=m.status,
)
for m in result.metrics
]
return GoldMetricsResponse(
timestamp=result.timestamp,
service_name=result.service_name,
metrics=metrics_list,
raw_data=raw_data,
raw_data=result.raw_data,
)
@@ -259,17 +124,5 @@ async def metrics_health() -> dict:
快速檢查 SignOz 連線狀態
"""
try:
signoz = get_signoz_client()
# 嘗試執行簡單查詢
results = await signoz._query_clickhouse("SELECT 1")
clickhouse_ok = len(results) > 0
except Exception as e:
clickhouse_ok = False
logger.warning("clickhouse_health_check_failed", error=str(e))
return {
"status": "healthy" if clickhouse_ok else "degraded",
"clickhouse": "connected" if clickhouse_ok else "disconnected",
"timestamp": datetime.now(UTC).isoformat(),
}
service = get_metrics_service()
return await service.check_health()

8
apps/api/src/repositories/__init__.py
View File

@@ -24,18 +24,26 @@ from src.repositories.incident_repository import (
from src.repositories.interfaces import (
IApprovalRepository,
IIncidentRepository,
IMetricsRepository,
ITimelineRepository,
)
from src.repositories.metrics_repository import (
MetricsDBRepository,
get_metrics_repository,
)
__all__ = [
# Interfaces
"IApprovalRepository",
"IIncidentRepository",
"IMetricsRepository",
"ITimelineRepository",
# Implementations
"ApprovalDBRepository",
"IncidentDBRepository",
"MetricsDBRepository",
# Getters
"get_approval_repository",
"get_incident_repository",
"get_metrics_repository",
]

46
apps/api/src/repositories/interfaces.py
View File

@@ -125,3 +125,49 @@ class ITimelineRepository(Protocol):
) -> list[dict]:
"""取得最近的 Timeline 事件"""
...
@runtime_checkable
class IMetricsRepository(Protocol):
"""
Metrics Repository Protocol
職責: Metrics 相關 DB 查詢 (AI Success Rate)
實作: MetricsDBRepository (PostgreSQL)
版本: v1.0
建立: 2026-03-26 (台北時區)
建立者: Claude Code (Phase 17 技術債修復)
"""
async def get_ai_success_rate(
self,
hours: int = 24,
) -> tuple[float, int, int]:
"""
計算 AI 提案成功執行率
Args:
hours: 統計時間範圍 (小時)
Returns:
(success_rate_percent, executed_count, total_count)
"""
...
async def get_ai_success_trend(
self,
hours: int = 24,
points: int = 10,
) -> list[float]:
"""
取得 AI 成功率趨勢 (Sparkline 用)
Args:
hours: 統計時間範圍 (小時)
points: 趨勢點數量
Returns:
list[float]: 每小時成功率列表 (由舊到新)
"""
...

186
apps/api/src/repositories/metrics_repository.py Normal file
View File

@@ -0,0 +1,186 @@
"""
Metrics Repository - PostgreSQL 實作
=====================================
Phase 17 技術債修復: Router 層違規抽取
職責: Metrics 相關 DB 查詢 (AI Success Rate 統計)
設計: 實作 IMetricsRepository Protocol
版本: v1.0
建立: 2026-03-26 (台北時區)
建立者: Claude Code (Phase 17 技術債修復)
"""
from datetime import UTC, datetime, timedelta
import structlog
from sqlalchemy import text
from src.db.base import get_db_context
from src.repositories.interfaces import IMetricsRepository
logger = structlog.get_logger(__name__)
# =============================================================================
# MetricsDBRepository
# =============================================================================
class MetricsDBRepository(IMetricsRepository):
"""
Metrics Repository - PostgreSQL 實作
職責:
- AI Success Rate 統計查詢
- 趨勢數據查詢 (Sparkline)
純 CRUD 操作，不含業務邏輯
業務邏輯請放在 Service 層
"""
async def get_ai_success_rate(
self,
hours: int = 24,
) -> tuple[float, int, int]:
"""
計算 AI 提案成功執行率
統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假
Args:
hours: 統計時間範圍 (小時)
Returns:
(success_rate_percent, executed_count, total_count)
"""
try:
async with get_db_context() as session:
cutoff = datetime.now(UTC) - timedelta(hours=hours)
cutoff_str = cutoff.isoformat()
# Query: 統計 executed vs total (approved + executed + execution_failed)
query = text("""
SELECT
COUNT(CASE WHEN status = 'executed' THEN 1 END) as executed_count,
COUNT(*) as total_count
FROM approval_records
WHERE created_at >= :cutoff
AND status IN ('approved', 'executed', 'execution_failed')
""")
result = await session.execute(query, {"cutoff": cutoff_str})
row = result.fetchone()
if row and row.total_count > 0:
executed = row.executed_count or 0
total = row.total_count
success_rate = (executed / total) * 100
else:
executed = 0
total = 0
success_rate = 0.0
logger.debug(
"ai_success_rate_queried",
hours=hours,
executed=executed,
total=total,
rate=success_rate,
)
return success_rate, executed, total
except Exception as e:
logger.exception(
"ai_success_rate_query_error",
hours=hours,
error=str(e),
)
# 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據
return 0.0, 0, 0
async def get_ai_success_trend(
self,
hours: int = 24,
points: int = 10,
) -> list[float]:
"""
取得 AI 成功率趨勢 (Sparkline 用)
統帥鐵律: 若無數據，回傳真實的 [0.0] * points，嚴禁造假
Args:
hours: 統計時間範圍 (小時)
points: 趨勢點數量
Returns:
list[float]: 每小時成功率列表 (由舊到新)
"""
try:
async with get_db_context() as session:
cutoff = datetime.now(UTC) - timedelta(hours=hours)
cutoff_str = cutoff.isoformat()
# Trend: 過去 N 個時間點的成功率 (每小時一點)
trend_query = text("""
SELECT
strftime('%Y-%m-%d %H:00:00', created_at) as hour_bucket,
COUNT(CASE WHEN status = 'executed' THEN 1 END) * 100.0 /
NULLIF(COUNT(*), 0) as hourly_rate
FROM approval_records
WHERE created_at >= :cutoff
AND status IN ('approved', 'executed', 'execution_failed')
GROUP BY hour_bucket
ORDER BY hour_bucket DESC
LIMIT :limit
""")
result = await session.execute(
trend_query,
{"cutoff": cutoff_str, "limit": points},
)
rows = result.fetchall()
if rows:
# 由舊到新排列
trend_values = [float(r.hourly_rate or 0) for r in reversed(rows)]
# 如果點數不足，前面補 0
if len(trend_values) < points:
trend_values = [0.0] * (points - len(trend_values)) + trend_values
else:
trend_values = [0.0] * points
logger.debug(
"ai_success_trend_queried",
hours=hours,
points=points,
actual_points=len(trend_values),
)
return trend_values
except Exception as e:
logger.exception(
"ai_success_trend_query_error",
hours=hours,
points=points,
error=str(e),
)
# 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據
return [0.0] * points
# =============================================================================
# Singleton
# =============================================================================
_metrics_repository: MetricsDBRepository | None = None
def get_metrics_repository() -> MetricsDBRepository:
"""取得 MetricsDBRepository 實例 (Singleton)"""
global _metrics_repository
if _metrics_repository is None:
_metrics_repository = MetricsDBRepository()
return _metrics_repository

381
apps/api/src/services/metrics_service.py Normal file
View File

@@ -0,0 +1,381 @@
"""
Metrics Service - 黃金指標服務層
=================================
Phase 17 技術債修復: Router 層違規抽取
職責:
- 聚合 SignOz + DB 數據
- 計算健康狀態 (healthy/warning/critical)
- 生成趨勢數據
設計原則:
- Service 透過 Protocol 依賴 Repository (DI)
- 業務邏輯集中於此層
- Router 只做 HTTP 轉發
版本: v1.0
建立: 2026-03-26 (台北時區)
建立者: Claude Code (Phase 17 技術債修復)
"""
from datetime import UTC, datetime
from typing import Any
import structlog
from src.repositories.interfaces import IMetricsRepository
from src.repositories.metrics_repository import get_metrics_repository
from src.services.signoz_client import get_signoz_client
logger = structlog.get_logger(__name__)
# =============================================================================
# Data Classes
# =============================================================================
class GoldMetricResult:
"""單一黃金指標結果"""
def __init__(
self,
label: str,
value: float | str,
unit: str | None,
trend: list[float],
status: str,
) -> None:
self.label = label
self.value = value
self.unit = unit
self.trend = trend
self.status = status
class GoldMetricsResult:
"""黃金指標聚合結果"""
def __init__(
self,
timestamp: datetime,
service_name: str,
metrics: list[GoldMetricResult],
raw_data: dict[str, Any] | None = None,
) -> None:
self.timestamp = timestamp
self.service_name = service_name
self.metrics = metrics
self.raw_data = raw_data
# =============================================================================
# MetricsService
# =============================================================================
class MetricsService:
"""
黃金指標服務
職責:
1. 從 SignOz 取得 RPS, Error Rate, P99
2. 從 Repository 取得 AI Success Rate
3. 計算健康狀態
4. 組合成統一的 GoldMetrics 回應
使用方式:
service = MetricsService()
result = await service.get_gold_metrics("awoooi-api", 10)
"""
def __init__(
self,
metrics_repo: IMetricsRepository | None = None,
) -> None:
"""
依賴注入建構函數
Args:
metrics_repo: Metrics Repository (預設使用 Singleton)
"""
self._metrics_repo = metrics_repo or get_metrics_repository()
# =========================================================================
# Gold Metrics
# =========================================================================
async def get_gold_metrics(
self,
service_name: str = "awoooi-api",
time_window_minutes: int = 10,
) -> GoldMetricsResult:
"""
獲取黃金指標 (Gold Metrics)
統帥鐵律:
- 所有數據必須來自 SignOz 真實血脈
- AI Success 來自 AuditLog 真實統計
- 無數據時顯示 0，嚴禁造假
Returns:
GoldMetricsResult with RPS, Error Rate, P99, AI Success
"""
logger.info(
"gold_metrics_fetch",
service=service_name,
window_minutes=time_window_minutes,
)
metrics_list: list[GoldMetricResult] = []
raw_data: dict[str, Any] = {}
# =====================================================================
# 1. SignOz Gold Metrics (RPS, Error Rate, P99)
# =====================================================================
signoz_metrics = await self._fetch_signoz_metrics(
service_name,
time_window_minutes,
)
metrics_list.extend(signoz_metrics["metrics"])
raw_data.update(signoz_metrics["raw_data"])
# =====================================================================
# 2. AI Success Rate (from Repository)
# =====================================================================
ai_metrics = await self._fetch_ai_success_metrics(hours=24)
metrics_list.append(ai_metrics["metric"])
raw_data.update(ai_metrics["raw_data"])
return GoldMetricsResult(
timestamp=datetime.now(UTC),
service_name=service_name,
metrics=metrics_list,
raw_data=raw_data,
)
async def _fetch_signoz_metrics(
self,
service_name: str,
time_window_minutes: int,
) -> dict[str, Any]:
"""
從 SignOz 取得 RPS, Error Rate, P99
統帥鐵律: SignOz 斷線時顯示 0，非假數據
"""
metrics: list[GoldMetricResult] = []
raw_data: dict[str, Any] = {}
try:
signoz = get_signoz_client()
gold = await signoz.get_gold_metrics(
service_name=service_name,
time_window_minutes=time_window_minutes,
)
# RPS
rps_status = self._calculate_rps_status(gold.rps)
rps_trend = self._generate_simulated_trend(gold.rps)
metrics.append(GoldMetricResult(
label="RPS",
value=round(gold.rps, 1),
unit="req/s",
trend=rps_trend,
status=rps_status,
))
# Error Rate
error_status = self._calculate_error_status(gold.error_rate)
error_trend = self._generate_simulated_trend(gold.error_rate)
metrics.append(GoldMetricResult(
label="Error Rate",
value=round(gold.error_rate, 2),
unit="%",
trend=error_trend,
status=error_status,
))
# P99 Latency
p99_status = self._calculate_latency_status(gold.p99_latency_ms)
p99_trend = self._generate_simulated_trend(gold.p99_latency_ms)
metrics.append(GoldMetricResult(
label="P99 Latency",
value=round(gold.p99_latency_ms, 0),
unit="ms",
trend=p99_trend,
status=p99_status,
))
raw_data["signoz"] = {
"rps": gold.rps,
"error_rate": gold.error_rate,
"p99_latency_ms": gold.p99_latency_ms,
"total_requests": gold.total_requests,
"error_count": gold.error_count,
}
except Exception as e:
logger.warning("signoz_metrics_error", error=str(e))
# 統帥鐵律: SignOz 斷線時顯示 0，非假數據
metrics.extend([
GoldMetricResult(
label="RPS",
value=0,
unit="req/s",
trend=[0] * 10,
status="critical",
),
GoldMetricResult(
label="Error Rate",
value=0,
unit="%",
trend=[0] * 10,
status="critical",
),
GoldMetricResult(
label="P99 Latency",
value=0,
unit="ms",
trend=[0] * 10,
status="critical",
),
])
raw_data["signoz_error"] = str(e)
return {"metrics": metrics, "raw_data": raw_data}
async def _fetch_ai_success_metrics(
self,
hours: int = 24,
) -> dict[str, Any]:
"""
從 Repository 取得 AI Success Rate
統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假
"""
# 從 Repository 取得數據
success_rate, executed, total = await self._metrics_repo.get_ai_success_rate(
hours=hours,
)
trend = await self._metrics_repo.get_ai_success_trend(
hours=hours,
points=10,
)
ai_status = self._calculate_ai_success_status(success_rate)
metric = GoldMetricResult(
label="AI Success",
value=round(success_rate, 1),
unit="%",
trend=trend,
status=ai_status,
)
raw_data = {
"ai_success": {
"rate": success_rate,
"executed": executed,
"total": total,
"hours": hours,
}
}
logger.info(
"ai_success_rate_calculated",
success_rate=success_rate,
executed=executed,
total=total,
hours=hours,
)
return {"metric": metric, "raw_data": raw_data}
# =========================================================================
# Health Check
# =========================================================================
async def check_health(self) -> dict[str, Any]:
"""
Metrics 子系統健康檢查
快速檢查 SignOz 連線狀態
"""
try:
signoz = get_signoz_client()
results = await signoz._query_clickhouse("SELECT 1")
clickhouse_ok = len(results) > 0
except Exception as e:
clickhouse_ok = False
logger.warning("clickhouse_health_check_failed", error=str(e))
return {
"status": "healthy" if clickhouse_ok else "degraded",
"clickhouse": "connected" if clickhouse_ok else "disconnected",
"timestamp": datetime.now(UTC).isoformat(),
}
# =========================================================================
# Status Calculation Helpers
# =========================================================================
@staticmethod
def _calculate_rps_status(rps: float) -> str:
"""計算 RPS 健康狀態"""
if rps < 1000:
return "healthy"
if rps < 5000:
return "warning"
return "critical"
@staticmethod
def _calculate_error_status(error_rate: float) -> str:
"""計算 Error Rate 健康狀態"""
if error_rate < 1:
return "healthy"
if error_rate < 5:
return "warning"
return "critical"
@staticmethod
def _calculate_latency_status(p99_ms: float) -> str:
"""計算 P99 Latency 健康狀態"""
if p99_ms < 200:
return "healthy"
if p99_ms < 500:
return "warning"
return "critical"
@staticmethod
def _calculate_ai_success_status(success_rate: float) -> str:
"""計算 AI Success Rate 健康狀態"""
if success_rate >= 90:
return "healthy"
if success_rate >= 70:
return "warning"
return "critical"
@staticmethod
def _generate_simulated_trend(base_value: float, points: int = 10) -> list[float]:
"""
生成模擬趨勢數據 (SignOz 不提供歷史數據時使用)
注意: 這是暫時方案，未來應從 SignOz 取得真實歷史數據
"""
return [base_value * (0.9 + i * 0.02) for i in range(points)]
# =============================================================================
# Singleton
# =============================================================================
_metrics_service: MetricsService | None = None
def get_metrics_service() -> MetricsService:
"""取得 MetricsService 實例 (Singleton)"""
global _metrics_service
if _metrics_service is None:
_metrics_service = MetricsService()
return _metrics_service