i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 添加 Dashboard 供应商成本统计功能

Browse Source 
- 新增 stats_daily_provider 表存储每日供应商统计数据
- 实现供应商维度的数据聚合服务
- Dashboard API 返回 provider_summary 供应商汇总数据
- 前端新增 DoughnutChart 环形图组件
- Dashboard 新增供应商成本分布可视化卡片
- 移除重复的请求次数/费用趋势折线图

Closes #110

Co-authored-by: RWDai <27391645+RWDai@users.noreply.github.com>
This commit is contained in:
fawney19
2026-01-19 20:23:55 +08:00
parent 6ae862980d
commit c29d57622f
8 changed files with 487 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
src/api/dashboard/routes.py
View File

@@ -14,7 +14,7 @@ from src.api.base.pipeline import ApiRequestPipeline
from src.config.constants import CacheTTL
from src.core.enums import UserRole
from src.database import get_db
from src.models.database import ApiKey, Provider, RequestCandidate, StatsDaily, StatsDailyModel, Usage
from src.models.database import ApiKey, Provider, RequestCandidate, StatsDaily, StatsDailyModel, StatsDailyProvider, Usage
from src.models.database import User as DBUser
from src.services.system.stats_aggregator import StatsAggregatorService
from src.utils.cache_decorator import cache_result
@@ -1147,9 +1147,97 @@ class DashboardDailyStatsAdapter(DashboardAdapter):
for item in formatted:
item["model_breakdown"] = breakdown.get(item["date"], [])
# ==================== 供应商统计 ====================
if is_admin:
# 管理员：使用预聚合数据 + 今日实时数据
# 历史数据从 stats_daily_provider 获取
historical_provider_stats = (
db.query(StatsDailyProvider)
.filter(and_(StatsDailyProvider.date >= start_date, StatsDailyProvider.date < today))
.all()
)
# 按供应商汇总历史数据
provider_agg: dict[str, dict[str, int | float]] = {}
for stat in historical_provider_stats:
provider = stat.provider_name or "Unknown"
if provider not in provider_agg:
provider_agg[provider] = {"requests": 0, "tokens": 0, "cost": 0.0}
provider_agg[provider]["requests"] += stat.total_requests
tokens = (stat.input_tokens + stat.output_tokens +
stat.cache_creation_tokens + stat.cache_read_tokens)
provider_agg[provider]["tokens"] += tokens
provider_agg[provider]["cost"] += stat.total_cost
# 今日实时供应商统计
today_provider_stats = (
db.query(
Usage.provider_name,
func.count(Usage.id).label("requests"),
func.sum(Usage.total_tokens).label("tokens"),
func.sum(Usage.total_cost_usd).label("cost"),
)
.filter(Usage.created_at >= today)
.group_by(Usage.provider_name)
.all()
)
for stat in today_provider_stats:
provider = stat.provider_name or "Unknown"
if provider not in provider_agg:
provider_agg[provider] = {"requests": 0, "tokens": 0, "cost": 0.0}
provider_agg[provider]["requests"] += stat.requests or 0
provider_agg[provider]["tokens"] += int(stat.tokens or 0)
provider_agg[provider]["cost"] += float(stat.cost or 0)
# 构建 provider_summary
provider_summary = [
{
"provider": provider,
"requests": agg["requests"],
"tokens": agg["tokens"],
"cost": agg["cost"],
}
for provider, agg in provider_agg.items()
]
provider_summary.sort(key=lambda x: x["cost"], reverse=True)
else:
# 普通用户：实时查询
provider_stats = (
db.query(
Usage.provider_name,
func.count(Usage.id).label("requests"),
func.sum(Usage.total_tokens).label("tokens"),
func.sum(Usage.total_cost_usd).label("cost"),
)
.filter(
and_(
Usage.user_id == user.id,
Usage.created_at >= start_date,
Usage.created_at <= end_date
)
)
.group_by(Usage.provider_name)
.order_by(func.sum(Usage.total_cost_usd).desc())
.all()
)
provider_summary = [
{
"provider": stat.provider_name or "Unknown",
"requests": stat.requests or 0,
"tokens": int(stat.tokens or 0),
"cost": float(stat.cost or 0),
}
for stat in provider_stats
]
return {
"daily_stats": formatted,
"model_summary": model_summary,
"provider_summary": provider_summary,
"period": {
"start_date": start_date.date().isoformat(),
"end_date": end_date.date().isoformat(),

44
src/models/database.py
View File

@@ -1706,6 +1706,50 @@ class StatsDailyModel(Base):
)
class StatsDailyProvider(Base):
"""每日供应商统计快照 - 用于快速查询每日供应商维度数据"""
__tablename__ = "stats_daily_provider"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
# 统计日期 (UTC)
date = Column(DateTime(timezone=True), nullable=False, index=True)
# 供应商名称
provider_name = Column(String(100), nullable=False)
# 请求统计
total_requests = Column(Integer, default=0, nullable=False)
# Token 统计
input_tokens = Column(BigInteger, default=0, nullable=False)
output_tokens = Column(BigInteger, default=0, nullable=False)
cache_creation_tokens = Column(BigInteger, default=0, nullable=False)
cache_read_tokens = Column(BigInteger, default=0, nullable=False)
# 成本统计 (USD)
total_cost = Column(Float, default=0.0, nullable=False)
# 时间戳
created_at = Column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
)
updated_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
nullable=False,
)
# 唯一约束：每个供应商每天只有一条记录
__table_args__ = (
UniqueConstraint("date", "provider_name", name="uq_stats_daily_provider"),
Index("idx_stats_daily_provider_date", "date"),
Index("idx_stats_daily_provider_date_provider", "date", "provider_name"),
)
class StatsSummary(Base):
"""全局统计汇总 - 单行记录，存储截止到昨天的累计数据"""

