🚀 Auto-deploy backend from GitHub (cffaf6f)

config/ai_pricing.py

@@ -0,0 +1,60 @@
+# backend/config/ai_pricing.py
+# DeepSeek V4 API Pricing Configuration
+# TODO: Review pricing after 2026-05-31
+
+from datetime import datetime, timezone
+
+DEEPSEEK_PRICING = {
+    "deepseek-v4-pro": {
+        "promotional": {
+            "active": True,
+            "expires_utc": datetime(2026, 5, 31, 15, 59, 0, tzinfo=timezone.utc),
+            "input_cache_hit_per_1m": 0.003625,
+            "input_cache_miss_per_1m": 0.435,
+            "output_per_1m": 0.87,
+        },
+        "full_price": {
+            "input_cache_hit_per_1m": 0.0145,
+            "input_cache_miss_per_1m": 1.74,
+            "output_per_1m": 3.48,
+        },
+    },
+    "deepseek-v4-flash": {
+        "input_cache_hit_per_1m": 0.0028,
+        "input_cache_miss_per_1m": 0.14,
+        "output_per_1m": 0.28,
+    },
+}
+
+
+def get_active_pricing(model_id: str) -> dict:
+    """Returns the currently active pricing tier for a given model."""
+    model = DEEPSEEK_PRICING.get(model_id)
+    if not model:
+        raise ValueError(f"Unknown model: {model_id}")
+    if "promotional" in model:
+        promo = model["promotional"]
+        if promo["active"] and datetime.now(timezone.utc) < promo["expires_utc"]:
+            return {
+                "input_cache_hit_per_1m": promo["input_cache_hit_per_1m"],
+                "input_cache_miss_per_1m": promo["input_cache_miss_per_1m"],
+                "output_per_1m": promo["output_per_1m"],
+                "is_promotional": True,
+                "promo_expires_utc": promo["expires_utc"].isoformat(),
+            }
+        return {**model["full_price"], "is_promotional": False}
+    return {**model, "is_promotional": False}
+
+
+def get_full_pricing(model_id: str) -> dict:
+    """Returns the full (non-promotional) pricing for a model."""
+    model = DEEPSEEK_PRICING.get(model_id)
+    if not model:
+        raise ValueError(f"Unknown model: {model_id}")
+    if "full_price" in model:
+        return model["full_price"]
+    return {
+        "input_cache_hit_per_1m": model["input_cache_hit_per_1m"],
+        "input_cache_miss_per_1m": model["input_cache_miss_per_1m"],
+        "output_per_1m": model["output_per_1m"],
+    }

main.py

@@ -103,6 +103,7 @@ from routes.class_records_router import router as class_records_router
 from routes.risk_router import router as risk_router
 from routes.tutor_checkin import router as tutor_checkin_router
 from routes.practice import router as practice_router
+from routes.ai_monitoring import router as ai_monitoring_router
 
 # Rate limiting (slowapi)
 try:
@@ -1145,6 +1146,7 @@ app.include_router(class_records_router)
 app.include_router(risk_router)
 app.include_router(tutor_checkin_router)
 app.include_router(practice_router)
+app.include_router(ai_monitoring_router)
 
 
 # ─── Global Exception Handler ─────────────────────────────────

routes/ai_monitoring.py

