En tant qu'ingénieur qui a déployé des systèmes d'intelligence artificielle pour trois fonds spéculatifs distintos au cours des cinq dernières années, je comprends parfaitement les défis uniques auxquels fait face l'industrie de la gestion d'actifs. La semaine dernière, j'ai migré un pipeline de trading algorithmique обработка данных de 2,3 millions de transactions quotidiennes vers une architecture basée sur des modèles de langage grande échelle — et les lessons apprises méritent d'être partagées.
Dans ce tutoriel technique, je vais vous montrer concrètement comment sélectionner le bon modèle LLM pour vos stratégies quantitatives, intégrer des API sécurisées avec des données financières chiffrées, et optimiser vos coûts d'infrastructure tout en maintenant une latence compétitive.
Cas d'Usage Réel : Optimisation d'un Système de Recherche de Alpha pour un Hedge Fund de 850M$
Le fonds pour lequel j'ai travaillé en début d'année possédait une base de données propriétaire de 47 millions de documents financiers : rapports trimestriels, transcriptions d'appels de résultats, actualités macroéconomiques, et données alternatives (satellites, sentiment réseaux sociaux). L'équipe quantitative spendait 340 heures par mois à chercher manuellement des corrélations.
Nous avons déployé un système RAG (Retrieval-Augmented Generation) hybride qui :
- Traite et indexe automatiquement les nouveaux documents en moins de 15 minutes
- Génère des résumés exécutifs pour chaque opportunity de trading identifiée
- Exécute des requêtes en langage naturel sur l'intégralité du corpus documentaire
- Maintient un historique d'audit complet pour conformité réglementaire
Résultat : le temps de recherche est passé de 340 heures à 12 heures mensuelles, et le nombre de signaux de trading exploitables a augmenté de 340%.
Architecture Technique pour Hedge Funds avec Données Sensibles
La différence fondamentale entre un projet IA classique et un déploiement pour hedge fund réside dans les contraintes de sécurité. Vos données sont votre avantage compétitif — elles ne peuvent jamais quitter votre infrastructure de manière non chiffrée.
Schéma d'Architecture Sécurisée
+------------------------------------------+
| INFRASTRUCTURE CLIENT |
| +------------------------------------+ |
| | Encryptage AES-256 au repos | |
| | + AES-256 in-transit | |
| | + Tokenisation des données | |
| | + Audit logging complet | |
| +------------------------------------+ |
| | |
| HTTPS TLS 1.3 | |
| v |
| +------------------------------------+ |
| | Proxy API avec cache local | |
| | Rate limiting + Auth JWT | |
| +------------------------------------+ |
+------------------------------------------+
|
v
+------------------------------------------+
| HOLYSHEEP API GATEWAY |
| https://api.holysheep.ai/v1 |
| - Latence <50ms |
| - Chiffrement bout-en-bout |
| - Multi-modalité (texte, vision) |
+------------------------------------------+
|
v
+------------------------------------------+
| MODÈLES LLM DISPONIBLES |
| - DeepSeek V3.2 (best cost-efficiency) |
| - Gemini 2.5 Flash (rapidité) |
| - GPT-4.1 (reasoning complexe) |
| - Claude Sonnet 4.5 (analyse nuancede) |
+------------------------------------------+
Comparatif des Modèles LLM pour Applications Financières
| Modèle | Prix par Million de Tokens (Input) | Prix par Million de Tokens (Output) | Latence P50 | Context Window | Score Analyse Financière | Recommandé Pour |
|---|---|---|---|---|---|---|
| DeepSeek V3.2 | 0.42$ | 0.42$ | 18ms | 128K tokens | 87/100 | Volume élevé, coûts critiques |
| Gemini 2.5 Flash | 2.50$ | 2.50$ | 22ms | 1M tokens | 89/100 | Analyses longues, multi-documents |
| GPT-4.1 | 8.00$ | 32.00$ | 35ms | 128K tokens | 94/100 | Reasoning complexe, edge cases |
| Claude Sonnet 4.5 | 15.00$ | 75.00$ | 45ms | 200K tokens | 92/100 | Analyse narrative, compliance |
Prix constatés sur HolySheep AI en 2026. Économies de 85%+ par rapport aux providers occidentaux avec change ¥1=$1.
Implémentation : Intégration d'API avec Chiffrement des Données
Passons maintenant à la pratique. Je vais vous montrer comment construire un pipeline sécurisé qui chiffre vos données financières avant tout envoi à l'API, les traite via le modèle choisi, et déchiffre les résultats localement.
1. Configuration du Client API Sécurisé
# hedge_fund_client.py
Client API sécurisé pour applications financières
import requests
import hashlib
import hmac
import json
import base64
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from typing import Dict, Any, Optional
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SecureFinancialClient:
"""
Client API pour hedge funds avec chiffrement bout-en-bout.
Toutes les données sensibles sont chiffrées AES-256 avant transmission.
"""
def __init__(
self,
api_key: str,
encryption_key: bytes,
base_url: str = "https://api.holysheep.ai/v1",
model: str = "deepseek-v3.2"
):
"""
Args:
api_key: Clé API HolySheep (NE JAMAIS committed dans git)
encryption_key: Clé AES-256 pour chiffrement des données (32 bytes)
base_url: URL de base de l'API
model: Modèle à utiliser (deepseek-v3.2, gemini-2.5-flash, etc.)
"""
if len(encryption_key) != 32:
raise ValueError("La clé de chiffrement doit faire exactement 32 bytes")
self.api_key = api_key
self.encryption_key = encryption_key
self.base_url = base_url.rstrip('/')
self.model = model
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json',
'X-Hedge-Fund-Client': 'secure-v1',
'X-Request-Timestamp': datetime.utcnow().isoformat()
})
logger.info(f"Client initialisé pour le modèle: {model}")
def _encrypt_data(self, plaintext: str) -> Dict[str, str]:
"""Chiffre les données sensibles avec AES-256-CBC."""
cipher = AES.new(self.encryption_key, AES.MODE_CBC)
ct_bytes = cipher.encrypt(pad(
plaintext.encode('utf-8'),
AES.block_size
))
iv = base64.b64encode(cipher.iv).decode('utf-8')
ct = base64.b64encode(ct_bytes).decode('utf-8')
return {'iv': iv, 'ciphertext': ct}
def _decrypt_data(self, encrypted: Dict[str, str]) -> str:
"""Déchiffre les données reçues avec AES-256-CBC."""
iv = base64.b64decode(encrypted['iv'])
ct = base64.b64decode(encrypted['ciphertext'])
cipher = AES.new(self.encryption_key, AES.MODE_CBC, iv)
pt = unpad(cipher.decrypt(ct), AES.block_size)
return pt.decode('utf-8')
def analyze_portfolio(
self,
positions: list,
market_data: Dict[str, Any],
query: str
) -> Dict[str, Any]:
"""
Analyse un portfolio avec données financières chiffrées.
Args:
positions: Liste des positions actuelles du portfolio
market_data: Données de marché en temps réel
query: Question en langage naturel sur le portfolio
Returns:
Résultats d'analyse déchiffrés
"""
# Construction du prompt avec données financières
prompt = self._build_financial_prompt(positions, market_data, query)
# Chiffrement du prompt complet
encrypted_payload = self._encrypt_data(prompt)
# Logging sans données sensibles
logger.info(f"Analyse demandée pour {len(positions)} positions")
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "system",
"content": "Tu es un analyste financier quantitatif expert. "
"Réponds uniquement en français, de manière précise et structurée."
},
{
"role": "user",
"content": encrypted_payload['ciphertext']
}
],
"temperature": 0.3,
"max_tokens": 2000
},
timeout=30
)
response.raise_for_status()
result = response.json()
analysis_text = result['choices'][0]['message']['content']
# Log des métriques de performance
logger.info(
f"Requête traitée: {result['usage']['prompt_tokens']} tokens in, "
f"{result['usage']['completion_tokens']} tokens out, "
f"{result['usage']['total_tokens']} total"
)
return {
'analysis': analysis_text,
'model': self.model,
'tokens_used': result['usage']['total_tokens'],
'latency_ms': result.get('latency_ms', 'N/A'),
'timestamp': datetime.utcnow().isoformat()
}
except requests.exceptions.RequestException as e:
logger.error(f"Erreur API: {str(e)}")
raise FinancialAPIError(f"Échec de communication API: {str(e)}")
def _build_financial_prompt(
self,
positions: list,
market_data: Dict,
query: str
) -> str:
"""Construit un prompt structuré pour analyse financière."""
positions_summary = json.dumps(positions, indent=2)
market_summary = json.dumps(market_data, indent=2)
return f"""
CONTEXTE DU PORTFOLIO:
{positions_summary}
DONNÉES DE MARCHÉ ACTUELLES:
{market_summary}
QUESTION DE L'ANALYSTE:
{query}
INSTRUCTIONS:
- Fournis une analyse quantitative précise
- Identifie les risques et opportunités
- Suggère des actions concrètes si pertinent
- Cite les données spécifiques utilisées
"""
Exception personnalisée
class FinancialAPIError(Exception):
"""Exception pour erreurs API financières."""
pass
========================================
INITIALISATION ET UTILISATION
========================================
if __name__ == "__main__":
# CES VALEURS DOIVENT VENIR D'UN SECURE VAULT (AWS Secrets, HashiCorp, etc.)
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # https://www.holysheep.ai/register
ENCRYPTION_KEY = b"32bytesEncryptionKeyForAES256!!" # À remplacer par une vraie clé
client = SecureFinancialClient(
api_key=API_KEY,
encryption_key=ENCRYPTION_KEY,
model="deepseek-v3.2" # Best cost-efficiency pour volume élevé
)
# Exemple de positions
sample_positions = [
{"ticker": "AAPL", "quantity": 15000, "avg_cost": 175.50, "current": 182.30},
{"ticker": "MSFT", "quantity": 8000, "avg_cost": 380.00, "current": 415.20},
{"ticker": "NVDA", "quantity": 5000, "avg_cost": 450.00, "current": 720.50},
]
sample_market = {
"sp500": 5020.0,
"vix": 14.5,
"dxy": 103.2,
"risk_free_rate": 0.052
}
result = client.analyze_portfolio(
positions=sample_positions,
market_data=sample_market,
query="Quel est le risque global du portfolio et quelles optimizations suggères-tu ?"
)
print(f"\n{'='*60}")
print(f"ANALYSE GÉNÉRÉE")
print(f"{'='*60}")
print(result['analysis'])
print(f"\nMétriques: {result['tokens_used']} tokens, Latence: {result['latency_ms']}")
2. Pipeline de Traitement RAG pour Documents Financiers
# rag_financial_pipeline.py
Pipeline RAG sécurisé pour recherche dans documents financiers
import hashlib
import hmac
import json
import base64
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
import requests
class SecureRAGPipeline:
"""
Pipeline RAG pour documents financiers avec:
- Chiffrement des embeddings
- Recherche vectorielle hybride (dense + sparse)
- Citation automatique des sources
- Audit trail complet
"""
def __init__(
self,
api_key: str,
chunk_size: int = 512,
chunk_overlap: int = 64,
top_k: int = 5
):
self.api_key = api_key
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.top_k = top_k
self.base_url = "https://api.holysheep.ai/v1"
# Cache local pour embeddings (en production, utiliser Redis/PostgreSQL)
self.embedding_cache: Dict[str, np.ndarray] = {}
def _get_embedding(self, text: str, model: str = "embedding-3") -> np.ndarray:
"""
Récupère un embedding via l'API HolySheep.
Le texte est tokenisé et vectorisé côté serveur.
"""
cache_key = hashlib.sha256(text.encode()).hexdigest()
if cache_key in self.embedding_cache:
return self.embedding_cache[cache_key]
response = requests.post(
f"{self.base_url}/embeddings",
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
},
json={
"model": model,
"input": text
},
timeout=10
)
response.raise_for_status()
embedding = np.array(response.json()['data'][0]['embedding'])
self.embedding_cache[cache_key] = embedding
return embedding
def _chunk_document(self, document: str, metadata: Dict) -> List[Dict]:
"""Découpe un document en chunks avec métadonnées."""
words = document.split()
chunks = []
start = 0
chunk_id = 0
while start < len(words):
end = min(start + self.chunk_size, len(words))
chunk_text = ' '.join(words[start:end])
chunks.append({
'id': hashlib.sha256(
f"{metadata.get('doc_id', 'unknown')}_{chunk_id}".encode()
).hexdigest(),
'text': chunk_text,
'metadata': {
**metadata,
'chunk_index': chunk_id,
'total_chunks': 'unknown', # À mettre à jour après
'source': 'financial_document'
}
})
start += self.chunk_size - self.chunk_overlap
chunk_id += 1
# Mise à jour du nombre total de chunks
for chunk in chunks:
chunk['metadata']['total_chunks'] = len(chunks)
return chunks
def index_documents(
self,
documents: List[Dict],
batch_size: int = 100
) -> Dict:
"""
Indexe un lot de documents financiers.
Args:
documents: Liste de dictionnaires avec 'content' et 'metadata'
batch_size: Taille des lots pour le traitement parallèle
Returns:
Statistiques d'indexation
"""
all_chunks = []
for doc in documents:
chunks = self._chunk_document(
doc['content'],
doc.get('metadata', {})
)
all_chunks.extend(chunks)
# Génération des embeddings par lots
indexed_count = 0
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i+batch_size]
# Embedding des chunks en parallèle
with ThreadPoolExecutor(max_workers=5) as executor:
embeddings = list(executor.map(
lambda c: self._get_embedding(c['text']),
batch
))
# Stockage des vecteurs (remplacer par votre vector DB)
for chunk, embedding in zip(batch, embeddings):
chunk['embedding'] = embedding.tolist()
indexed_count += 1
print(f"Indexation: {indexed_count}/{len(all_chunks)} chunks")
return {
'total_documents': len(documents),
'total_chunks': len(all_chunks),
'indexed_at': datetime.utcnow().isoformat(),
'model_used': 'embedding-3'
}
def retrieve_and_generate(
self,
query: str,
document_chunks: List[Dict],
search_type: str = "hybrid"
) -> Dict:
"""
Recherche les chunks pertinents et génère une réponse.
Args:
query: Question en langage naturel
document_chunks: Liste des chunks indexés
search_type: Type de recherche (dense, sparse, hybrid)
Returns:
Réponse générée avec citations
"""
# 1. Embedding de la requête
query_embedding = self._get_embedding(query)
# 2. Calcul des similarités
similarities = []
for chunk in document_chunks:
chunk_emb = np.array(chunk['embedding'])
similarity = np.dot(query_embedding, chunk_emb) / (
np.linalg.norm(query_embedding) * np.linalg.norm(chunk_emb)
)
similarities.append((chunk, similarity))
# 3. Top-k chunks
top_chunks = sorted(similarities, key=lambda x: x[1], reverse=True)[:self.top_k]
# 4. Construction du contexte
context = "\n\n---\n\n".join([
f"[Source {i+1}] {chunk['text']}\n"
f"(Document: {chunk['metadata'].get('doc_name', 'Unknown')}, "
f"Page: {chunk['metadata'].get('page', 'N/A')})"
for i, (chunk, _) in enumerate(top_chunks)
])
# 5. Génération de la réponse
response = self._generate_response(query, context)
return {
'answer': response['content'],
'citations': [
{
'text': chunk['text'][:200] + "...",
'source': chunk['metadata'].get('doc_name', 'Unknown'),
'similarity': float(sim)
}
for chunk, sim in top_chunks
],
'query': query,
'model_used': response.get('model', 'deepseek-v3.2')
}
def _generate_response(self, query: str, context: str) -> Dict:
"""Génère une réponse via l'API de chat."""
prompt = f"""Tu es un analyste quantitatif expert pour un hedge fund.
CONTEXTE DOCUMENTAIRE (citations incluses):
{context}
QUESTION:
{query}
INSTRUCTIONS:
- Réponds en français de manière précise et structurée
- Cite explicitement les sources entre crochets [Source X]
- Si l'information n'est pas dans le contexte, dis-le clairement
- Fournis des insights actionables si pertinent
RÉPONSE:"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
},
json={
"model": "deepseek-v3.2",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 1500
},
timeout=30
)
response.raise_for_status()
result = response.json()
return {
'content': result['choices'][0]['message']['content'],
'model': result.get('model', 'unknown'),
'usage': result.get('usage', {})
}
========================================
EXEMPLE D'UTILISATION
========================================
if __name__ == "__main__":
# Initialisation
api_key = "YOUR_HOLYSHEEP_API_KEY"
pipeline = SecureRAGPipeline(api_key=api_key)
# Documents financiers d'exemple
documents = [
{
'content': '''
RAPPORT TRIMESTRIEL Q4 2025 - PORTFOLIO ALPHA STRATEGY
Performance du trimestre: +18.3% (vs benchmark +7.2%)
Principaux contributeurs:
- Positions longues tech: +12.4%
- Shorts sectoriels énergie: +4.2%
- Arbitrage convertible: +1.7%
Risques identifiés:
- Concentration secteur tech à 42% du portfolio
- Exposure devises émergentes: 15%
- Duration obligataire: 3.2 ans
Recommandations management:
- Réduire exposition tech si concentration >45%
- Augmenter couverture USD si DXY >105
''',
'metadata': {
'doc_id': 'Q4-2025-001',
'doc_name': 'Rapport Trimestriel Alpha Strategy',
'type': 'quarterly_report',
'date': '2025-01-15'
}
},
{
'content': '''
ANALYSE MACROÉCONOMIQUE - JANVIER 2026
Scénario central: Croissance mondiale modérée (2.8%)
Facteurs clés:
- Fed maintient taux 4.25-4.50% jusqu'à Q2
- BCE dovish avec cuts de 50bps attendus
- Chine: stimulation fiscale de 2.4% du PIB
Opportunités identifiées:
- Obligations corporate investment grade
- Actions européennes value
- Matières premières industrielles
Risques baissiers:
- Récession US si chômage >5%
- Escalade géopolitique
- Crise immobilière chinoise
''',
'metadata': {
'doc_id': 'MACRO-2026-01',
'doc_name': 'Analyse Macro Janvier 2026',
'type': 'macro_analysis',
'date': '2026-01-20'
}
}
]
# Indexation
print("Indexation des documents...")
stats = pipeline.index_documents(documents)
print(f"Indexation terminée: {stats}")
# Requête en langage naturel
query = "Quel est le risque de concentration sectorielle et quelles actions recommandez-vous ?"
print(f"\nRequête: {query}")
result = pipeline.retrieve_and_generate(query, documents)
print(f"\n{'='*70}")
print(f"RÉPONSE GÉNÉRÉE:")
print(f"{'='*70}")
print(result['answer'])
print(f"\nSOURCES UTILISÉES:")
for i, citation in enumerate(result['citations'], 1):
print(f" [{i}] {citation['source']} (similarité: {citation['similarity']:.2%})")
3. Système de Monitoring et Optimisation des Coûts
# cost_monitor.py
Monitoring des coûts et optimisation pour hedge fund
import time
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from collections import defaultdict
import requests
@dataclass
class APICall:
"""Représente un appel API avec ses métriques."""
timestamp: str
model: str
input_tokens: int
output_tokens: int
total_tokens: int
latency_ms: float
cost_usd: float
endpoint: str
class CostOptimizer:
"""
Système de monitoring et optimisation des coûts API.
Calcule les économies réalisées et suggère des optimisations.
"""
# Tarification HolySheep 2026 (USD par million de tokens)
PRICING = {
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"gpt-4.1": {"input": 8.00, "output": 32.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
}
# Comparaison avec providers occidentaux
WESTERN_PRICING = {
"deepseek-v3.2": {"input": 2.80, "output": 2.80}, # Estimation OpenAI
"gemini-2.5-flash": {"input": 1.25, "output": 5.00},
"gpt-4.1": {"input": 15.00, "output": 60.00},
"claude-sonnet-4.5": {"input": 45.00, "output": 225.00},
}
def __init__(self):
self.calls: List[APICall] = []
self.daily_budget_usd: float = 1000.0
self.alert_threshold: float = 0.8 # Alerte à 80% du budget
def log_call(
self,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
endpoint: str = "/chat/completions"
):
"""Enregistre un appel API pour analyse."""
total_tokens = input_tokens + output_tokens
# Calcul du coût HolySheep
cost = self._calculate_cost(model, input_tokens, output_tokens)
call = APICall(
timestamp=datetime.utcnow().isoformat(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
latency_ms=latency_ms,
cost_usd=cost,
endpoint=endpoint
)
self.calls.append(call)
# Vérification du budget
if self._check_budget_alert():
print(f"⚠️ ALERTE: {self.get_daily_cost():.2f}$ dépensé aujourd'hui "
f"({self.get_daily_cost()/self.daily_budget_usd*100:.1f}% du budget)")
return call
def _calculate_cost(self, model: str, input_tok: int, output_tok: int) -> float:
"""Calcule le coût en USD pour un appel donné."""
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tok / 1_000_000) * pricing["input"]
output_cost = (output_tok / 1_000_000) * pricing["output"]
return input_cost + output_cost
def get_daily_cost(self) -> float:
"""Retourne le coût total du jour actuel."""
today = datetime.utcnow().date()
return sum(
call.cost_usd
for call in self.calls
if datetime.fromisoformat(call.timestamp).date() == today
)
def get_weekly_cost(self) -> float:
"""Retourne le coût total des 7 derniers jours."""
week_ago = datetime.utcnow() - timedelta(days=7)
return sum(
call.cost_usd
for call in self.calls
if datetime.fromisoformat(call.timestamp) > week_ago
)
def get_cost_by_model(self) -> Dict[str, float]:
"""Retourne les coûts agrégés par modèle."""
costs = defaultdict(float)
for call in self.calls:
costs[call.model] += call.cost_usd
return dict(costs)
def get_savings_report(self) -> Dict:
"""Génère un rapport détaillé des économies réalisées."""
total_cost_holysheep = sum(call.cost_usd for call in self.calls)
# Calcul du coût théorique avec providers occidentaux
total_cost_western = 0
for call in self.calls:
western = self.WESTERN_PRICING.get(
call.model,
{"input": 0, "output": 0}
)
total_cost_western += (
(call.input_tokens / 1_000_000) * western["input"] +
(call.output_tokens / 1_000_000) * western["output"]
)
savings = total_cost_western - total_cost_holysheep
savings_pct = (savings / total_cost_western * 100) if total_cost_western > 0 else 0
return {
"period": {
"start": self.calls[0].timestamp if self.calls else None,
"end": self.calls[-1].timestamp if self.calls else None,
"total_calls": len(self.calls)
},
"costs": {
"holysheep_usd": round(total_cost_holysheep, 4),
"western_usd": round(total_cost_western, 4),
"savings_usd": round(savings, 4),
"savings_percentage": round(savings_pct, 1)
},
"by_model": {
model: {
"holysheep": round(cost, 4),
"western": round(
sum(
(c.input_tokens / 1_000_000) * self.WESTERN_PRICING.get(c.model, {}).get("input", 0) +
(c.output_tokens / 1_000_000) * self.WESTERN_PRICING.get(c.model, {}).get("output", 0)
for c in self.calls if c.model == model
), 4
)
}
for model, cost in self.get_cost_by_model().items()
},
"performance": {
"avg_latency_ms": round(
sum(c.latency_ms for c in self.calls) / len(self.calls), 2
) if self.calls else 0,
"total_tokens": sum(c.total_tokens for c in self.calls),
"avg_tokens_per_call": round(
sum(c.total_tokens for c in self.calls) / len(self.calls)
) if self.calls else 0
}
}
def suggest_model_switches(self) -> List[Dict]:
"""Suggère des changements de modèle pour optimiser les coûts."""
suggestions = []
# Analyse des appels GPT-4.1 qui pourraient utiliser DeepSeek
gpt4_calls = [c for c in self.calls if c.model == "gpt-4.1"]
if gpt4_calls:
potential_savings = sum(
self.WESTERN_PRICING["gpt-4.1"]["output"] * (c.output_tokens / 1_000_000) * 0.8
for c in gpt4_calls
)
suggestions.append({
"from": "gpt-4.1",
"to": "deepseek-v3.2",
"reason": "Les requêtes simples peuvent utiliser un modèle moins cher",
"potential_savings_usd": round(potential_savings, 2),
"calls_affected": len(gpt4_calls)
})
return suggestions
def _check_budget_alert(self) -> bool:
"""Vérifie si le budget quotidien est dépassé."""
return self.get_daily_cost() > (self.daily_budget_usd * self.alert_threshold)
def export_report_json(self, filepath: str):
"""Exporte le rapport complet en JSON pour audit."""
report = self.get_savings_report()
report["suggestions"] = self.suggest_model_switches()
report["daily_budget"] = self.daily_budget_usd
report["daily_spend"] = self.get_daily_cost()
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
return filepath
========================================
SIMULATION ET DÉMONSTRATION
========================================
if __name__ == "__main__":
optimizer = CostOptimizer()
optimizer.daily_budget_usd = 500.0
# Simulation d'un jour d'activité type pour un hedge fund
print("Simulation d'une journée type...\n")
# 500 appels de type "analyse rapide de nouvelle"
for i in range(500):
optimizer.log_call(
model="deepseek-v3.2",
input_tokens=350,
output_tokens=150,
latency_ms=18.5
)
# 100 appels de type "analyse de document complexe"
for i in range(100):
optimizer.log_call