Leveraging Gemini API's Cost Advantage for SaaS — How to Undercut Competitors by 50% and Still Profit

from google import genai
from enum import Enum
import re
 
class ModelChoice(Enum):
    LITE = "gemini-2.5-lite"
    FLASH = "gemini-2.5-flash"
    PRO = "gemini-2.5-pro"
 
def estimate_complexity(prompt: str, context_length: int) -> ModelChoice:
    """
    Estimate request complexity to auto-select the right model.
    
    Scoring:
    - Presence of reasoning keywords (analyze, compare, design, optimize, etc.)
    - Input context size (longer = more complex)
    - Task type (judgment-free classification vs. creative/strategic)
    """
    complexity_score = 0
    
    reasoning_keywords = [
        "analyze", "compare", "trade-off", "reasoning", "strategy",
        "design", "optimize", "evaluate", "justify", "explain why",
        "pros and cons", "recommendation", "insight", "pattern"
    ]
    
    # Count reasoning keywords
    prompt_lower = prompt.lower()
    complexity_score += sum(
        1 for kw in reasoning_keywords 
        if kw in prompt_lower
    ) * 2
    
    # Context size penalty
    if context_length > 50000:
        complexity_score += 4
    elif context_length > 20000:
        complexity_score += 3
    elif context_length > 5000:
        complexity_score += 2
    elif context_length > 1000:
        complexity_score += 1
    
    # Simple task detection (high priority override)
    simple_keywords = ["translate", "format", "convert", "extract", "rewrite"]
    if any(kw in prompt_lower for kw in simple_keywords):
        if complexity_score < 2:
            return ModelChoice.LITE
    
    # Final routing
    if complexity_score >= 5:
        return ModelChoice.PRO
    elif complexity_score >= 2:
        return ModelChoice.FLASH
    else:
        return ModelChoice.LITE
 
async def call_gemini(
    prompt: str,
    context: str = "",
    system_prompt: str = "",
    user_id: str = None
) -> dict:
    """
    Call Gemini with intelligent model routing.
    Returns response + metadata for cost tracking.
    """
    client = genai.Client()
    
    # Estimate complexity and select model
    model_choice = estimate_complexity(prompt, len(context))
    
    # Build full prompt
    full_text = f"{system_prompt}\n\n{context}\n\n{prompt}"
    
    # API call
    response = await client.aio.models.generate_content(
        model=model_choice.value,
        contents=[{"role": "user", "parts": [{"text": full_text}]}],
        config=genai.types.GenerateContentConfig(
            temperature=0.7,
            max_output_tokens=2000,
        ),
    )
    
    usage = response.usage_metadata
    
    return {
        "response": response.text,
        "model_used": model_choice.value,
        "input_tokens": usage.prompt_token_count,
        "output_tokens": usage.candidates_token_count,
        "estimated_cost_usd": calculate_cost(
            model_choice.value,
            usage.prompt_token_count,
            usage.candidates_token_count,
        ),
    }
 
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """Calculate actual API cost in USD."""
    rates = {
        "gemini-2.5-lite": {"in": 0.0375 / 1_000_000, "out": 0.15 / 1_000_000},
        "gemini-2.5-flash": {"in": 0.075 / 1_000_000, "out": 0.30 / 1_000_000},
        "gemini-2.5-pro": {"in": 1.25 / 1_000_000, "out": 5.00 / 1_000_000},
    }
    rate = rates.get(model, rates["gemini-2.5-flash"])
    return (input_tokens * rate["in"]) + (output_tokens * rate["out"])

async def analyze_with_cache(
    user_id: str,
    document: str,
    question: str,
    system_instructions: str
) -> dict:
    """
    Analyze a document repeatedly, leveraging context cache
    to reduce token costs by 90% on cache hits.
    """
    client = genai.Client()
    
    # System prompt (cacheable)
    system_content = [
        {
            "type": "text",
            "text": system_instructions,
            "cache_control": {"type": "ephemeral"}  # Cache for 5 minutes
        },
        {
            "type": "text",
            "text": f"Document to analyze:\n\n{document}",
            "cache_control": {"type": "ephemeral"}
        }
    ]
    
    # Make request with cache
    response = await client.aio.models.generate_content(
        model="gemini-2.5-flash",
        contents=[
            {
                "role": "user",
                "parts": [
                    *system_content,
                    {"text": question}
                ]
            }
        ]
    )
    
    usage = response.usage_metadata
    
    # Extract cache statistics
    cache_read_tokens = getattr(usage, 'cache_read_input_token_count', 0)
    cache_hit = cache_read_tokens > 0
    
    # Cost breakdown
    input_cost = usage.prompt_token_count * (0.075 / 1_000_000)
    cache_cost = cache_read_tokens * (0.075 / 1_000_000) * 0.1
    output_cost = usage.candidates_token_count * (0.30 / 1_000_000)
    
    return {
        "response": response.text,
        "cache_hit": cache_hit,
        "cache_tokens": cache_read_tokens,
        "total_cost_usd": input_cost + cache_cost + output_cost,
        "cost_without_cache_usd": (
            usage.prompt_token_count * (0.075 / 1_000_000) +
            usage.candidates_token_count * (0.30 / 1_000_000)
        ),
        "savings_usd": (
            (usage.prompt_token_count - cache_read_tokens) * 
            (0.075 / 1_000_000) * 0.9
        )
    }