@@ -0,0 +1,160 @@
+# backend/routes/ai_monitoring.py
+# TODO: Review pricing after 2026-05-31
+from datetime import datetime, timezone
+from fastapi import APIRouter, Depends, HTTPException, Request
+import logging
+
+from config.ai_pricing import get_active_pricing, get_full_pricing, DEEPSEEK_PRICING
+from services.cost_calculator import calculate_feature_cost, calculate_full_price_cost
+
+logger = logging.getLogger("mathpulse.ai_monitoring")
+
+router = APIRouter(prefix="/api/admin/ai-monitoring", tags=["admin", "ai-monitoring"])
+
+
+def require_admin(request: Request):
+    user = getattr(request.state, "user", None)
+    if user is None:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    if user.role not in ("admin", "superadmin"):
+        raise HTTPException(status_code=403, detail="Admin access required")
+    return user
+
+
+def _build_pricing_meta(model_id: str = "deepseek-v4-pro") -> dict:
+    """Build pricingMeta block for response."""
+    pricing = get_active_pricing(model_id)
+    full = get_full_pricing(model_id)
+    now = datetime.now(timezone.utc)
+    promo_config = DEEPSEEK_PRICING.get(model_id, {}).get("promotional", {})
+    expires = promo_config.get("expires_utc", now)
+    days_remaining = max(0, (expires - now).days) if pricing.get("is_promotional") else 0
+
+    return {
+        "activeModel": model_id,
+        "isPromotional": pricing.get("is_promotional", False),
+        "promoExpiresUtc": expires.isoformat() if pricing.get("is_promotional") else None,
+        "daysUntilPromoEnds": days_remaining,
+        "currentInputCacheMissRate": pricing["input_cache_miss_per_1m"],
+        "currentOutputRate": pricing["output_per_1m"],
+        "fullPriceInputRate": full["input_cache_miss_per_1m"],
+        "fullPriceOutputRate": full["output_per_1m"],
+    }
+
+
+def _aggregate_summary() -> dict:
+    """
+    Aggregate AI monitoring summary from in-memory/mock data.
+    In production, this reads from Firestore ai_usage_logs collection.
+    """
+    # TODO: Replace with actual Firestore aggregation when usage logging is wired
+    model_id = "deepseek-v4-pro"
+    pricing = get_active_pricing(model_id)
+
+    # Feature definitions with estimated token distributions
+    features_config = [
+        {"id": "ai_chat_tutor", "name": "AI Chat Tutor", "model": model_id, "share": 0.35, "cache_hit_rate": 0.62, "icon": "MessageCircle"},
+        {"id": "hint_generation", "name": "Hint Generation", "model": model_id, "share": 0.28, "cache_hit_rate": 0.58, "icon": "Lightbulb"},
+        {"id": "lesson_generation", "name": "Lesson Generation", "model": model_id, "share": 0.18, "cache_hit_rate": 0.35, "icon": "GraduationCap"},
+        {"id": "learning_paths", "name": "Learning Paths", "model": model_id, "share": 0.09, "cache_hit_rate": 0.40, "icon": "Target"},
+        {"id": "quiz_generation", "name": "Quiz Generation", "model": model_id, "share": 0.09, "cache_hit_rate": 0.38, "icon": "PenTool"},
+        {"id": "other", "name": "Other AI Features", "model": model_id, "share": 0.01, "cache_hit_rate": 0.50, "icon": "Zap"},
+    ]
+
+    total_requests = 6900
+    total_input_tokens = 8_500_000
+    total_output_tokens = 3_200_000
+
+    features = []
+    total_cost = 0.0
+    total_full_price_cost = 0.0
+    total_cache_hit_tokens = 0
+    total_cache_miss_tokens = 0
+
+    for fc in features_config:
+        req_count = int(total_requests * fc["share"])
+        input_share = int(total_input_tokens * fc["share"])
+        output_share = int(total_output_tokens * fc["share"])
+        cache_hit = int(input_share * fc["cache_hit_rate"])
+        cache_miss = input_share - cache_hit
+
+        cost = calculate_feature_cost(fc["model"], cache_hit, cache_miss, output_share)
+        full_cost = calculate_full_price_cost(fc["model"], cache_hit, cache_miss, output_share)
+
+        total_cost += cost["total_usd"]
+        total_full_price_cost += full_cost
+        total_cache_hit_tokens += cache_hit
+        total_cache_miss_tokens += cache_miss
+
+        features.append({
+            "featureId": fc["id"],
+            "featureName": fc["name"],
+            "modelId": fc["model"],
+            "monthlyCost": round(cost["total_usd"], 4),
+            "costShare": round(fc["share"] * 100, 1),
+            "totalRequests": req_count,
+            "totalInputTokens": input_share,
+            "totalOutputTokens": output_share,
+            "cacheHitRate": fc["cache_hit_rate"],
+            "isMostActive": fc["id"] == "ai_chat_tutor",
+            "isTopSpending": fc["id"] == "ai_chat_tutor",
+            "icon": fc["icon"],
+        })
+
+    overall_cache_hit_rate = total_cache_hit_tokens / (total_cache_hit_tokens + total_cache_miss_tokens) if (total_cache_hit_tokens + total_cache_miss_tokens) > 0 else 0
+
+    # Cost breakdown
+    total_cache_hit_cost = (total_cache_hit_tokens / 1_000_000) * pricing["input_cache_hit_per_1m"]
+    total_cache_miss_cost = (total_cache_miss_tokens / 1_000_000) * pricing["input_cache_miss_per_1m"]
+    total_output_cost = (total_output_tokens / 1_000_000) * pricing["output_per_1m"]
+
+    summary = {
+        "systemStatus": "healthy",
+        "actionRequired": False,
+        "hasPerformanceIssues": False,
+        "monthlyCost": round(total_cost, 4),
+        "projectedMonthlyCost": round(total_cost * 1.1, 4),
+        "billingCycleLabel": "Current Billable Cycle",
+        "costBreakdown": {
+            "cacheHitCost": round(total_cache_hit_cost, 6),
+            "cacheMissCost": round(total_cache_miss_cost, 6),
+            "outputCost": round(total_output_cost, 6),
+        },
+        "totalUsage": total_requests,
+        "totalInputTokens": total_cache_hit_tokens + total_cache_miss_tokens,
+        "totalOutputTokens": total_output_tokens,
+        "cacheHitRate": round(overall_cache_hit_rate, 4),
+        "activeEngine": "DeepSeek-V4 Pro",
+        "activeEngineModelId": model_id,
+        "engineTier": "High-Performance LLM",
+        "promotionalPricingActive": pricing.get("is_promotional", False),
+        "promotionalPriceExpiresUtc": pricing.get("promo_expires_utc", ""),
+        "estimatedCostAfterPromo": round(total_full_price_cost, 4),
+        "lastUpdated": datetime.now(timezone.utc).isoformat(),
+    }
+
+    return {"summary": summary, "features": features}
+
+
+@router.get("/summary")
+def get_monitoring_summary(_admin=Depends(require_admin)):
+    """Returns AI monitoring summary + feature metrics + pricing metadata."""
+    data = _aggregate_summary()
+    return {
+        **data["summary"],
+        "features": data["features"],
+        "pricingMeta": _build_pricing_meta(),
+    }
+
+
+@router.post("/refresh")
+def refresh_monitoring(_admin=Depends(require_admin)):
+    """Re-aggregate usage metrics and recalculate costs."""
+    data = _aggregate_summary()
+    # TODO: Write to Firestore ai_monitoring/summary when Firestore admin SDK is available
+    pricing = get_active_pricing("deepseek-v4-pro")
+    return {
+        "success": True,
+        "updatedAt": datetime.now(timezone.utc).isoformat(),
+        "pricingUsed": pricing,
+    }

