diff --git a/build/lib/sage_datamart/__init__.py b/build/lib/sage_datamart/__init__.py
new file mode 100644
index 0000000..736c30c
--- /dev/null
+++ b/build/lib/sage_datamart/__init__.py
@@ -0,0 +1,2 @@
+"""sage_datamart — Sage 数据集市模块"""
+from .init import load_sage_datamart
diff --git a/build/lib/sage_datamart/dashboards.py b/build/lib/sage_datamart/dashboards.py
new file mode 100644
index 0000000..589d618
--- /dev/null
+++ b/build/lib/sage_datamart/dashboards.py
@@ -0,0 +1,561 @@
+"""
+统一 Dashboard API — 从集市表 / llmusage 查询运营指标 + 性能指标
+支持客户(customer)和分销商(distributor)两种视图
+实时指标直接查 llmusage，非实时从集市表取
+"""
+from sqlor.dbpools import get_sor_context
+from datetime import datetime, timedelta
+import redis.asyncio as redis
+
+# ── 角色/范围辅助 ──
+
+CUSTOMER = 'customer'
+DISTRIBUTOR = 'distributor'
+
+
+def _fact_where(scope, org_filter):
+    """根据角色返回 dm_model_call_fact 的 WHERE 条件"""
+    if scope == DISTRIBUTOR and org_filter:
+        return "distributor_orgid = ${org}$"
+    elif scope == CUSTOMER and org_filter:
+        return "userorgid = ${org}$"
+    return "1=1"
+
+
+def _usage_where(scope, org_filter):
+    """根据角色返回 llmusage 的 WHERE 条件"""
+    if scope == DISTRIBUTOR and org_filter:
+        return "distributor_orgid = ${org}$"
+    elif scope == CUSTOMER and org_filter:
+        return "userorgid = ${org}$"
+    return "1=1"
+
+
+# ── 实时指标卡片（直接查 llmusage，无 ETL 延迟）──
+
+async def get_realtime_calls(sor, today, scope=None, org_filter=None):
+    """实时今日调用笔数"""
+    w = _usage_where(scope, org_filter)
+    sql = ("SELECT COUNT(*) as cnt FROM llmusage "
+           "WHERE use_date = ${today}$ AND " + w)
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_realtime_amount(sor, today, scope=None, org_filter=None):
+    """实时今日交易金额（原币汇总）"""
+    w = _usage_where(scope, org_filter)
+    sql = ("SELECT COALESCE(SUM(amount), 0) as total FROM llmusage "
+           "WHERE use_date = ${today}$ AND " + w)
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return round(float(recs[0].total), 2) if recs else 0
+
+
+async def get_total_users_count(sor, scope=None, org_filter=None):
+    """用户总数"""
+    if scope == CUSTOMER and org_filter:
+        sql = "SELECT COUNT(*) as cnt FROM users WHERE orgid = ${org}$"
+        ns = {'org': org_filter}
+    elif scope == DISTRIBUTOR and org_filter:
+        # 分销商下所有客户机构的用户
+        sql = ("SELECT COUNT(DISTINCT u.id) as cnt FROM users u "
+               "JOIN organization o ON o.id = u.orgid "
+               "WHERE o.resellerid = ${org}$")
+        ns = {'org': org_filter}
+    else:
+        sql = "SELECT COUNT(*) as cnt FROM users"
+        ns = {}
+    recs = await sor.sqlExe(sql, ns)
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_realtime_active_users(sor, today, scope=None, org_filter=None):
+    """实时今日活跃用户数"""
+    w = _usage_where(scope, org_filter)
+    sql = ("SELECT COUNT(DISTINCT userid) as cnt FROM llmusage "
+           "WHERE use_date = ${today}$ AND " + w)
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_realtime_stats(sor, today, scope=None, org_filter=None):
+    """一次查llmusage返回全部实时指标，确保数据一致"""
+    w = _usage_where(scope, org_filter)
+    sql = ("SELECT "
+           "COUNT(*) as calls, "
+           "COALESCE(SUM(amount), 0) as amount, "
+           "COUNT(DISTINCT userid) as active_users, "
+           "SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success, "
+           "SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail "
+           "FROM llmusage "
+           "WHERE use_date = ${today}$ AND " + w)
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    if recs:
+        r = recs[0]
+        total = int(r.calls or 0)
+        success = int(r.success or 0)
+        fail = int(r.fail or 0)
+        return {
+            'calls': total,
+            'amount': round(float(r.amount or 0), 2),
+            'active_users': int(r.active_users or 0),
+            'success_rate': round(success * 100.0 / total, 1) if total else 0,
+            'fail_count': fail,
+        }
+    return {'calls': 0, 'amount': 0, 'active_users': 0, 'success_rate': 0, 'fail_count': 0}
+
+
+async def get_online_users(env, scope=None, org_filter=None):
+    """在线用户数 — 从 Redis session 中统计"""
+    try:
+        config = env.get_config()
+        redis_url = getattr(config, 'redis_url', 'redis://127.0.0.1:6379')
+        r = redis.from_url(redis_url, decode_responses=True)
+        count = 0
+        cursor = 0
+        while True:
+            cursor, keys = await r.scan(cursor, match='AIOHTTP_SESSION_*', count=100)
+            for k in keys:
+                data = await r.get(k)
+                if data:
+                    count += 1
+            if cursor == 0:
+                break
+        await r.close()
+        return count
+    except Exception:
+        return 0
+
+
+async def get_online_users_by_org(env, org_filter, scope=None):
+    """按机构统计在线用户数"""
+    try:
+        config = env.get_config()
+        redis_url = getattr(config, 'redis_url', 'redis://127.0.0.1:6379')
+        r = redis.from_url(redis_url, decode_responses=True)
+        count = 0
+        cursor = 0
+        while True:
+            cursor, keys = await r.scan(cursor, match='AIOHTTP_SESSION_*', count=100)
+            for k in keys:
+                data = await r.get(k)
+                if data and org_filter in data:
+                    count += 1
+            if cursor == 0:
+                break
+        await r.close()
+        return count
+    except Exception:
+        return 0
+
+
+# ── 统计卡片（集市表，ETL 更新）──
+
+async def get_today_usage(env, sor):
+    """当天调用总数"""
+    sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$"
+    recs = await sor.sqlExe(sql, {'today': env.curDateString()})
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_today_amount(env, sor):
+    """当天总金额(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
+
+
+async def get_success_rate(env, sor):
+    """当天成功率"""
+    sql = """SELECT
+        COUNT(*) as total,
+        SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success
+    FROM dm_model_call_fact WHERE call_date = ${today}$"""
+    recs = await sor.sqlExe(sql, {'today': env.curDateString()})
+    if recs and recs[0].total:
+        return round(recs[0].success * 100.0 / recs[0].total, 1)
+    return 0
+
+
+async def get_fail_count(env, sor):
+    """当天失败数"""
+    sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND status = 'FAILED'"
+    recs = await sor.sqlExe(sql, {'today': env.curDateString()})
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_active_users(env, sor):
+    """当天活跃用户数"""
+    sql = "SELECT COUNT(DISTINCT userid) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$"
+    recs = await sor.sqlExe(sql, {'today': env.curDateString()})
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_avg_ttft(env, sor):
+    """当天平均 TTFT"""
+    sql = "SELECT AVG(avg_ttft_ms) as v FROM dm_model_perf_daily WHERE stat_date = ${today}$"
+    recs = await sor.sqlExe(sql, {'today': env.curDateString()})
+    if recs and recs[0].v:
+        return round(float(recs[0].v), 0)
+    return 0
+
+
+async def get_currency_breakdown(env, sor, org_filter=None):
+    """各币种分布（全量+可选org过滤）"""
+    w = ""
+    ns = {}
+    if org_filter:
+        w = " AND userorgid = ${org}$"
+        ns['org'] = org_filter
+    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 1=1""" + w + """
+    GROUP BY dm_model_call_fact.currency
+    ORDER BY call_count DESC"""
+    recs = await sor.sqlExe(sql, ns)
+    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, org_filter=None):
+    """Top N 模型（按调用次数，全量+可选org过滤）"""
+    w = ""
+    ns = {'limit': limit}
+    if org_filter:
+        w = " AND userorgid = ${org}$"
+        ns['org'] = org_filter
+    sql = ("SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount "
+           "FROM dm_model_call_fact WHERE 1=1" + w +
+           " GROUP BY model ORDER BY cnt DESC LIMIT ${limit}$")
+    recs = await sor.sqlExe(sql, ns)
+    return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 2)} for r in recs]
+
+
+async def get_top_providers(env, sor, limit=5, org_filter=None):
+    """Top N 供应商（按金额，全量+可选org过滤）"""
+    w = ""
+    ns = {'limit': limit}
+    if org_filter:
+        w = " AND f.userorgid = ${org}$"
+        ns['org'] = org_filter
+    sql = """SELECT f.providerid, COUNT(*) as cnt, COALESCE(SUM(f.amount_cny), 0) as total_amount,
+             COALESCE(o.orgname, f.providerid) as provider_name
+    FROM dm_model_call_fact f
+    LEFT JOIN organization o ON o.id = f.providerid
+    WHERE f.providerid IS NOT NULL""" + w + """
+    GROUP BY f.providerid ORDER BY total_amount DESC LIMIT ${limit}$"""
+    recs = await sor.sqlExe(sql, ns)
+    return [{'provider_name': r.provider_name or r.providerid or 'Unknown',
+             'cnt': int(r.cnt), 'total_amount': round(float(r.total_amount), 2)} for r in recs]
+
+
+async def get_top_users(env, sor, limit=5, org_filter=None):
+    """全量 Top N 用户（按金额+可选org过滤）"""
+    w = ""
+    ns = {'limit': limit}
+    if org_filter:
+        w = " AND f.userorgid = ${org}$"
+        ns['org'] = org_filter
+    sql = """SELECT f.userid, COUNT(*) as cnt, COALESCE(SUM(f.amount_cny), 0) as total_amount,
+             COALESCE(u.username, f.userid) as user_name
+    FROM dm_model_call_fact f
+    LEFT JOIN users u ON u.id = f.userid
+    WHERE 1=1""" + w + """
+    GROUP BY f.userid ORDER BY total_amount DESC LIMIT ${limit}$"""
+    recs = await sor.sqlExe(sql, ns)
+    return [{'user_name': r.user_name or r.userid, 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 2)} for r in recs]
+
+
+# ── 客户维度 ──
+
+async def get_customer_daily_models(env, sor, userorgid):
+    """客户当天各模型用量"""
+    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})
+    return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 4)} for r in recs]
+
+
+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_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})
+    return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 4)} for r in recs]
+
+
+async def get_customer_user_today(env, sor, userorgid):
+    """客户当天各用户用量"""
+    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})
+    return [{'user_name': r.userid, 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 4)} for r in recs]
+
+
+# ── 趋势 ──
+
+async def get_daily_trend(env, sor, days=7):
+    """最近 N 天调用量和金额趋势"""
+    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"""
+    start = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
+    recs = await sor.sqlExe(sql, {'start': start})
+    return [{'date': r.call_date, 'cnt': int(r.cnt),
+             'amount': round(float(r.total_amount), 2)} for r in recs]
+
+
+async def get_hourly_concurrency(env, sor):
+    """每小时并发（全量）"""
+    sql = """SELECT call_hour, COUNT(DISTINCT userid) as users, COUNT(*) as calls
+    FROM dm_model_call_fact
+    GROUP BY call_hour ORDER BY call_hour"""
+    recs = await sor.sqlExe(sql, {})
+    return [{'hour': r.call_hour, 'users': int(r.users), 'calls': int(r.calls)} for r in recs]
+
+
+# ── 分销商专用 ──
+
+async def get_distributor_customers_count(sor, distributor_orgid):
+    """分销商旗下客户数"""
+    sql = """SELECT COUNT(DISTINCT userorgid) as cnt FROM dm_model_call_fact
+             WHERE distributor_orgid = ${org}$"""
+    recs = await sor.sqlExe(sql, {'org': distributor_orgid})
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_distributor_active_customers(sor, today, distributor_orgid):
+    """分销商今日活跃客户数"""
+    sql = """SELECT COUNT(DISTINCT userorgid) as cnt FROM dm_model_call_fact
+             WHERE call_date = ${today}$ AND distributor_orgid = ${org}$"""
+    recs = await sor.sqlExe(sql, {'today': today, 'org': distributor_orgid})
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_distributor_top_customers(sor, distributor_orgid, limit=5):
+    """分销商旗下客户排行（按金额）"""
+    sql = """SELECT userorgid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
+    FROM dm_model_call_fact WHERE distributor_orgid = ${org}$
+    GROUP BY userorgid ORDER BY total_amount DESC LIMIT ${limit}$"""
+    recs = await sor.sqlExe(sql, {'org': distributor_orgid, 'limit': limit})
+    return [{'org_id': r.userorgid, 'cnt': int(r.cnt),
+             'total_amount': round(float(r.total_amount), 2)} for r in recs]
+
+
+# ── scope 过滤版（在全局函数基础上加 org_filter）──
+
+async def get_scoped_today_usage(sor, today, scope=None, org_filter=None):
+    w = _fact_where(scope, org_filter)
+    sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return int(recs[0].cnt) if recs else 0
+
+
+async def get_scoped_today_amount(sor, today, scope=None, org_filter=None):
+    w = _fact_where(scope, org_filter)
+    sql = "SELECT COALESCE(SUM(amount_cny), 0) as total FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return round(float(recs[0].total), 2) if recs else 0
+
+
+async def get_scoped_success_rate(sor, today, scope=None, org_filter=None):
+    w = _fact_where(scope, org_filter)
+    sql = ("SELECT COUNT(*) as total, SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success "
+           "FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w)
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    if recs and recs[0].total:
+        return round(recs[0].success * 100.0 / recs[0].total, 1)
+    return 0
+
+
+async def get_scoped_fail_count(sor, today, scope=None, org_filter=None):
+    w = _fact_where(scope, org_filter)
+    sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND status = 'FAILED' AND " + w
+    ns = {'today': today}
+    if org_filter:
+        ns['org'] = org_filter
+    recs = await sor.sqlExe(sql, ns)
+    return int(recs[0].cnt) if recs else 0
+
+
+# ── 分销商财务统计 ──
+
+from datetime import date as _date_module_date
+
+async def get_distributor_financial(sor, org_filter, period='today'):
+    """分销商财务概览：按币种统计收入/成本/毛利
+    period: 'today' | 'month' | 'year' | 'all'
+    收入含直接客户 + 二级分销商
+    """
+    today_str = _date_module_date.today().isoformat()
+
+    date_sql = ""
+    ns = {'org': org_filter, 'today': today_str}
+    if period == 'month':
+        date_sql = "AND f.call_date BETWEEN ${month_start}$ AND ${today}$"
+        ns['month_start'] = today_str[:7] + '-01'
+    elif period == 'year':
+        date_sql = "AND f.call_date BETWEEN ${year_start}$ AND ${today}$"
+        ns['year_start'] = today_str[:4] + '-01-01'
+    elif period == 'today':
+        date_sql = "AND f.call_date = ${today}$"
+
+    # 收入按币种（含直接客户 + 二级分销商）
+    sql_rev = f"""SELECT
+        COALESCE(f.currency, 'CNY') as currency,
+        COUNT(*) as call_count,
+        COALESCE(SUM(f.amount), 0) as revenue,
+        COALESCE(SUM(f.amount_cny), 0) as revenue_cny
+    FROM dm_model_call_fact f
+    WHERE f.distributor_orgid = ${{org}}$ {date_sql}
+    GROUP BY f.currency ORDER BY revenue_cny DESC"""
+
+    recs = await sor.sqlExe(sql_rev, ns)
+
+    revenue_by_currency = []
+    total_revenue_cny = 0.0
+    total_calls = 0
+    for r in recs:
+        total_revenue_cny += float(r.revenue_cny or 0)
+        total_calls += int(r.call_count or 0)
+        revenue_by_currency.append({
+            'currency': r.currency,
+            'call_count': int(r.call_count or 0),
+            'revenue': round(float(r.revenue or 0), 2),
+            'revenue_cny': round(float(r.revenue_cny or 0), 2),
+        })
+
+    # 成本（尝试从provider_cost归集，关联distributor_orgid）
+    total_cost_cny = 0.0
+    try:
+        sql_cost = f"""SELECT COALESCE(SUM(c.total_amount_cny), 0) as total_cost_cny
+        FROM dm_provider_cost_daily c
+        JOIN dm_model_call_fact f ON f.call_date = c.stat_date
+            AND f.providerid = c.providerid AND f.catelogid = c.catelogid
+        WHERE f.distributor_orgid = ${{org}}$ {date_sql}"""
+        cost_recs = await sor.sqlExe(sql_cost, ns)
+        if cost_recs:
+            total_cost_cny = round(float(cost_recs[0].total_cost_cny or 0), 2)
+    except Exception:
+        pass
+
+    gross_profit_cny = round(total_revenue_cny - total_cost_cny, 2)
+
+    return {
+        'revenue_by_currency': revenue_by_currency,
+        'total_revenue_cny': total_revenue_cny,
+        'total_cost_cny': total_cost_cny,
+        'gross_profit_cny': gross_profit_cny,
+        'total_calls': total_calls,
+        'period': period,
+    }
+
+
+async def get_customer_currency_stats(sor, org_filter, period='today'):
+    """客户费用统计：按币种+时间维度"""
+    today_str = _date_module_date.today().isoformat()
+    date_sql = ""
+    ns = {'org': org_filter, 'today': today_str}
+    if period == 'month':
+        date_sql = "AND f.call_date BETWEEN ${month_start}$ AND ${today}$"
+        ns['month_start'] = today_str[:7] + '-01'
+    elif period == 'year':
+        date_sql = "AND f.call_date BETWEEN ${year_start}$ AND ${today}$"
+        ns['year_start'] = today_str[:4] + '-01-01'
+    elif period == 'today':
+        date_sql = "AND f.call_date = ${today}$"
+
+    sql = f"""SELECT COALESCE(f.currency, 'CNY') as currency,
+        COUNT(*) as call_count,
+        COALESCE(SUM(f.amount), 0) as revenue,
+        COALESCE(SUM(f.amount_cny), 0) as revenue_cny
+    FROM dm_model_call_fact f
+    WHERE f.userorgid = ${{org}}$ {date_sql}
+    GROUP BY f.currency ORDER BY revenue_cny DESC"""
+async def get_financial_summary(sor, period='all'):
+    """平台级财务汇总：收入/成本/毛利（不分机构）
+    period: 'today' | 'month' | 'year' | 'all'
+    """
+    today_str = _date_module_date.today().isoformat()
+    ns = {'today': today_str}
+    date_sql = ""
+    if period == 'month':
+        date_sql = "AND call_date BETWEEN ${month_start}$ AND ${today}$"
+        ns['month_start'] = today_str[:7] + '-01'
+    elif period == 'year':
+        date_sql = "AND call_date BETWEEN ${year_start}$ AND ${today}$"
+        ns['year_start'] = today_str[:4] + '-01-01'
+    elif period == 'today':
+        date_sql = "AND call_date = ${today}$"
+
+    sql_rev = f"""SELECT
+        COALESCE(SUM(amount_cny), 0) as revenue_cny,
+        COUNT(*) as call_count
+    FROM dm_model_call_fact
+    WHERE 1=1 {date_sql}"""
+    recs = await sor.sqlExe(sql_rev, ns)
+    total_revenue_cny = round(float(recs[0].revenue_cny or 0), 2) if recs else 0
+    total_calls = int(recs[0].call_count or 0) if recs else 0
+
+    total_cost_cny = 0.0
+    try:
+        cost_sql = f"""SELECT COALESCE(SUM(c.total_amount_cny), 0) as total_cost_cny
+        FROM dm_provider_cost_daily c
+        WHERE 1=1 {date_sql.replace('call_date', 'c.stat_date')}"""
+        cost_recs = await sor.sqlExe(cost_sql, ns)
+        if cost_recs:
+            total_cost_cny = round(float(cost_recs[0].total_cost_cny or 0), 2)
+    except Exception:
+        pass
+
+    gross_profit_cny = round(total_revenue_cny - total_cost_cny, 2)
+
+    return {
+        'total_revenue_cny': total_revenue_cny,
+        'total_cost_cny': total_cost_cny,
+        'gross_profit_cny': gross_profit_cny,
+        'total_calls': total_calls,
+        'period': period,
+    }
diff --git a/build/lib/sage_datamart/etl.py b/build/lib/sage_datamart/etl.py
new file mode 100644
index 0000000..31c1adf
--- /dev/null
+++ b/build/lib/sage_datamart/etl.py
@@ -0,0 +1,283 @@
+"""
+数据集市 ETL — 增量同步 llmusage → 集市表
+建表由 models/*.json + json2ddl + build.sh 部署时完成，ETL 只负责数据读写。
+"""
+import json
+from datetime import datetime, timedelta
+from appPublic.uniqueID import getID
+from appPublic.dictObject import DictObject
+from appPublic.log import info
+
+MODULE_NAME = "sage_datamart"
+
+
+def parse_usages(usages_str):
+    """从 llmusage.usages JSON 中提取 token 数"""
+    try:
+        u = json.loads(usages_str) if isinstance(usages_str, str) else usages_str
+        if isinstance(u, dict):
+            return (
+                u.get('prompt_tokens', 0) or 0,
+                u.get('completion_tokens', 0) or 0,
+                u.get('total_tokens', 0) or 0
+            )
+    except:
+        pass
+    return 0, 0, 0
+
+
+async def sync_call_fact(sor, last_sync=None):
+    """增量同步 llmusage → dm_model_call_fact（含币种折算）
+    
+    金额优先从 product_usage_log.sell_price 取实际计费金额，
+    取不到时 fallback 到 llmusage.amount。
+    币种从 llmusage.amount_currency 取，按 exchange_rate 折算 CNY。
+    """
+    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)
+
+    # 预取 product_usage_log 实际计费金额（按 source_ref_table='llmusage' 关联）
+    usage_log_map = await _load_usage_log_bills(sor, last_sync)
+
+    sql = """
+        SELECT lu.*, l.model as llm_model, l.providerid,
+            (SELECT llmcatelogid FROM llm_api_map 
+             WHERE llmid = lu.llmid AND isdefaultcatelog = '1' LIMIT 1) as catelogid
+        FROM llmusage lu
+        JOIN llm l ON l.id = lu.llmid
+        WHERE lu.use_time >= ${last_sync}$
+        ORDER BY lu.use_time
+    """
+    rows = await sor.sqlExe(sql, {'last_sync': last_sync})
+    info('sage_datamart ETL: syncing ' + str(len(rows)) + ' rows from ' + last_sync)
+
+    count = 0
+    for r in rows:
+        r = DictObject(r)
+        # Skip existing
+        existing = await sor.sqlExe(
+            'SELECT id FROM dm_model_call_fact WHERE luid=${luid}$',
+            {'luid': r.id})
+        if existing:
+            continue
+
+        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'
+
+        # 优先取 product_usage_log.sell_price（实际计费金额），fallback 到 llmusage.amount
+        bill_data = usage_log_map.get(r.id)
+        if bill_data:
+            amount = bill_data.get('sell_price', 0) or 0
+            # product_usage_log 的币种来自 product_subscription.purchase_currency
+            bill_currency = bill_data.get('purchase_currency')
+            if bill_currency:
+                currency = bill_currency
+        else:
+            amount = r.amount or 0
+
+        rate = rates.get(currency, 1.0)
+        amount_cny = round(float(amount) * rate, 4)
+
+        ns = {
+            'id': getID(),
+            'luid': r.id,
+            'llmid': r.llmid,
+            'model': r.llm_model or r.model,
+            'catelogid': r.catelogid,
+            'userid': r.userid,
+            'userorgid': r.userorgid,
+            'ownerid': r.ownerid,
+            'providerid': r.providerid,
+            'call_date': r.use_date or call_time.strftime('%Y-%m-%d'),
+            'call_hour': call_time.hour,
+            'call_time': r.use_time,
+            'ttft_ms': int(r.responsed_seconds * 1000) if r.responsed_seconds else None,
+            'ttot_ms': int(r.finish_seconds * 1000) if r.finish_seconds else None,
+            'prompt_tokens': prompt_tokens,
+            'completion_tokens': completion_tokens,
+            'status': r.status or 'SUCCEEDED',
+            'amount': amount,
+            'currency': currency,
+            'amount_cny': amount_cny,
+        }
+        await sor.C('dm_model_call_fact', ns)
+        count += 1
+
+    info('sage_datamart ETL: inserted ' + str(count) + ' new rows')
+    return count
+
+
+async def _load_usage_log_bills(sor, last_sync):
+    """从 product_usage_log 预取实际计费金额"""
+    usage_log_map = {}
+    try:
+        sql = """
+            SELECT pul.source_ref_id, pul.sell_price, ps.purchase_currency
+            FROM product_usage_log pul
+            LEFT JOIN product_subscription ps ON ps.id = pul.subscription_id
+            WHERE pul.source_ref_table = 'llmusage'
+              AND pul.use_time >= ${last_sync}$
+        """
+        rows = await sor.sqlExe(sql, {'last_sync': last_sync})
+        for r in rows:
+            ref_id = r.source_ref_id if hasattr(r, 'source_ref_id') else r['source_ref_id']
+            sell_price = r.sell_price if hasattr(r, 'sell_price') else r['sell_price']
+            purchase_currency = r.purchase_currency if hasattr(r, 'purchase_currency') else r.get('purchase_currency')
+            if ref_id:
+                usage_log_map[ref_id] = {
+                    'sell_price': float(sell_price) if sell_price else 0.0,
+                    'purchase_currency': purchase_currency,
+                }
+    except Exception as e:
+        info('sage_datamart: product_usage_log lookup failed: ' + str(e))
+    return usage_log_map
+
+
+async def _get_exchange_rates(sor, stat_date):
+    """查询汇率表"""
+    rates = {'CNY': 1.0}
+    try:
+        sql = """SELECT from_currency, mid_rate FROM exchange_rate
+                 WHERE to_currency = 'CNY' AND effective_date <= ${d}$
+                 ORDER BY effective_date DESC"""
+        rows = await 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（按币种分组）"""
+    if stat_date is None:
+        stat_date = datetime.now().strftime('%Y-%m-%d')
+
+    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,
+            AVG(ttft_ms) as avg_ttft,
+            AVG(ttot_ms) as avg_ttot,
+            SUM(prompt_tokens) as total_prompt,
+            SUM(completion_tokens) as total_completion,
+            SUM(amount) as total_amount
+        FROM dm_model_call_fact
+        WHERE call_date = ${stat_date}$
+        GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, dm_model_call_fact.userorgid, dm_model_call_fact.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)
+        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(currency,"CNY")=${c}$ AND COALESCE(userorgid,"")=${u}$',
+            {'d': stat_date, 'l': r.llmid, 'c': cur, 'u': r.userorgid or ''})
+
+        ns = {
+            'id': getID(),
+            'stat_date': stat_date,
+            'llmid': r.llmid,
+            'model': r.model,
+            'catelogid': r.catelogid,
+            'providerid': r.providerid,
+            'ownerid': r.ownerid,
+            'total_calls': r.total_calls,
+            'success_calls': r.success_calls,
+            'fail_calls': r.fail_calls,
+            'success_rate': round(r.success_calls * 100.0 / r.total_calls, 2) if r.total_calls else 0,
+            'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
+            'avg_ttot_ms': round(r.avg_ttot, 1) if r.avg_ttot else 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)
+
+    return len(rows)
+
+
+async def aggregate_provider_cost(sor, stat_date=None):
+    """聚合同模型供应商性价比 — 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, dm_model_call_fact.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}$ AND COALESCE(currency,"CNY")=${c}$',
+            {'d': stat_date, 'm': r.model, 'p': r.providerid, 'c': cur})
+
+        ns = {
+            'id': getID(),
+            'stat_date': stat_date,
+            'model': r.model,
+            'providerid': r.providerid,
+            '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,
+        }
+        await sor.C('dm_provider_cost_daily', ns)
+
+    return len(rows)
+
+
+async def run_etl_sync(sor):
+    """每5分钟运行: 增量同步"""
+    await sync_call_fact(sor)
+
+
+async def run_etl_aggregate(sor):
+    """每小时运行: 聚合天级性能 + 并发统计"""
+    await aggregate_daily_perf(sor)
+
+
+async def run_etl_provider_cost(sor):
+    """每天运行: 供应商性价比"""
+    await aggregate_provider_cost(sor)
diff --git a/build/lib/sage_datamart/i18n/__init__.py b/build/lib/sage_datamart/i18n/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/build/lib/sage_datamart/i18n/en/__init__.py b/build/lib/sage_datamart/i18n/en/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/build/lib/sage_datamart/i18n/en/i18n.json b/build/lib/sage_datamart/i18n/en/i18n.json
new file mode 100644
index 0000000..8894cf5
--- /dev/null
+++ b/build/lib/sage_datamart/i18n/en/i18n.json
@@ -0,0 +1,26 @@
+{
+  "API Key排行": "API Key Ranking",
+  "供应商性价比": "Provider Cost-Performance",
+  "供应商排行": "Provider Ranking",
+  "币种分布": "Currency Distribution",
+  "我的模型(按金额Top5)": "My Models (Top 5 by Amount)",
+  "模型性能排行": "Model Performance Ranking",
+  "热门模型": "Popular Models",
+  "用户排行": "User Ranking",
+  "财务概览": "Financial Overview",
+  "今天调用": "Today",
+  "用户": "Users",
+  "今日调用笔数": "Today Calls",
+  "今日交易金额": "Today Amount",
+  "用户总数": "Total Users",
+  "今日活跃": "Today Active",
+  "在线用户": "Online Users",
+  "成功率": "Success Rate",
+  "失败数": "Failed",
+  "币种成本收入": "Currency Cost & Revenue",
+  "币种费用": "Currency Expenses",
+  "今天": "Today",
+  "本月": "This Month",
+  "本年": "This Year",
+  "累计": "Cumulative"
+}
\ No newline at end of file
diff --git a/build/lib/sage_datamart/i18n/jp/__init__.py b/build/lib/sage_datamart/i18n/jp/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/build/lib/sage_datamart/i18n/jp/i18n.json b/build/lib/sage_datamart/i18n/jp/i18n.json
new file mode 100644
index 0000000..976f2d3
--- /dev/null
+++ b/build/lib/sage_datamart/i18n/jp/i18n.json
@@ -0,0 +1,26 @@
+{
+  "API Key排行": "APIキーランキング",
+  "供应商性价比": "プロバイダーコストパフォーマンス",
+  "供应商排行": "プロバイダーランキング",
+  "币种分布": "通貨分布",
+  "我的模型(按金额Top5)": "マイモデル（金額Top5）",
+  "模型性能排行": "モデルパフォーマンスランキング",
+  "热门模型": "人気モデル",
+  "用户排行": "ユーザーランキング",
+  "财务概览": "財務概要",
+  "今天调用": "本日",
+  "用户": "ユーザー",
+  "今日调用笔数": "本日コール数",
+  "今日交易金额": "本日取引額",
+  "用户总数": "総ユーザー数",
+  "今日活跃": "本日アクティブ",
+  "在线用户": "オンラインユーザー",
+  "成功率": "成功率",
+  "失败数": "失敗数",
+  "币种成本收入": "通貨コスト収入",
+  "币种费用": "通貨費用",
+  "今天": "今日",
+  "本月": "今月",
+  "本年": "今年",
+  "累计": "累計"
+}
\ No newline at end of file
diff --git a/build/lib/sage_datamart/i18n/ko/__init__.py b/build/lib/sage_datamart/i18n/ko/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/build/lib/sage_datamart/i18n/ko/i18n.json b/build/lib/sage_datamart/i18n/ko/i18n.json
new file mode 100644
index 0000000..a1c1f34
--- /dev/null
+++ b/build/lib/sage_datamart/i18n/ko/i18n.json
@@ -0,0 +1,26 @@
+{
+  "API Key排行": "API 키 랭킹",
+  "供应商性价比": "공급업체 가성비",
+  "供应商排行": "공급업체 랭킹",
+  "币种分布": "통화 분포",
+  "我的模型(按金额Top5)": "내 모델 (금액 Top5)",
+  "模型性能排行": "모델 성능 랭킹",
+  "热门模型": "인기 모델",
+  "用户排行": "사용자 랭킹",
+  "财务概览": "재무 개요",
+  "今天调用": "오늘",
+  "用户": "사용자",
+  "今日调用笔数": "오늘 호출",
+  "今日交易金额": "오늘 거래액",
+  "用户总数": "전체 사용자",
+  "今日活跃": "오늘 활성",
+  "在线用户": "온라인 사용자",
+  "成功率": "성공률",
+  "失败数": "실패수",
+  "币种成本收入": "통화 비용 수익",
+  "币种费用": "통화 비용",
+  "今天": "오늘",
+  "本月": "이번 달",
+  "本年": "올해",
+  "累计": "누계"
+}
\ No newline at end of file
diff --git a/build/lib/sage_datamart/i18n/zh/__init__.py b/build/lib/sage_datamart/i18n/zh/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/build/lib/sage_datamart/i18n/zh/i18n.json b/build/lib/sage_datamart/i18n/zh/i18n.json
new file mode 100644
index 0000000..9db34c4
--- /dev/null
+++ b/build/lib/sage_datamart/i18n/zh/i18n.json
@@ -0,0 +1,26 @@
+{
+  "API Key排行": "API Key排行",
+  "供应商性价比": "供应商性价比",
+  "供应商排行": "供应商排行",
+  "币种分布": "币种分布",
+  "我的模型(按金额Top5)": "我的模型(按金额Top5)",
+  "模型性能排行": "模型性能排行",
+  "热门模型": "热门模型",
+  "用户排行": "用户排行",
+  "财务概览": "财务概览",
+  "今天调用": "今天调用",
+  "用户": "用户",
+  "今日调用笔数": "今日调用笔数",
+  "今日交易金额": "今日交易金额",
+  "用户总数": "用户总数",
+  "今日活跃": "今日活跃",
+  "在线用户": "在线用户",
+  "成功率": "成功率",
+  "失败数": "失败数",
+  "币种成本收入": "币种成本收入",
+  "币种费用": "币种费用",
+  "今天": "今天",
+  "本月": "本月",
+  "本年": "本年",
+  "累计": "累计"
+}
\ No newline at end of file
diff --git a/build/lib/sage_datamart/init.py b/build/lib/sage_datamart/init.py
new file mode 100644
index 0000000..087c652
--- /dev/null
+++ b/build/lib/sage_datamart/init.py
@@ -0,0 +1,621 @@
+"""sage_datamart 模块初始化 — 统一 Dashboard API，支持客户/分销商角色"""
+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
+
+from .dashboards import (
+    CUSTOMER, DISTRIBUTOR,
+    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_currency_breakdown,
+    get_realtime_calls, get_realtime_amount, get_total_users_count,
+    get_realtime_active_users, get_online_users, get_realtime_stats,
+    get_distributor_customers_count, get_distributor_active_customers,
+    get_distributor_top_customers,
+    get_scoped_today_usage, get_scoped_today_amount,
+    get_scoped_success_rate, get_scoped_fail_count,
+    get_distributor_financial, get_customer_currency_stats,
+    get_financial_summary,
+)
+
+MODULE_NAME = "sage_datamart"
+
+
+async def _resolve_view(request):
+    """解析当前用户角色和机构过滤条件
+    返回 (scope, org_filter) — scope: 'customer'|'distributor'|None, org_filter: org_id|None
+    org_filter=None 表示不按机构过滤（平台管理员看全部数据）
+    """
+    env = request._run_ns
+    userorgid = await env.get_userorgid()
+    scope = CUSTOMER
+    org_filter = userorgid
+    # 平台根机构('0')无过滤，看全部数据
+    if not userorgid or userorgid == '0':
+        return None, None
+    try:
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            recs = await sor.sqlExe(
+                "SELECT org_type FROM organization WHERE id = ${id}$",
+                {'id': userorgid})
+            if recs:
+                org_type = (recs[0].org_type if hasattr(recs[0], 'org_type')
+                            else recs[0].get('org_type', ''))
+                if org_type == 'reseller':
+                    scope = DISTRIBUTOR
+    except Exception as e:
+        debug("sage_datamart _resolve_view error: " + str(e))
+
+        pass
+    return scope, org_filter
+
+
+# ── 性能指标 API ──
+
+async def api_model_perf(sor, params_kw=None):
+    try:
+        stat_date = (params_kw or {}).get('date')
+        sql = """SELECT COALESCE(l.name, f.model) as model,
+            CASE WHEN SUM(total_calls)>0 THEN ROUND(SUM(total_calls*COALESCE(avg_ttft_ms,avg_ttot_ms))/SUM(total_calls),1) ELSE NULL END as avg_latency_ms,
+            CASE WHEN SUM(total_calls)>0 THEN ROUND(SUM(total_calls*avg_ttot_ms)/SUM(total_calls),1) ELSE NULL END as avg_ttot_ms,
+            ROUND(SUM(success_calls)*100.0/GREATEST(SUM(total_calls),1),1) as success_rate
+        FROM dm_model_perf_daily f
+        LEFT JOIN llm l ON l.id = f.llmid
+        WHERE 1=1"""
+        ns = {}
+        if stat_date:
+            sql += " AND f.stat_date=${stat_date}$"
+            ns['stat_date'] = stat_date
+        sql += " GROUP BY f.model, f.catelogid, f.providerid"
+        sql += " ORDER BY avg_latency_ms ASC LIMIT 5"
+        rows = await sor.sqlExe(sql, ns)
+        return [dict(r) for r in rows]
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_provider_roi(sor, params_kw=None):
+    try:
+        model = (params_kw or {}).get('model')
+        stat_date = (params_kw or {}).get('date')
+        sql = """SELECT p.*, COALESCE(o.orgname, p.providerid) as provider_name
+        FROM dm_provider_cost_daily p
+        LEFT JOIN organization o ON o.id = p.providerid
+        WHERE 1=1"""
+        ns = {}
+        if stat_date:
+            sql += " AND stat_date=${stat_date}$"
+            ns['stat_date'] = stat_date
+        if model:
+            sql += " AND model=${model}$"
+            ns['model'] = model
+        sql += " ORDER BY total_amount DESC LIMIT 50"
+        rows = await sor.sqlExe(sql, ns)
+        return [dict(r) for r in rows]
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+# ── 运营指标 API ──
+
+async def api_stats(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return {
+                'today_usage': await get_today_usage(env, sor),
+                'today_amount': await get_today_amount(env, sor),
+                'success_rate': await get_success_rate(env, sor),
+                'fail_count': await get_fail_count(env, sor),
+                'active_users': await get_active_users(env, sor),
+            }
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return {}
+
+
+async def api_top_models(request):
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_models(env, sor, org_filter=org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_top_providers(request):
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_providers(env, sor, org_filter=org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_top_users(request):
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_users(env, sor, org_filter=org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_customer_models(request):
+    try:
+        env = request._run_ns
+        userorgid = await env.get_userorgid()
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return {
+                'daily': await get_customer_daily_models(env, sor, userorgid),
+                'monthly': await get_customer_monthly_models(env, sor, userorgid),
+            }
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return {'daily': [], 'monthly': []}
+
+
+async def api_daily_trend(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_daily_trend(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_hourly_concurrency(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_hourly_concurrency(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+async def api_currency_stats(request):
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_currency_breakdown(env, sor, org_filter=org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+# ── 实时指标 API（角色感知）──
+
+async def api_realtime_cards(request):
+    """实时统计卡片 — /sage_datamart/api/realtime_cards.dspy"""
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        today = env.curDateString()
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            if scope == DISTRIBUTOR:
+                return {
+                    'scope': scope,
+                    'calls': await get_realtime_calls(sor, today, scope, org_filter),
+                    'amount': await get_realtime_amount(sor, today, scope, org_filter),
+                    'total_users': await get_distributor_customers_count(sor, org_filter),
+                    'active_customers': await get_distributor_active_customers(sor, today, org_filter),
+                    'online_users': await get_online_users(env),
+                    'success_rate': await get_scoped_success_rate(sor, today, scope, org_filter),
+                    'fail_count': await get_scoped_fail_count(sor, today, scope, org_filter),
+                }
+            else:
+                return {
+                    'scope': scope,
+                    'calls': await get_realtime_calls(sor, today, scope, org_filter),
+                    'amount': await get_realtime_amount(sor, today, scope, org_filter),
+                    'total_users': await get_total_users_count(sor, scope, org_filter),
+                    'active_users': await get_realtime_active_users(sor, today, scope, org_filter),
+                    'online_users': await get_online_users(env),
+                    'success_rate': await get_scoped_success_rate(sor, today, scope, org_filter),
+                    'fail_count': await get_scoped_fail_count(sor, today, scope, org_filter),
+                }
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return {}
+
+
+# ── 分销商专用 API ──
+
+async def api_distributor_customers(request):
+    """分销商客户排行 — /sage_datamart/api/distributor_customers.dspy"""
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_distributor_top_customers(sor, org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return []
+
+
+# ── Jinja2 模板函数（实时，角色感知）──
+
+async def _get_realtime_snapshot(request):
+    """一次查询返回全部实时指标(含角色)，同一请求内缓存复用"""
+    cache_key = '_sage_datamart_rt_cache'
+    if cache_key in request:
+        return request[cache_key]
+    env = request._run_ns
+    scope, org_filter = await _resolve_view(request)
+    today = env.curDateString()
+    async with get_sor_context(env, MODULE_NAME) as sor:
+        result = await get_realtime_stats(sor, today, scope, org_filter)
+    request[cache_key] = result
+    return result
+
+
+async def j2_realtime_calls(request):
+    try:
+        s = await _get_realtime_snapshot(request)
+        return s.get('calls', 0)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_realtime_amount(request):
+    try:
+        s = await _get_realtime_snapshot(request)
+        return s.get('amount', 0)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_total_users(request):
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            if scope == DISTRIBUTOR:
+                return await get_distributor_customers_count(sor, org_filter)
+            return await get_total_users_count(sor, scope, org_filter)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_active_today(request):
+    try:
+        s = await _get_realtime_snapshot(request)
+        return s.get('active_users', 0)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_online_users(request):
+    try:
+        env = request._run_ns
+        return await get_online_users(env)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_scoped_success_rate(request):
+    try:
+        s = await _get_realtime_snapshot(request)
+        return s.get('success_rate', 0)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+async def j2_scoped_fail_count(request):
+    try:
+        s = await _get_realtime_snapshot(request)
+        return s.get('fail_count', 0)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return 0
+
+
+# ── 向后兼容 Jinja2 函数 ──
+
+async def j2_today_usage(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_today_usage(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+async def j2_today_amount(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_today_amount(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+async def j2_active_users(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_active_users(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+async def j2_fail_count(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_fail_count(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+async def j2_success_rate(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_success_rate(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+async def j2_avg_ttft(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_avg_ttft(env, sor)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+
+        return 0
+
+
+
+async def api_user_apikeys(request):
+    """当前用户每个 API Key 的使用排行"""
+    try:
+        env = request._run_ns
+        userid = await env.get_userid()
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            sql = """SELECT uak.name as apikey_name, COUNT(*) as cnt, COALESCE(SUM(lu.amount), 0) as total_amount
+            FROM llmusage lu
+            JOIN userapikey uak ON uak.apikey = lu.apikey
+            WHERE lu.userid = ${userid}$
+            GROUP BY uak.name ORDER BY cnt DESC LIMIT 10"""
+            recs = await sor.sqlExe(sql, {'userid': userid})
+            return [{'apikey_name': r.apikey_name or 'Unknown',
+                     'cnt': int(r.cnt or 0),
+                     'total_amount': round(float(r.total_amount or 0), 2)} for r in recs]
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return []
+
+
+async def api_distributor_financial(request):
+    """分销商财务统计 API"""
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        period = (request._run_ns.params_kw or {}).get('period', 'today')
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_distributor_financial(sor, org_filter, period)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return {}
+
+
+async def j2_distributor_financial(request, period='today'):
+    """Jinja2: 分销商财务数据"""
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_distributor_financial(sor, org_filter, period)
+    except Exception:
+        return {'total_revenue_cny': 0, 'total_cost_cny': 0, 'gross_profit_cny': 0,
+                'revenue_by_currency': [], 'total_calls': 0, 'period': period}
+
+
+async def j2_is_distributor(request):
+    """Jinja2: 是否分销商角色"""
+    try:
+        scope, _ = await _resolve_view(request)
+        return scope == DISTRIBUTOR
+    except Exception:
+        return False
+
+
+async def j2_customer_currency(request, period='today'):
+    """Jinja2: 客户币种费用"""
+    try:
+        env = request._run_ns
+        scope, org_filter = await _resolve_view(request)
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_customer_currency_stats(sor, org_filter, period)
+    except Exception:
+        return []
+
+# ── 监控组件 j2 函数，每段独立 .ui 通过 RefreshWidget 加载 ──
+
+async def j2_model_perf_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await api_model_perf(sor)
+    except Exception:
+        return []
+
+
+async def j2_currency_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_currency_breakdown(env, sor)
+    except Exception:
+        return []
+
+
+async def j2_provider_roi_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await api_provider_roi(sor)
+    except Exception:
+        return []
+
+
+async def j2_top_models_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_models(env, sor)
+    except Exception:
+        return []
+
+
+async def j2_top_users_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_users(env, sor)
+    except Exception:
+        return []
+
+
+async def j2_top_providers_data(request):
+    try:
+        env = request._run_ns
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_top_providers(env, sor)
+    except Exception:
+        return []
+
+
+async def api_financial_summary(request):
+    """平台级财务汇总 API"""
+    try:
+        env = request._run_ns
+        period = (request._run_ns.params_kw or {}).get('period', 'all')
+        async with get_sor_context(env, MODULE_NAME) as sor:
+            return await get_financial_summary(sor, period)
+    except Exception as e:
+        debug("sage_datamart API error: " + str(e))
+        return {}
+
+
+# ── Cron ETL ──
+
+async def cron_etl_sync(request):
+    ip = request['client_ip']
+    if ip not in ['127.0.0.1']:
+        return {'error': f'IP {ip} not allowed'}
+    env = request._run_ns
+    async with get_sor_context(env, MODULE_NAME) as sor:
+        count = await sync_call_fact(sor)
+        return {'status': 'ok', 'synced': count}
+
+
+async def cron_etl_aggregate(request):
+    ip = request['client_ip']
+    if ip not in ['127.0.0.1']:
+        return {'error': f'IP {ip} not allowed'}
+    env = request._run_ns
+    async with get_sor_context(env, MODULE_NAME) as sor:
+        count = await aggregate_daily_perf(sor)
+        return {'status': 'ok', 'aggregated': count}
+
+
+async def cron_etl_provider_cost(request):
+    ip = request['client_ip']
+    if ip not in ['127.0.0.1']:
+        return {'error': f'IP {ip} not allowed'}
+    env = request._run_ns
+    async with get_sor_context(env, MODULE_NAME) as sor:
+        count = await aggregate_provider_cost(sor)
+        return {'status': 'ok', 'aggregated': count}
+
+
+def load_sage_datamart():
+    env = ServerEnv()
+    env.api_model_perf = api_model_perf
+    env.api_provider_roi = api_provider_roi
+    env.api_stats = api_stats
+    env.api_top_models = api_top_models
+    env.api_top_providers = api_top_providers
+    env.api_top_users = api_top_users
+    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.api_realtime_cards = api_realtime_cards
+    env.api_distributor_customers = api_distributor_customers
+    env.j2_today_usage = j2_today_usage
+    env.j2_today_amount = j2_today_amount
+    env.j2_active_users = j2_active_users
+    env.j2_fail_count = j2_fail_count
+    env.j2_success_rate = j2_success_rate
+    env.j2_avg_ttft = j2_avg_ttft
+    env.j2_realtime_calls = j2_realtime_calls
+    env.j2_realtime_amount = j2_realtime_amount
+    env.j2_total_users = j2_total_users
+    env.j2_active_today = j2_active_today
+    env.j2_online_users = j2_online_users
+    env.j2_scoped_success_rate = j2_scoped_success_rate
+    env.j2_scoped_fail_count = j2_scoped_fail_count
+    env.api_user_apikeys = api_user_apikeys
+    env.api_distributor_financial = api_distributor_financial
+    env.j2_distributor_financial = j2_distributor_financial
+    env.j2_is_distributor = j2_is_distributor
+    env.j2_customer_currency = j2_customer_currency
+    env.j2_model_perf_data = j2_model_perf_data
+    env.j2_currency_data = j2_currency_data
+    env.j2_provider_roi_data = j2_provider_roi_data
+    env.j2_top_models_data = j2_top_models_data
+    env.j2_top_users_data = j2_top_users_data
+    env.j2_top_providers_data = j2_top_providers_data
+    env.api_financial_summary = api_financial_summary
+    env.cron_etl_sync = cron_etl_sync
+    env.cron_etl_aggregate = cron_etl_aggregate
+    env.cron_etl_provider_cost = cron_etl_provider_cost
+
+    debug("[sage_datamart] registered all API + realtime + role-aware endpoints")
diff --git a/sage_datamart/dashboards.py b/sage_datamart/dashboards.py
index 589d618..443c339 100644
--- a/sage_datamart/dashboards.py
+++ b/sage_datamart/dashboards.py
@@ -514,6 +514,11 @@ async def get_customer_currency_stats(sor, org_filter, period='today'):
     FROM dm_model_call_fact f
     WHERE f.userorgid = ${{org}}$ {date_sql}
     GROUP BY f.currency ORDER BY revenue_cny DESC"""
+    recs = await sor.sqlExe(sql, ns)
+    return [{'currency': r.currency, 'call_count': int(r.call_count or 0),
+             'revenue': round(float(r.revenue or 0), 2),
+             'revenue_cny': round(float(r.revenue_cny or 0), 2)} for r in recs]
+
 async def get_financial_summary(sor, period='all'):
     """平台级财务汇总：收入/成本/毛利（不分机构）
     period: 'today' | 'month' | 'year' | 'all'