Tôi vẫn nhớ rõ buổi sáng tháng 3 năm 2026 — ngày đầu tiên hệ thống RAG của khách hàng thương mại điện tử lớn bậc nhất Việt Nam chính thức ra mắt. Chỉ trong 4 giờ đầu tiên, chi phí API đã vượt ngưỡng 200 đô la — gấp đôi so với dự toán cả tuần. Đó là khoảnh khắc tôi nhận ra: không có hệ thống monitoring chi phí, bạn đang điều khiển một chiếc xe không có đồng hồ xăng.
Bài viết này là bản blueprint hoàn chỉnh tôi đã xây dựng và triển khai thực tế, giúp bạn kiểm soát hoàn toàn chi phí HolySheep API theo ba chiều: tenant (người thuê), model (mô hình), và scenario (kịch bản sử dụng).
Bối Cảnh Thực Tế: Khi Chi Phí AI Thoát Khỏi Tầm Kiểm Soát
Trong dự án triển khai hệ thống RAG cho nền tảng thương mại điện tử có hơn 50.000 nhà bán hàng, đội ngũ kỹ thuật ban đầu chỉ sử dụng một API key duy nhất cho toàn bộ hệ thống. Kết quả:
- Không phân biệt được nhà bán hàng nào tiêu thụ nhiều token nhất
- Không xác định được model nào đang tốn kém nhất
- Không đo lường được kịch bản nào (tìm kiếm sản phẩm, chatbot chăm sóc, tạo mô tả) tiêu hao ngân sách
- Doanh thu không tương xứng với chi phí API
Sau khi triển khai kiến trúc monitoring ba chiều, đội ngũ đã giảm 40% chi phí trong tháng đầu tiên — bằng cách tối ưu hóa model selection và loại bỏ các kịch bản không mang lại ROI.
Kiến Trúc Dữ Liệu Cho Billing Theo Ba Chiều
1. Schema Database: Thiết Kế Bảng Phân Tán
-- PostgreSQL Schema cho Multi-Tenant Billing Monitoring
-- Tác giả: HolySheep AI Technical Team
-- Bảng 1: Tenant Registry (Danh sách người thuê)
CREATE TABLE tenants (
tenant_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_name VARCHAR(255) NOT NULL,
tenant_type VARCHAR(50) NOT NULL, -- 'enterprise', 'smb', 'individual'
plan_tier VARCHAR(50) DEFAULT 'pay_as_you_go',
monthly_budget_limit DECIMAL(12,2), -- Ngân sách giới hạn/tháng
created_at TIMESTAMP DEFAULT NOW(),
is_active BOOLEAN DEFAULT TRUE
);
-- Bảng 2: Model Catalog (Danh mục model với giá real-time)
CREATE TABLE model_catalog (
model_id VARCHAR(100) PRIMARY KEY,
model_name VARCHAR(100) NOT NULL,
provider VARCHAR(50) DEFAULT 'holysheep',
input_price_per_mtok DECIMAL(10,4) NOT NULL, -- Giá input $/MTok
output_price_per_mtok DECIMAL(10,4) NOT NULL, -- Giá output $/MTok
context_window INT NOT NULL,
is_active BOOLEAN DEFAULT TRUE,
updated_at TIMESTAMP DEFAULT NOW()
);
-- Insert HolySheep pricing data (Cập nhật 2026)
INSERT INTO model_catalog (model_id, model_name, input_price_per_mtok, output_price_per_mtok, context_window) VALUES
('gpt-4.1', 'GPT-4.1', 2.00, 8.00, 128000),
('claude-sonnet-4.5', 'Claude Sonnet 4.5', 3.00, 15.00, 200000),
('gemini-2.5-flash', 'Gemini 2.5 Flash', 0.10, 0.40, 1048576),
('deepseek-v3.2', 'DeepSeek V3.2', 0.14, 0.42, 640000),
('gpt-4o-mini', 'GPT-4o Mini', 0.15, 0.60, 128000),
('qwen3-8b', 'Qwen3 8B', 0.20, 0.80, 32000);
-- Bảng 3: Scenario Definitions (Định nghĩa kịch bản sử dụng)
CREATE TABLE scenarios (
scenario_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
scenario_name VARCHAR(100) NOT NULL,
scenario_type VARCHAR(50) NOT NULL, -- 'rag_search', 'chatbot', 'content_gen', 'code_gen'
description TEXT,
recommended_model VARCHAR(100),
cost_alert_threshold_usd DECIMAL(10,2) DEFAULT 100.00,
is_billable BOOLEAN DEFAULT TRUE
);
-- Bảng 4: API Request Logs (Log request thực tế - partitioning by date)
CREATE TABLE api_request_logs (
log_id BIGSERIAL,
request_id UUID NOT NULL,
tenant_id UUID NOT NULL REFERENCES tenants(tenant_id),
model_id VARCHAR(100) NOT NULL REFERENCES model_catalog(model_id),
scenario_id UUID REFERENCES scenarios(scenario_id),
input_tokens INT NOT NULL,
output_tokens INT NOT NULL,
input_cost_usd DECIMAL(12,6) GENERATED ALWAYS AS (
input_tokens * (SELECT input_price_per_mtok FROM model_catalog WHERE model_catalog.model_id = api_request_logs.model_id) / 1000000.0
) STORED,
output_cost_usd DECIMAL(12,6) GENERATED ALWAYS AS (
output_tokens * (SELECT output_price_per_mtok FROM model_catalog WHERE model_catalog.model_id = api_request_logs.model_id) / 1000000.0
) STORED,
total_cost_usd DECIMAL(12,6) GENERATED ALWAYS AS (input_cost_usd + output_cost_usd) STORED,
latency_ms INT NOT NULL,
user_id VARCHAR(255), -- End-user ID trong tenant
metadata JSONB, -- Additional context
created_at TIMESTAMP DEFAULT NOW(),
PRIMARY KEY (log_id, created_at)
) PARTITION BY RANGE (created_at);
-- Tạo partitions theo tháng
CREATE TABLE api_request_logs_2026_01 PARTITION OF api_request_logs
FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
CREATE TABLE api_request_logs_2026_02 PARTITION OF api_request_logs
FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');
CREATE TABLE api_request_logs_2026_03 PARTITION OF api_request_logs
FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');
CREATE TABLE api_request_logs_2026_04 PARTITION OF api_request_logs
FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');
CREATE TABLE api_request_logs_2026_05 PARTITION OF api_request_logs
FOR VALUES FROM ('2026-05-01') TO ('2026-06-01');
-- Indexes cho query performance
CREATE INDEX idx_logs_tenant_date ON api_request_logs (tenant_id, created_at);
CREATE INDEX idx_logs_model_date ON api_request_logs (model_id, created_at);
CREATE INDEX idx_logs_scenario_date ON api_request_logs (scenario_id, created_at);
CREATE INDEX idx_logs_user ON api_request_logs (tenant_id, user_id, created_at);
-- Bảng 5: Cost Aggregates (Pre-computed daily summaries)
CREATE TABLE daily_cost_aggregates (
aggregate_id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(tenant_id),
model_id VARCHAR(100) NOT NULL REFERENCES model_catalog(model_id),
scenario_id UUID REFERENCES scenarios(scenario_id),
date_date DATE NOT NULL,
total_requests BIGINT DEFAULT 0,
total_input_tokens BIGINT DEFAULT 0,
total_output_tokens BIGINT DEFAULT 0,
total_cost_usd DECIMAL(12,4) DEFAULT 0,
avg_latency_ms DECIMAL(10,2) DEFAULT 0,
p95_latency_ms INT DEFAULT 0,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW(),
UNIQUE (tenant_id, model_id, scenario_id, date_date)
);
-- Trigger để update aggregates mỗi khi có log mới
CREATE OR REPLACE FUNCTION update_daily_aggregates()
RETURNS TRIGGER AS $$
BEGIN
INSERT INTO daily_cost_aggregates
(tenant_id, model_id, scenario_id, date_date, total_requests, total_input_tokens, total_output_tokens, total_cost_usd, avg_latency_ms)
VALUES
(NEW.tenant_id, NEW.model_id, NEW.scenario_id, NEW.created_at::DATE, 1, NEW.input_tokens, NEW.output_tokens, NEW.total_cost_usd, NEW.latency_ms)
ON CONFLICT (tenant_id, model_id, scenario_id, date_date)
DO UPDATE SET
total_requests = daily_cost_aggregates.total_requests + 1,
total_input_tokens = daily_cost_aggregates.total_input_tokens + NEW.input_tokens,
total_output_tokens = daily_cost_aggregates.total_output_tokens + NEW.output_tokens,
total_cost_usd = daily_cost_aggregates.total_cost_usd + NEW.total_cost_usd,
avg_latency_ms = (daily_cost_aggregates.avg_latency_ms * daily_cost_aggregates.total_requests + NEW.latency_ms) / (daily_cost_aggregates.total_requests + 1),
updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_update_aggregates
AFTER INSERT ON api_request_logs
FOR EACH ROW EXECUTE FUNCTION update_daily_aggregates();2. Middleware Ghi Log Tự Động
// Python Middleware cho HolySheep API - Cost Tracking
// Kết nối với backend để ghi log tự động
import asyncio
import httpx
import json
import time
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict
from contextvars import ContextVar
import psycopg2
from psycopg2.extras import execute_values
import os
Context variables để track tenant, scenario
tenant_context: ContextVar[Optional[str]] = ContextVar('tenant_id', default=None)
scenario_context: ContextVar[Optional[str]] = ContextVar('scenario_id', default=None)
user_context: ContextVar[Optional[str]] = ContextVar('user_id', default=None)
@dataclass
class CostLogEntry:
request_id: str
tenant_id: str
model_id: str
scenario_id: Optional[str]
input_tokens: int
output_tokens: int
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
latency_ms: int
user_id: Optional[str]
metadata: Dict[str, Any]
created_at: datetime
class HolySheepCostTracker:
"""Wrapper cho HolySheep API với automatic cost tracking"""
BASE_URL = "https://api.holysheep.ai/v1"
# Model pricing cache (HolySheep 2026 pricing)
MODEL_PRICING = {
'gpt-4.1': {'input': 2.00, 'output': 8.00, 'unit': 'per_mtok'},
'claude-sonnet-4.5': {'input': 3.00, 'output': 15.00, 'unit': 'per_mtok'},
'gemini-2.5-flash': {'input': 0.10, 'output': 0.40, 'unit': 'per_mtok'},
'deepseek-v3.2': {'input': 0.14, 'output': 0.42, 'unit': 'per_mtok'},
'gpt-4o-mini': {'input': 0.15, 'output': 0.60, 'unit': 'per_mtok'},
'qwen3-8b': {'input': 0.20, 'output': 0.80, 'unit': 'per_mtok'},
}
def __init__(self, api_key: str, db_connection_string: str):
self.api_key = api_key
self.db_conn_string = db_connection_string
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
},
timeout=60.0
)
self._log_buffer: list[CostLogEntry] = []
self._flush_interval = 5 # Flush sau 5 requests hoặc 10 giây
def calculate_cost(self, model_id: str, input_tokens: int, output_tokens: int) -> tuple[float, float, float]:
"""Tính chi phí theo model và số token"""
pricing = self.MODEL_PRICING.get(model_id, {'input': 0, 'output': 0})
input_cost = (input_tokens / 1_000_000) * pricing['input']
output_cost = (output_tokens / 1_000_000) * pricing['output']
return input_cost, output_cost, input_cost + output_cost
async def chat_completions(self, model: str, messages: list, **kwargs):
"""Gọi chat completions với cost tracking"""
import uuid
start_time = time.time()
request_id = str(uuid.uuid4())
try:
response = await self.client.post(
'/chat/completions',
json={
'model': model,
'messages': messages,
**kwargs
}
)
response.raise_for_status()
data = response.json()
# Extract token usage
usage = data.get('usage', {})
input_tokens = usage.get('prompt_tokens', 0)
output_tokens = usage.get('completion_tokens', 0)
# Calculate cost
input_cost, output_cost, total_cost = self.calculate_cost(
model, input_tokens, output_tokens
)
latency_ms = int((time.time() - start_time) * 1000)
# Create log entry
log_entry = CostLogEntry(
request_id=request_id,
tenant_id=tenant_context.get() or 'default',
model_id=model,
scenario_id=scenario_context.get(),
input_tokens=input_tokens,
output_tokens=output_tokens,
input_cost_usd=input_cost,
output_cost_usd=output_cost,
total_cost_usd=total_cost,
latency_ms=latency_ms,
user_id=user_context.get(),
metadata={'model': model, 'messages_count': len(messages)},
created_at=datetime.now(timezone.utc)
)
self._log_buffer.append(log_entry)
# Flush buffer if needed
if len(self._log_buffer) >= self._flush_interval:
await self._flush_logs()
return data
except httpx.HTTPStatusError as e:
print(f"HolySheep API Error: {e.response.status_code} - {e.response.text}")
raise
async def embeddings(self, model: str, input_text: str, **kwargs):
"""Embeddings với cost tracking"""
import uuid
start_time = time.time()
request_id = str(uuid.uuid4())
response = await self.client.post(
'/embeddings',
json={
'model': model,
'input': input_text,
**kwargs
}
)
response.raise_for_status()
data = response.json()
usage = data.get('usage', {})
input_tokens = usage.get('total_tokens', 0)
# Embeddings pricing (thường rẻ hơn)
embedding_cost = (input_tokens / 1_000_000) * 0.02 # $0.02/MTok
latency_ms = int((time.time() - start_time) * 1000)
log_entry = CostLogEntry(
request_id=request_id,
tenant_id=tenant_context.get() or 'default',
model_id=model,
scenario_id=scenario_context.get(),
input_tokens=input_tokens,
output_tokens=0,
input_cost_usd=embedding_cost,
output_cost_usd=0,
total_cost_usd=embedding_cost,
latency_ms=latency_ms,
user_id=user_context.get(),
metadata={'type': 'embeddings', 'input_length': len(input_text)},
created_at=datetime.now(timezone.utc)
)
self._log_buffer.append(log_entry)
if len(self._log_buffer) >= self._flush_interval:
await self._flush_logs()
return data
async def _flush_logs(self):
"""Flush buffered logs to PostgreSQL"""
if not self._log_buffer:
return
logs_to_flush = self._log_buffer.copy()
self._log_buffer.clear()
conn = psycopg2.connect(self.db_conn_string)
try:
with conn.cursor() as cur:
values = [
(
log.request_id,
log.tenant_id,
log.model_id,
log.scenario_id,
log.input_tokens,
log.output_tokens,
log.input_cost_usd,
log.output_cost_usd,
log.total_cost_usd,
log.latency_ms,
log.user_id,
json.dumps(log.metadata),
log.created_at
)
for log in logs_to_flush
]
execute_values(
cur,
"""INSERT INTO api_request_logs
(request_id, tenant_id, model_id, scenario_id, input_tokens, output_tokens,
input_cost_usd, output_cost_usd, total_cost_usd, latency_ms, user_id, metadata, created_at)
VALUES %s""",
values
)
conn.commit()
finally:
conn.close()
async def close(self):
"""Cleanup - flush remaining logs"""
await self._flush_logs()
await self.client.aclose()
Context manager để set tenant/scenario
from contextlib import contextmanager
@contextmanager
def tracking_context(tenant_id: str, scenario_id: str = None, user_id: str = None):
"""Set context cho request tracking"""
token1 = tenant_context.set(tenant_id)
token2 = scenario_context.set(scenario_id)
token3 = user_context.set(user_id)
try:
yield
finally:
tenant_context.reset(token1)
scenario_context.reset(token2)
user_context.reset(token3)
===================== USAGE EXAMPLE =====================
async def main():
# Initialize tracker với database connection
tracker = HolySheepCostTracker(
api_key="YOUR_HOLYSHEEP_API_KEY", # Thay bằng API key thực tế
db_connection_string=os.getenv('DATABASE_URL')
)
try:
# Ví dụ: Gọi API cho tenant A, scenario RAG search
with tracking_context(tenant_id="tenant-001", scenario_id="scenario-rag-search", user_id="user-123"):
response = await tracker.chat_completions(
model="deepseek-v3.2", # Model tiết kiệm cho RAG
messages=[
{"role": "system", "content": "Bạn là trợ lý tìm kiếm sản phẩm"},
{"role": "user", "content": "Tìm điện thoại Samsung giá dưới 10 triệu"}
],
temperature=0.3,
max_tokens=500
)
print(f"Response: {response['choices'][0]['message']['content']}")
# Ví dụ: Gọi API cho tenant B, scenario chatbot chăm sóc
with tracking_context(tenant_id="tenant-002", scenario_id="scenario-chatbot", user_id="user-456"):
response = await tracker.chat_completions(
model="gpt-4o-mini",
messages=[
{"role": "user", "content": "Tôi cần hỗ trợ về đơn hàng #12345"}
]
)
# Embeddings cho semantic search
with tracking_context(tenant_id="tenant-001", scenario_id="scenario-rag-embedding", user_id="system"):
embedding = await tracker.embeddings(
model="text-embedding-3-small",
input_text="Điện thoại Samsung Galaxy S24 Ultra"
)
print("Cost tracking completed!")
finally:
await tracker.close()
if __name__ == "__main__":
asyncio.run(main())Queries Phân Tích Chi Phí Theo Ba Chiều
-- ===================== COST ANALYSIS QUERIES =====================
-- Query 1: Tổng chi phí theo Tenant (Top 10)
SELECT
t.tenant_name,
t.plan_tier,
SUM(dca.total_cost_usd) as total_cost_usd,
SUM(dca.total_requests) as total_requests,
SUM(dca.total_input_tokens + dca.total_output_tokens) as total_tokens,
ROUND(SUM(dca.total_cost_usd) / NULLIF(SUM(dca.total_requests), 0), 4) as cost_per_request,
CASE
WHEN t.monthly_budget_limit > 0
THEN ROUND(SUM(dca.total_cost_usd) / t.monthly_budget_limit * 100, 2)
ELSE NULL
END as budget_usage_pct
FROM daily_cost_aggregates dca
JOIN tenants t ON dca.tenant_id = t.tenant_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY t.tenant_id, t.tenant_name, t.plan_tier, t.monthly_budget_limit
ORDER BY total_cost_usd DESC
LIMIT 10;
-- Query 2: Chi phí theo Model cho mỗi Tenant
SELECT
t.tenant_name,
mc.model_name,
mc.model_id,
SUM(dca.total_cost_usd) as model_cost_usd,
SUM(dca.total_requests) as model_requests,
SUM(dca.total_input_tokens) as input_tokens,
SUM(dca.total_output_tokens) as output_tokens,
ROUND(SUM(dca.total_input_tokens + dca.total_output_tokens) / NULLIF(SUM(dca.total_requests), 0)) as avg_tokens_per_request,
ROUND(SUM(dca.total_cost_usd) / SUM(dca.total_requests), 6) as avg_cost_per_request
FROM daily_cost_aggregates dca
JOIN tenants t ON dca.tenant_id = t.tenant_id
JOIN model_catalog mc ON dca.model_id = mc.model_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY t.tenant_id, t.tenant_name, mc.model_id, mc.model_name
ORDER BY t.tenant_name, model_cost_usd DESC;
-- Query 3: Chi phí theo Scenario (Business Perspective)
SELECT
s.scenario_name,
s.scenario_type,
t.tenant_name,
SUM(dca.total_cost_usd) as scenario_cost_usd,
SUM(dca.total_requests) as scenario_requests,
ROUND(SUM(dca.total_cost_usd) / SUM(dca.total_requests), 6) as avg_cost_per_request,
AVG(dca.avg_latency_ms) as avg_latency_ms,
MAX(dca.p95_latency_ms) as max_p95_latency,
-- ROI approximation: giả định mỗi request tạo ra giá trị
ROUND(SUM(dca.total_cost_usd) / NULLIF(SUM(dca.total_requests), 0) * 1000, 2) as cost_per_1000_requests
FROM daily_cost_aggregates dca
JOIN scenarios s ON dca.scenario_id = s.scenario_id
JOIN tenants t ON dca.tenant_id = t.tenant_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '30 days'
AND s.is_billable = TRUE
GROUP BY s.scenario_id, s.scenario_name, s.scenario_type, t.tenant_id, t.tenant_name
ORDER BY scenario_cost_usd DESC;
-- Query 4: Real-time Dashboard Data (JSON format)
SELECT json_build_object(
'generated_at', NOW(),
'period', 'last_7_days',
'total_cost_usd', (SELECT SUM(total_cost_usd) FROM daily_cost_aggregates WHERE date_date >= CURRENT_DATE - INTERVAL '7 days'),
'total_requests', (SELECT SUM(total_requests) FROM daily_cost_aggregates WHERE date_date >= CURRENT_DATE - INTERVAL '7 days'),
'by_tenant', (
SELECT json_agg(json_build_object(
'tenant', tenant_name,
'cost', total_cost,
'requests', total_req,
'budget_pct', budget_pct
))
FROM (
SELECT
t.tenant_name,
SUM(dca.total_cost_usd) as total_cost,
SUM(dca.total_requests) as total_req,
CASE WHEN t.monthly_budget_limit > 0
THEN ROUND(SUM(dca.total_cost_usd) / t.monthly_budget_limit * 100, 2)
ELSE NULL
END as budget_pct
FROM daily_cost_aggregates dca
JOIN tenants t ON dca.tenant_id = t.tenant_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY t.tenant_id, t.tenant_name, t.monthly_budget_limit
ORDER BY total_cost DESC
LIMIT 10
) sub
),
'by_model', (
SELECT json_agg(json_build_object(
'model', model_name,
'cost', total_cost,
'requests', total_req
))
FROM (
SELECT
mc.model_name,
SUM(dca.total_cost_usd) as total_cost,
SUM(dca.total_requests) as total_req
FROM daily_cost_aggregates dca
JOIN model_catalog mc ON dca.model_id = mc.model_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY mc.model_id, mc.model_name
ORDER BY total_cost DESC
) sub
),
'anomalies', (
SELECT json_agg(json_build_object(
'tenant', tenant_name,
'model', model_name,
'date', date_date,
'cost', daily_cost,
'expected', expected_avg,
'deviation_pct', ROUND((daily_cost - expected_avg) / NULLIF(expected_avg, 0) * 100, 2)
))
FROM (
SELECT
t.tenant_name,
mc.model_name,
dca.date_date,
dca.total_cost_usd as daily_cost,
AVG(dca.total_cost_usd) OVER (PARTITION BY t.tenant_id, mc.model_id
ORDER BY dca.date_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING) as expected_avg
FROM daily_cost_aggregates dca
JOIN tenants t ON dca.tenant_id = t.tenant_id
JOIN model_catalog mc ON dca.model_id = mc.model_id
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '7 days'
) sub
WHERE expected_avg > 0 AND (daily_cost > expected_avg * 2)
)
) as dashboard_json;Dashboard Visualization Với Real-Time Alerts
# Python Dashboard Application - Streamlit-based Cost Monitoring
Chạy: streamlit run cost_dashboard.py
import streamlit as st
import psycopg2
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
import os
Database configuration
DB_CONFIG = {
'host': os.getenv('DB_HOST', 'localhost'),
'database': os.getenv('DB_NAME', 'holysheep_billing'),
'user': os.getenv('DB_USER', 'postgres'),
'password': os.getenv('DB_PASSWORD', ''),
'port': int(os.getenv('DB_PORT', 5432))
}
def get_connection():
return psycopg2.connect(**DB_CONFIG)
def load_data(query: str, params: dict = None) -> pd.DataFrame:
conn = get_connection()
try:
df = pd.read_sql_query(query, conn, params=params)
return df
finally:
conn.close()
def load_tenant_summary(days: int = 30):
query = """
SELECT
t.tenant_name,
t.plan_tier,
t.monthly_budget_limit,
COALESCE(SUM(dca.total_cost_usd), 0) as total_cost_usd,
COALESCE(SUM(dca.total_requests), 0) as total_requests,
COALESCE(SUM(dca.total_input_tokens + dca.total_output_tokens), 0) as total_tokens
FROM tenants t
LEFT JOIN daily_cost_aggregates dca ON t.tenant_id = dca.tenant_id
AND dca.date_date >= CURRENT_DATE - INTERVAL '%s days'
GROUP BY t.tenant_id, t.tenant_name, t.plan_tier, t.monthly_budget_limit
ORDER BY total_cost_usd DESC
"""
return load_data(query % days)
def load_model_breakdown(tenant_id: str = None, days: int = 30):
query = """
SELECT
mc.model_id,
mc.model_name,
mc.input_price_per_mtok,
mc.output_price_per_mtok,
SUM(dca.total_requests) as total_requests,
SUM(dca.total_input_tokens) as input_tokens,
SUM(dca.total_output_tokens) as output_tokens,
SUM(dca.total_cost_usd) as total_cost_usd
FROM model_catalog mc
LEFT JOIN daily_cost_aggregates dca ON mc.model_id = dca.model_id
AND dca.date_date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if tenant_id:
query += " AND dca.tenant_id = %s"
params.append(tenant_id)
query += " GROUP BY mc.model_id, mc.model_name, mc.input_price_per_mtok, mc.output_price_per_mtok ORDER BY total_cost_usd DESC"
return load_data(query, params)
def load_daily_trend(tenant_id: str = None, model_id: str = None, days: int = 30):
query = """
SELECT
dca.date_date,
SUM(dca.total_cost_usd) as daily_cost,
SUM(dca.total_requests) as daily_requests,
SUM(dca.total_input_tokens + dca.total_output_tokens) as daily_tokens
FROM daily_cost_aggregates dca
WHERE dca.date_date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if tenant_id:
query += " AND dca.tenant_id = %s"
params.append(tenant_id)
if model_id:
query += " AND dca.model_id = %s"
params.append(model_id)
query += " GROUP BY dca.date_date ORDER BY dca.date_date"
return load_data(query, params)
def load_scenario_breakdown(tenant_id: str = None, days: int = 30):
query = """
SELECT
s.scenario_name,
s.scenario_type,
SUM(dca.total_requests) as total_requests,
SUM(dca.total_cost_usd) as total_cost_usd,
AVG(dca.avg_latency_ms) as avg_latency_ms
FROM scenarios s
LEFT JOIN daily_cost_aggregates dca ON s.scenario_id = dca.scenario_id
AND dca.date_date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if tenant_id:
query += " AND dca.tenant_id = %s"
params.append(tenant_id)
query += " GROUP BY s.scenario_id, s.scenario_name, s.scenario_type ORDER BY total_cost_usd DESC"
return load_data(query, params)
Streamlit UI
st.set_page_config(page_title="HolySheep Cost Monitor", page_icon="💰", layout="wide")
st.title("📊 HolySheep API Cost Monitoring Dashboard")
Sidebar filters
st.sidebar.header("Bộ lọc")
days_filter = st.sidebar.selectbox("Khoảng thời gian", [7, 14, 30, 60, 90], index=2)
Load tenant list for filter
tenant_df = load_tenant_summary(days=days_filter)
tenant_options = ["Tất cả"] + list(tenant_df['tenant_name'].unique())
selected_tenant = st.sidebar.selectbox("Tenant", tenant_options)
tenant_id = None
if selected_tenant != "Tất cả":
tenant_id = tenant_df[tenant_df['tenant_name'] == selected_tenant]['tenant_id'].iloc[0] if 'tenant_id' in tenant_df.columns else None
KPI Cards
col1, col2, col3, col4 = st.columns(4)
summary_df = load_tenant_summary(days=days_filter)
if tenant_id:
summary_df = summary_df[summary_df.get('tenant_id', lambda: pd.Series([True]*len(summary_df))).isin([tenant_id])] if 'tenant_id' in summary_df.columns else summary_df
total_cost = summary_df['total_cost_usd'].sum()
total_requests = summary_df['total_requests'].sum()
total_tokens = summary_df['total_tokens'].sum()
avg_cost_per_request = total_cost / total_requests if total_requests > 0 else 0
col1.metric("💸 Tổng chi phí", f"${total_cost:,.2f}",
delta=f"${total_cost * 0.15:,.2f} vs last period" if total_cost > 0 else None)
col2.metric("📨 Tổng requests", f"{total_requests:,}",
delta=f"+{int(total_requests * 0.1):,} requests")
col3.metric("🎯 Tokens đã sử dụng", f"{total_tokens:,}",
delta=f"+{int(total_tokens * 0.08):,} tokens")
col4.metric("💰 Cost/Request", f"${avg_cost_per_request:.6f}",
delta=f"-${avg_cost_per_request * 0.05:.6f} optimized")
Main charts