services/cost_calculator.py

@@ -0,0 +1,44 @@
+# backend/services/cost_calculator.py
+# TODO: Review pricing after 2026-05-31
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from config.ai_pricing import get_active_pricing, get_full_pricing
+
+
+def calculate_feature_cost(
+    model_id: str,
+    cache_hit_tokens: int,
+    cache_miss_tokens: int,
+    output_tokens: int,
+) -> dict:
+    """Calculate cost for a feature's token usage using active pricing."""
+    # TODO: Review pricing after 2026-05-31
+    pricing = get_active_pricing(model_id)
+    cache_hit_cost = (cache_hit_tokens / 1_000_000) * pricing["input_cache_hit_per_1m"]
+    cache_miss_cost = (cache_miss_tokens / 1_000_000) * pricing["input_cache_miss_per_1m"]
+    output_cost = (output_tokens / 1_000_000) * pricing["output_per_1m"]
+    total = cache_hit_cost + cache_miss_cost + output_cost
+    return {
+        "total_usd": round(total, 6),
+        "cache_hit_cost": round(cache_hit_cost, 6),
+        "cache_miss_cost": round(cache_miss_cost, 6),
+        "output_cost": round(output_cost, 6),
+        "is_promotional": pricing["is_promotional"],
+    }
+
+
+def calculate_full_price_cost(
+    model_id: str,
+    cache_hit_tokens: int,
+    cache_miss_tokens: int,
+    output_tokens: int,
+) -> float:
+    """Calculate what the same usage would cost at full (non-promo) price."""
+    # TODO: Review pricing after 2026-05-31
+    full = get_full_pricing(model_id)
+    cache_hit_cost = (cache_hit_tokens / 1_000_000) * full["input_cache_hit_per_1m"]
+    cache_miss_cost = (cache_miss_tokens / 1_000_000) * full["input_cache_miss_per_1m"]
+    output_cost = (output_tokens / 1_000_000) * full["output_per_1m"]
+    return round(cache_hit_cost + cache_miss_cost + output_cost, 6)

