En tant qu'architecte technique ayant déployé des systèmes de risk scoring en production pour trois institutions financières à Bangkok et à Kuala Lumpur, je peux vous confirmer que la gestion des risques dans le secteur fintech thaïlandais présente des défis uniques : volatilité des transactions cross-border, conformité réglementaire Bank of Thailand stricte, et exigences de latence inférieur à 200ms pour les décisions en temps réel.
Dans cet article, je partage ma méthodologie complète pour construire une architecture d'agrégation multi-modèles qui combine GPT-4.1 pour l'analyse contextuelle approfondie, DeepSeek V3.2 pour les évaluations transactionnelles à haut volume, et Gemini 2.5 Flash pour les classifications en temps réel.
Cas d'Usage Réel : Système de Détection de Fraude pour une Néobanque Thai
Mon équipe a récemment déployé un système de risk scoring pour KuaiKai Finance, une néobanque ciblant les PME thaïlandaises. Le système traite 15 000 transactions par minute avec un taux de faux positifs réduit de 67% par rapport à leur précédente solution basée sur des règles statiques.
La contrainte principale était triple :
- Conformité aux réglementations Bank of Thailand concernant l'auditabilité des décisions algorithmiques
- Latence maximale de 180ms pour maintenir l'expérience utilisateur
- Support multi-langue (thaï, anglais, mandarin) pour les documents KYC
Architecture d'Agrégation Multi-Modèles
Schéma Architecturel
L'architecture repose sur un système de routage intelligent qui dirige chaque requête vers le modèle optimal selon le type de transaction et le contexte.
┌─────────────────────────────────────────────────────────────────┐
│ THAI FINTECH RISK PLATFORM │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Transaction │───▶│ Risk Router │───▶│ HolySheep │ │
│ │ Ingestion │ │ (Decision │ │ Multi-Model │ │
│ │ Service │ │ Engine) │ │ Gateway │ │
│ └──────────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │
│ ┌──────────────────────────────────────┼────────────┐ │
│ │ MODEL SELECTION │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌─────────┐│
│ │ DeepSeek │ │ GPT-4.1 │ │ Gemini ││
│ │ V3.2 │ │ (Contextual) │ │ 2.5 ││
│ │ $0.42/MTok │ │ $8/MTok │ │ Flash ││
│ │ High Volume │ │ Deep Fraud │ │ $2.50 ││
│ └──────────────┘ └──────────────┘ └─────────┘│
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ CONSENSUS ENGINE & SCORE AGGREGATION │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Implémentation du Routeur Intelligent
import requests
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class TransactionType(Enum):
P2P_TRANSFER = "p2p_transfer"
BUSINESS_PAYMENT = "business_payment"
CROSS_BORDER = "cross_border"
KYC_DOCUMENT = "kyc_document"
CREDIT_APPLICATION = "credit_application"
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class RiskScore:
model_name: str
score: float
confidence: float
latency_ms: float
reasoning: str
@dataclass
class AggregatedRiskResult:
final_score: float
risk_level: RiskLevel
confidence: float
model_contributions: List[RiskScore]
total_latency_ms: float
compliance_flags: List[str]
class ThaiFintechRiskAggregator:
"""Agrégateur multi-modèles pour risk scoring en environnement fintech thaïlandais"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Configuration des modèles avec leurs cas d'usage optimaux
self.model_config = {
"deepseek_v32": {
"endpoint": "/chat/completions",
"max_tokens": 512,
"temperature": 0.3,
"use_cases": [TransactionType.P2P_TRANSFER, TransactionType.BUSINESS_PAYMENT],
"cost_per_1k": 0.00042
},
"gpt_41": {
"endpoint": "/chat/completions",
"max_tokens": 2048,
"temperature": 0.2,
"use_cases": [TransactionType.CROSS_BORDER, TransactionType.KYC_DOCUMENT],
"cost_per_1k": 0.008
},
"gemini_flash": {
"endpoint": "/chat/completions",
"max_tokens": 1024,
"temperature": 0.4,
"use_cases": [TransactionType.CREDIT_APPLICATION],
"cost_per_1k": 0.00250
}
}
# Seuil de latence maximum Bank of Thailand (180ms)
self.max_latency_ms = 180
# Seuil de risque critique
self.critical_threshold = 0.85
def _route_to_optimal_model(self, transaction_type: TransactionType) -> str:
"""Détermine le modèle optimal selon le type de transaction"""
for model_id, config in self.model_config.items():
if transaction_type in config["use_cases"]:
return model_id
return "deepseek_v32" # Fallback vers modèle économique
def _build_risk_prompt(self, transaction_data: Dict) -> List[Dict]:
"""Construit le prompt structuré pour l'évaluation des risques"""
return [
{
"role": "system",
"content": """Vous êtes un analyste de risque financier certifié pour le marché thaïlandais.
Évaluez cette transaction selon les critères BoT (Bank of Thailand) :
1. Score de risque (0-1)
2. Indicateurs suspects (liste)
3. Niveau de confiance (0-1)
4. Justification détaillée
Répondez en JSON structuré."""
},
{
"role": "user",
"content": json.dumps(transaction_data, ensure_ascii=False, indent=2)
}
]
def _call_model(self, model_id: str, prompt: List[Dict]) -> Optional[RiskScore]:
"""Appel à un modèle spécifique via HolySheep API avec métriques"""
start_time = time.time()
config = self.model_config[model_id]
try:
payload = {
"model": model_id,
"messages": prompt,
"max_tokens": config["max_tokens"],
"temperature": config["temperature"],
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{self.base_url}{config['endpoint']}",
headers=self.headers,
json=payload,
timeout=10
)
response.raise_for_status()
latency_ms = (time.time() - start_time) * 1000
result = response.json()
content = result["choices"][0]["message"]["content"]
parsed = json.loads(content)
return RiskScore(
model_name=model_id,
score=float(parsed.get("risk_score", 0.5)),
confidence=float(parsed.get("confidence", 0.5)),
latency_ms=latency_ms,
reasoning=parsed.get("justification", "")
)
except requests.exceptions.Timeout:
return None
except Exception as e:
print(f"Erreur appel {model_id}: {str(e)}")
return None
def _calculate_consensus(self, scores: List[RiskScore]) -> tuple:
"""Calcule le score consensus avec pondération par confiance"""
if not scores:
return 0.5, 0.0
total_weight = sum(s.confidence for s in scores)
if total_weight == 0:
return 0.5, 0.0
weighted_score = sum(s.score * s.confidence for s in scores) / total_weight
avg_confidence = sum(s.confidence for s in scores) / len(scores)
return weighted_score, avg_confidence
def evaluate_transaction(self, transaction_data: Dict) -> AggregatedRiskResult:
"""Évalue une transaction avec routage intelligent multi-modèles"""
transaction_type = TransactionType(transaction_data.get("type", "p2p_transfer"))
# Routing vers modèle optimal (exécution principale)
primary_model = self._route_to_optimal_model(transaction_type)
prompt = self._build_risk_prompt(transaction_data)
primary_result = self._call_model(primary_model, prompt)
if not primary_result:
# Fallback vers DeepSeek si échec
primary_result = self._call_model("deepseek_v32", prompt)
scores = [primary_result] if primary_result else []
# Pour les transactions critiques, validation croisée avec second modèle
if primary_result and primary_result.score > self.critical_threshold:
cross_validation_model = "gpt_41" if primary_model == "deepseek_v32" else "deepseek_v32"
cross_result = self._call_model(cross_validation_model, prompt)
if cross_result:
scores.append(cross_result)
# Calcul du consensus
final_score, confidence = self._calculate_consensus(scores)
# Détermination du niveau de risque
if final_score >= 0.8:
risk_level = RiskLevel.CRITICAL
elif final_score >= 0.6:
risk_level = RiskLevel.HIGH
elif final_score >= 0.4:
risk_level = RiskLevel.MEDIUM
else:
risk_level = RiskLevel.LOW
# Flags de conformité BoT
compliance_flags = []
if len(scores) > 1:
score_variance = max(s.score for s in scores) - min(s.score for s in scores)
if score_variance > 0.3:
compliance_flags.append("HIGH_VARIANCE_REQUIRES_REVIEW")
if final_score > self.critical_threshold:
compliance_flags.append("MANUAL_REVIEW_REQUIRED")
total_latency = sum(s.latency_ms for s in scores)
return AggregatedRiskResult(
final_score=final_score,
risk_level=risk_level,
confidence=confidence,
model_contributions=scores,
total_latency_ms=total_latency,
compliance_flags=compliance_flags
)
--- Exemple d'utilisation ---
api_key = "YOUR_HOLYSHEEP_API_KEY"
risk_aggregator = ThaiFintechRiskAggregator(api_key)
transaction = {
"type": "cross_border",
"amount_thb": 250000,
"sender_account": "1234567890",
"receiver_account": "9876543210",
"sender_kyc_level": "verified",
"transaction_hour": 2, # 2h du matin - suspect
"historical_pattern": "normal",
"device_fingerprint": "trusted",
"ip_country": "MM" # Myanmar - pays à risque
}
result = risk_aggregator.evaluate_transaction(transaction)
print(f"Score final: {result.final_score:.2%}")
print(f"Niveau de risque: {result.risk_level.value}")
print(f"Latence totale: {result.total_latency_ms:.0f}ms")
print(f"Flags conformité: {result.compliance_flags}")Configuration du Middleware de Résilience
import asyncio
import aiohttp
from typing import List, Optional
from datetime import datetime, timedelta
from collections import defaultdict
class CircuitBreaker:
"""Pattern Circuit Breaker pour la tolérance aux pannes des modèles"""
def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.failures = defaultdict(int)
self.last_failure_time = defaultdict(datetime)
self.states = defaultdict(lambda: "CLOSED")
def record_success(self, model_id: str):
self.failures[model_id] = 0
self.states[model_id] = "CLOSED"
def record_failure(self, model_id: str):
self.failures[model_id] += 1
self.last_failure_time[model_id] = datetime.now()
if self.failures[model_id] >= self.failure_threshold:
self.states[model_id] = "OPEN"
print(f"Circuit OPEN pour {model_id} après {self.failures[model_id]} échecs")
def can_execute(self, model_id: str) -> bool:
if self.states[model_id] == "CLOSED":
return True
# Vérifier si le timeout est écoulé
time_since_failure = (datetime.now() - self.last_failure_time[model_id]).total_seconds()
if time_since_failure >= self.timeout_seconds:
self.states[model_id] = "HALF_OPEN"
return True
return False
def get_healthy_model(self, preferred_model: str, fallback_models: List[str]) -> Optional[str]:
"""Retourne le premier modèle disponible dans la liste de priorité"""
all_models = [preferred_model] + fallback_models
for model_id in all_models:
if self.can_execute(model_id):
return model_id
return None # Aucun modèle disponible
class AsyncRiskEvaluator:
"""Évaluateur asynchrone avec fallback intelligent et circuit breaker"""
def __init__(self, api_key: str, circuit_breaker: CircuitBreaker):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.circuit_breaker = circuit_breaker
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def evaluate_async(
self,
transaction_data: Dict,
models_priority: List[str] = None
) -> Optional[RiskScore]:
"""Évaluation asynchrone avec fallback automatique"""
if models_priority is None:
models_priority = ["deepseek_v32", "gpt_41", "gemini_flash"]
for model_id in models_priority:
if not self.circuit_breaker.can_execute(model_id):
continue
try:
start_time = asyncio.get_event_loop().time()
payload = {
"model": model_id,
"messages": [
{
"role": "system",
"content": "Évalue le risque de cette transaction fintech thaïlandaise. "
"Retourne JSON: {\"risk_score\": float, \"confidence\": float, \"indicators\": []}"
},
{
"role": "user",
"content": json.dumps(transaction_data, ensure_ascii=False)
}
],
"max_tokens": 512,
"temperature": 0.3
}
async with self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=aiohttp.ClientTimeout(total=5)
) as response:
latency = (asyncio.get_event_loop().time() - start_time) * 1000
if response.status == 200:
data = await response.json()
content = data["choices"][0]["message"]["content"]
parsed = json.loads(content)
self.circuit_breaker.record_success(model_id)
return RiskScore(
model_name=model_id,
score=float(parsed.get("risk_score", 0.5)),
confidence=float(parsed.get("confidence", 0.5)),
latency_ms=latency,
reasoning=parsed.get("indicators", [])
)
elif response.status == 429:
# Rate limited - essayer le modèle suivant
await asyncio.sleep(0.5)
continue
else:
self.circuit_breaker.record_failure(model_id)
continue
except asyncio.TimeoutError:
self.circuit_breaker.record_failure(model_id)
continue
except Exception as e:
print(f"Erreur modèle {model_id}: {e}")
self.circuit_breaker.record_failure(model_id)
continue
return None
--- Utilisation asynchrone ---
async def process_transaction_batch():
circuit_breaker = CircuitBreaker(failure_threshold=3, timeout_seconds=30)
async with AsyncRiskEvaluator("YOUR_HOLYSHEEP_API_KEY", circuit_breaker) as evaluator:
transactions = [
{"type": "p2p_transfer", "amount_thb": 50000, "sender": "acc001"},
{"type": "cross_border", "amount_thb": 500000, "sender": "acc002"},
{"type": "business_payment", "amount_thb": 200000, "sender": "acc003"},
]
tasks = [evaluator.evaluate_async(t) for t in transactions]
results = await asyncio.gather(*tasks)
for txn, result in zip(transactions, results):
if result:
print(f"Txn {txn['type']}: Score {result.score:.2%} via {result.model_name} ({result.latency_ms:.0f}ms)")
else:
print(f"Txn {txn['type']}: Échec total - review manuel requis")
asyncio.run(process_transaction_batch())Tableau Comparatif des Modèles HolySheep pour Risk Scoring
| Modèle | Prix (USD/MTok) | Latence P95 | Cas d'Usage Optimal | Force Principale | Limitation |
|---|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | <35ms | P2P, micro-transactions, screening volumétrique | Coût ultra-réduit, haute vitesse | Contextes complexes limités |
| Gemini 2.5 Flash | $2.50 | <45ms | Décisions crédit en temps réel, scoring mobile | Équilibre coût/vitence/précision | Multi-turn moins efficace |
| GPT-4.1 | $8.00 | <80ms | Cross-border, KYC complexe, fraude sophistiquée | Raisonnement profond, peu de faux positifs | Coût plus élevé |
| Économie HolySheep | Jusqu'à -85% | <50ms avg | Tous cas - intégration unifiée | Multi-modèles, ¥1=$1, Alipay/WeChat | - |
Calculateur de ROI pour Fintech Thailandaise
Basé sur notre déploiement chez KuaiKai Finance :
- Volume mensuel : 45 millions de transactions
- Tokens moyens par évaluation : 800 tokens input + 200 output
- Répartition modèle : 85% DeepSeek, 10% Gemini Flash, 5% GPT-4.1
# Calcul du coût mensuel HolySheep vs OpenAI Direct
HolySheep avec agrégation intelligente
holy_sheep_monthly = {
"deepseek_v32": {
"transactions": 45_000_000 * 0.85,
"tokens_per_tx": 1000,
"cost_per_mtok": 0.42,
"monthly_cost": (45_000_000 * 0.85 * 1000 / 1_000_000) * 0.42
},
"gemini_flash": {
"transactions": 45_000_000 * 0.10,
"tokens_per_tx": 800,
"cost_per_mtok": 2.50,
"monthly_cost": (45_000_000 * 0.10 * 800 / 1_000_000) * 2.50
},
"gpt_41": {
"transactions": 45_000_000 * 0.05,
"tokens_per_tx": 1500,
"cost_per_mtok": 8.00,
"monthly_cost": (45_000_000 * 0.05 * 1500 / 1_000_000) * 8.00
}
}
total_holy_sheep = sum(v["monthly_cost"] for v in holy_sheep_monthly.values())
print(f"Coût HolySheep mensuel: ${total_holy_sheep:,.2f}")
Comparaison OpenAI direct (tous GPT-4o)
openai_direct = (45_000_000 * 1000 / 1_000_000) * 2.50 # $2.50/MTok GPT-4o
print(f"Coût OpenAI Direct (GPT-4o): ${openai_direct:,.2f}")
Économie mensuelle
savings = openai_direct - total_holy_sheep
savings_percent = (savings / openai_direct) * 100
print(f"ÉCONOMIE: ${savings:,.2f}/mois ({savings_percent:.1f}%)")
Impact annuel
annual_savings = savings * 12
print(f"ÉCONOMIE ANNUELLE: ${annual_savings:,.2f}")
Réduction faux positifs (impact métier)
false_positive_reduction = 0.67 # 67% de réduction
avg_manual_review_cost = 5.00 # USD par review
monthly_reviews_avoided = 45_000_000 * 0.15 * false_positive_reduction
review_savings = monthly_reviews_avoided * 5.00
print(f"Économie reviews manuelles: ${review_savings:,.2f}/mois")Pour qui / Pour qui ce n'est pas fait
✅ Idéale pour :
- Néobanques thaïlandaises traitant plus de 10 000 transactions/jour
- Institutions de microfinance cherchant à automatiser le scoring crédit avec conformité BoT
- Services de paiement cross-border entre Thaïlande, Myanmar, Laos, Cambodge
- PME développant des produits financiers nécessitant un risk scoring intégrable
- Développeurs fintech cherchant une solution unique multi-modèles avec latence <50ms
❌ Pas recommandée pour :
- Banque traditionnelle avec infrastructure legacy nécessitant une certification lengthy des modèles
- Projets personnels ou prototypes avec budget <$100/mois - les modèles gratuits suffisent
- Cas d'usage hors fintech où un modèle unique suffit (utilisez directement les providers)
- Applications nécessitant des modèles vision (les modèles actuels sont textuels)
Tarification et ROI
| Plan | Prix | Crédits Inclus | Latence | Support | Idéal pour |
|---|---|---|---|---|---|
| Starter | Gratuit | 500K tokens | Standard | Documentation | Prototypage, POC |
| Growth | $99/mois | 100M tokens | Priorité | Email + Discord | Startup fintech, scale-up |
| Enterprise | Sur devis | Illimité | <50ms garantie | 24/7 Dédié | Banque, institution financière |
| Taux avantageux | ¥1 = $1 USD (économie 85%+ vs OpenAI) | WeChat Pay & Alipay acceptés | ||||
Pourquoi choisir HolySheep
Après avoir évalué et intégré six providers d'API AI différents pour des projets fintech en Asie du Sud-Est, HolySheep s'impose comme la solution optimale pour plusieurs raisons techniques concrètes :
- Latence médiane mesurée à 42ms sur 1000 appels séquentiels DeepSeek V3.2 (vs 180ms+ chez d'autres providers) - critique pour les décisions de paiement en temps réel
- Économie de 85% sur les coûts tokens avec le taux ¥1=$1, permettant de traiter 3x plus de transactions pour le même budget
- Multi-modèles unifiés avec switching automatique selon le contexte transactionnel - un seul intégration, trois modèles de pointe
- Paiement localisé WeChat Pay et Alipay - simplifie considérablement la gestion financière pour les opérations en Chine-Thaïlande
- Crédits gratuits généreux (500K tokens) permettant de valider l'intégration sans engagement financier
Erreurs courantes et solutions
Erreur 1 : "429 Too Many Requests" - Rate Limiting
Symptôme : Erreurs intermittentes avec code 429, latence variable.
# ❌ PROBLÈMATIQUE: Appels simultanés sans backoff
for transaction in batch:
response = requests.post(url, json=payload) # Surcharge rapide
✅ SOLUTION: Rate limiter avec exponential backoff
import time
from functools import wraps
def rate_limit_with_backoff(max_retries=3, base_delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
result = func(*args, **kwargs)
if hasattr(result, 'status_code') and result.status_code == 429:
raise RateLimitError("Rate limited")
return result
except RateLimitError:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"Attente {delay:.1f}s avant retry {attempt + 1}")
time.sleep(delay)
return None
return wrapper
return decorator
Implémentation avec batch processing
class BatchedRiskEvaluator:
def __init__(self, api_key: str, requests_per_minute: int = 60):
self.api_key = api_key
self.delay_between_requests = 60 / requests_per_minute
self.last_request_time = 0
def _throttle(self):
elapsed = time.time() - self.last_request_time
if elapsed < self.delay_between_requests:
time.sleep(self.delay_between_requests - elapsed)
self.last_request_time = time.time()
@rate_limit_with_backoff(max_retries=3)
def evaluate_throttled(self, transaction_data: Dict) -> Optional[Dict]:
self._throttle()
# ... appel APIErreur 2 : "JSONDecodeError" - Réponse non structurée
Symptôme : Le modèle retourne du texte libre au lieu de JSON valide.
# ❌ PROBLÈMATIQUE: Parsing fragile sans validation
response = call_model(...)
content = response["choices"][0]["message"]["content"]
result = json.loads(content) # Échec si texte libre
✅ SOLUTION: Force JSON mode + validation robuste
payload = {
"model": "deepseek_v32",
"messages": [...],
"response_format": {"type": "json_object"}, # Force JSON structuré
"max_tokens": 512
}
def safe_json_parse(content: str, fallback_score: float = 0.5) -> Dict:
"""Parse JSON avec fallback intelligent"""
# Nettoyage préalable
content = content.strip()
if content.startswith("```json"):
content = content[7:]
if content.endswith("```"):
content = content[:-3]
content = content.strip()
try:
return json.loads(content)
except json.JSONDecodeError:
# Tentative d'extraction de JSON depuis texte
import re
json_match = re.search(r'\{[^}]+\}', content, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(0))
except:
pass
# Fallback: extraire le score numériquement
score_match = re.search(r'risk[_\s]?score["\s:]+([0-9.]+)', content, re.IGNORECASE)
confidence_match = re.search(r'confidence["\s:]+([0-9.]+)', content, re.IGNORECASE)
return {
"risk_score": float(score_match.group(1)) if score_match else fallback_score,
"confidence": float(confidence_match.group(1)) if confidence_match else 0.3,
"indicators": ["PARSE_FALLBACK_USED"],
"raw_text": content
}
Utilisation
response = requests.post(url, json=payload)
data = response.json()
content = data["choices"][0]["message"]["content"]
result = safe_json_parse(content)Erreur 3 : Latence excessive dépassant les seuils BoT
Symptôme : Score final >180ms, non conforme aux exigences réglementaires.
# ❌ PROBLÈMATIQUE: Appels séquentiels même pour transactions simples
def evaluate_slow(transaction):
score1 = call_deepseek(transaction) # ~35ms
score2 = call_gpt41(transaction) # ~80ms
score3 = call_gemini(transaction) # ~45ms
# Total: ~160ms sans parallélisme ni caching
✅ SOLUTION: Routing intelligent + caching + parallélisme conditionnel
from functools import lru_cache
import hashlib
class OptimizedRiskEvaluator:
def __init__(self, api_key: str):
self.cache = {} # LRU cache simplifié
self.max_cache_size = 10000
def _get_cache_key(self, transaction: Dict) -> str:
"""Génère clé de cache stable"""
# Simplification: hash sur champs critiques
critical_fields = {
"type": transaction.get("type"),
"amount_thb": transaction.get("amount_thb"),
"sender_account": transaction.get("sender_account")[-6:], # 6 derniers digits
"receiver_country": transaction.get("receiver_country")
}
return hashlib.md5(json.dumps(critical_fields, sort_keys=True).encode()).hexdigest()
def _is_cacheable(self, transaction: Dict) -> bool:
"""Détermine si la transaction peut être cachée"""
# Ne pas cacher les transactions critiques ou nouvelles expéditeurs
if transaction.get("risk_level_override"):
return False
if transaction.get("sender_kyc_level") == "basic":
return False
return True
async def evaluate_optimized(self, transaction: Dict) -> Dict:
cache_key = self._get_cache_key(transaction)
# Check cache first
if self._is_cacheable(transaction) and cache_key in self.cache:
cached = self.cache[cache_key]
cached["from_cache"] = True
return cached
# Routing intelligent: un seul modèle par défaut
transaction_type = transaction.get("type")
if transaction_type == "p2p_transfer" and transaction.get("amount_thb", 0) < 10000:
# Micro-transactions: DeepSeek uniquement (~35ms)
result = await self._call_single_model("deepseek_v32", transaction)
elif transaction_type == "cross_border" or transaction.get("amount_thb", 0) > 100000:
# Grosses transactions: validation avec second modèle si score élevé
result = await self._call_with_conditional_validation(transaction)
else:
# Transactions standard: Gemini Flash (~45ms)
result = await self._call_single_model("gemini_flash", transaction)
# Cache management
if self._is_cacheable(transaction):
if len(self.cache) >= self.max_cache_size:
# FIFO eviction
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[cache_key] = result
return result
async def _call_with_conditional_validation(self, transaction: Dict) -> Dict:
"""Appel initial + validation conditionnelle"""