無料ユーザー一人あたりのトークン予算を決める — AdMob 収益と AI 機能のコストを釣り合わせる設計

# gemini_budget.py
from dataclasses import dataclass
from google import genai
 
client = genai.Client()
 
# 自分の請求から逆算した実効単価（円 / 100万トークン）。四半期ごとに見直します。
PRICE_JPY_PER_MTOK = {
    "input": 45.0,
    "cached": 11.0,
    "output": 360.0,
    "thoughts": 360.0,
}
 
 
@dataclass
class Usage:
    fresh_input: int
    cached_input: int
    output: int
    thoughts: int
 
    @property
    def cost_jpy(self) -> float:
        p = PRICE_JPY_PER_MTOK
        return (
            self.fresh_input * p["input"]
            + self.cached_input * p["cached"]
            + self.output * p["output"]
            + self.thoughts * p["thoughts"]
        ) / 1_000_000
 
    @property
    def weighted_tokens(self) -> int:
        # 予算は「入力換算トークン」で持つと、モデルを替えても台帳が壊れません
        return int(self.cost_jpy / (PRICE_JPY_PER_MTOK["input"] / 1_000_000))
 
 
def extract_usage(response) -> Usage:
    u = response.usage_metadata
    cached = getattr(u, "cached_content_token_count", 0) or 0
    thoughts = getattr(u, "thoughts_token_count", 0) or 0
    prompt = u.prompt_token_count or 0
    output = u.candidates_token_count or 0
    return Usage(
        fresh_input=max(prompt - cached, 0),
        cached_input=cached,
        output=output,
        thoughts=thoughts,
    )

# budget_guard.py
SOFT_CAP = 0.70
MONTHLY_BUDGET_TOKENS = 7_500
 
MODEL_LADDER = {
    "normal": ("gemini-3-pro", {"thinking_level": "high"}),
    "reduced": ("gemini-flash-latest", {"thinking_level": "low"}),
}
 
 
class BudgetExceeded(Exception):
    pass
 
 
def select_tier(spent_tokens: int) -> str:
    ratio = spent_tokens / MONTHLY_BUDGET_TOKENS
    if ratio >= 1.0:
        raise BudgetExceeded(f"budget exhausted: {ratio:.0%}")
    return "reduced" if ratio >= SOFT_CAP else "normal"
 
 
def generate_for_user(ledger, user_id: str, prompt: str):
    spent = ledger.spent_this_period(user_id)
    tier = select_tier(spent)              # ここで 100% 超は例外
    model, config = MODEL_LADDER[tier]
 
    response = client.models.generate_content(
        model=model, contents=prompt, config=config
    )
    usage = extract_usage(response)
    ledger.record(user_id, usage.weighted_tokens, usage.cost_jpy, tier)
    return response.text, tier

# concentration.py
import statistics
 
 
def top_share(spend_by_user: dict[str, float], top_pct: float = 0.01) -> float:
    values = sorted(spend_by_user.values(), reverse=True)
    if not values:
        return 0.0
    k = max(1, int(len(values) * top_pct))
    return sum(values[:k]) / sum(values)
 
 
def anomaly_report(spend_by_user: dict[str, float]) -> dict:
    values = list(spend_by_user.values())
    median = statistics.median(values)
    outliers = {
        uid: v for uid, v in spend_by_user.items()
        if median > 0 and v > median * 20
    }
    return {
        "top1_share": round(top_share(spend_by_user), 3),
        "median_tokens": median,
        "outlier_users": sorted(outliers, key=outliers.get, reverse=True)[:20],
    }