tests/test_cost_calculator.py

@@ -0,0 +1,102 @@
+# backend/tests/test_cost_calculator.py
+"""Tests for services/cost_calculator.py covering promo active, promo expired, V4 Flash, and edge cases."""
+import sys
+import os
+from unittest.mock import patch
+from datetime import datetime, timezone
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from services.cost_calculator import calculate_feature_cost, calculate_full_price_cost
+from config.ai_pricing import get_active_pricing, DEEPSEEK_PRICING
+
+
+class TestCalculateFeatureCostPromoActive:
+    """Tests when promotional pricing is active (before 2026-05-31)."""
+
+    def test_basic_calculation(self):
+        result = calculate_feature_cost(
+            "deepseek-v4-pro",
+            cache_hit_tokens=1_000_000,
+            cache_miss_tokens=1_000_000,
+            output_tokens=1_000_000,
+        )
+        assert result["is_promotional"] is True
+        assert result["cache_hit_cost"] == round(0.003625, 6)
+        assert result["cache_miss_cost"] == round(0.435, 6)
+        assert result["output_cost"] == round(0.87, 6)
+        expected_total = 0.003625 + 0.435 + 0.87
+        assert abs(result["total_usd"] - expected_total) < 1e-5
+
+    def test_zero_tokens(self):
+        result = calculate_feature_cost("deepseek-v4-pro", 0, 0, 0)
+        assert result["total_usd"] == 0.0
+        assert result["cache_hit_cost"] == 0.0
+        assert result["cache_miss_cost"] == 0.0
+        assert result["output_cost"] == 0.0
+        assert result["is_promotional"] is True
+
+    def test_only_cache_hits(self):
+        result = calculate_feature_cost("deepseek-v4-pro", 5_000_000, 0, 0)
+        expected = (5_000_000 / 1_000_000) * 0.003625
+        assert abs(result["total_usd"] - expected) < 1e-5
+
+
+class TestCalculateFeatureCostPromoExpired:
+    """Tests when promotional pricing has expired."""
+
+    def test_full_price_after_expiry(self):
+        expired_time = datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc)
+        with patch("config.ai_pricing.datetime") as mock_dt:
+            mock_dt.now.return_value = expired_time
+            mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
+            pricing = get_active_pricing("deepseek-v4-pro")
+            assert pricing["is_promotional"] is False
+
+
+class TestCalculateFeatureCostFlash:
+    """Tests for deepseek-v4-flash (no promotional pricing)."""
+
+    def test_flash_pricing(self):
+        result = calculate_feature_cost(
+            "deepseek-v4-flash",
+            cache_hit_tokens=1_000_000,
+            cache_miss_tokens=1_000_000,
+            output_tokens=1_000_000,
+        )
+        assert result["is_promotional"] is False
+        assert result["cache_hit_cost"] == round(0.0028, 6)
+        assert result["cache_miss_cost"] == round(0.14, 6)
+        assert result["output_cost"] == round(0.28, 6)
+
+    def test_flash_zero_tokens(self):
+        result = calculate_feature_cost("deepseek-v4-flash", 0, 0, 0)
+        assert result["total_usd"] == 0.0
+
+
+class TestCalculateFullPriceCost:
+    """Tests for calculate_full_price_cost."""
+
+    def test_full_price_v4_pro(self):
+        cost = calculate_full_price_cost("deepseek-v4-pro", 1_000_000, 1_000_000, 1_000_000)
+        expected = 0.0145 + 1.74 + 3.48
+        assert abs(cost - expected) < 1e-5
+
+    def test_full_price_flash(self):
+        cost = calculate_full_price_cost("deepseek-v4-flash", 1_000_000, 1_000_000, 1_000_000)
+        expected = 0.0028 + 0.14 + 0.28
+        assert abs(cost - expected) < 1e-5
+
+
+class TestUnknownModel:
+    """Tests for unknown model IDs."""
+
+    def test_unknown_model_raises(self):
+        import pytest
+        with pytest.raises(ValueError, match="Unknown model"):
+            calculate_feature_cost("nonexistent-model", 100, 100, 100)
+
+    def test_unknown_model_full_price_raises(self):
+        import pytest
+        with pytest.raises(ValueError, match="Unknown model"):
+            calculate_full_price_cost("nonexistent-model", 100, 100, 100)