import json
import time
from google import genai
 
async def batch_process_documents(
    documents: list[dict],
    analysis_prompt: str
) -> list[dict]:
    """
    Process multiple documents via Batch API.
    Results available in seconds to minutes.
    50% cost savings vs. regular API.
    """
    client = genai.Client()
    
    # Build batch requests
    requests = []
    for i, doc in enumerate(documents):
        requests.append({
            "custom_id": f"doc_{i}",
            "request_body": {
                "contents": [
                    {
                        "role": "user",
                        "parts": [
                            {
                                "text": f"{analysis_prompt}\n\n{doc['content']}"
                            }
                        ]
                    }
                ],
                "generationConfig": {
                    "temperature": 0.7,
                    "maxOutputTokens": 500
                }
            }
        })
    
    # Submit batch
    batch = await client.aio.batches.create(
        requests=requests,
        model="gemini-2.5-flash"
    )
    
    # Poll until complete (usually <1 minute)
    max_wait = 300  # 5 minutes
    elapsed = 0
    while elapsed < max_wait:
        batch_status = await client.aio.batches.get(batch.name)
        
        if batch_status.state == "COMPLETED":
            break
        
        time.sleep(2)
        elapsed += 2
    
    # Collect results
    results = []
    if hasattr(batch_status, 'results'):
        for result in batch_status.results:
            results.append({
                "doc_id": result.custom_id,
                "analysis": result.response.candidates[0].content.parts[0].text,
                "cost_savings_percent": 50
            })
    
    return results

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import asyncio
 
@dataclass
class APICall:
    user_id: str
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    is_batch: bool = False
    
    def cost_usd(self) -> float:
        """Calculate actual cost for this call."""
        rates = {
            "gemini-2.5-lite": {"in": 0.0375, "out": 0.15},
            "gemini-2.5-flash": {"in": 0.075, "out": 0.30},
            "gemini-2.5-pro": {"in": 1.25, "out": 5.00},
        }
        
        r = rates.get(self.model, rates["gemini-2.5-flash"])
        
        # Input cost (cache is 90% off)
        input_cost = (self.input_tokens * r["in"]) / 1_000_000
        cache_cost = (self.cache_read_tokens * r["in"]) * 0.1 / 1_000_000
        
        # Output cost
        output_cost = (self.output_tokens * r["out"]) / 1_000_000
        
        total = input_cost + cache_cost + output_cost
        
        # Batch is 50% off
        if self.is_batch:
            total *= 0.5
        
        return total
 
class UsageDB:
    """Simple interface for tracking API usage."""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    async def log_call(self, call: APICall) -> None:
        """Log an API call for cost tracking."""
        await self.db.execute("""
            INSERT INTO api_calls 
            (user_id, timestamp, model, input_tokens, output_tokens,
             cache_read_tokens, is_batch, cost_usd)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            call.user_id, call.timestamp, call.model,
            call.input_tokens, call.output_tokens,
            call.cache_read_tokens, call.is_batch,
            call.cost_usd()
        ))
    
    async def monthly_summary(self, year: int, month: int) -> dict:
        """Get cost breakdown by model for a month."""
        rows = await self.db.fetch("""
            SELECT 
                model,
                COUNT(*) as call_count,
                SUM(input_tokens) as total_input,
                SUM(output_tokens) as total_output,
                SUM(cache_read_tokens) as cache_tokens,
                SUM(cost_usd) as total_cost
            FROM api_calls
            WHERE YEAR(timestamp) = ? AND MONTH(timestamp) = ?
            GROUP BY model
            ORDER BY total_cost DESC
        """, (year, month))
        
        total_cost = sum(r['total_cost'] for r in rows)
        
        return {
            "period": f"{year}-{month:02d}",
            "by_model": [dict(r) for r in rows],
            "total_cost_usd": total_cost,
            "total_calls": sum(r['call_count'] for r in rows),
            "avg_cost_per_call": total_cost / sum(r['call_count'] for r in rows),
        }
    
    async def user_profitability(self, year: int, month: int) -> dict:
        """Identify which users are profitable."""
        rows = await self.db.fetch("""
            SELECT 
                u.user_id,
                u.plan,
                SUM(ac.cost_usd) as api_cost,
                CASE 
                    WHEN u.plan = 'pro' THEN 49
                    WHEN u.plan = 'standard' THEN 24
                    ELSE 0
                END as monthly_mrr
            FROM users u
            LEFT JOIN api_calls ac 
                ON u.user_id = ac.user_id
                AND YEAR(ac.timestamp) = ?
                AND MONTH(ac.timestamp) = ?
            GROUP BY u.user_id
        """, (year, month))
        
        profitable = [
            r for r in rows 
            if r['monthly_mrr'] and r['api_cost'] < r['monthly_mrr'] * 0.4
        ]
        
        return {
            "total_users": len(rows),
            "profitable": len(profitable),
            "profitability_rate": len(profitable) / len(rows) if rows else 0,
            "avg_api_cost": sum(r['api_cost'] for r in rows) / len(rows) if rows else 0,
        }