26
src/services/system/cleanup_scheduler.py
View File

@@ -208,7 +208,7 @@ class CleanupScheduler:
return
# 非首次运行，检查最近是否有缺失的日期需要回填
from src.models.database import StatsDailyModel
from src.models.database import StatsDailyModel, StatsDailyProvider
yesterday_business_date = today_local.date() - timedelta(days=1)
max_backfill_days: int = SystemConfigService.get_config(
@@ -223,6 +223,7 @@ class CleanupScheduler:
# 获取 StatsDaily 和 StatsDailyModel 中已有数据的日期集合
existing_daily_dates = set()
existing_model_dates = set()
existing_provider_dates = set()
daily_stats = (
db.query(StatsDaily.date)
@@ -245,6 +246,17 @@ class CleanupScheduler:
stat_date = stat_date.replace(tzinfo=timezone.utc)
existing_model_dates.add(stat_date.astimezone(app_tz).date())
provider_stats = (
db.query(StatsDailyProvider.date)
.filter(StatsDailyProvider.date >= check_start_date.isoformat())
.distinct()
.all()
)
for (stat_date,) in provider_stats:
if stat_date.tzinfo is None:
stat_date = stat_date.replace(tzinfo=timezone.utc)
existing_provider_dates.add(stat_date.astimezone(app_tz).date())
# 找出需要回填的日期
all_dates = set()
current = check_start_date
@@ -256,15 +268,18 @@ class CleanupScheduler:
missing_daily_dates = all_dates - existing_daily_dates
# 需要回填 StatsDailyModel 的日期
missing_model_dates = all_dates - existing_model_dates
# 需要回填 StatsDailyProvider 的日期
missing_provider_dates = all_dates - existing_provider_dates
# 合并所有需要处理的日期
dates_to_process = missing_daily_dates | missing_model_dates
dates_to_process = missing_daily_dates | missing_model_dates | missing_provider_dates
if dates_to_process:
sorted_dates = sorted(dates_to_process)
logger.info(
f"检测到 {len(dates_to_process)} 天的统计数据需要回填 "
f"(StatsDaily 缺失 {len(missing_daily_dates)} 天, "
f"StatsDailyModel 缺失 {len(missing_model_dates)} 天)"
f"StatsDailyModel 缺失 {len(missing_model_dates)} 天, "
f"StatsDailyProvider 缺失 {len(missing_provider_dates)} 天)"
)
users = (
@@ -288,6 +303,10 @@ class CleanupScheduler:
StatsAggregatorService.aggregate_daily_model_stats(
db, current_date_local
)
if current_date in missing_provider_dates:
StatsAggregatorService.aggregate_daily_provider_stats(
db, current_date_local
)
# 用户统计在任一缺失时都回填
for (user_id,) in users:
try:
@@ -329,6 +348,7 @@ class CleanupScheduler:
StatsAggregatorService.aggregate_daily_stats(db, yesterday_local)
StatsAggregatorService.aggregate_daily_model_stats(db, yesterday_local)
StatsAggregatorService.aggregate_daily_provider_stats(db, yesterday_local)
users = db.query(DBUser.id).filter(DBUser.is_active.is_(True)).all()
for (user_id,) in users:

64
src/services/system/stats_aggregator.py
View File

@@ -17,6 +17,7 @@ from src.models.database import (
RequestCandidate,
StatsDaily,
StatsDailyModel,
StatsDailyProvider,
StatsSummary,
StatsUserDaily,
Usage,
@@ -286,6 +287,68 @@ class StatsAggregatorService:
)
return results
@staticmethod
def aggregate_daily_provider_stats(db: Session, date: datetime) -> list[StatsDailyProvider]:
"""聚合指定日期的供应商维度统计数据
Args:
db: 数据库会话
date: 要聚合的业务日期
Returns:
StatsDailyProvider 记录列表
"""
day_start, day_end = _get_business_day_range(date)
# 按供应商分组统计
provider_name_expr = func.coalesce(Usage.provider_name, "Unknown")
provider_stats = (
db.query(
provider_name_expr.label("provider_name"),
func.count(Usage.id).label("total_requests"),
func.sum(Usage.input_tokens).label("input_tokens"),
func.sum(Usage.output_tokens).label("output_tokens"),
func.sum(Usage.cache_creation_input_tokens).label("cache_creation_tokens"),
func.sum(Usage.cache_read_input_tokens).label("cache_read_tokens"),
func.sum(Usage.total_cost_usd).label("total_cost"),
)
.filter(and_(Usage.created_at >= day_start, Usage.created_at < day_end))
.group_by(provider_name_expr)
.all()
)
results = []
for stat in provider_stats:
existing = (
db.query(StatsDailyProvider)
.filter(and_(StatsDailyProvider.date == day_start, StatsDailyProvider.provider_name == stat.provider_name))
.first()
)
if existing:
record = existing
else:
record = StatsDailyProvider(
id=str(uuid.uuid4()), date=day_start, provider_name=stat.provider_name
)
record.total_requests = stat.total_requests or 0
record.input_tokens = int(stat.input_tokens or 0)
record.output_tokens = int(stat.output_tokens or 0)
record.cache_creation_tokens = int(stat.cache_creation_tokens or 0)
record.cache_read_tokens = int(stat.cache_read_tokens or 0)
record.total_cost = float(stat.total_cost or 0)
if not existing:
db.add(record)
results.append(record)
db.commit()
logger.info(
f"[StatsAggregator] 聚合日期 {date.date()} 供应商统计完成: {len(results)} 个供应商"
)
return results
@staticmethod
def get_daily_model_stats(db: Session, start_date: datetime, end_date: datetime) -> list[dict]:
"""获取日期范围内的模型统计数据（优先使用预聚合）
@@ -613,6 +676,7 @@ class StatsAggregatorService:
while current_date < today_local:
StatsAggregatorService.aggregate_daily_stats(db, current_date)
StatsAggregatorService.aggregate_daily_model_stats(db, current_date)
StatsAggregatorService.aggregate_daily_provider_stats(db, current_date)
count += 1
current_date += timedelta(days=1)