yumoqing/sage_datamart
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完整修复币种逻辑：

Browse Source 
- dm_model_call_fact 增加 amount_cny 字段（CNY折算金额）
- ETL sync_call_fact 同步时计算并写入 amount_cny
- dm_model_perf_daily/dm_provider_cost_daily 唯一键加入 currency（按币种分组）
- ETL 聚合函数按 currency 分组，保留原币种维度
- Dashboard 查询统一使用 amount_cny 进行跨币种比较
- 汇率查询从 accounting 模块获取（修正库名）
- 新增币种分布统计接口和UI展示
This commit is contained in:
yumoqing 2026-07-20 15:44:36 +08:00
parent e84fc8882b
commit 4d8e93dc87
8 changed files with 185 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
models/dm_model_call_fact.json
View File

@ -121,7 +121,7 @@
},
{
"name": "amount",
"title": "金额",
"title": "金额(原币)",
"type": "double",
"length": 10,
"dec": 4,
@ -134,6 +134,14 @@
"length": 5,
"default": "CNY"
},
{
"name": "amount_cny",
"title": "金额(CNY)",
"type": "double",
"length": 10,
"dec": 4,
"default": "0"
},
{
"name": "distributor_orgid",
"title": "分销商机构ID",

19
models/dm_model_perf_daily.json
View File

@ -150,9 +150,24 @@
"type": "bigint",
"default": "0"
},
{
"name": "currency",
"title": "货币",
"type": "str",
"length": 5,
"default": "CNY"
},
{
"name": "total_amount",
"title": "总金额",
"title": "总金额(原币)",
"type": "double",
"length": 12,
"dec": 4,
"default": "0"
},
{
"name": "total_amount_cny",
"title": "总金额(CNY)",
"type": "double",
"length": 12,
"dec": 4,
@ -175,7 +190,7 @@
{
"name": "uk_date_model_org",
"idxtype": "unique",
"idxfields": ["stat_date", "llmid", "userorgid"]
"idxfields": ["stat_date", "llmid", "currency", "userorgid"]
},
{
"name": "idx_date",

19
models/dm_provider_cost_daily.json
View File

@ -59,9 +59,24 @@
"type": "bigint",
"default": "0"
},
{
"name": "currency",
"title": "货币",
"type": "str",
"length": 5,
"default": "CNY"
},
{
"name": "total_amount",
"title": "总金额",
"title": "总金额(原币)",
"type": "double",
"length": 12,
"dec": 4,
"default": "0"
},
{
"name": "total_amount_cny",
"title": "总金额(CNY)",
"type": "double",
"length": 12,
"dec": 4,
@ -86,7 +101,7 @@
{
"name": "uk_date_model_provider",
"idxtype": "unique",
"idxfields": ["stat_date", "model", "providerid"]
"idxfields": ["stat_date", "model", "providerid", "currency"]
},
{
"name": "idx_model",

38
sage_datamart/dashboards.py
View File

@ -16,8 +16,8 @@ async def get_today_usage(env, sor):
async def get_today_amount(env, sor):
"""当天总金额"""
sql = "SELECT COALESCE(SUM(amount), 0) as total FROM dm_model_call_fact WHERE call_date = ${today}$"
"""当天总金额(CNY)"""
sql = "SELECT COALESCE(SUM(amount_cny), 0) as total FROM dm_model_call_fact WHERE call_date = ${today}$"
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
return round(float(recs[0].total), 2) if recs else 0
@ -57,11 +57,31 @@ async def get_avg_ttft(env, sor):
return 0
async def get_currency_breakdown(env, sor):
"""当天各币种分布"""
sql = """SELECT
COALESCE(currency, 'CNY') as currency,
COUNT(*) as call_count,
COALESCE(SUM(amount), 0) as amount,
COALESCE(SUM(amount_cny), 0) as amount_cny
FROM dm_model_call_fact
WHERE call_date = ${today}$
GROUP BY currency
ORDER BY call_count DESC"""
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
return [{
'currency': r.currency or 'CNY',
'call_count': int(r.call_count),
'amount': round(float(r.amount), 2),
'amount_cny': round(float(r.amount_cny), 2)
} for r in recs]
# ── 排行榜 ──
async def get_top_models(env, sor, limit=5):
"""当天 Top N 模型（按调用次数）"""
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact WHERE call_date = ${today}$
GROUP BY model ORDER BY cnt DESC LIMIT ${limit}$"""
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'limit': str(limit)})
@ -71,7 +91,7 @@ async def get_top_models(env, sor, limit=5):
async def get_top_providers(env, sor, limit=5):
"""当天 Top N 供应商（按金额）"""
sql = """SELECT providerid, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT providerid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact WHERE call_date = ${today}$ AND providerid IS NOT NULL
GROUP BY providerid ORDER BY total_amount DESC LIMIT ${limit}$"""
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'limit': str(limit)})
@ -81,7 +101,7 @@ async def get_top_providers(env, sor, limit=5):
async def get_top_users(env, sor, limit=5):
"""全量 Top N 用户（按金额）"""
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact GROUP BY userid ORDER BY total_amount DESC LIMIT ${limit}$"""
recs = await sor.sqlExe(sql, {'limit': str(limit)})
return [{'user_name': r.userid, 'cnt': int(r.cnt),
@ -92,7 +112,7 @@ async def get_top_users(env, sor, limit=5):
async def get_customer_daily_models(env, sor, userorgid):
"""客户当天各模型用量"""
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact WHERE call_date = ${today}$ AND userorgid = ${orgid}$
GROUP BY model ORDER BY cnt DESC"""
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'orgid': userorgid})
@ -104,7 +124,7 @@ async def get_customer_monthly_models(env, sor, userorgid):
"""客户当月各模型用量"""
now = datetime.now()
month_start = now.strftime('%Y-%m-01')
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact WHERE call_date >= ${start}$ AND userorgid = ${orgid}$
GROUP BY model ORDER BY cnt DESC"""
recs = await sor.sqlExe(sql, {'start': month_start, 'orgid': userorgid})
@ -114,7 +134,7 @@ async def get_customer_monthly_models(env, sor, userorgid):
async def get_customer_user_today(env, sor, userorgid):
"""客户当天各用户用量"""
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact WHERE call_date = ${today}$ AND userorgid = ${orgid}$
GROUP BY userid ORDER BY cnt DESC"""
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'orgid': userorgid})
@ -126,7 +146,7 @@ async def get_customer_user_today(env, sor, userorgid):
async def get_daily_trend(env, sor, days=7):
"""最近 N 天调用量和金额趋势"""
sql = """SELECT call_date, COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total_amount
sql = """SELECT call_date, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
FROM dm_model_call_fact
WHERE call_date >= ${start}$
GROUP BY call_date ORDER BY call_date"""

71
sage_datamart/etl.py
View File

@ -27,10 +27,14 @@ def parse_usages(usages_str):
async def sync_call_fact(sor, last_sync=None):
"""增量同步 llmusage → dm_model_call_fact"""
"""增量同步 llmusage → dm_model_call_fact（含币种折算）"""
if last_sync is None:
last_sync = (datetime.now() - timedelta(minutes=10)).strftime('%Y-%m-%d %H:%M:%S')
# 预取汇率
stat_date = datetime.now().strftime('%Y-%m-%d')
rates = await _get_exchange_rates(sor, stat_date)
sql = """
SELECT lu.*, l.model as llm_model, l.providerid,
(SELECT llmcatelogid FROM llm_api_map
@ -55,6 +59,10 @@ async def sync_call_fact(sor, last_sync=None):
prompt_tokens, completion_tokens, _ = parse_usages(r.usages)
call_time = datetime.strptime(r.use_time[:19], '%Y-%m-%d %H:%M:%S')
currency = getattr(r, 'amount_currency', None) or 'CNY'
amount = r.amount or 0
rate = rates.get(currency, 1.0)
amount_cny = round(float(amount) * rate, 4)
ns = {
'id': getID(),
@ -74,8 +82,9 @@ async def sync_call_fact(sor, last_sync=None):
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens,
'status': r.status or 'SUCCEEDED',
'amount': r.amount or 0,
'currency': getattr(r, 'amount_currency', None) or 'CNY',
'amount': amount,
'currency': currency,
'amount_cny': amount_cny,
}
await sor.C('dm_model_call_fact', ns)
count += 1
@ -84,16 +93,38 @@ async def sync_call_fact(sor, last_sync=None):
return count
async def _get_exchange_rates(sor, stat_date):
"""从 sage 库获取汇率表，返回 {currency: mid_rate} 字典"""
from sqlor.dbpools import get_sor_context
rates = {'CNY': 1.0}
try:
async with get_sor_context(sor.env if hasattr(sor, 'env') else None, 'accounting') as sage_sor:
sql = """SELECT from_currency, mid_rate FROM exchange_rate
WHERE to_currency = 'CNY' AND effective_date <= ${d}$
ORDER BY effective_date DESC"""
rows = await sage_sor.sqlExe(sql, {'d': stat_date})
for r in rows:
cur = r.from_currency if hasattr(r, 'from_currency') else r['from_currency']
rate = r.mid_rate if hasattr(r, 'mid_rate') else r['mid_rate']
if cur not in rates and rate:
rates[cur] = float(rate)
except Exception as e:
info('sage_datamart: exchange_rate lookup failed: ' + str(e))
return rates
async def aggregate_daily_perf(sor, stat_date=None):
"""聚合当天 dm_model_call_fact → dm_model_perf_daily"""
"""聚合当天 dm_model_call_fact → dm_model_perf_daily（按币种分组）"""
if stat_date is None:
stat_date = datetime.now().strftime('%Y-%m-%d')
# 按 llmid + userorgid 聚合
rates = await _get_exchange_rates(sor, stat_date)
sql = """
SELECT
call_date, llmid, model, catelogid, providerid, ownerid,
COALESCE(userorgid, '') as userorgid,
COALESCE(currency, 'CNY') as currency,
COUNT(*) as total_calls,
SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success_calls,
SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail_calls,
@ -104,17 +135,20 @@ async def aggregate_daily_perf(sor, stat_date=None):
SUM(amount) as total_amount
FROM dm_model_call_fact
WHERE call_date = ${stat_date}$
GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, userorgid
GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, userorgid, currency
"""
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
info('sage_datamart AGG: aggregating ' + str(len(rows)) + ' groups for ' + stat_date)
for r in rows:
r = DictObject(r)
# Delete old row, insert new
cur = r.currency or 'CNY'
rate = rates.get(cur, 1.0)
amount_cny = round(float(r.total_amount or 0) * rate, 4)
await sor.sqlExe(
'DELETE FROM dm_model_perf_daily WHERE stat_date=${d}$ AND llmid=${l}$ AND COALESCE(userorgid,\"\")=${u}$',
{'d': stat_date, 'l': r.llmid, 'u': r.userorgid or ''})
'DELETE FROM dm_model_perf_daily WHERE stat_date=${d}$ AND llmid=${l}$ AND COALESCE(currency,"CNY")=${c}$ AND COALESCE(userorgid,"")=${u}$',
{'d': stat_date, 'l': r.llmid, 'c': cur, 'u': r.userorgid or ''})
ns = {
'id': getID(),
@ -133,6 +167,8 @@ async def aggregate_daily_perf(sor, stat_date=None):
'total_prompt_tokens': r.total_prompt or 0,
'total_completion_tokens': r.total_completion or 0,
'total_amount': r.total_amount or 0,
'currency': cur,
'total_amount_cny': amount_cny,
'userorgid': r.userorgid or None,
}
await sor.C('dm_model_perf_daily', ns)
@ -141,28 +177,35 @@ async def aggregate_daily_perf(sor, stat_date=None):
async def aggregate_provider_cost(sor, stat_date=None):
"""聚合同模型供应商性价比 — dm_provider_cost_daily"""
"""聚合同模型供应商性价比 — dm_provider_cost_daily（按币种分组）"""
if stat_date is None:
stat_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
rates = await _get_exchange_rates(sor, stat_date)
sql = """
SELECT
model, providerid,
COALESCE(currency, 'CNY') as currency,
COUNT(*) as total_calls,
SUM(prompt_tokens + completion_tokens) as total_tokens,
SUM(amount) as total_amount,
AVG(ttft_ms) as avg_ttft
FROM dm_model_call_fact
WHERE call_date = ${stat_date}$
GROUP BY model, providerid
GROUP BY model, providerid, currency
"""
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
for r in rows:
r = DictObject(r)
cur = r.currency or 'CNY'
rate = rates.get(cur, 1.0)
amount_cny = round(float(r.total_amount or 0) * rate, 4)
await sor.sqlExe(
'DELETE FROM dm_provider_cost_daily WHERE stat_date=${d}$ AND model=${m}$ AND providerid=${p}$',
{'d': stat_date, 'm': r.model, 'p': r.providerid})
'DELETE FROM dm_provider_cost_daily WHERE stat_date=${d}$ AND model=${m}$ AND providerid=${p}$ AND COALESCE(currency,"CNY")=${c}$',
{'d': stat_date, 'm': r.model, 'p': r.providerid, 'c': cur})
ns = {
'id': getID(),
@ -172,6 +215,8 @@ async def aggregate_provider_cost(sor, stat_date=None):
'total_calls': r.total_calls,
'total_tokens': r.total_tokens or 0,
'total_amount': r.total_amount or 0,
'currency': cur,
'total_amount_cny': amount_cny,
'unit_price': round(float(r.total_amount) / r.total_tokens, 8) if r.total_tokens else None,
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
}

15
sage_datamart/init.py
View File

@ -2,14 +2,14 @@
from ahserver.serverenv import ServerEnv
from appPublic.log import debug
from sqlor.dbpools import get_sor_context
from .etl import sync_call_fact, aggregate_daily_perf, aggregate_provider_cost, ensure_tables
from .etl import sync_call_fact, aggregate_daily_perf, aggregate_provider_cost
from .dashboards import (
get_today_usage, get_today_amount, get_success_rate, get_fail_count,
get_active_users, get_top_models, get_top_providers, get_top_users,
get_customer_daily_models, get_customer_monthly_models,
get_customer_user_today, get_daily_trend, get_hourly_concurrency,
get_avg_ttft,
get_avg_ttft, get_currency_breakdown,
)
MODULE_NAME = "sage_datamart"
@ -110,6 +110,13 @@ async def api_hourly_concurrency(request):
return await get_hourly_concurrency(env, sor)
async def api_currency_stats(request):
"""币种分布统计 — /sage_datamart/api/currency_stats.dspy"""
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
return await get_currency_breakdown(env, sor)
# ── Jinja2 模板函数（给 stat_*.ui 调用）──
async def j2_today_usage(request):
@ -157,7 +164,6 @@ async def cron_etl_sync(request):
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await sync_call_fact(sor)
return {'status': 'ok', 'synced': count}
@ -169,7 +175,6 @@ async def cron_etl_aggregate(request):
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await aggregate_daily_perf(sor)
return {'status': 'ok', 'aggregated': count}
@ -181,7 +186,6 @@ async def cron_etl_provider_cost(request):
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await aggregate_provider_cost(sor)
return {'status': 'ok', 'aggregated': count}
@ -200,6 +204,7 @@ def load_sage_datamart():
env.api_customer_models = api_customer_models
env.api_daily_trend = api_daily_trend
env.api_hourly_concurrency = api_hourly_concurrency
env.api_currency_stats = api_currency_stats
env.j2_today_usage = j2_today_usage
env.j2_today_amount = j2_today_amount
env.j2_active_users = j2_active_users

7
wwwroot/api/currency_stats.dspy Normal file
View File

@ -0,0 +1,7 @@
{
"python": {
"method": "get",
"import": "sage_datamart.init",
"call": "api_currency_stats"
}
}

45
wwwroot/index.ui
View File

@ -34,7 +34,7 @@
"subwidgets": [
{
"widgettype": "VBox",
"options": {"css": "card", "width": "50%", "padding": "16px"},
"options": {"css": "card", "width": "60%", "padding": "16px"},
"subwidgets": [
{"widgettype": "Title4", "options": {"fontWeight": "600", "otext": "模型性能排行", "i18n": true, "marginBottom": "8px"}},
{
@ -51,13 +51,44 @@
{"widgettype": "Text", "options": {"text": "${avg_ttft_ms}ms", "width": "15%"}},
{"widgettype": "Text", "options": {"text": "${success_rate}%", "width": "15%"}},
{"widgettype": "Text", "options": {"text": "${total_calls}", "width": "15%"}},
{"widgettype": "Text", "options": {"text": "¥${total_amount}", "width": "20%"}}
{"widgettype": "Text", "options": {"text": "${total_amount_cny}", "width": "20%"}}
]
}
}
}
]
},
{
"widgettype": "VBox",
"options": {"css": "card", "width": "40%", "padding": "16px"},
"subwidgets": [
{"widgettype": "Title4", "options": {"fontWeight": "600", "otext": "币种分布", "i18n": true, "marginBottom": "8px"}},
{
"widgettype": "Cols",
"options": {
"col_cwidth": 24,
"data_url": "{{entire_url('api/currency_stats.dspy')}}",
"data_params": {},
"record_view": {
"widgettype": "HBox",
"options": {"cheight": 3, "width": "100%"},
"subwidgets": [
{"widgettype": "Text", "options": {"text": "${currency}", "width": "20%"}},
{"widgettype": "Text", "options": {"text": "${call_count}次", "width": "25%"}},
{"widgettype": "Text", "options": {"text": "${amount}", "width": "25%"}},
{"widgettype": "Text", "options": {"text": "${amount_cny}", "width": "30%"}}
]
}
}
}
]
}
]
},
{
"widgettype": "HBox",
"options": {"width": "100%", "gap": "16px", "marginBottom": "16px"},
"subwidgets": [
{
"widgettype": "VBox",
"options": {"css": "card", "width": "50%", "padding": "16px"},
@ -74,10 +105,10 @@
"options": {"cheight": 3, "width": "100%"},
"subwidgets": [
{"widgettype": "Text", "options": {"text": "${model}", "width": "35%"}},
{"widgettype": "Text", "options": {"text": "¥${unit_price}/t", "width": "20%"}},
{"widgettype": "Text", "options": {"text": "${unit_price}/t", "width": "20%"}},
{"widgettype": "Text", "options": {"text": "${avg_ttft_ms}ms", "width": "15%"}},
{"widgettype": "Text", "options": {"text": "${total_calls}次", "width": "15%"}},
{"widgettype": "Text", "options": {"text": "¥${total_amount}", "width": "15%"}}
{"widgettype": "Text", "options": {"text": "${total_amount_cny}", "width": "15%"}}
]
}
}
@ -107,7 +138,7 @@
"subwidgets": [
{"widgettype": "Text", "options": {"text": "${model_name}", "width": "50%"}},
{"widgettype": "Text", "options": {"text": "${cnt}次", "width": "25%"}},
{"widgettype": "Text", "options": {"text": "¥${total_amount}", "width": "25%"}}
{"widgettype": "Text", "options": {"text": "${total_amount}", "width": "25%"}}
]
}
}
@ -131,7 +162,7 @@
"subwidgets": [
{"widgettype": "Text", "options": {"text": "${user_name}", "width": "50%"}},
{"widgettype": "Text", "options": {"text": "${cnt}次", "width": "25%"}},
{"widgettype": "Text", "options": {"text": "¥${total_amount}", "width": "25%"}}
{"widgettype": "Text", "options": {"text": "${total_amount}", "width": "25%"}}
]
}
}
@ -155,7 +186,7 @@
"subwidgets": [
{"widgettype": "Text", "options": {"text": "${provider_name}", "width": "50%"}},
{"widgettype": "Text", "options": {"text": "${cnt}次", "width": "25%"}},
{"widgettype": "Text", "options": {"text": "¥${total_amount}", "width": "25%"}}
{"widgettype": "Text", "options": {"text": "${total_amount}", "width": "25%"}}
]
}
}