Introduction : Le Scandale d'une Fuite à 2 847 $ en 72 Heures
Il y a six mois, j'ai vécu ce que tout développeur SaaS redoute. Un dimanche soir, à 23h47, mon téléphone vibre avec une alerte Slack : « Budget mensuel dépassé de 340% ». En打开ant le dashboard HolySheep AI, je vois le chiffre glaçant : 2 847,23 $ engloutis en exactement 71 heures et 23 minutes. Le culprit ? Une boucle infinie dans notre système de récapitulatifs IA qui relançait la même requête GPT-4.1 sans condition d'arrêt. Chaque appel coûtait 0,008 $ — mais avec 356 000 appels ratés, la facture avait explosé.
Cette expérience douloureuse m'a poussé à concevoir une architecture de gouvernance des coûts béton. Aujourd'hui, je vais partager avec vous chaque ligne de code, chaque stratégie, chaque leçon apprise — pour que vous n'ayez jamais à subir ce que j'ai vécu. Ce guide est le résultat de 6 mois d'optimisation intensive sur HolySheep AI, avec des données réelles issues de notre production处理 2 millions de tokens par jour.
Comprendre l'Écosystème des Coûts HolySheep AI
Architecture de Facturation par Token
Avant d'aborder les stratégies d'optimisation, comprenons la anatomie d'une requête sur HolySheep AI. Chaque interaction avec l'API génère deux types de tokens :
- Tokens d'entrée (input) : le texte que vous envoyez au modèle (prompts, contexte, historique)
- Tokens de sortie (output) : le texte généré par le modèle (réponses, code, analyses)
HolySheep AI facture ces deux types séparément, avec des taux qui varient significativement selon le modèle choisi. La latence moyenne observée sur leurs serveurs est inférieure à 50ms pour les appels synchrones — un avantage compétitif majeur pour les applications temps réel.
Tableau Comparatif : Prix par Million de Tokens (Mai 2026)
| Modèle | Input ($/M tokens) | Output ($/M tokens) | Ratio Input/Output | Latence Moyenne | Cas d'Usage Optimal |
|---|---|---|---|---|---|
| GPT-4.1 | 8,00 $ | 24,00 $ | 1:3 | 1 250 ms | Raisonnement complexe, code advanced |
| Claude Sonnet 4.5 | 15,00 $ | 75,00 $ | 1:5 | 1 800 ms | Rédaction longue, analyses nuancées |
| Gemini 2.5 Flash | 2,50 $ | 10,00 $ | 1:4 | 380 ms | Traitement par lots, haute fréquence |
| DeepSeek V3.2 | 0,42 $ | 1,68 $ | 1:4 | 320 ms | Tâches simples, maximiser volume |
Source : Grille tarifaire HolySheep AI, mise à jour Mai 2026. Tous les prix en dollars USD.
Stratégie #1 : Implémentation d'un Système d'Alertes Budget en Temps Réel
La première ligne de défense contre les factures surprises est un système d'alertes granulaire. J'ai développé un module Python complet qui monitore l'utilisation en temps réel et envoie des notifications à plusieurs seuils critiques.
#!/usr/bin/env python3
"""
HolySheep AI — Budget Alert System v2.1
Surveillance temps réel avec seuils configurables
"""
import requests
import time
import smtplib
from datetime import datetime, timedelta
from collections import deque
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class BudgetConfig:
"""Configuration des seuils d'alerte"""
daily_limit_usd: float = 100.0 # Limite quotidienne
weekly_limit_usd: float = 500.0 # Limite hebdomadaire
monthly_limit_usd: float = 1500.0 # Limite mensuelle
warning_threshold: float = 0.75 # Alerte à 75% d'utilisation
critical_threshold: float = 0.90 # Alerte critique à 90%
cooldown_seconds: int = 300 # 5 minutes entre notifications
class HolySheepBudgetMonitor:
"""Moniteur de budget HolySheep AI avec alertes intelligentes"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, config: Optional[BudgetConfig] = None):
self.api_key = api_key
self.config = config or BudgetConfig()
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Historique des coûts (dernières 24h)
self.cost_history = deque(maxlen=288) # 288 * 5 minutes = 24h
self.last_alert_time = {}
self.session_costs = 0.0
def get_usage_stats(self) -> dict:
"""Récupère les statistiques d'utilisation depuis l'API"""
try:
response = requests.get(
f"{self.BASE_URL}/usage",
headers=self.headers,
timeout=10
)
response.raise_for_status()
data = response.json()
return {
"daily_cost": data.get("daily_cost", 0.0),
"weekly_cost": data.get("weekly_cost", 0.0),
"monthly_cost": data.get("monthly_cost", 0.0),
"total_tokens_today": data.get("total_tokens", 0),
"request_count": data.get("request_count", 0),
"timestamp": datetime.now().isoformat()
}
except requests.exceptions.RequestException as e:
print(f"Erreur récupération stats : {e}")
return self._estimate_costs_from_history()
def _estimate_costs_from_history(self) -> dict:
"""Fallback : estimation basée sur l'historique local"""
total = sum(item["cost"] for item in self.cost_history)
return {
"daily_cost": total,
"weekly_cost": total * 7,
"monthly_cost": total * 30,
"total_tokens_today": 0,
"request_count": len(self.cost_history)
}
def track_request(self, input_tokens: int, output_tokens: int, model: str):
"""Enregistre une requête et calcule son coût"""
# Tarifs HolySheep AI (à jour Mai 2026)
pricing = {
"gpt-4.1": {"input": 8.00, "output": 24.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00},
"deepseek-v3.2": {"input": 0.42, "output": 1.68}
}
model_key = model.lower()
if model_key not in pricing:
print(f"Modèle {model} non reconnu, utilisation DeepSeek par défaut")
model_key = "deepseek-v3.2"
input_cost = (input_tokens / 1_000_000) * pricing[model_key]["input"]
output_cost = (output_tokens / 1_000_000) * pricing[model_key]["output"]
total_cost = input_cost + output_cost
self.session_costs += total_cost
self.cost_history.append({
"timestamp": datetime.now(),
"cost": total_cost,
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens
})
return total_cost
def check_thresholds(self, usage: dict) -> list:
"""Vérifie les seuils et retourne les alertes nécessaires"""
alerts = []
checks = [
("daily", usage["daily_cost"], self.config.daily_limit_usd),
("weekly", usage["weekly_cost"], self.config.weekly_limit_usd),
("monthly", usage["monthly_cost"], self.config.monthly_limit_usd)
]
for period, cost, limit in checks:
ratio = cost / limit
if ratio >= self.config.critical_threshold:
alerts.append({
"level": "CRITICAL",
"period": period,
"cost": cost,
"limit": limit,
"percentage": ratio * 100,
"message": f"⚠️ CRITIQUE : {period} {cost:.2f}$ / {limit:.2f}$ ({ratio*100:.1f}%)"
})
elif ratio >= self.config.warning_threshold:
alerts.append({
"level": "WARNING",
"period": period,
"cost": cost,
"limit": limit,
"percentage": ratio * 100,
"message": f"🔔 Alerte : {period} {cost:.2f}$ / {limit:.2f}$ ({ratio*100:.1f}%)"
})
return alerts
def send_alert(self, alert: dict):
"""Envoie une notification (email, Slack, webhook)"""
# Évite le spam : respect du cooldown
alert_key = f"{alert['level']}_{alert['period']}"
last_time = self.last_alert_time.get(alert_key, datetime.min)
if (datetime.now() - last_time).total_seconds() < self.config.cooldown_seconds:
return # Pas d'envoi, en cooldown
self.last_alert_time[alert_key] = datetime.now()
# Formatage du message
message = f"""
╔══════════════════════════════════════════════════════╗
║ 🚨 HOLYSHEEP AI — ALERTE BUDGET {alert['level']} ║
╠══════════════════════════════════════════════════════╣
║ Période : {alert['period'].upper():<33} ║
║ Coût actuel : {alert['cost']:<8.2f}$ ║
║ Limite : {alert['limit']:<8.2f}$ ║
║ Utilisation : {alert['percentage']:<5.1f}% ║
║ Timestamp : {datetime.now().strftime('%Y-%m-%d %H:%M:%S'):<20} ║
╚══════════════════════════════════════════════════════╝
"""
# Sortie console (dev)
print(message)
# TODO: Implémenter l'envoi email/Slack ici
# self._send_email_alert(alert)
# self._send_slack_notification(alert)
def run_monitoring_loop(self, check_interval: int = 60):
"""Boucle principale de monitoring"""
print(f"🎯 Démarrage HolySheep Budget Monitor")
print(f" Limites — Daily: {self.config.daily_limit_usd}$ | "
f"Weekly: {self.config.weekly_limit_usd}$ | "
f"Monthly: {self.config.monthly_limit_usd}$")
while True:
try:
usage = self.get_usage_stats()
alerts = self.check_thresholds(usage)
for alert in alerts:
self.send_alert(alert)
time.sleep(check_interval)
except KeyboardInterrupt:
print("\n⏹️ Monitoring arrêté")
break
except Exception as e:
print(f"❌ Erreur monitoring : {e}")
time.sleep(check_interval)
=== UTILISATION ===
if __name__ == "__main__":
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
config = BudgetConfig(
daily_limit_usd=50.0, # 50$ / jour pour dev
weekly_limit_usd=300.0, # 300$ / semaine
monthly_limit_usd=1000.0, # 1000$ / mois
warning_threshold=0.75,
critical_threshold=0.90
)
monitor = HolySheepBudgetMonitor(API_KEY, config)
monitor.run_monitoring_loop(check_interval=60)Ce système surveille en temps réel votre consommation et alerte via console (ou email/Slack si configuré) dès que vous atteignez 75% ou 90% de vos limites. La fonctionnalité de cooldown évite le spam de notifications.
Stratégie #2 : Architecture de Downgrade Automatique avec Fallback Intelligent
Le concept clé ici : ne jamais utiliser un modèle surdimensionné quand un modèle plus économique fait l'affaire. J'ai implémenté un système de routing intelligent qui sélectionne automatiquement le modèle optimal selon la complexité de la tâche.
#!/usr/bin/env python3
"""
HolySheep AI — Smart Model Router avec Downgrade Automatique
Sélectionne le modèle optimal selon le type de tâche
"""
import requests
import json
import re
from enum import Enum
from typing import Optional, Callable
from dataclasses import dataclass
import hashlib
class TaskComplexity(Enum):
"""Niveaux de complexité de tâche"""
TRIVIAL = 1 # Q&A simple, classifications basiques
SIMPLE = 2 # Formatting, réécriture légère
MODERATE = 3 # Summarisation, traductions
COMPLEX = 4 # Code intermédiaire, analyses
ADVANCED = 5 # Raisonnement profond, code complexe
@dataclass
class ModelConfig:
"""Configuration d'un modèle"""
name: str
provider: str
complexity_range: tuple # (min, max)
cost_per_1k_input: float
cost_per_1k_output: float
max_tokens: int
supports_streaming: bool = True
class HolySheepModelRouter:
"""Router intelligent avec fallback automatique"""
BASE_URL = "https://api.holysheep.ai/v1"
# Annuaire des modèles HolySheep AI
MODELS = {
"deepseek-v3.2": ModelConfig(
name="DeepSeek V3.2",
provider="HolySheep",
complexity_range=(1, 2),
cost_per_1k_input=0.42,
cost_per_1k_output=1.68,
max_tokens=8192
),
"gemini-2.5-flash": ModelConfig(
name="Gemini 2.5 Flash",
provider="HolySheep",
complexity_range=(2, 3),
cost_per_1k_input=2.50,
cost_per_1k_output=10.00,
max_tokens=32768
),
"gpt-4.1": ModelConfig(
name="GPT-4.1",
provider="HolySheep",
complexity_range=(4, 5),
cost_per_1k_input=8.00,
cost_per_1k_output=24.00,
max_tokens=128000
),
"claude-sonnet-4.5": ModelConfig(
name="Claude Sonnet 4.5",
provider="HolySheep",
complexity_range=(3, 5),
cost_per_1k_input=15.00,
cost_per_1k_output=75.00,
max_tokens=200000
)
}
def __init__(self, api_key: str, budget_controller=None):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.budget_controller = budget_controller
self.usage_stats = {"requests": 0, "total_cost": 0.0}
def analyze_complexity(self, prompt: str, context: Optional[str] = None) -> TaskComplexity:
"""Analyse la complexité d'une tâche basée sur le prompt"""
prompt_lower = prompt.lower()
context = context or ""
combined = (prompt + " " + context).lower()
# Indices de haute complexité
advanced_indicators = [
r'\b(fonction|algorithme|architecture|refactor|optimize)\b',
r'\b(debug|fix|error|exception|stack trace)\b',
r'\b(analyse|research|compare|evaluate|assess)\b',
r'\b(write code|implement|develop|build system)\b',
r'(?s)long context.*multiple.*files',
r'\b(step by step|reasoning|explain.*why)\b'
]
# Indices de basse complexité
trivial_indicators = [
r'\b(translate|summarize|list|what is|who is)\b',
r'\b(yes|no|true|false|confirm)\b',
r'\b(spell check|grammar|fix typo)\b',
r'^.{1,100}[?]$', # Questions courtes
r'\b(classify|categorize|tag)\s+\w+\s+(into|as)\b'
]
# Compteur de sophistication
score = 2 # Par défaut MODERATE
for pattern in advanced_indicators:
if re.search(pattern, combined):
score += 1
for pattern in trivial_indicators:
if re.search(pattern, combined):
score -= 1
# Limiter le score
score = max(1, min(5, score))
return TaskComplexity(score)
def select_model(self, complexity: TaskComplexity, force_model: Optional[str] = None) -> str:
"""Sélectionne le modèle optimal selon la complexité"""
if force_model and force_model in self.MODELS:
return force_model
# Mapping complexité -> modèle par défaut
complexity_model_map = {
TaskComplexity.TRIVIAL: "deepseek-v3.2",
TaskComplexity.SIMPLE: "deepseek-v3.2",
TaskComplexity.MODERATE: "gemini-2.5-flash",
TaskComplexity.COMPLEX: "gpt-4.1",
TaskComplexity.ADVANCED: "gpt-4.1"
}
selected = complexity_model_map.get(complexity, "gemini-2.5-flash")
# Vérification budget : downgrade si proche des limites
if self.budget_controller:
remaining = self.budget_controller.get_remaining_budget()
if remaining < 50: # Moins de 50$ restants
# Downgrade systématique vers modèle économique
selected = "deepseek-v3.2"
print(f"💰 Budget bas détecté ({remaining:.2f}$), downgrade vers DeepSeek")
elif remaining < 200:
selected = "gemini-2.5-flash"
print(f"💰 Budget modéré ({remaining:.2f}$), sélection Gemini Flash")
return selected
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Estime le coût d'une requête"""
if model not in self.MODELS:
return 0.0
config = self.MODELS[model]
input_cost = (input_tokens / 1000) * config.cost_per_1k_input
output_cost = (output_tokens / 1000) * config.cost_per_1k_output
return input_cost + output_cost
def make_request(self, prompt: str, context: Optional[str] = None,
force_model: Optional[str] = None,
max_output_tokens: int = 2048) -> dict:
"""Effectue une requête avec routing intelligent"""
# Étape 1 : Analyser la complexité
complexity = self.analyze_complexity(prompt, context)
print(f"🎯 Complexité détectée : {complexity.name} (score: {complexity.value})")
# Étape 2 : Sélectionner le modèle
model = self.select_model(complexity, force_model)
print(f"🤖 Modèle sélectionné : {self.MODELS[model].name}")
# Étape 3 : Vérifier le budget
estimated_tokens = len(prompt.split()) * 1.3 # Approximation
estimated_cost = self.estimate_cost(model, estimated_tokens, max_output_tokens)
if self.budget_controller:
if not self.budget_controller.can_afford(estimated_cost):
# Fallback vers modèle moins cher
model = "deepseek-v3.2"
print(f"⚠️ Budget insuffisant, fallback vers DeepSeek V3.2")
# Étape 4 : Exécuter la requête
try:
response = self._execute_completion(model, prompt, context, max_output_tokens)
# Étape 5 : Tracker les stats
self._update_stats(response, model)
return {
"success": True,
"model_used": model,
"model_name": self.MODELS[model].name,
"complexity": complexity.name,
"response": response["content"],
"tokens_used": response.get("usage", {}),
"actual_cost": response.get("cost", 0.0)
}
except Exception as e:
print(f"❌ Erreur avec {model}: {e}")
# Tentative de fallback automatique
return self._handle_failure_and_fallback(prompt, context, model, e)
def _execute_completion(self, model: str, prompt: str, context: Optional[str],
max_tokens: int) -> dict:
"""Appel effectif à l'API HolySheep"""
messages = []
if context:
messages.append({"role": "system", "content": context})
messages.append({"role": "user", "content": prompt})
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": 0.7
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code == 401:
raise Exception("401 Unauthorized — Clé API invalide ou expiré")
elif response.status_code == 429:
raise Exception("429 Rate Limited — Trop de requêtes")
elif response.status_code != 200:
raise Exception(f"HTTP {response.status_code}: {response.text}")
data = response.json()
# Extraction des données
content = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
# Calcul du coût réel
cost = self.estimate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
return {
"content": content,
"usage": {
"input_tokens": usage.get("prompt_tokens", 0),
"output_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0)
},
"cost": cost
}
def _handle_failure_and_fallback(self, prompt: str, context: Optional[str],
failed_model: str, error: Exception) -> dict:
"""Gère les échecs avec fallback intelligent"""
# Ordre de fallback : gpt-4.1 -> gemini -> deepseek
fallback_order = ["deepseek-v3.2"]
if "429" in str(error):
fallback_order = ["deepseek-v3.2", "gemini-2.5-flash"]
for fallback_model in fallback_order:
if fallback_model == failed_model:
continue
print(f"🔄 Tentative fallback vers {self.MODELS[fallback_model].name}...")
try:
response = self._execute_completion(
fallback_model, prompt, context, 2048
)
self._update_stats(response, fallback_model)
return {
"success": True,
"model_used": fallback_model,
"model_name": self.MODELS[fallback_model].name,
"complexity": "FALLBACK",
"response": response["content"],
"tokens_used": response.get("usage", {}),
"actual_cost": response.get("cost", 0.0),
"fallback_from": failed_model
}
except Exception:
continue
return {
"success": False,
"error": str(error),
"all_models_failed": True
}
def _update_stats(self, response: dict, model: str):
"""Met à jour les statistiques d'utilisation"""
self.usage_stats["requests"] += 1
self.usage_stats["total_cost"] += response.get("cost", 0.0)
def get_cost_report(self) -> dict:
"""Génère un rapport de coûts détaillé"""
return {
"total_requests": self.usage_stats["requests"],
"total_cost_usd": self.usage_stats["total_cost"],
"average_cost_per_request": (
self.usage_stats["total_cost"] / self.usage_stats["requests"]
if self.usage_stats["requests"] > 0 else 0
),
"currency_note": "Tous les montants en USD. Taux avantageux pour utilisateurs CNY."
}
=== CLASSE BUDGET CONTROLLER (pour le routing) ===
class SimpleBudgetController:
"""Contrôleur de budget minimal pour le router"""
def __init__(self, daily_limit: float = 100.0):
self.daily_limit = daily_limit
self.spent_today = 0.0
def can_afford(self, amount: float) -> bool:
return (self.spent_today + amount) <= self.daily_limit
def get_remaining_budget(self) -> float:
return max(0, self.daily_limit - self.spent_today)
def record_spending(self, amount: float):
self.spent_today += amount
=== EXEMPLE D'UTILISATION ===
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# Initialisation avec contrôleur de budget
budget = SimpleBudgetController(daily_limit=50.0)
router = HolySheepModelRouter(API_KEY, budget)
# Exemple 1 : Question simple (détournera vers DeepSeek)
result1 = router.make_request(
prompt="Quelle est la capitale du Japon ?",
max_output_tokens=100
)
print(f"\n📝 Résultat 1 : {result1['model_name']}")
# Exemple 2 : Tâche complexe (utilisera GPT-4.1)
result2 = router.make_request(
prompt="Implémente un algorithme de tri rapide en Python avec tests unitaires",
context="Langage: Python 3.10+",
max_output_tokens=2048
)
print(f"\n📝 Résultat 2 : {result2['model_name']}")
# Rapport de coûts
print(f"\n💰 Rapport de coûts : {router.get_cost_report()}")Stratégie #3 : Configuration de Rate Limiting et Quotas par Clé API
Une autre couche de protection essentielle : limiter les dommages quand quelque chose tourne mal. Voici comment configurer des quotas stricts sur vos clés API HolySheep.
#!/usr/bin/env python3
"""
HolySheep AI — Rate Limiter & Quota Manager
Protection contre les surcharges et les abus
"""
import time
import threading
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional
import hashlib
@dataclass
class QuotaConfig:
"""Configuration des quotas par clé API"""
requests_per_minute: int = 60
requests_per_hour: int = 1000
requests_per_day: int = 10000
tokens_per_minute: int = 100000
tokens_per_day: int = 5000000
max_cost_per_day_usd: float = 100.0
enabled: bool = True
class RateLimiter:
"""Rate limiter thread-safe avec quotas configurables"""
def __init__(self):
self.quotas: Dict[str, QuotaConfig] = {}
self.request_history: Dict[str, list] = defaultdict(list)
self.token_usage: Dict[str, list] = defaultdict(list)
self.lock = threading.Lock()
def register_key(self, api_key: str, quota: QuotaConfig):
"""Enregistre une clé API avec ses quotas"""
with self.lock:
# Hasher la clé pour le stockage (sécurité)
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
self.quotas[key_hash] = quota
self.request_history[key_hash] = []
self.token_usage[key_hash] = []
def check_request_allowed(self, api_key: str, estimated_tokens: int = 0) -> tuple:
"""Vérifie si une requête est autorisée. Retourne (allowed, reason, wait_seconds)"""
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
if key_hash not in self.quotas:
return True, "Aucune restriction", 0
quota = self.quotas[key_hash]
if not quota.enabled:
return True, "Quotas désactivés", 0
now = datetime.now()
history = self.request_history[key_hash]
tokens = self.token_usage[key_hash]
# Nettoyage des entrées expirées
self._cleanup_old_entries(key_hash, now)
# Vérification RPM (requests per minute)
one_minute_ago = now - timedelta(minutes=1)
recent_requests = [t for t in history if t > one_minute_ago]
if len(recent_requests) >= quota.requests_per_minute:
oldest = min(recent_requests)
wait = (oldest - one_minute_ago).total_seconds() + 1
return False, f"RPM limit ({quota.requests_per_minute}/min)", max(0, wait)
# Vérification RPH (requests per hour)
one_hour_ago = now - timedelta(hours=1)
hour_requests = [t for t in history if t > one_hour_ago]
if len(hour_requests) >= quota.requests_per_hour:
oldest = min(hour_requests)
wait = (oldest - one_hour_ago).total_seconds() + 60
return False, f"RPH limit ({quota.requests_per_hour}/hour)", max(0, wait)
# Vérification RPD (requests per day)
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
day_requests = [t for t in history if t > today_start]
if len(day_requests) >= quota.requests_per_day:
return False, f"RPD limit reached ({quota.requests_per_day}/day)", 86400
# Vérification TPD (tokens per day)
day_tokens = sum(
t for t, _ in tokens
if datetime.fromtimestamp(t) > today_start
)
if day_tokens + estimated_tokens > quota.tokens_per_day:
return False, f"TPD limit ({quota.tokens_per_day:,}/day)", 86400
# Tout est bon
return True, "Allowed", 0
def record_request(self, api_key: str, tokens_used: int, cost_usd: float):
"""Enregistre une requête complétée"""
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
now = datetime.now()
with self.lock:
self.request_history[key_hash].append(now)
self.token_usage[key_hash].append((time.time(), tokens_used))
def _cleanup_old_entries(self, key_hash: str, now: datetime):
"""Supprime les entrées plus anciennes que 24h"""
one_day_ago = now - timedelta(days=1)
# Nettoyer l'historique des requêtes
self.request_history[key_hash] = [
t for t in self.request_history[key_hash]
if t > one_day_ago
]
# Nettoyer l'utilisation des tokens
cutoff = time.time() - 86400
self.token_usage[key_hash]