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

fix(dashboard): 修复仪表盘与明细统计数值不一致并重建 cost_savings 聚合

Browse Source 
- 统一 dashboard 聚合与 raw 查询的 token 计算，拆分今日节省和周期节省
- cache savings 改用 input price 估算未命中成本，修复历史 cost_savings 偏高
- raw 查询 total_tokens 改用 effective_input + output + cache_creation + cache_read 公式
- 新增独立 backfill 重建历史 cost_savings 聚合表，保留已发布 backfill 不变
This commit is contained in:
mayrain
2026-05-05 17:33:16 +08:00
parent 2f915b33c7
commit 53fa33b0c5
10 changed files with 487 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/aether-admin/src/observability/stats.rs
View File

@@ -1429,7 +1429,7 @@ pub fn build_admin_stats_cost_savings_response(
let mut estimated_full_cost: f64 = usage
.iter()
.map(|item| {
item.settlement_output_price_per_1m().unwrap_or(0.0)
item.settlement_input_price_per_1m().unwrap_or(0.0)
* item.cache_read_input_tokens as f64
/ 1_000_000.0
})

261
crates/aether-data/backfills/20260504120000_rebuild_cost_savings_input_price.sql Normal file
View File

@@ -0,0 +1,261 @@
CREATE TEMP TABLE tmp_rebuild_cost_savings_context ON COMMIT DROP AS
SELECT
NOW() AS now_utc,
(date_trunc('day', NOW() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS current_day_utc;
CREATE TEMP TABLE tmp_rebuild_cost_savings_source ON COMMIT DROP AS
SELECT
usage.user_id,
usage.username,
COALESCE(usage.provider_name, '') AS provider_name,
COALESCE(usage.model, '') AS model,
(date_trunc('day', usage.created_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS day_utc,
GREATEST(COALESCE(usage.cache_read_input_tokens, 0), 0)::BIGINT AS cache_read_tokens,
COALESCE(CAST(usage.cache_read_cost_usd AS DOUBLE PRECISION), 0) AS cache_read_cost,
COALESCE(CAST(usage.cache_creation_cost_usd AS DOUBLE PRECISION), 0) AS cache_creation_cost,
COALESCE(
CAST(usage_settlement_snapshots.input_price_per_1m AS DOUBLE PRECISION),
CAST(usage.input_price_per_1m AS DOUBLE PRECISION),
0
) * GREATEST(COALESCE(usage.cache_read_input_tokens, 0), 0)::DOUBLE PRECISION / 1000000.0
AS estimated_full_cost
FROM usage_billing_facts AS usage
LEFT JOIN usage_settlement_snapshots
ON usage_settlement_snapshots.request_id = usage.request_id
CROSS JOIN tmp_rebuild_cost_savings_context AS context
WHERE usage.created_at < context.current_day_utc;
TRUNCATE TABLE
stats_daily_cost_savings,
stats_daily_cost_savings_provider,
stats_daily_cost_savings_model,
stats_daily_cost_savings_model_provider,
stats_user_daily_cost_savings,
stats_user_daily_cost_savings_provider,
stats_user_daily_cost_savings_model,
stats_user_daily_cost_savings_model_provider;
INSERT INTO stats_daily_cost_savings (
id,
date,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-daily-cost-savings:', CAST(source.day_utc AS TEXT))),
source.day_utc,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
GROUP BY source.day_utc, context.now_utc;
INSERT INTO stats_daily_cost_savings_provider (
id,
date,
provider_name,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-daily-cost-savings-provider:', CAST(source.day_utc AS TEXT), ':', source.provider_name)),
source.day_utc,
source.provider_name,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
GROUP BY source.day_utc, source.provider_name, context.now_utc;
INSERT INTO stats_daily_cost_savings_model (
id,
date,
model,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-daily-cost-savings-model:', CAST(source.day_utc AS TEXT), ':', source.model)),
source.day_utc,
source.model,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
GROUP BY source.day_utc, source.model, context.now_utc;
INSERT INTO stats_daily_cost_savings_model_provider (
id,
date,
model,
provider_name,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-daily-cost-savings-model-provider:', CAST(source.day_utc AS TEXT), ':', source.model, ':', source.provider_name)),
source.day_utc,
source.model,
source.provider_name,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
GROUP BY source.day_utc, source.model, source.provider_name, context.now_utc;
INSERT INTO stats_user_daily_cost_savings (
id,
user_id,
username,
date,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-user-daily-cost-savings:', source.user_id, ':', CAST(source.day_utc AS TEXT))),
source.user_id,
MAX(source.username),
source.day_utc,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
WHERE source.user_id IS NOT NULL
GROUP BY source.user_id, source.day_utc, context.now_utc;
INSERT INTO stats_user_daily_cost_savings_provider (
id,
user_id,
username,
date,
provider_name,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-user-daily-cost-savings-provider:', source.user_id, ':', CAST(source.day_utc AS TEXT), ':', source.provider_name)),
source.user_id,
MAX(source.username),
source.day_utc,
source.provider_name,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
WHERE source.user_id IS NOT NULL
GROUP BY source.user_id, source.day_utc, source.provider_name, context.now_utc;
INSERT INTO stats_user_daily_cost_savings_model (
id,
user_id,
username,
date,
model,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-user-daily-cost-savings-model:', source.user_id, ':', CAST(source.day_utc AS TEXT), ':', source.model)),
source.user_id,
MAX(source.username),
source.day_utc,
source.model,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
WHERE source.user_id IS NOT NULL
GROUP BY source.user_id, source.day_utc, source.model, context.now_utc;
INSERT INTO stats_user_daily_cost_savings_model_provider (
id,
user_id,
username,
date,
model,
provider_name,
cache_read_tokens,
cache_read_cost,
cache_creation_cost,
estimated_full_cost,
created_at,
updated_at
)
SELECT
md5(CONCAT('stats-user-daily-cost-savings-model-provider:', source.user_id, ':', CAST(source.day_utc AS TEXT), ':', source.model, ':', source.provider_name)),
source.user_id,
MAX(source.username),
source.day_utc,
source.model,
source.provider_name,
COALESCE(SUM(source.cache_read_tokens), 0)::BIGINT,
CAST(COALESCE(SUM(source.cache_read_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.cache_creation_cost), 0) AS DOUBLE PRECISION),
CAST(COALESCE(SUM(source.estimated_full_cost), 0) AS DOUBLE PRECISION),
context.now_utc,
context.now_utc
FROM tmp_rebuild_cost_savings_source AS source
CROSS JOIN tmp_rebuild_cost_savings_context AS context
WHERE source.user_id IS NOT NULL
GROUP BY
source.user_id,
source.day_utc,
source.model,
source.provider_name,
context.now_utc;

12
crates/aether-data/src/backfill.rs
View File

@@ -549,6 +549,7 @@ mod tests {
cache_read_cost_usd,
total_cost_usd,
actual_total_cost_usd,
input_price_per_1m,
output_price_per_1m,
status,
billing_status,
@@ -576,6 +577,7 @@ mod tests {
1.25,
1.10,
50.0,
50.0,
'completed',
'settled',
TIMESTAMPTZ '2024-05-06 07:18:09+00',
@@ -595,9 +597,10 @@ mod tests {
let pending_before = pending_backfills(&pool)
.await
.expect("pending backfills should load");
assert_eq!(pending_before.len(), 2);
assert_eq!(pending_before.len(), 3);
assert_eq!(pending_before[0].version, 20260422110000);
assert_eq!(pending_before[1].version, 20260422120000);
assert_eq!(pending_before[2].version, 20260504120000);
run_backfills(&pool)
.await
@@ -613,7 +616,10 @@ mod tests {
.fetch_all(&pool)
.await
.expect("applied backfill versions should load");
assert_eq!(applied_versions, vec![20260422110000, 20260422120000]);
assert_eq!(
applied_versions,
vec![20260422110000, 20260422120000, 20260504120000]
);
let api_key_total_requests: i64 = query_scalar(
"SELECT COALESCE(total_requests, 0)::BIGINT FROM public.api_keys WHERE id = 'api-key-backfill-1'",
@@ -1265,6 +1271,6 @@ mod tests {
.fetch_one(&pool)
.await
.expect("backfill count should load");
assert_eq!(applied_count, 1);
assert_eq!(applied_count, 3);
}
}

2
crates/aether-data/src/repository/usage/memory.rs
View File

@@ -1915,7 +1915,7 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
.saturating_add(item.cache_read_input_tokens);
summary.cache_read_cost_usd += item.cache_read_cost_usd;
summary.cache_creation_cost_usd += item.cache_creation_cost_usd;
summary.estimated_full_cost_usd += item.settlement_output_price_per_1m().unwrap_or(0.0)
summary.estimated_full_cost_usd += item.settlement_input_price_per_1m().unwrap_or(0.0)
* item.cache_read_input_tokens as f64
/ 1_000_000.0;
}

64
crates/aether-data/src/repository/usage/sql/mod.rs
View File

@@ -1394,7 +1394,7 @@ SELECT
COALESCE(SUM(input_tokens), 0)::BIGINT AS input_tokens,
COALESCE(SUM(effective_input_tokens), 0)::BIGINT AS effective_input_tokens,
COALESCE(SUM(output_tokens), 0)::BIGINT AS output_tokens,
COALESCE(SUM(input_tokens + output_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(cache_creation_tokens), 0)::BIGINT AS cache_creation_tokens,
COALESCE(SUM(cache_read_tokens), 0)::BIGINT AS cache_read_tokens,
COALESCE(SUM(total_input_context), 0)::BIGINT AS total_input_context,
@@ -1425,7 +1425,7 @@ SELECT
COALESCE(SUM(input_tokens), 0)::BIGINT AS input_tokens,
COALESCE(SUM(effective_input_tokens), 0)::BIGINT AS effective_input_tokens,
COALESCE(SUM(output_tokens), 0)::BIGINT AS output_tokens,
COALESCE(SUM(input_tokens + output_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(cache_creation_tokens), 0)::BIGINT AS cache_creation_tokens,
COALESCE(SUM(cache_read_tokens), 0)::BIGINT AS cache_read_tokens,
COALESCE(SUM(total_input_context), 0)::BIGINT AS total_input_context,
@@ -1482,7 +1482,33 @@ SELECT
END
), 0)::BIGINT AS effective_input_tokens,
COALESCE(SUM(GREATEST(COALESCE("usage".output_tokens, 0), 0)), 0)::BIGINT AS output_tokens,
COALESCE(SUM(GREATEST(COALESCE("usage".total_tokens, 0), 0)), 0)::BIGINT AS total_tokens,
COALESCE(SUM(
CASE
WHEN GREATEST(COALESCE("usage".input_tokens, 0), 0) <= 0 THEN 0
WHEN GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0) <= 0
THEN GREATEST(COALESCE("usage".input_tokens, 0), 0)
WHEN split_part(lower(COALESCE(COALESCE("usage".endpoint_api_format, "usage".api_format), '')), ':', 1)
IN ('openai', 'gemini', 'google')
THEN GREATEST(
GREATEST(COALESCE("usage".input_tokens, 0), 0)
- GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0),
0
)
ELSE GREATEST(COALESCE("usage".input_tokens, 0), 0)
END
+ GREATEST(COALESCE("usage".output_tokens, 0), 0)
+ CASE
WHEN COALESCE("usage".cache_creation_input_tokens, 0) = 0
AND (
COALESCE("usage".cache_creation_input_tokens_5m, 0)
+ COALESCE("usage".cache_creation_input_tokens_1h, 0)
) > 0
THEN COALESCE("usage".cache_creation_input_tokens_5m, 0)
+ COALESCE("usage".cache_creation_input_tokens_1h, 0)
ELSE COALESCE("usage".cache_creation_input_tokens, 0)
END
+ GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0)
), 0)::BIGINT AS total_tokens,
COALESCE(SUM(
CASE
WHEN COALESCE("usage".cache_creation_input_tokens, 0) = 0
@@ -3740,7 +3766,33 @@ SELECT
"usage".model AS model,
"usage".provider_name AS provider,
COUNT(*)::BIGINT AS requests,
COALESCE(SUM(GREATEST(COALESCE("usage".total_tokens, 0), 0)), 0)::BIGINT AS total_tokens,
COALESCE(SUM(
CASE
WHEN GREATEST(COALESCE("usage".input_tokens, 0), 0) <= 0 THEN 0
WHEN GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0) <= 0
THEN GREATEST(COALESCE("usage".input_tokens, 0), 0)
WHEN split_part(lower(COALESCE(COALESCE("usage".endpoint_api_format, "usage".api_format), '')), ':', 1)
IN ('openai', 'gemini', 'google')
THEN GREATEST(
GREATEST(COALESCE("usage".input_tokens, 0), 0)
- GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0),
0
)
ELSE GREATEST(COALESCE("usage".input_tokens, 0), 0)
END
+ GREATEST(COALESCE("usage".output_tokens, 0), 0)
+ CASE
WHEN COALESCE("usage".cache_creation_input_tokens, 0) = 0
AND (
COALESCE("usage".cache_creation_input_tokens_5m, 0)
+ COALESCE("usage".cache_creation_input_tokens_1h, 0)
) > 0
THEN COALESCE("usage".cache_creation_input_tokens_5m, 0)
+ COALESCE("usage".cache_creation_input_tokens_1h, 0)
ELSE COALESCE("usage".cache_creation_input_tokens, 0)
END
+ GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0)
), 0)::BIGINT AS total_tokens,
COALESCE(SUM(COALESCE(CAST("usage".total_cost_usd AS DOUBLE PRECISION), 0)), 0)
AS total_cost_usd,
COALESCE(SUM(
@@ -3944,7 +3996,7 @@ SELECT
{group_column} AS group_key,
COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
COALESCE(SUM(input_tokens), 0)::BIGINT AS input_tokens,
COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(output_tokens), 0)::BIGINT AS output_tokens,
COALESCE(SUM(effective_input_tokens), 0)::BIGINT AS effective_input_tokens,
COALESCE(SUM(total_input_context), 0)::BIGINT AS total_input_context,
@@ -4951,7 +5003,7 @@ SELECT
AS cache_creation_cost_usd,
COALESCE(SUM(
COALESCE(
CAST("usage".output_price_per_1m AS DOUBLE PRECISION),
CAST("usage".input_price_per_1m AS DOUBLE PRECISION),
0
) * GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0)::DOUBLE PRECISION / 1000000.0
), 0) AS estimated_full_cost_usd